Coverage for docs_src/security/tutorial005.py: 100%

1from datetime import datetime, timedelta, timezone 7 ctx1abcdefg

2from typing import List, Union 7 ctx1abcdefg

3 

4import jwt 7 ctx1abcdefg

5from fastapi import Depends, FastAPI, HTTPException, Security, status 7 ctx1abcdefg

6from fastapi.security import ( 7 ctx1abcdefg

7 OAuth2PasswordBearer, 

8 OAuth2PasswordRequestForm, 

9 SecurityScopes, 

10) 

11from jwt.exceptions import InvalidTokenError 7 ctx1abcdefg

12from pwdlib import PasswordHash 7 ctx1abcdefg

13from pydantic import BaseModel, ValidationError 7 ctx1abcdefg

14 

15# to get a string like this run: 

16# openssl rand -hex 32 

17SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 7 ctx1abcdefg

18ALGORITHM = "HS256" 7 ctx1abcdefg

19ACCESS_TOKEN_EXPIRE_MINUTES = 30 7 ctx1abcdefg

20 

21 

22fake_users_db = { 7 ctx1abcdefg

23 "johndoe": { 

24 "username": "johndoe", 

25 "full_name": "John Doe", 

26 "email": "johndoe@example.com", 

27 "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc", 

28 "disabled": False, 

29 }, 

30 "alice": { 

31 "username": "alice", 

32 "full_name": "Alice Chains", 

33 "email": "alicechains@example.com", 

34 "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$g2/AV1zwopqUntPKJavBFw$BwpRGDCyUHLvHICnwijyX8ROGoiUPwNKZ7915MeYfCE", 

35 "disabled": True, 

36 }, 

37} 

38 

39 

40class Token(BaseModel): 7 ctx1abcdefg

41 access_token: str 7 ctx1abcdefg

42 token_type: str 7 ctx1abcdefg

43 

44 

45class TokenData(BaseModel): 7 ctx1abcdefg

46 username: Union[str, None] = None 7 ctx1abcdefg

47 scopes: List[str] = [] 7 ctx1abcdefg

48 

49 

50class User(BaseModel): 7 ctx1abcdefg

51 username: str 7 ctx1abcdefg

52 email: Union[str, None] = None 7 ctx1abcdefg

53 full_name: Union[str, None] = None 7 ctx1abcdefg

54 disabled: Union[bool, None] = None 7 ctx1abcdefg

55 

56 

57class UserInDB(User): 7 ctx1abcdefg

58 hashed_password: str 7 ctx1abcdefg

59 

60 

61password_hash = PasswordHash.recommended() 7 ctx1abcdefg

62 

63oauth2_scheme = OAuth2PasswordBearer( 7 ctx1abcdefg

64 tokenUrl="token", 

65 scopes={"me": "Read information about the current user.", "items": "Read items."}, 

66) 

67 

68app = FastAPI() 7 ctx1abcdefg

69 

70 

71def verify_password(plain_password, hashed_password): 7 ctx1abcdefg

72 return password_hash.verify(plain_password, hashed_password) 56 ctx2Q # h C i v D jbR $ j E k w F kbS % l G m x H lbT ' n I o y J mbU ( p K q z L nbV ) r M s A N obW * t O u B P pb

73 

74 

75def get_password_hash(password): 7 ctx1abcdefg

76 return password_hash.hash(password) 7 ctx2qbrbsbtbubvbwb

77 

78 

79def get_user(db, username: str): 7 ctx1abcdefg

80 if username in db: 70 ctx1Q#=hCivDXYR$?jEkwFZ0S%@lGmxH12T'[nIoyJ34U(]pKqzL56V)^rMsAN78W*_tOuBP9!

81 user_dict = db[username] 49 ctx1Q#hCivDR$jEkwFS%lGmxHT'nIoyJU(pKqzLV)rMsANW*tOuBP

82 return UserInDB(**user_dict) 49 ctx1Q#hCivDR$jEkwFS%lGmxHT'nIoyJU(pKqzLV)rMsANW*tOuBP

83 

84 

85def authenticate_user(fake_db, username: str, password: str): 7 ctx1abcdefg

