Coverage for docs_src / security / tutorial004_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, status 3 ctx1abc

6from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm 3 ctx1abc

7from jwt.exceptions import InvalidTokenError 3 ctx1abc

8from pwdlib import PasswordHash 3 ctx1abc

9from pydantic import BaseModel 3 ctx1abc

10 

11# to get a string like this run: 

12# openssl rand -hex 32 

13SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 3 ctx1abc

14ALGORITHM = "HS256" 3 ctx1abc

15ACCESS_TOKEN_EXPIRE_MINUTES = 30 3 ctx1abc

16 

17 

18fake_users_db = { 3 ctx1abc

19 "johndoe": { 

20 "username": "johndoe", 

21 "full_name": "John Doe", 

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

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

24 "disabled": False, 

25 } 

26} 

27 

28 

29class Token(BaseModel): 3 ctx1abc

30 access_token: str 3 ctx1abc

31 token_type: str 3 ctx1abc

32 

33 

34class TokenData(BaseModel): 3 ctx1abc

35 username: str | None = None 3 ctx1abc

36 

37 

38class User(BaseModel): 3 ctx1abc

39 username: str 3 ctx1abc

40 email: str | None = None 3 ctx1abc

41 full_name: str | None = None 3 ctx1abc

42 disabled: bool | None = None 3 ctx1abc

43 

44 

45class UserInDB(User): 3 ctx1abc

46 hashed_password: str 3 ctx1abc

47 

48 

49password_hash = PasswordHash.recommended() 3 ctx1abc

50 

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

52 

53oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") 3 ctx1abc

54 

55app = FastAPI() 3 ctx1abc

56 

57 

58def verify_password(plain_password, hashed_password): 3 ctx1abc

59 return password_hash.verify(plain_password, hashed_password) 21 ctx1mpydefKnqzghiLorAjklM

60 

61 

62def get_password_hash(password): 3 ctx1abc

63 return password_hash.hash(password) 6 ctx1NfOiPl

64 

65 

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

67 if username in db: 24 ctx1mpydefstnqzghiuvorAjklwx

68 user_dict = db[username] 15 ctx1mpdefnqghiorjkl

69 return UserInDB(**user_dict) 15 ctx1mpdefnqghiorjkl

70 

71 

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

73 user = get_user(fake_db, username) 18 ctx1mpydefnqzghiorAjkl

74 if not user: 18 ctx1mpydefnqzghiorAjkl

75 verify_password(password, DUMMY_HASH) 3 ctx1yzA

76 return False 3 ctx1yzA

77 if not verify_password(password, user.hashed_password): 15 ctx1mpdefnqghiorjkl

78 return False 3 ctx1pqr

79 return user 12 ctx1mdefnghiojkl

80 

81 

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

83 to_encode = data.copy() 15 ctx1EmdefFnghiGojkl

84 if expires_delta: 15 ctx1EmdefFnghiGojkl

85 expire = datetime.now(timezone.utc) + expires_delta 12 ctx1mdefnghiojkl

86 else: 

87 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 3 ctx1EFG

88 to_encode.update({"exp": expire}) 15 ctx1EmdefFnghiGojkl

89 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 15 ctx1EmdefFnghiGojkl

90 return encoded_jwt 15 ctx1EmdefFnghiGojkl

91 

92 

93async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): 3 ctx1abc

94 credentials_exception = HTTPException( 21 ctx1HdefBstIghiCuvJjklDwx

95 status_code=status.HTTP_401_UNAUTHORIZED, 

96 detail="Could not validate credentials", 

97 headers={"WWW-Authenticate": "Bearer"}, 

98 ) 

99 try: 21 ctx1HdefBstIghiCuvJjklDwx

100 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 21 ctx1HdefBstIghiCuvJjklDwx

101 username = payload.get("sub") 18 ctx1defBstghiCuvjklDwx

102 if username is None: 18 ctx1defBstghiCuvjklDwx

103 raise credentials_exception 3 ctx1BCD

104 token_data = TokenData(username=username) 15 ctx1defstghiuvjklwx

105 except InvalidTokenError: 6 ctx1HBICJD

106 raise credentials_exception 3 ctx1HIJ

107 user = get_user(fake_users_db, username=token_data.username) 15 ctx1defstghiuvjklwx

108 if user is None: 15 ctx1defstghiuvjklwx

109 raise credentials_exception 6 ctx1stuvwx

110 return user 9 ctx1defghijkl

111 

112 

113async def get_current_active_user( 3 ctx1abc

114 current_user: Annotated[User, Depends(get_current_user)], 

115): 

116 if current_user.disabled: 9 ctx1defghijkl

117 raise HTTPException(status_code=400, detail="Inactive user") 3 ctx1fil

118 return current_user 6 ctx1deghjk

119 

120 

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

122async def login_for_access_token( 3 ctx1abc

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

124) -> Token: 

125 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 18 ctx1mpydefnqzghiorAjkl

126 if not user: 18 ctx1mpydefnqzghiorAjkl

127 raise HTTPException( 6 ctx1pyqzrA

128 status_code=status.HTTP_401_UNAUTHORIZED, 

129 detail="Incorrect username or password", 

130 headers={"WWW-Authenticate": "Bearer"}, 

131 ) 

132 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 12 ctx1mdefnghiojkl

133 access_token = create_access_token( 12 ctx1mdefnghiojkl

134 data={"sub": user.username}, expires_delta=access_token_expires 

135 ) 

136 return Token(access_token=access_token, token_type="bearer") 12 ctx1mdefnghiojkl

137 

138 

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

140async def read_users_me( 3 ctx1abc

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

142) -> User: 

143 return current_user 3 ctx1ehk

144 

145 

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

147async def read_own_items( 3 ctx1abc

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

149): 

150 return [{"item_id": "Foo", "owner": current_user.username}] 3 ctx1dgj