"""Authentication dependencies for FastAPI routes.""" from dataclasses import dataclass from typing import Optional from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from app.core.config import settings security = HTTPBearer(auto_error=False) @dataclass class AuthenticatedUser: """Authenticated user extracted from JWT token.""" google_id: str email: str name: str picture: Optional[str] = None def _decode_token(token: str) -> dict: """Decode and validate a JWT token.""" try: payload = jwt.decode( token, settings.nextauth_secret, algorithms=["HS256"], ) return payload except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token", headers={"WWW-Authenticate": "Bearer"}, ) async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> AuthenticatedUser: """Require a valid JWT and return the authenticated user.""" if credentials is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required", headers={"WWW-Authenticate": "Bearer"}, ) payload = _decode_token(credentials.credentials) return AuthenticatedUser( google_id=payload.get("sub", ""), email=payload.get("email", ""), name=payload.get("name", ""), picture=payload.get("picture"), ) async def get_optional_user( credentials: Optional[HTTPAuthorizationCredentials] = Depends(security), ) -> Optional[AuthenticatedUser]: """Return authenticated user if token is present, otherwise None.""" if credentials is None: return None try: payload = _decode_token(credentials.credentials) return AuthenticatedUser( google_id=payload.get("sub", ""), email=payload.get("email", ""), name=payload.get("name", ""), picture=payload.get("picture"), ) except HTTPException: return None