Spaces:
Running
Running
| """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) | |
| 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 | |