from dataclasses import dataclass from typing import Annotated import jwt from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from app.core.config import Settings, get_settings bearer = HTTPBearer(auto_error=False) @dataclass(slots=True) class CurrentUser: id: str email: str | None = None role: str = "authenticated" async def get_current_user( credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)], settings: Annotated[Settings, Depends(get_settings)], ) -> CurrentUser: if credentials is None: if settings.auth_required: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required", ) return CurrentUser(id="00000000-0000-0000-0000-000000000000", role="anonymous") try: if not settings.supabase_url: if settings.environment == "production": raise ValueError("SUPABASE_URL is not configured") claims = jwt.decode( credentials.credentials, options={"verify_signature": False}, algorithms=["HS256", "RS256"], ) else: jwks_client = jwt.PyJWKClient( f"{settings.supabase_url.rstrip('/')}/auth/v1/.well-known/jwks.json", cache_jwk_set=True, lifespan=3600, ) signing_key = jwks_client.get_signing_key_from_jwt(credentials.credentials) claims = jwt.decode( credentials.credentials, signing_key.key, algorithms=["ES256", "RS256"], audience=settings.supabase_jwt_audience, options={"require": ["exp", "sub"]}, ) except (jwt.PyJWTError, ValueError) as exc: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired access token", ) from exc return CurrentUser( id=str(claims["sub"]), email=claims.get("email"), role=claims.get("role", "authenticated"), )