feature: add oidc login flow

backend/oidc.py

@@ -0,0 +1,140 @@
+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")