118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""OIDC/JWT authentication helpers for backend API requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from functools import lru_cache
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
import jwt
|
|
import requests
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
from jwt import PyJWKClient
|
|
from jwt.exceptions import InvalidTokenError
|
|
|
|
from media_library_viewer_api.config import Settings, get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EXEMPT_PATHS = {
|
|
"/api/health",
|
|
"/docs",
|
|
"/openapi.json",
|
|
"/redoc",
|
|
}
|
|
|
|
|
|
def _normalize_issuer_url(issuer_url: str) -> str:
|
|
return issuer_url.rstrip("/") + "/" if issuer_url else ""
|
|
|
|
|
|
@lru_cache
|
|
def get_oidc_metadata(issuer_url: str) -> dict[str, Any]:
|
|
normalized = _normalize_issuer_url(issuer_url)
|
|
discovery_url = urljoin(normalized, ".well-known/openid-configuration")
|
|
response = requests.get(discovery_url, timeout=10)
|
|
response.raise_for_status()
|
|
metadata = response.json()
|
|
if not isinstance(metadata, dict):
|
|
raise RuntimeError("OIDC discovery response was not a JSON object")
|
|
return metadata
|
|
|
|
|
|
@lru_cache
|
|
def get_jwk_client(jwks_url: str) -> PyJWKClient:
|
|
return PyJWKClient(jwks_url)
|
|
|
|
|
|
def _split_audience(audience: str) -> list[str]:
|
|
return [item.strip() for item in audience.split(",") if item.strip()]
|
|
|
|
|
|
def validate_auth_settings(settings: Settings) -> None:
|
|
if not settings.auth_enabled:
|
|
return
|
|
if not settings.oidc_issuer_url:
|
|
raise RuntimeError("AUTH_ENABLED is true but OIDC_ISSUER_URL is not configured")
|
|
if not settings.oidc_audience:
|
|
raise RuntimeError("AUTH_ENABLED is true but OIDC_AUDIENCE is not configured")
|
|
|
|
|
|
def validate_bearer_jwt(authorization: str | None, settings: Settings | None = None) -> dict[str, Any]:
|
|
settings = settings or get_settings()
|
|
validate_auth_settings(settings)
|
|
if not settings.auth_enabled:
|
|
return {}
|
|
|
|
if not authorization:
|
|
raise PermissionError("Missing Authorization header")
|
|
|
|
scheme, _, token = authorization.partition(" ")
|
|
if scheme.lower() != "bearer" or not token.strip():
|
|
raise PermissionError("Authorization header must use Bearer token format")
|
|
|
|
issuer_url = _normalize_issuer_url(settings.oidc_issuer_url)
|
|
metadata = get_oidc_metadata(issuer_url)
|
|
jwks_url = settings.oidc_jwks_url.strip() or str(metadata.get("jwks_uri") or "")
|
|
if not jwks_url:
|
|
raise RuntimeError("OIDC discovery metadata does not include a JWKS URL")
|
|
|
|
jwk_client = get_jwk_client(jwks_url)
|
|
signing_key = jwk_client.get_signing_key_from_jwt(token).key
|
|
audience = _split_audience(settings.oidc_audience)
|
|
claims = jwt.decode(
|
|
token,
|
|
signing_key,
|
|
algorithms=list(metadata.get("id_token_signing_alg_values_supported") or ["RS256"]),
|
|
audience=audience[0] if len(audience) == 1 else audience,
|
|
issuer=issuer_url,
|
|
leeway=int(settings.oidc_clock_skew_seconds or 0),
|
|
options={"require": ["exp", "iss"]},
|
|
)
|
|
return claims
|
|
|
|
|
|
async def require_jwt_auth(request: Request, call_next):
|
|
settings = get_settings()
|
|
path = request.url.path
|
|
if request.method == "OPTIONS" or path in EXEMPT_PATHS or not path.startswith("/api"):
|
|
return await call_next(request)
|
|
|
|
try:
|
|
claims = validate_bearer_jwt(request.headers.get("authorization"), settings)
|
|
except PermissionError as exc:
|
|
logger.warning("JWT auth rejected path=%s reason=%s", path, exc)
|
|
return JSONResponse(status_code=401, content={"detail": str(exc)})
|
|
except InvalidTokenError as exc:
|
|
logger.warning("JWT auth token invalid path=%s error=%s", path, exc)
|
|
return JSONResponse(status_code=401, content={"detail": "Invalid bearer token"})
|
|
except Exception as exc: # pragma: no cover - safety net for OIDC/JWKS failures
|
|
logger.exception("JWT auth validation failed path=%s", path)
|
|
return JSONResponse(status_code=500, content={"detail": str(exc)})
|
|
|
|
request.state.jwt_claims = claims
|
|
request.state.jwt_subject = claims.get("sub") if isinstance(claims, dict) else None
|
|
return await call_next(request)
|