File size: 1,788 Bytes
310260a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from typing import Dict
from ..config import settings
# HTTP Bearer security scheme
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> Dict[str, any]:
"""
Verify JWT token and extract user identity.
This dependency extracts and validates the JWT token from the Authorization header.
User identity is extracted from the token payload, never from client input.
Args:
credentials: HTTP Bearer credentials containing JWT token
Returns:
Dict containing user_id and email from token payload
Raises:
HTTPException: 401 if token is missing, invalid, or expired
"""
try:
# Decode and verify JWT token
payload = jwt.decode(
credentials.credentials,
settings.JWT_SECRET,
algorithms=[settings.JWT_ALGORITHM]
)
# Extract user ID from token payload
user_id = payload.get("sub")
email = payload.get("email")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token: missing user ID"
)
return {
"user_id": int(user_id),
"email": email
}
except JWTError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Invalid token: {str(e)}"
)
except ValueError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token: malformed user ID"
)
|