File size: 3,457 Bytes
1d9bd9b | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """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
|