86 user = get_user(fake_db, username) 56 ctx1Q#=hCivDR$?jEkwFS%@lGmxHT'[nIoyJU(]pKqzLV)^rMsANW*_tOuBP

87 if not user: 56 ctx1Q#=hCivDR$?jEkwFS%@lGmxHT'[nIoyJU(]pKqzLV)^rMsANW*_tOuBP

88 return False 7 ctx1=?@[]^_

89 if not verify_password(password, user.hashed_password): 49 ctx1Q#hCivDR$jEkwFS%lGmxHT'nIoyJU(pKqzLV)rMsANW*tOuBP

90 return False 7 ctx1#$%'()*

91 return user 42 ctx1QhCivDRjEkwFSlGmxHTnIoyJUpKqzLVrMsANWtOuBP

92 

93 

94def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None): 7 ctx1abcdefg

95 to_encode = data.copy() 49 ctx2cbQ h C i v D dbR j E k w F ebS l G m x H fbT n I o y J gbU p K q z L hbV r M s A N ibW t O u B P

96 if expires_delta: 49 ctx2cbQ h C i v D dbR j E k w F ebS l G m x H fbT n I o y J gbU p K q z L hbV r M s A N ibW t O u B P

97 expire = datetime.now(timezone.utc) + expires_delta 42 ctx1QhCivDRjEkwFSlGmxHTnIoyJUpKqzLVrMsANWtOuBP

98 else: 

99 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 7 ctx2cbdbebfbgbhbib

100 to_encode.update({"exp": expire}) 49 ctx2cbQ h C i v D dbR j E k w F ebS l G m x H fbT n I o y J gbU p K q z L hbV r M s A N ibW t O u B P

101 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 49 ctx2cbQ h C i v D dbR j E k w F ebS l G m x H fbT n I o y J gbU p K q z L hbV r M s A N ibW t O u B P

102 return encoded_jwt 49 ctx2cbQ h C i v D dbR j E k w F ebS l G m x H fbT n I o y J gbU p K q z L hbV r M s A N ibW t O u B P

103 

104 

105async def get_current_user( 7 ctx1abcdefg

106 security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme) 

107): 

108 if security_scopes.scopes: 63 ctx2` h C i v D + X Y { j E k w F , Z 0 | l G m x H - 1 2 } n I o y J . 3 4 ~ p K q z L / 5 6 abr M s A N : 7 8 bbt O u B P ; 9 !

109 authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' 56 ctx2` h i v D + X Y { j k w F , Z 0 | l m x H - 1 2 } n o y J . 3 4 ~ p q z L / 5 6 abr s A N : 7 8 bbt u B P ; 9 !

110 else: 

111 authenticate_value = "Bearer" 7 ctx1CEGIKMO

