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

1from datetime import datetime, timedelta, timezone 4 ctx1abcd

2from typing import Annotated 4 ctx1abcd

3 

4import jwt 4 ctx1abcd

5from fastapi import Depends, FastAPI, HTTPException, Security, status 4 ctx1abcd

6from fastapi.security import ( 4 ctx1abcd

7 OAuth2PasswordBearer, 

8 OAuth2PasswordRequestForm, 

9 SecurityScopes, 

10) 

11from jwt.exceptions import InvalidTokenError 4 ctx1abcd

12from passlib.context import CryptContext 4 ctx1abcd

13from pydantic import BaseModel, ValidationError 4 ctx1abcd

14 

15# to get a string like this run: 

16# openssl rand -hex 32 

17SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 4 ctx1abcd

18ALGORITHM = "HS256" 4 ctx1abcd

19ACCESS_TOKEN_EXPIRE_MINUTES = 30 4 ctx1abcd

20 

21 

22fake_users_db = { 4 ctx1abcd

23 "johndoe": { 

24 "username": "johndoe", 

25 "full_name": "John Doe", 

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

27 "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", 

28 "disabled": False, 

29 }, 

30 "alice": { 

31 "username": "alice", 

32 "full_name": "Alice Chains", 

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

34 "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm", 

35 "disabled": True, 

36 }, 

37} 

38 

39 

40class Token(BaseModel): 4 ctx1abcd

41 access_token: str 4 ctx1abcd

42 token_type: str 4 ctx1abcd

43 

44 

45class TokenData(BaseModel): 4 ctx1abcd

46 username: str | None = None 4 ctx1abcd

47 scopes: list[str] = [] 4 ctx1abcd

48 

49 

50class User(BaseModel): 4 ctx1abcd

51 username: str 4 ctx1abcd

52 email: str | None = None 4 ctx1abcd

53 full_name: str | None = None 4 ctx1abcd

54 disabled: bool | None = None 4 ctx1abcd

55 

56 

57class UserInDB(User): 4 ctx1abcd

58 hashed_password: str 4 ctx1abcd

59 

60 

61pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 4 ctx1abcd

62 

63oauth2_scheme = OAuth2PasswordBearer( 4 ctx1abcd

64 tokenUrl="token", 

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

66) 

67 

68app = FastAPI() 4 ctx1abcd

69 

70 

71def verify_password(plain_password, hashed_password): 4 ctx1abcd

72 return pwd_context.verify(plain_password, hashed_password) 32 ctx1yKeqfmr4zLgshnt5AMiujov6BNkwlpx7

73 

74 

75def get_password_hash(password): 4 ctx1abcd

76 return pwd_context.hash(password) 4 ctx189!#

77 

78 

79def get_user(db, username: str): 4 ctx1abcd

80 if username in db: 40 ctx1yKSeqfmrCDzLTgshntEFAMUiujovGHBNVkwlpxIJ

81 user_dict = db[username] 28 ctx1yKeqfmrzLgshntAMiujovBNkwlpx

82 return UserInDB(**user_dict) 28 ctx1yKeqfmrzLgshntAMiujovBNkwlpx

83 

84 

85def authenticate_user(fake_db, username: str, password: str): 4 ctx1abcd

86 user = get_user(fake_db, username) 32 ctx1yKSeqfmrzLTgshntAMUiujovBNVkwlpx

87 if not user: 32 ctx1yKSeqfmrzLTgshntAMUiujovBNVkwlpx

88 return False 4 ctx1STUV

89 if not verify_password(password, user.hashed_password): 28 ctx1yKeqfmrzLgshntAMiujovBNkwlpx

90 return False 4 ctx1KLMN

91 return user 24 ctx1yeqfmrzgshntAiujovBkwlpx

92 

93 

94def create_access_token(data: dict, expires_delta: timedelta | None = None): 4 ctx1abcd

95 to_encode = data.copy() 28 ctx10yeqfmr1zgshnt2Aiujov3Bkwlpx

96 if expires_delta: 28 ctx10yeqfmr1zgshnt2Aiujov3Bkwlpx

97 expire = datetime.now(timezone.utc) + expires_delta 24 ctx1yeqfmrzgshntAiujovBkwlpx

98 else: 

99 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 4 ctx10123

100 to_encode.update({"exp": expire}) 28 ctx10yeqfmr1zgshnt2Aiujov3Bkwlpx

