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

1from datetime import datetime, timedelta, timezone 3 ctx1abc

2from typing import Annotated 3 ctx1abc

3 

4import jwt 3 ctx1abc

5from fastapi import Depends, FastAPI, HTTPException, Security, status 3 ctx1abc

6from fastapi.security import ( 3 ctx1abc

7 OAuth2PasswordBearer, 

8 OAuth2PasswordRequestForm, 

9 SecurityScopes, 

10) 

11from jwt.exceptions import InvalidTokenError 3 ctx1abc

12from pwdlib import PasswordHash 3 ctx1abc

13from pydantic import BaseModel, ValidationError 3 ctx1abc

14 

15# to get a string like this run: 

16# openssl rand -hex 32 

17SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 3 ctx1abc

18ALGORITHM = "HS256" 3 ctx1abc

19ACCESS_TOKEN_EXPIRE_MINUTES = 30 3 ctx1abc

20 

21 

22fake_users_db = { 3 ctx1abc

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): 3 ctx1abc

41 access_token: str 3 ctx1abc

42 token_type: str 3 ctx1abc

43 

44 

45class TokenData(BaseModel): 3 ctx1abc

46 username: str | None = None 3 ctx1abc

47 scopes: list[str] = [] 3 ctx1abc

48 

49 

50class User(BaseModel): 3 ctx1abc

51 username: str 3 ctx1abc

52 email: str | None = None 3 ctx1abc

53 full_name: str | None = None 3 ctx1abc

54 disabled: bool | None = None 3 ctx1abc

55 

56 

57class UserInDB(User): 3 ctx1abc

58 hashed_password: str 3 ctx1abc

59 

60 

61password_hash = PasswordHash.recommended() 3 ctx1abc

62 

63DUMMY_HASH = password_hash.hash("dummypassword") 3 ctx1abc

64 

65oauth2_scheme = OAuth2PasswordBearer( 3 ctx1abc

66 tokenUrl="token", 

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

68) 

69 

70app = FastAPI() 3 ctx1abc

71 

72 

73def verify_password(plain_password, hashed_password): 3 ctx1abc

74 return password_hash.verify(plain_password, hashed_password) 27 ctx1sBEdmejnQtCFfogkpRuDGhqilrS

75 

76 

77def get_password_hash(password): 3 ctx1abc

78 return password_hash.hash(password) 3 ctx1TUV

79 

80 

81def get_user(db, username: str): 3 ctx1abc

82 if username in db: 30 ctx1sBEdmejnvwtCFfogkpxyuDGhqilrzA

83 user_dict = db[username] 21 ctx1sBdmejntCfogkpuDhqilr

84 return UserInDB(**user_dict) 21 ctx1sBdmejntCfogkpuDhqilr

85 

86 

87def authenticate_user(fake_db, username: str, password: str): 3 ctx1abc

88 user = get_user(fake_db, username) 24 ctx1sBEdmejntCFfogkpuDGhqilr

89 if not user: 24 ctx1sBEdmejntCFfogkpuDGhqilr

90 verify_password(password, DUMMY_HASH) 3 ctx1EFG

91 return False 3 ctx1EFG

92 if not verify_password(password, user.hashed_password): 21 ctx1sBdmejntCfogkpuDhqilr

93 return False 3 ctx1BCD

94 return user 18 ctx1sdmejntfogkpuhqilr

95 

96 

97def create_access_token(data: dict, expires_delta: timedelta | None = None): 3 ctx1abc

98 to_encode = data.copy() 21 ctx1NsdmejnOtfogkpPuhqilr

99 if expires_delta: 21 ctx1NsdmejnOtfogkpPuhqilr

100 expire = datetime.now(timezone.utc) + expires_delta 18 ctx1sdmejntfogkpuhqilr

101 else: 

102 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 3 ctx1NOP

103 to_encode.update({"exp": expire}) 21 ctx1NsdmejnOtfogkpPuhqilr

104 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 21 ctx1NsdmejnOtfogkpPuhqilr

