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

1from datetime import datetime, timedelta, timezone 3 ctx1abc

2 

3import jwt 3 ctx1abc

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

5from fastapi.security import ( 3 ctx1abc

6 OAuth2PasswordBearer, 

7 OAuth2PasswordRequestForm, 

8 SecurityScopes, 

9) 

10from jwt.exceptions import InvalidTokenError 3 ctx1abc

11from passlib.context import CryptContext 3 ctx1abc

12from pydantic import BaseModel, ValidationError 3 ctx1abc

13 

14# to get a string like this run: 

15# openssl rand -hex 32 

16SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 3 ctx1abc

17ALGORITHM = "HS256" 3 ctx1abc

18ACCESS_TOKEN_EXPIRE_MINUTES = 30 3 ctx1abc

19 

20 

21fake_users_db = { 3 ctx1abc

22 "johndoe": { 

23 "username": "johndoe", 

24 "full_name": "John Doe", 

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

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

27 "disabled": False, 

28 }, 

29 "alice": { 

30 "username": "alice", 

31 "full_name": "Alice Chains", 

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

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

34 "disabled": True, 

35 }, 

36} 

37 

38 

39class Token(BaseModel): 3 ctx1abc

40 access_token: str 3 ctx1abc

41 token_type: str 3 ctx1abc

42 

43 

44class TokenData(BaseModel): 3 ctx1abc

45 username: str | None = None 3 ctx1abc

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

47 

48 

49class User(BaseModel): 3 ctx1abc

50 username: str 3 ctx1abc

51 email: str | None = None 3 ctx1abc

52 full_name: str | None = None 3 ctx1abc

53 disabled: bool | None = None 3 ctx1abc

54 

55 

56class UserInDB(User): 3 ctx1abc

57 hashed_password: str 3 ctx1abc

58 

59 

60pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 3 ctx1abc

61 

62oauth2_scheme = OAuth2PasswordBearer( 3 ctx1abc

63 tokenUrl="token", 

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

65) 

66 

67app = FastAPI() 3 ctx1abc

68 

69 

70def verify_password(plain_password, hashed_password): 3 ctx1abc

71 return pwd_context.verify(plain_password, hashed_password) 24 ctx1sBdmejnQtCfogkpRuDhqilrS

72 

73 

74def get_password_hash(password): 3 ctx1abc

75 return pwd_context.hash(password) 3 ctx1TUV

76 

77 

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

79 if username in db: 30 ctx1sBHdmejnvwtCIfogkpxyuDJhqilrzA

80 user_dict = db[username] 21 ctx1sBdmejntCfogkpuDhqilr

81 return UserInDB(**user_dict) 21 ctx1sBdmejntCfogkpuDhqilr

82 

83 

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

85 user = get_user(fake_db, username) 24 ctx1sBHdmejntCIfogkpuDJhqilr

86 if not user: 24 ctx1sBHdmejntCIfogkpuDJhqilr

87 return False 3 ctx1HIJ

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

89 return False 3 ctx1BCD

90 return user 18 ctx1sdmejntfogkpuhqilr

91 

92 

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

94 to_encode = data.copy() 21 ctx1NsdmejnOtfogkpPuhqilr

95 if expires_delta: 21 ctx1NsdmejnOtfogkpPuhqilr

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

97 else: 

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

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

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

101 return encoded_jwt 21 ctx1NsdmejnOtfogkpPuhqilr

102 

103 

104async def get_current_user( 3 ctx1abc

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

106): 

107 if security_scopes.scopes: 27 ctx1KdmejnEvwLfogkpFxyMhqilrGzA

108 authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' 24 ctx1KdejnEvwLfgkpFxyMhilrGzA

109 else: 

110 authenticate_value = "Bearer" 3 ctx1moq

111 credentials_exception = HTTPException( 27 ctx1KdmejnEvwLfogkpFxyMhqilrGzA

112 status_code=status.HTTP_401_UNAUTHORIZED, 

113 detail="Could not validate credentials", 

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

115 ) 

116 try: 27 ctx1KdmejnEvwLfogkpFxyMhqilrGzA

117 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 27 ctx1KdmejnEvwLfogkpFxyMhqilrGzA

118 username: str = payload.get("sub") 24 ctx1dmejnEvwfogkpFxyhqilrGzA

119 if username is None: 24 ctx1dmejnEvwfogkpFxyhqilrGzA

120 raise credentials_exception 3 ctx1EFG

121 token_scopes = payload.get("scopes", []) 21 ctx1dmejnvwfogkpxyhqilrzA

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

123 except (InvalidTokenError, ValidationError): 6 ctx1KELFMG

124 raise credentials_exception 3 ctx1KLM

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

126 if user is None: 21 ctx1dmejnvwfogkpxyhqilrzA

127 raise credentials_exception 6 ctx1vwxyzA

128 for scope in security_scopes.scopes: 15 ctx1dmejnfogkphqilr

129 if scope not in token_data.scopes: 12 ctx1dejnfgkphilr

130 raise HTTPException( 3 ctx1npr

131 status_code=status.HTTP_401_UNAUTHORIZED, 

132 detail="Not enough permissions", 

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

134 ) 

135 return user 12 ctx1dmejfogkhqil

136 

137 

138async def get_current_active_user( 3 ctx1abc

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

140): 

141 if current_user.disabled: 9 ctx1dejfgkhil

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

143 return current_user 6 ctx1defghi

144 

145 

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

147async def login_for_access_token( 3 ctx1abc

148 form_data: OAuth2PasswordRequestForm = Depends(), 

149) -> Token: 

150 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 24 ctx1sBHdmejntCIfogkpuDJhqilr

151 if not user: 24 ctx1sBHdmejntCIfogkpuDJhqilr

152 raise HTTPException(status_code=400, detail="Incorrect username or password") 6 ctx1BHCIDJ

153 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 18 ctx1sdmejntfogkpuhqilr

154 access_token = create_access_token( 18 ctx1sdmejntfogkpuhqilr

155 data={"sub": user.username, "scopes": form_data.scopes}, 

156 expires_delta=access_token_expires, 

157 ) 

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

159 

160 

161@app.get("/users/me/", response_model=User) 3 ctx1abc

162async def read_users_me(current_user: User = Depends(get_current_active_user)): 3 ctx1abc

163 return current_user 3 ctx1egi

164 

165 

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

167async def read_own_items( 3 ctx1abc

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

169): 

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

171 

172 

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

174async def read_system_status(current_user: User = Depends(get_current_user)): 3 ctx1abc

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