112 credentials_exception = HTTPException( 63 ctx2` h C i v D + X Y { j E k w F , Z 0 | l G m x H - 1 2 } n I o y J . 3 4 ~ p K q z L / 5 6 abr M s A N : 7 8 bbt O u B P ; 9 !

113 status_code=status.HTTP_401_UNAUTHORIZED, 

114 detail="Could not validate credentials", 

115 headers={"WWW-Authenticate": authenticate_value}, 

116 ) 

117 try: 63 ctx2` h C i v D + X Y { j E k w F , Z 0 | l G m x H - 1 2 } n I o y J . 3 4 ~ p K q z L / 5 6 abr M s A N : 7 8 bbt O u B P ; 9 !

118 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 63 ctx2` h C i v D + X Y { j E k w F , Z 0 | l G m x H - 1 2 } n I o y J . 3 4 ~ p K q z L / 5 6 abr M s A N : 7 8 bbt O u B P ; 9 !

119 username: str = payload.get("sub") 56 ctx1hCivD+XYjEkwF,Z0lGmxH-12nIoyJ.34pKqzL/56rMsAN:78tOuBP;9!

120 if username is None: 56 ctx1hCivD+XYjEkwF,Z0lGmxH-12nIoyJ.34pKqzL/56rMsAN:78tOuBP;9!

121 raise credentials_exception 7 ctx1+,-./:;

122 scope: str = payload.get("scope", "") 49 ctx1hCivDXYjEkwFZ0lGmxH12nIoyJ34pKqzL56rMsAN78tOuBP9!

123 token_scopes = scope.split(" ") 49 ctx1hCivDXYjEkwFZ0lGmxH12nIoyJ34pKqzL56rMsAN78tOuBP9!

124 token_data = TokenData(scopes=token_scopes, username=username) 49 ctx1hCivDXYjEkwFZ0lGmxH12nIoyJ34pKqzL56rMsAN78tOuBP9!

125 except (InvalidTokenError, ValidationError): 14 ctx2` + { , | - } . ~ / ab: bb;

126 raise credentials_exception 7 ctx2` { | } ~ abbb

127 user = get_user(fake_users_db, username=token_data.username) 49 ctx1hCivDXYjEkwFZ0lGmxH12nIoyJ34pKqzL56rMsAN78tOuBP9!

128 if user is None: 49 ctx1hCivDXYjEkwFZ0lGmxH12nIoyJ34pKqzL56rMsAN78tOuBP9!

129 raise credentials_exception 14 ctx1XYZ0123456789!

130 for scope in security_scopes.scopes: 35 ctx1hCivDjEkwFlGmxHnIoyJpKqzLrMsANtOuBP

131 if scope not in token_data.scopes: 28 ctx1hivDjkwFlmxHnoyJpqzLrsANtuBP

132 raise HTTPException( 7 ctx1DFHJLNP

133 status_code=status.HTTP_401_UNAUTHORIZED, 

134 detail="Not enough permissions", 

135 headers={"WWW-Authenticate": authenticate_value}, 

136 ) 

137 return user 28 ctx1hCivjEkwlGmxnIoypKqzrMsAtOuB

138 

139 

140async def get_current_active_user( 7 ctx1abcdefg

141 current_user: User = Security(get_current_user, scopes=["me"]), 

142): 

143 if current_user.disabled: 21 ctx1hivjkwlmxnoypqzrsAtuB

144 raise HTTPException(status_code=400, detail="Inactive user") 7 ctx1vwxyzAB

145 return current_user 14 ctx1hijklmnopqrstu

146 

147 

148@app.post("/token") 7 ctx1abcdefg

149async def login_for_access_token( 7 ctx1abcdefg

150 form_data: OAuth2PasswordRequestForm = Depends(), 

151) -> Token: 

152 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 56 ctx1Q#=hCivDR$?jEkwFS%@lGmxHT'[nIoyJU(]pKqzLV)^rMsANW*_tOuBP

153 if not user: 56 ctx1Q#=hCivDR$?jEkwFS%@lGmxHT'[nIoyJU(]pKqzLV)^rMsANW*_tOuBP

154 raise HTTPException(status_code=400, detail="Incorrect username or password") 14 ctx1#=$?%@'[(])^*_

155 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 42 ctx1QhCivDRjEkwFSlGmxHTnIoyJUpKqzLVrMsANWtOuBP

156 access_token = create_access_token( 42 ctx1QhCivDRjEkwFSlGmxHTnIoyJUpKqzLVrMsANWtOuBP

157 data={"sub": user.username, "scope": " ".join(form_data.scopes)}, 

158 expires_delta=access_token_expires, 

159 ) 

160 return Token(access_token=access_token, token_type="bearer") 42 ctx1QhCivDRjEkwFSlGmxHTnIoyJUpKqzLVrMsANWtOuBP

161 

162 

163@app.get("/users/me/", response_model=User) 7 ctx1abcdefg

164async def read_users_me(current_user: User = Depends(get_current_active_user)): 7 ctx1abcdefg

165 return current_user 7 ctx1ikmoqsu

166 

167 

168@app.get("/users/me/items/") 7 ctx1abcdefg

169async def read_own_items( 7 ctx1abcdefg

170 current_user: User = Security(get_current_active_user, scopes=["items"]), 

171): 

172 return [{"item_id": "Foo", "owner": current_user.username}] 7 ctx1hjlnprt

173 

174 

175@app.get("/status/") 7 ctx1abcdefg

176async def read_system_status(current_user: User = Depends(get_current_user)): 7 ctx1abcdefg

177 return {"status": "ok"} 7 ctx1CEGIKMO