Coverage for docs_src/security/tutorial005_py39.py: 100%
93 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 03:53 +0000
1from datetime import datetime, timedelta, timezone 1abcd
2from typing import Union 1abcd
4import jwt 1abcd
5from fastapi import Depends, FastAPI, HTTPException, Security, status 1abcd
6from fastapi.security import ( 1abcd
7 OAuth2PasswordBearer,
8 OAuth2PasswordRequestForm,
9 SecurityScopes,
10)
11from jwt.exceptions import InvalidTokenError 1abcd
12from passlib.context import CryptContext 1abcd
13from pydantic import BaseModel, ValidationError 1abcd
15# to get a string like this run:
16# openssl rand -hex 32
17SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" 1abcd
18ALGORITHM = "HS256" 1abcd
19ACCESS_TOKEN_EXPIRE_MINUTES = 30 1abcd
22fake_users_db = { 1abcd
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}
40class Token(BaseModel): 1abcd
41 access_token: str 1abcd
42 token_type: str 1abcd
45class TokenData(BaseModel): 1abcd
46 username: Union[str, None] = None 1abcd
47 scopes: list[str] = [] 1abcd
50class User(BaseModel): 1abcd
51 username: str 1abcd
52 email: Union[str, None] = None 1abcd
53 full_name: Union[str, None] = None 1abcd
54 disabled: Union[bool, None] = None 1abcd
57class UserInDB(User): 1abcd
58 hashed_password: str 1abcd
61pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 1abcd
63oauth2_scheme = OAuth2PasswordBearer( 1abcd
64 tokenUrl="token",
65 scopes={"me": "Read information about the current user.", "items": "Read items."},
66)
68app = FastAPI() 1abcd
71def verify_password(plain_password, hashed_password): 1abcd
72 return pwd_context.verify(plain_password, hashed_password) 1abcd
75def get_password_hash(password): 1abcd
76 return pwd_context.hash(password) 1abcd
79def get_user(db, username: str): 1abcd
80 if username in db: 1abcd
81 user_dict = db[username] 1abcd
82 return UserInDB(**user_dict) 1abcd
85def authenticate_user(fake_db, username: str, password: str): 1abcd
86 user = get_user(fake_db, username) 1abcd
87 if not user: 1abcd
88 return False 1abcd
89 if not verify_password(password, user.hashed_password): 1abcd
90 return False 1abcd
91 return user 1abcd
94def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None): 1abcd
95 to_encode = data.copy() 1abcd
96 if expires_delta: 1abcd
97 expire = datetime.now(timezone.utc) + expires_delta 1abcd
98 else:
99 expire = datetime.now(timezone.utc) + timedelta(minutes=15) 1abcd
100 to_encode.update({"exp": expire}) 1abcd
101 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) 1abcd
102 return encoded_jwt 1abcd
105async def get_current_user( 1abcd
106 security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
107):
108 if security_scopes.scopes: 1abcd
109 authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' 1abcd
110 else:
111 authenticate_value = "Bearer" 1abcd
112 credentials_exception = HTTPException( 1abcd
113 status_code=status.HTTP_401_UNAUTHORIZED,
114 detail="Could not validate credentials",
115 headers={"WWW-Authenticate": authenticate_value},
116 )
117 try: 1abcd
118 payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) 1abcd
119 username: str = payload.get("sub") 1abcd
120 if username is None: 1abcd
121 raise credentials_exception 1abcd
122 token_scopes = payload.get("scopes", []) 1abcd
123 token_data = TokenData(scopes=token_scopes, username=username) 1abcd
124 except (InvalidTokenError, ValidationError): 1abcd
125 raise credentials_exception 1abcd
126 user = get_user(fake_users_db, username=token_data.username) 1abcd
127 if user is None: 1abcd
128 raise credentials_exception 1abcd
129 for scope in security_scopes.scopes: 1abcd
130 if scope not in token_data.scopes: 1abcd
131 raise HTTPException( 1abcd
132 status_code=status.HTTP_401_UNAUTHORIZED,
133 detail="Not enough permissions",
134 headers={"WWW-Authenticate": authenticate_value},
135 )
136 return user 1abcd
139async def get_current_active_user( 1abcd
140 current_user: User = Security(get_current_user, scopes=["me"]),
141):
142 if current_user.disabled: 1abcd
143 raise HTTPException(status_code=400, detail="Inactive user") 1abcd
144 return current_user 1abcd
147@app.post("/token") 1abcd
148async def login_for_access_token( 1abcd
149 form_data: OAuth2PasswordRequestForm = Depends(),
150) -> Token:
151 user = authenticate_user(fake_users_db, form_data.username, form_data.password) 1abcd
152 if not user: 1abcd
153 raise HTTPException(status_code=400, detail="Incorrect username or password") 1abcd
154 access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) 1abcd
155 access_token = create_access_token( 1abcd
156 data={"sub": user.username, "scopes": form_data.scopes},
157 expires_delta=access_token_expires,
158 )
159 return Token(access_token=access_token, token_type="bearer") 1abcd
162@app.get("/users/me/", response_model=User) 1abcd
163async def read_users_me(current_user: User = Depends(get_current_active_user)): 1abcd
164 return current_user 1abcd
167@app.get("/users/me/items/") 1abcd
168async def read_own_items( 1abcd
169 current_user: User = Security(get_current_active_user, scopes=["items"]),
170):
171 return [{"item_id": "Foo", "owner": current_user.username}] 1abcd
174@app.get("/status/") 1abcd
175async def read_system_status(current_user: User = Depends(get_current_user)): 1abcd
176 return {"status": "ok"} 1abcd