"""Clerk authentication helpers for the Moonley API.""" import os from dataclasses import dataclass from clerk_backend_api import AuthenticateRequestOptions, authenticate_request from fastapi import Request from fastapi.responses import JSONResponse PUBLIC_PATHS = { "/", "/api/v2/auth/config", "/api/v2/health", "/api/v2/ready", } def _csv_env(name: str) -> list[str]: return [item.strip().rstrip("/") for item in os.environ.get(name, "").split(",") if item.strip()] def _pem_env(name: str) -> str | None: value = os.environ.get(name, "").strip() return value.replace("\\n", "\n") if value else None @dataclass(frozen=True) class ClerkSettings: publishable_key: str secret_key: str jwt_key: str | None authorized_parties: list[str] @property def configured(self) -> bool: return bool(self.publishable_key and self.secret_key and self.authorized_parties) def clerk_settings() -> ClerkSettings: return ClerkSettings( publishable_key=os.environ.get("CLERK_PUBLISHABLE_KEY", "").strip(), secret_key=os.environ.get("CLERK_SECRET_KEY", "").strip(), jwt_key=_pem_env("CLERK_JWT_KEY"), authorized_parties=_csv_env("CLERK_AUTHORIZED_PARTIES"), ) def cors_origins() -> list[str]: """Use the same explicit browser allow-list as Clerk's azp validation.""" return clerk_settings().authorized_parties or ["*"] def frontend_auth_config() -> JSONResponse: settings = clerk_settings() if not settings.publishable_key: return JSONResponse( {"error": "authentication_not_configured"}, status_code=503, headers={"Cache-Control": "no-store"}, ) return JSONResponse( {"publishable_key": settings.publishable_key, "configured": settings.configured}, headers={"Cache-Control": "no-store"}, ) def authenticate_clerk_request(request: Request) -> JSONResponse | None: """Verify a Clerk session token and attach its claims to request.state.""" settings = clerk_settings() if not settings.configured: return JSONResponse( {"error": "authentication_not_configured"}, status_code=503, headers={"Cache-Control": "no-store"}, ) try: state = authenticate_request( request, AuthenticateRequestOptions( secret_key=settings.secret_key, jwt_key=settings.jwt_key, authorized_parties=settings.authorized_parties, accepts_token=["session_token"], ), ) except Exception as exc: # Never log the Authorization header or token. The exception type is enough # to distinguish SDK/network failures operationally. print(f"[clerk] verification unavailable: {type(exc).__name__}", flush=True) return JSONResponse( {"error": "authentication_unavailable"}, status_code=503, headers={"Cache-Control": "no-store"}, ) if not state.is_signed_in: reason = state.reason.name if state.reason else "unauthorized" return JSONResponse( {"error": "unauthorized", "reason": reason}, status_code=401, headers={"WWW-Authenticate": "Bearer", "Cache-Control": "no-store"}, ) request.state.clerk_auth = state request.state.clerk_user_id = state.payload["sub"] return None