app/core/security.py

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"),
    )