1- import secrets
1+ """Security utilities for creating and validating JWT tokens."""
2+
23from datetime import datetime , timedelta , timezone
3- from typing import Optional
4+ from typing import Any , Dict , Optional
5+
46from fastapi import Depends , HTTPException , status
57from fastapi .security import OAuth2PasswordBearer
6- from jose import jwt , JWTError
8+ from jose import JWTError , jwt
79
810from app .core .config import settings
911
1012ALGORITHM = "HS256"
1113
1214oauth2_scheme = OAuth2PasswordBearer (tokenUrl = "/auth/login" )
1315
14- def create_access_token (data : dict , expires_delta : Optional [timedelta ] = None ) -> str :
16+
17+ def create_access_token (data : Dict [str , Any ], expires_delta : Optional [timedelta ] = None ) -> str :
18+ """Create a signed JWT access token."""
19+
1520 to_encode = data .copy ()
16- expire = datetime .now (timezone .utc ) + (expires_delta or timedelta (minutes = settings .ACCESS_TOKEN_EXPIRE_MINUTES ))
21+ expire = datetime .now (timezone .utc ) + (
22+ expires_delta or timedelta (minutes = settings .ACCESS_TOKEN_EXPIRE_MINUTES )
23+ )
1724 to_encode .update ({"exp" : expire })
1825 return jwt .encode (to_encode , settings .JWT_SECRET_KEY , algorithm = ALGORITHM )
1926
2027
21- def create_refresh_token (data : dict , expires_delta : Optional [timedelta ] = None ) -> str :
28+ def create_refresh_token (data : Dict [str , Any ], expires_delta : Optional [timedelta ] = None ) -> str :
29+ """Create a signed JWT refresh token."""
30+
2231 to_encode = data .copy ()
23- expire = datetime .now (timezone .utc ) + (expires_delta or timedelta (days = settings .REFRESH_TOKEN_EXPIRE_DAYS ))
32+ expire = datetime .now (timezone .utc ) + (
33+ expires_delta or timedelta (days = settings .REFRESH_TOKEN_EXPIRE_DAYS )
34+ )
2435 to_encode .update ({"exp" : expire , "type" : "refresh" })
2536 return jwt .encode (to_encode , settings .JWT_SECRET_KEY , algorithm = ALGORITHM )
2637
2738
28- def verify_token (token : str ) -> Optional [dict ]:
39+ def verify_token (token : str ) -> Optional [Dict [str , Any ]]:
40+ """Decode a JWT token and return its payload, if valid."""
41+
2942 try :
30- payload = jwt .decode (token , settings .JWT_SECRET_KEY , algorithms = [ALGORITHM ])
31- return payload
43+ return jwt .decode (token , settings .JWT_SECRET_KEY , algorithms = [ALGORITHM ])
3244 except JWTError :
3345 return None
3446
35- def get_current_user (token : str = Depends (oauth2_scheme )):
47+
48+ def get_current_user (token : str = Depends (oauth2_scheme )) -> str :
49+ """Return the username encoded in the access token."""
50+
3651 payload = verify_token (token )
3752 if not payload :
3853 raise HTTPException (
@@ -43,6 +58,9 @@ def get_current_user(token: str = Depends(oauth2_scheme)):
4358
4459 username = payload .get ("sub" )
4560 if not username :
46- raise HTTPException (status_code = status .HTTP_401_UNAUTHORIZED , detail = "Invalid token payload." )
61+ raise HTTPException (
62+ status_code = status .HTTP_401_UNAUTHORIZED ,
63+ detail = "Invalid token payload." ,
64+ )
4765
48- return username
66+ return username
0 commit comments