141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
from functools import lru_cache
|
|
from typing import Any, Dict, Optional
|
|
|
|
import httpx
|
|
import jwt
|
|
from fastapi import HTTPException, status
|
|
from jwt import PyJWKClient
|
|
|
|
from backend.settings import Settings
|
|
|
|
|
|
@dataclass
|
|
class OIDCConfig:
|
|
issuer: str
|
|
authorization_endpoint: str
|
|
token_endpoint: str
|
|
jwks_uri: str
|
|
userinfo_endpoint: Optional[str] = None
|
|
|
|
|
|
class OIDCError(Exception):
|
|
"""Raised when the OIDC provider cannot be used."""
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _load_provider_config(issuer: str) -> OIDCConfig:
|
|
discovery_url = urllib.parse.urljoin(issuer.rstrip("/") + "/", ".well-known/openid-configuration")
|
|
try:
|
|
with httpx.Client(timeout=5.0) as client:
|
|
resp = client.get(discovery_url)
|
|
resp.raise_for_status()
|
|
except httpx.HTTPError as exc: # pragma: no cover - network failure branch
|
|
raise OIDCError(f"Failed to load discovery document: {exc}") from exc
|
|
|
|
data = resp.json()
|
|
required = ("issuer", "authorization_endpoint", "token_endpoint", "jwks_uri")
|
|
if not all(key in data for key in required):
|
|
raise OIDCError("Discovery document missing required fields")
|
|
|
|
return OIDCConfig(
|
|
issuer=data["issuer"],
|
|
authorization_endpoint=data["authorization_endpoint"],
|
|
token_endpoint=data["token_endpoint"],
|
|
jwks_uri=data["jwks_uri"],
|
|
userinfo_endpoint=data.get("userinfo_endpoint"),
|
|
)
|
|
|
|
|
|
class OIDCClient:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
if not settings.oidc_issuer:
|
|
raise OIDCError("SKD_OIDC_ISSUER not configured")
|
|
if not settings.oidc_client_id or not settings.oidc_client_secret:
|
|
raise OIDCError("SKD_OIDC_CLIENT_ID/SECRET must be set")
|
|
self.config = _load_provider_config(settings.oidc_issuer)
|
|
self.jwk_client = PyJWKClient(self.config.jwks_uri)
|
|
|
|
def build_state_token(self) -> str:
|
|
raw = secrets.token_urlsafe(24)
|
|
sig = hmac.new(self.settings.auth_secret.encode(), raw.encode(), hashlib.sha256).hexdigest()
|
|
return f"{raw}.{sig}"
|
|
|
|
def is_state_valid(self, provided: str, stored: str) -> bool:
|
|
if not provided or not stored or provided != stored:
|
|
return False
|
|
try:
|
|
raw, sig = provided.split(".", 1)
|
|
except ValueError:
|
|
return False
|
|
expected = hmac.new(self.settings.auth_secret.encode(), raw.encode(), hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(sig, expected)
|
|
|
|
def authorization_url(self, state: str) -> str:
|
|
params = {
|
|
"client_id": self.settings.oidc_client_id,
|
|
"redirect_uri": self.settings.oidc_redirect_uri,
|
|
"response_type": "code",
|
|
"scope": self.settings.oidc_scopes,
|
|
"state": state,
|
|
}
|
|
return f"{self.config.authorization_endpoint}?{urllib.parse.urlencode(params)}"
|
|
|
|
def exchange_code_for_claims(self, code: str) -> Dict[str, Any]:
|
|
payload = {
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": self.settings.oidc_redirect_uri,
|
|
"client_id": self.settings.oidc_client_id,
|
|
"client_secret": self.settings.oidc_client_secret,
|
|
}
|
|
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.post(self.config.token_endpoint, data=payload, headers=headers)
|
|
except httpx.HTTPError as exc: # pragma: no cover - network failure branch
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"OIDC token request failed: {exc}",
|
|
) from exc
|
|
|
|
if resp.status_code != status.HTTP_200_OK:
|
|
detail = resp.text or "token request failed"
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"OIDC token exchange failed: {detail}",
|
|
)
|
|
|
|
token_response = resp.json()
|
|
id_token = token_response.get("id_token")
|
|
if not id_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="OIDC response missing id_token",
|
|
)
|
|
|
|
signing_key = self.jwk_client.get_signing_key_from_jwt(id_token).key
|
|
try:
|
|
claims = jwt.decode(
|
|
id_token,
|
|
signing_key,
|
|
algorithms=["RS256"],
|
|
audience=self.settings.oidc_client_id,
|
|
issuer=self.config.issuer,
|
|
)
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"Invalid ID token: {exc}",
|
|
) from exc
|
|
return claims
|
|
|
|
@staticmethod
|
|
def extract_username(claims: Dict[str, Any]) -> Optional[str]:
|
|
return claims.get("preferred_username") or claims.get("email") or claims.get("sub")
|