File size: 2,195 Bytes
3e93464 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | 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"),
)
|