| """ |
| Authentication dependencies for FastAPI. |
| |
| Extracts and validates JWT from the Authorization header. |
| """ |
|
|
| from fastapi import Depends, HTTPException, status |
| from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials |
| from jose import JWTError, jwt |
| from sqlalchemy.orm import Session |
|
|
| from backend.config.settings import get_settings |
| from backend.db.database import get_db |
| from backend.db.models import User |
|
|
| security = HTTPBearer() |
|
|
|
|
| def get_current_user( |
| credentials: HTTPAuthorizationCredentials = Depends(security), |
| db: Session = Depends(get_db), |
| ) -> User: |
| """Decode the JWT and return the authenticated User, or raise 401.""" |
| settings = get_settings() |
| token = credentials.credentials |
|
|
| credentials_exception = HTTPException( |
| status_code=status.HTTP_401_UNAUTHORIZED, |
| detail="Invalid or expired token", |
| headers={"WWW-Authenticate": "Bearer"}, |
| ) |
|
|
| try: |
| payload = jwt.decode( |
| token, |
| settings.SECRET_KEY, |
| algorithms=[settings.JWT_ALGORITHM], |
| ) |
| user_id: int | None = payload.get("sub") |
| if user_id is None: |
| raise credentials_exception |
| except JWTError: |
| raise credentials_exception |
|
|
| user = db.query(User).filter(User.id == int(user_id)).first() |
| if user is None: |
| raise credentials_exception |
|
|
| return user |
|
|
