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

1from datetime import datetime, timedelta, timezone 9 ctx1abhicdefg

2 

3import jwt 9 ctx1abhicdefg

4from fastapi import Depends, FastAPI, HTTPException, status 9 ctx1abhicdefg

5from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm 9 ctx1abhicdefg

6from jwt.exceptions import InvalidTokenError 9 ctx1abhicdefg

7from pwdlib import PasswordHash 9 ctx1abhicdefg

8from pydantic import BaseModel 9 ctx1abhicdefg

9 

10# to get a string like this run: 

11# openssl rand -hex 32 

12SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 9 ctx1abhicdefg

13ALGORITHM = "HS256" 9 ctx1abhicdefg

14ACCESS_TOKEN_EXPIRE_MINUTES = 30 9 ctx1abhicdefg

15 

16 

17fake_users_db = { 9 ctx1abhicdefg

18 "johndoe": { 

19 "username": "johndoe", 

20 "full_name": "John Doe", 

21 "email": "[email protected]", 

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

23 "disabled": False, 

24 } 

25} 

26 

27 

28class Token(BaseModel): 9 ctx1abhicdefg

29 access_token: str 7 ctx1abcdefg

30 token_type: str 7 ctx1abcdefg

31 

32 

33class TokenData(BaseModel): 9 ctx1abhicdefg

34 username: str | None = None 9 ctx1abhicdefg

35 

36 

37class User(BaseModel): 9 ctx1abhicdefg

38 username: str 7 ctx1abcdefg

39 email: str | None = None 9 ctx1abhicdefg

40 full_name: str | None = None 9 ctx1abhicdefg

41 disabled: bool | None = None 9 ctx1abhicdefg

42 

43 

44class UserInDB(User): 9 ctx1abhicdefg

45 hashed_password: str 7 ctx1abcdefg

46 

47 

48password_hash = PasswordHash.recommended() 9 ctx1abhicdefg

49 

50DUMMY_HASH = password_hash.hash("dummypassword") 9 ctx1abhicdefg

51 

52oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") 9 ctx1abhicdefg

53 

54app = FastAPI() 9 ctx1abhicdefg

55 

56 

57def verify_password(plain_password, hashed_password): 9 ctx1abhicdefg

58 return password_hash.verify(plain_password, hashed_password) 23 ctx1txGjklYumvyHnopZwzIqrs0

59 

60 

61def get_password_hash(password): 9 ctx1abhicdefg

62 return password_hash.hash(password) 8 ctx11l2U3p4s

63 

64 

65def get_user(db, username: str): 9 ctx1abhicdefg

66 if username in db: 26 ctx1txGjklABumvyHnopCDwzIqrsEF

67 user_dict = db[username] 17 ctx1txjklumvynopwzqrs

68 return UserInDB(**user_dict) 17 ctx1txjklumvynopwzqrs

69 

70 

71def authenticate_user(fake_db, username: str, password: str): 9 ctx1abhicdefg

72 user = get_user(fake_db, username) 20 ctx1txGjklumvyHnopwzIqrs

73 if not user: 20 ctx1txGjklumvyHnopwzIqrs

74 verify_password(password, DUMMY_HASH) 4 ctx1GVHI

75 return False 4 ctx1GVHI

76 if not verify_password(password, user.hashed_password): 17 ctx1txjklumvynopwzqrs

77 return False 4 ctx1xWyz

78 return user 14 ctx1tjklumvnopwqrs

79 

80 

81def create_access_token(data: dict, expires_delta: timedelta | None = None): 9 ctx1abhicdefg

82 to_encode = data.copy() 17 ctx1NtjklumOvnopPwqrs

83 if expires_delta: 17 ctx1NtjklumOvnopPwqrs

84 expire = datetime.now(timezone.utc) + expires_delta 14 ctx1tjklumvnopwqrs

85 else: 

86 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 4 ctx1N5OP

87 to_encode.update({"exp": expire}) 17 ctx1NtjklumOvnopPwqrs

88 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 17 ctx1NtjklumOvnopPwqrs

89 return encoded_jwt 17 ctx1NtjklumOvnopPwqrs

90 

91 

92async def get_current_user(token: str = Depends(oauth2_scheme)): 9 ctx1abhicdefg

93 credentials_exception = HTTPException( 23 ctx1QjklKABmJRnopLCDSqrsMEF

94 status_code=status.HTTP_401_UNAUTHORIZED, 

95 detail="Could not validate credentials", 

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

97 ) 

98 try: 23 ctx1QjklKABmJRnopLCDSqrsMEF

99 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 23 ctx1QjklKABmJRnopLCDSqrsMEF

100 username = payload.get("sub") 20 ctx1jklKABmJnopLCDqrsMEF

101 if username is None: 20 ctx1jklKABmJnopLCDqrsMEF

102 raise credentials_exception 4 ctx1K6LM

103 token_data = TokenData(username=username) 17 ctx1jklABmJnopCDqrsEF

104 except InvalidTokenError: 7 ctx1QKXRLSM

105 raise credentials_exception 4 ctx1QXRS

106 user = get_user(fake_users_db, username=token_data.username) 17 ctx1jklABmJnopCDqrsEF

107 if user is None: 17 ctx1jklABmJnopCDqrsEF

108 raise credentials_exception 8 ctx1AB7JCDEF

109 return user 11 ctx1jklTmnopqrs

110 

111 

112async def get_current_active_user(current_user: User = Depends(get_current_user)): 9 ctx1abhicdefg

113 if current_user.disabled: 11 ctx1jklTmnopqrs

114 raise HTTPException(status_code=400, detail="Inactive user") 4 ctx1lUps

115 return current_user 8 ctx1jkTmnoqr

116 

117 

118@app.post("/token") 9 ctx1abhicdefg

119async def login_for_access_token( 9 ctx1abhicdefg

120 form_data: OAuth2PasswordRequestForm = Depends(), 

121) -> Token: 

122 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 20 ctx1txGjklumvyHnopwzIqrs

123 if not user: 20 ctx1txGjklumvyHnopwzIqrs

124 raise HTTPException( 7 ctx1xGWyHzI

125 status_code=status.HTTP_401_UNAUTHORIZED, 

126 detail="Incorrect username or password", 

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

128 ) 

129 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 14 ctx1tjklumvnopwqrs

130 access_token = create_access_token( 14 ctx1tjklumvnopwqrs

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

132 ) 

133 return Token(access_token=access_token, token_type="bearer") 14 ctx1tjklumvnopwqrs

134 

135 

136@app.get("/users/me/") 9 ctx1abhicdefg

137async def read_users_me(current_user: User = Depends(get_current_active_user)) -> User: 9 ctx1abhicdefg

138 return current_user 4 ctx1kmor

139 

140 

141@app.get("/users/me/items/") 9 ctx1abhicdefg

142async def read_own_items(current_user: User = Depends(get_current_active_user)): 9 ctx1abhicdefg

143 return [{"item_id": "Foo", "owner": current_user.username}] 4 ctx1jTnq