phase1/scripts/clerk_auth.py

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