105 return encoded_jwt 21 ctx1NsdmejnOtfogkpPuhqilr

106 

107 

108async def get_current_user( 3 ctx1abc

109 security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)] 

110): 

111 if security_scopes.scopes: 27 ctx1KdmejnHvwLfogkpIxyMhqilrJzA

112 authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' 24 ctx1KdejnHvwLfgkpIxyMhilrJzA

113 else: 

114 authenticate_value = "Bearer" 3 ctx1moq

115 credentials_exception = HTTPException( 27 ctx1KdmejnHvwLfogkpIxyMhqilrJzA

116 status_code=status.HTTP_401_UNAUTHORIZED, 

117 detail="Could not validate credentials", 

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

119 ) 

120 try: 27 ctx1KdmejnHvwLfogkpIxyMhqilrJzA

121 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 27 ctx1KdmejnHvwLfogkpIxyMhqilrJzA

122 username = payload.get("sub") 24 ctx1dmejnHvwfogkpIxyhqilrJzA

123 if username is None: 24 ctx1dmejnHvwfogkpIxyhqilrJzA

124 raise credentials_exception 3 ctx1HIJ

125 scope: str = payload.get("scope", "") 21 ctx1dmejnvwfogkpxyhqilrzA

126 token_scopes = scope.split(" ") 21 ctx1dmejnvwfogkpxyhqilrzA

127 token_data = TokenData(scopes=token_scopes, username=username) 21 ctx1dmejnvwfogkpxyhqilrzA

128 except (InvalidTokenError, ValidationError): 6 ctx1KHLIMJ

129 raise credentials_exception 3 ctx1KLM

130 user = get_user(fake_users_db, username=token_data.username) 21 ctx1dmejnvwfogkpxyhqilrzA

131 if user is None: 21 ctx1dmejnvwfogkpxyhqilrzA

132 raise credentials_exception 6 ctx1vwxyzA

133 for scope in security_scopes.scopes: 15 ctx1dmejnfogkphqilr

134 if scope not in token_data.scopes: 12 ctx1dejnfgkphilr

135 raise HTTPException( 3 ctx1npr

136 status_code=status.HTTP_401_UNAUTHORIZED, 

137 detail="Not enough permissions", 

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

139 ) 

140 return user 12 ctx1dmejfogkhqil

141 

142 

143async def get_current_active_user( 3 ctx1abc

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

145): 

146 if current_user.disabled: 9 ctx1dejfgkhil

147 raise HTTPException(status_code=400, detail="Inactive user") 3 ctx1jkl

148 return current_user 6 ctx1defghi

149 

150 

151@app.post("/token") 3 ctx1abc

152async def login_for_access_token( 3 ctx1abc

153 form_data: Annotated[OAuth2PasswordRequestForm, Depends()], 

154) -> Token: 

155 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 24 ctx1sBEdmejntCFfogkpuDGhqilr

156 if not user: 24 ctx1sBEdmejntCFfogkpuDGhqilr

157 raise HTTPException(status_code=400, detail="Incorrect username or password") 6 ctx1BECFDG

158 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 18 ctx1sdmejntfogkpuhqilr

159 access_token = create_access_token( 18 ctx1sdmejntfogkpuhqilr

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

161 expires_delta=access_token_expires, 

162 ) 

163 return Token(access_token=access_token, token_type="bearer") 18 ctx1sdmejntfogkpuhqilr

164 

165 

166@app.get("/users/me/") 3 ctx1abc

167async def read_users_me( 3 ctx1abc

168 current_user: Annotated[User, Depends(get_current_active_user)], 

169) -> User: 

170 return current_user 3 ctx1egi

171 

172 

173@app.get("/users/me/items/") 3 ctx1abc

174async def read_own_items( 3 ctx1abc

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

176): 

177 return [{"item_id": "Foo", "owner": current_user.username}] 3 ctx1dfh

178 

179 

180@app.get("/status/") 3 ctx1abc

181async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]): 3 ctx1abc

182 return {"status": "ok"} 3 ctx1moq