101 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 28 ctx10yeqfmr1zgshnt2Aiujov3Bkwlpx

102 return encoded_jwt 28 ctx10yeqfmr1zgshnt2Aiujov3Bkwlpx

103 

104 

105async def get_current_user( 4 ctx1abcd

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

107): 

108 if security_scopes.scopes: 36 ctx1WeqfmrOCDXgshntPEFYiujovQGHZkwlpxRIJ

109 authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' 32 ctx1WefmrOCDXghntPEFYijovQGHZklpxRIJ

110 else: 

111 authenticate_value = "Bearer" 4 ctx1qsuw

112 credentials_exception = HTTPException( 36 ctx1WeqfmrOCDXgshntPEFYiujovQGHZkwlpxRIJ

113 status_code=status.HTTP_401_UNAUTHORIZED, 

114 detail="Could not validate credentials", 

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

116 ) 

117 try: 36 ctx1WeqfmrOCDXgshntPEFYiujovQGHZkwlpxRIJ

118 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 36 ctx1WeqfmrOCDXgshntPEFYiujovQGHZkwlpxRIJ

119 username = payload.get("sub") 32 ctx1eqfmrOCDgshntPEFiujovQGHkwlpxRIJ

120 if username is None: 32 ctx1eqfmrOCDgshntPEFiujovQGHkwlpxRIJ

121 raise credentials_exception 4 ctx1OPQR

122 scope: str = payload.get("scope", "") 28 ctx1eqfmrCDgshntEFiujovGHkwlpxIJ

123 token_scopes = scope.split(" ") 28 ctx1eqfmrCDgshntEFiujovGHkwlpxIJ

124 token_data = TokenData(scopes=token_scopes, username=username) 28 ctx1eqfmrCDgshntEFiujovGHkwlpxIJ

125 except (InvalidTokenError, ValidationError): 8 ctx1WOXPYQZR

126 raise credentials_exception 4 ctx1WXYZ

127 user = get_user(fake_users_db, username=token_data.username) 28 ctx1eqfmrCDgshntEFiujovGHkwlpxIJ

128 if user is None: 28 ctx1eqfmrCDgshntEFiujovGHkwlpxIJ

129 raise credentials_exception 8 ctx1CDEFGHIJ

130 for scope in security_scopes.scopes: 20 ctx1eqfmrgshntiujovkwlpx

131 if scope not in token_data.scopes: 16 ctx1efmrghntijovklpx

132 raise HTTPException( 4 ctx1rtvx

133 status_code=status.HTTP_401_UNAUTHORIZED, 

134 detail="Not enough permissions", 

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

136 ) 

137 return user 16 ctx1eqfmgshniujokwlp

138 

139 

140async def get_current_active_user( 4 ctx1abcd

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

142): 

143 if current_user.disabled: 12 ctx1efmghnijoklp

144 raise HTTPException(status_code=400, detail="Inactive user") 4 ctx1mnop

145 return current_user 8 ctx1efghijkl

146 

147 

148@app.post("/token") 4 ctx1abcd

149async def login_for_access_token( 4 ctx1abcd

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

151) -> Token: 

152 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 32 ctx1yKSeqfmrzLTgshntAMUiujovBNVkwlpx

153 if not user: 32 ctx1yKSeqfmrzLTgshntAMUiujovBNVkwlpx

154 raise HTTPException(status_code=400, detail="Incorrect username or password") 8 ctx1KSLTMUNV

155 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 24 ctx1yeqfmrzgshntAiujovBkwlpx

156 access_token = create_access_token( 24 ctx1yeqfmrzgshntAiujovBkwlpx

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") 24 ctx1yeqfmrzgshntAiujovBkwlpx

161 

162 

163@app.get("/users/me/", response_model=User) 4 ctx1abcd

164async def read_users_me( 4 ctx1abcd

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

166): 

167 return current_user 4 ctx1fhjl

168 

169 

170@app.get("/users/me/items/") 4 ctx1abcd

171async def read_own_items( 4 ctx1abcd

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

173): 

174 return [{"item_id": "Foo", "owner": current_user.username}] 4 ctx1egik

175 

176 

177@app.get("/status/") 4 ctx1abcd

178async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]): 4 ctx1abcd

179 return {"status": "ok"} 4 ctx1qsuw