feat: v6.7.0 — SSO/SAML + OIDC entreprise (SP SAML signé + OIDC PKCE, auto-provisioning, groupes IdP → rôles workspace, SSO only, onglet Settings « SSO / Enterprise », API /api/v2/sso/*, help, migration 23, docs + OpenAPI 439 chemins) · 802 tests verts

app/auth/providers/__init__.py

@@ -0,0 +1,262 @@
+"""Multi-forge OAuth providers — Gitea + GitHub."""
+
+from __future__ import annotations
+
+import logging
+import time
+from abc import ABC, abstractmethod
+from urllib.parse import urlencode
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+class OAuthProvider(ABC):
+    """Abstract OAuth2 provider interface."""
+
+    name: str = ""
+    icon: str = "🔗"
+
+    @abstractmethod
+    def is_enabled(self) -> bool:
+        """Whether this provider is configured."""
+
+    @abstractmethod
+    def get_authorize_url(self, state: str, redirect_uri: str | None = None, force_login: bool = False) -> str:
+        """Build the authorization URL."""
+
+    @abstractmethod
+    async def exchange_code(self, code: str, redirect_uri: str | None = None) -> dict | None:
+        """Exchange authorization code for access token."""
+
+    @abstractmethod
+    async def get_user(self, access_token: str) -> dict | None:
+        """Fetch user profile from the provider."""
+
+    @abstractmethod
+    async def list_repositories(self, access_token: str) -> list[dict]:
+        """List all repositories accessible to this user."""
+
+
+class GiteaProvider(OAuthProvider):
+    """Gitea OAuth2 provider."""
+
+    name = "gitea"
+    icon = "🔗"
+
+    def __init__(self, base_url: str, client_id: str, client_secret: str, redirect_uri: str):
+        self.base = base_url.rstrip("/")
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.redirect_uri = redirect_uri
+
+    def is_enabled(self) -> bool:
+        return bool(self.client_id and self.client_secret)
+
+    def get_authorize_url(self, state: str, redirect_uri: str | None = None, force_login: bool = False) -> str:
+        params = {
+            "client_id": self.client_id,
+            "redirect_uri": redirect_uri or self.redirect_uri,
+            "response_type": "code",
+            "state": state,
+        }
+        if force_login:
+            # Gitea supports prompt=login param (undocumented but works)
+            # If not, fallback: add _force= timestamp cache-buster
+            params["_force"] = str(int(time.time()))
+        return f"{self.base}/login/oauth/authorize?" + urlencode(params)
+
+    async def exchange_code(self, code: str, redirect_uri: str | None = None) -> dict | None:
+        url = f"{self.base}/login/oauth/access_token"
+        data = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "code": code,
+            "grant_type": "authorization_code",
+            "redirect_uri": redirect_uri or self.redirect_uri,
+        }
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.post(url, json=data, headers={"Accept": "application/json"})
+            if r.status_code != 200:
+                logger.error("Gitea token exchange failed: %s", r.text)
+                return None
+            return r.json()
+
+    async def get_user(self, access_token: str) -> dict | None:
+        url = f"{self.base}/api/v1/user"
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.get(url, headers={"Authorization": f"token {access_token}"})
+            if r.status_code != 200:
+                return None
+            data = r.json()
+            return {
+                "login": data.get("login") or data.get("username", ""),
+                "full_name": data.get("full_name", ""),
+                "email": data.get("email", ""),
+                "avatar_url": data.get("avatar_url", ""),
+                "provider_id": str(data.get("id", "")),
+            }
+
+    async def list_repositories(self, access_token: str) -> list[dict]:
+        repos = []
+        async with httpx.AsyncClient(timeout=30) as client:
+            for page in range(1, 6):
+                r = await client.get(
+                    f"{self.base}/api/v1/user/repos",
+                    headers={"Authorization": f"token {access_token}"},
+                    params={"page": page, "limit": 50},
+                )
+                if r.status_code != 200:
+                    break
+                data = r.json()
+                if not data:
+                    break
+                for repo in data:
+                    repos.append({
+                        "id": str(repo.get("id", "")),
+                        "name": repo.get("name", ""),
+                        "full_name": repo.get("full_name", ""),
+                        "description": repo.get("description", ""),
+                        "html_url": repo.get("html_url", ""),
+                        "clone_url": repo.get("clone_url", ""),
+                        "default_branch": repo.get("default_branch", "main"),
+                        "language": repo.get("language", ""),
+                        "updated_at": repo.get("updated_at", ""),
+                        "private": repo.get("private", False),
+                        "forge": "gitea",
+                    })
+        return repos
+
+
+class GitHubProvider(OAuthProvider):
+    """GitHub OAuth2 provider."""
+
+    name = "github"
+    icon = "🐙"
+
+    def __init__(self, client_id: str, client_secret: str, redirect_uri: str):
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.redirect_uri = redirect_uri
+        self.authorize_url = "https://github.com/login/oauth/authorize"
+        self.token_url = "https://github.com/login/oauth/access_token"
+        self.api_url = "https://api.github.com"
+
+    def is_enabled(self) -> bool:
+        return bool(self.client_id and self.client_secret)
+
+    def get_authorize_url(self, state: str, redirect_uri: str | None = None, force_login: bool = False) -> str:
+        return (
+            f"{self.authorize_url}?"
+            + urlencode({
+                "client_id": self.client_id,
+                "redirect_uri": redirect_uri or self.redirect_uri,
+                "scope": "repo,user",
+                "state": state,
+            })
+        )
+
+    async def exchange_code(self, code: str, redirect_uri: str | None = None) -> dict | None:
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.post(
+                self.token_url,
+                data={
+                    "client_id": self.client_id,
+                    "client_secret": self.client_secret,
+                    "code": code,
+                    "redirect_uri": redirect_uri or self.redirect_uri,
+                },
+                headers={"Accept": "application/json"},
+            )
+            if r.status_code != 200:
+                logger.error("GitHub token exchange failed: %s", r.text)
+                return None
+            data = r.json()
+            if "access_token" not in data:
+                return None
+            return data
+
+    async def get_user(self, access_token: str) -> dict | None:
+        url = f"{self.api_url}/user"
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.get(
+                url,
+                headers={"Authorization": f"Bearer {access_token}", "Accept": "application/vnd.github.v3+json"},
+            )
+            if r.status_code != 200:
+                return None
+            data = r.json()
+            return {
+                "login": data.get("login", ""),
+                "full_name": data.get("name", "") or data.get("login", ""),
+                "email": data.get("email", ""),
+                "avatar_url": data.get("avatar_url", ""),
+                "provider_id": str(data.get("id", "")),
+            }
+
+    async def list_repositories(self, access_token: str) -> list[dict]:
+        repos = []
+        async with httpx.AsyncClient(timeout=30) as client:
+            for page in range(1, 6):
+                r = await client.get(
+                    f"{self.api_url}/user/repos",
+                    headers={"Authorization": f"Bearer {access_token}", "Accept": "application/vnd.github.v3+json"},
+                    params={"page": page, "per_page": 50, "sort": "updated"},
+                )
+                if r.status_code != 200:
+                    break
+                data = r.json()
+                if not data:
+                    break
+                for repo in data:
+                    repos.append({
+                        "id": str(repo.get("id", "")),
+                        "name": repo.get("name", ""),
+                        "full_name": repo.get("full_name", ""),
+                        "description": repo.get("description", ""),
+                        "html_url": repo.get("html_url", ""),
+                        "clone_url": repo.get("clone_url", ""),
+                        "default_branch": repo.get("default_branch", "main"),
+                        "language": repo.get("language", ""),
+                        "updated_at": repo.get("updated_at", ""),
+                        "private": repo.get("private", False),
+                        "forge": "github",
+                    })
+        return repos
+
+
+# ═══════════ Provider registry ═══════════
+
+def get_providers() -> list[OAuthProvider]:
+    """Return all configured OAuth providers."""
+    from app.config import settings
+
+    providers: list[OAuthProvider] = []
+
+    gitea = GiteaProvider(
+        base_url=settings.gitea_url,
+        client_id=settings.gitea_oauth_client_id,
+        client_secret=settings.gitea_oauth_client_secret,
+        redirect_uri=settings.oauth_redirect_uri,
+    )
+    if gitea.is_enabled():
+        providers.append(gitea)
+
+    github = GitHubProvider(
+        client_id=settings.github_oauth_client_id or "",
+        client_secret=settings.github_oauth_client_secret or "",
+        redirect_uri=settings.oauth_redirect_uri or "",
+    )
+    if github.is_enabled():
+        providers.append(github)
+
+    return providers
+
+
+def get_provider(name: str) -> OAuthProvider | None:
+    """Get a specific provider by name."""
+    for p in get_providers():
+        if p.name == name:
+            return p
+    return None

app/auth/providers/oidc_provider.py

@@ -0,0 +1,219 @@
+"""OIDC provider — authorization code flow with PKCE (v6.7.0).
+
+Discovery (``.well-known/openid-configuration``) is cached for an hour, the
+ID token signature is verified against the issuer JWKS via authlib's JOSE
+implementation, and ``iss`` / ``aud`` / ``exp`` / ``nonce`` are checked here
+explicitly so the rules are visible and unit-testable.
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import logging
+import secrets
+import time
+import warnings
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+#: Default attribute mapping (design doc §3.2) — OIDC claim names.
+DEFAULT_OIDC_MAPPING: dict[str, str] = {
+    "login": "sub",
+    "email": "email",
+    "full_name": "name",
+    "avatar_url": "picture",
+    "groups": "groups",
+}
+
+_DISCOVERY_TTL = 3600.0
+_discovery_cache: dict[str, tuple[float, dict]] = {}
+
+
+class OIDCError(Exception):
+    """OIDC processing failure — ``message`` is user-facing."""
+
+
+def pkce_pair() -> tuple[str, str]:
+    """Return ``(code_verifier, code_challenge)`` for the S256 method."""
+    verifier = secrets.token_urlsafe(64)
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+    return verifier, challenge
+
+
+def _b64url(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+def _b64url_decode(data: str) -> bytes:
+    return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))
+
+
+async def discover(issuer_url: str) -> dict:
+    """Fetch (and cache) the issuer's OIDC discovery document."""
+    issuer = issuer_url.rstrip("/")
+    url = f"{issuer}/.well-known/openid-configuration"
+    now = time.time()
+    hit = _discovery_cache.get(issuer)
+    if hit and now - hit[0] < _DISCOVERY_TTL:
+        return hit[1]
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.get(url)
+            r.raise_for_status()
+            doc = r.json()
+    except Exception as err:
+        raise OIDCError(f"OIDC discovery failed ({url}): {err}") from err
+    if not doc.get("authorization_endpoint") or not doc.get("token_endpoint"):
+        raise OIDCError("OIDC discovery document is missing authorization/token endpoints")
+    _discovery_cache[issuer] = (now, doc)
+    return doc
+
+
+def build_authorize_url(
+    doc: dict,
+    *,
+    client_id: str,
+    redirect_uri: str,
+    scope: str,
+    state: str,
+    nonce: str,
+    code_challenge: str,
+) -> str:
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": client_id,
+        "redirect_uri": redirect_uri,
+        "response_type": "code",
+        "scope": scope or "openid profile email",
+        "state": state,
+        "nonce": nonce,
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
+    }
+    sep = "&" if "?" in doc["authorization_endpoint"] else "?"
+    return doc["authorization_endpoint"] + sep + urlencode(params)
+
+
+async def exchange_code(
+    doc: dict, *, client_id: str, client_secret: str, code: str, redirect_uri: str, code_verifier: str
+) -> dict:
+    """Exchange the authorization code for tokens (PKCE, confidential client)."""
+    data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": redirect_uri,
+        "client_id": client_id,
+        "code_verifier": code_verifier,
+    }
+    auth = None
+    if client_secret:
+        auth = (client_id, client_secret)
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.post(doc["token_endpoint"], data=data, auth=auth)
+    except Exception as err:
+        raise OIDCError(f"OIDC token request failed: {err}") from err
+    if r.status_code != 200:
+        raise OIDCError(f"OIDC token endpoint returned {r.status_code}: {r.text[:300]}")
+    try:
+        tokens = r.json()
+    except Exception as err:
+        raise OIDCError(f"OIDC token endpoint returned a non-JSON body: {err}") from err
+    if "error" in tokens:
+        raise OIDCError(f"OIDC error: {tokens.get('error')} {tokens.get('error_description', '')}".strip())
+    return tokens
+
+
+async def fetch_userinfo(doc: dict, access_token: str) -> dict:
+    """Best-effort userinfo fetch (groups often only live there)."""
+    endpoint = doc.get("userinfo_endpoint")
+    if not endpoint or not access_token:
+        return {}
+    try:
+        async with httpx.AsyncClient(timeout=15) as client:
+            r = await client.get(endpoint, headers={"Authorization": f"Bearer {access_token}"})
+        if r.status_code != 200:
+            return {}
+        data = r.json()
+        return data if isinstance(data, dict) else {}
+    except Exception as err:  # userinfo is optional enrichment
+        logger.debug("userinfo fetch failed: %s", err)
+        return {}
+
+
+def validate_id_token(
+    id_token: str, *, issuer: str, client_id: str, nonce: str, jwks: dict
+) -> dict:
+    """Verify the ID token signature and claims. Returns the claims dict."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        from authlib.jose import JsonWebKey
+        from authlib.jose import jwt as jose_jwt
+
+    if isinstance(id_token, bytes):
+        # authlib's jose.jwt.encode() returns bytes; IdP token endpoints send
+        # str — accept both instead of crashing on ``bytes.count(".")``.
+        id_token = id_token.decode()
+    if not id_token or id_token.count(".") != 2:
+        raise OIDCError("Missing or malformed ID token")
+
+    try:
+        keyset = JsonWebKey.import_key_set(jwks)
+    except Exception as err:
+        raise OIDCError(f"Invalid issuer JWKS: {err}") from err
+
+    try:
+        # Pick the key matching the token header (kid) when several are offered.
+        header = json.loads(_b64url_decode(id_token.split(".")[0]))
+        kid = header.get("kid")
+        key = keyset.get_by_kid(kid) if kid and hasattr(keyset, "get_by_kid") else None
+        token_obj = jose_jwt.decode(id_token, key or keyset)
+    except Exception as err:
+        raise OIDCError(f"ID token signature verification failed: {err}") from err
+
+    claims = dict(token_obj)  # authlib's JWTClaims is a dict subclass
+    now = int(time.time())
+
+    if claims.get("iss") != issuer.rstrip("/") and claims.get("iss") != issuer:
+        raise OIDCError(f"ID token issuer mismatch: {claims.get('iss')!r}")
+    aud = claims.get("aud")
+    aud_list = aud if isinstance(aud, list) else [aud]
+    if client_id not in aud_list:
+        raise OIDCError("ID token audience does not include this client")
+    exp = claims.get("exp")
+    if not isinstance(exp, int) or exp < now:
+        raise OIDCError("ID token expired")
+    iat = claims.get("iat")
+    if isinstance(iat, int) and iat > now + 300:
+        raise OIDCError("ID token issued in the future")
+    if nonce and claims.get("nonce") != nonce:
+        raise OIDCError("ID token nonce mismatch")
+    if not claims.get("sub"):
+        raise OIDCError("ID token has no subject")
+    return claims
+
+
+def claims_to_identity(claims: dict, mapping: dict | None = None) -> dict:
+    """Map OIDC claims onto the shared ``{login, email, full_name, avatar_url, groups}`` shape."""
+    mapping = mapping or DEFAULT_OIDC_MAPPING
+    identity: dict = {"_raw": claims}
+    for field in ("login", "email", "full_name", "avatar_url"):
+        source = mapping.get(field) or field
+        value = claims.get(source, "")
+        if isinstance(value, list):
+            value = value[0] if value else ""
+        identity[field] = str(value or "").strip()
+    groups = claims.get(mapping.get("groups", "groups"), [])
+    if isinstance(groups, str):
+        groups = [groups]
+    identity["groups"] = [str(g) for g in groups if g]
+    if not identity["email"]:
+        identity["email"] = claims.get("email", "") or ""
+    if not identity["full_name"]:
+        identity["full_name"] = claims.get("name", "") or identity["email"]
+    return identity

app/auth/providers/saml_provider.py

@@ -0,0 +1,279 @@
+"""SAML 2.0 Service Provider — wrapper around python3-saml (OneLogin toolkit).
+
+v6.7.0. Adapts FastAPI's ``Request`` to the toolkit's flat ``request_data``
+dict and builds the SP settings from the ``sso_config`` row.
+
+What the toolkit validates in strict mode (all covered by tests):
+XML schema, signature of the assertion and/or the message against the IdP
+certificate, ``Conditions`` timestamps, ``Audience``, ``Destination``,
+``Issuer``, ``Status``, "exactly one assertion", and ``InResponseTo``
+against the AuthnRequest id we pass to ``process_response()`` — combined
+with the single-use ``sso_requests`` store that makes replay impossible.
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+
+from fastapi import Request
+
+logger = logging.getLogger(__name__)
+
+BINDING_HTTP_REDIRECT = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
+BINDING_HTTP_POST = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
+NAMEID_FORMAT_EMAIL = "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
+
+#: Default attribute mapping (design doc §3.2). ``nameid`` = the assertion's
+#: NameID; every other value is matched against attribute Name / FriendlyName
+#: / URI local part (so ``email`` finds both ``email`` and
+#: ``http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress``).
+DEFAULT_SAML_MAPPING: dict[str, str] = {
+    "login": "nameid",
+    "email": "nameid",
+    "full_name": "displayName",
+    "avatar_url": "avatar",
+    "groups": "groups",
+}
+
+
+class SAMLError(Exception):
+    """SAML processing failure — ``message`` is user-facing, ``reason`` is logged."""
+
+
+@dataclass
+class SAMLIdentity:
+    """What a validated assertion tells us about the user."""
+
+    name_id: str
+    name_id_format: str = ""
+    session_index: str = ""
+    attributes: dict[str, list[str]] = field(default_factory=dict)
+    friendly_attributes: dict[str, list[str]] = field(default_factory=dict)
+
+    def resolve(self, source: str) -> str:
+        """Resolve one mapped source (``nameid`` or an attribute name) → first value."""
+        if not source or source == "nameid":
+            return self.name_id or ""
+        if source in self.attributes and self.attributes[source]:
+            return (self.attributes[source][0] or "").strip()
+        # FriendlyName match (case-insensitive)
+        lower = {k.lower(): v for k, v in self.friendly_attributes.items()}
+        if source.lower() in lower and lower[source.lower()]:
+            return (lower[source.lower()][0] or "").strip()
+        # URI local part match: ".../claims/emailaddress" ~ "emailaddress", and
+        # a mapping of "email" must still find ".../claims/emailaddress".
+        want = source.lower().lstrip("./")
+        for name, values in self.attributes.items():
+            if not values:
+                continue
+            local = name.rsplit("/", 1)[-1].rsplit("}", 1)[-1].lower()
+            if local == want or local.endswith(want) or want.endswith(local):
+                return (values[0] or "").strip()
+        return ""
+
+
+def external_base_url(request: Request) -> str:
+    """Scheme://host the user actually used (proxy-aware, like OAuth redirects)."""
+    proto = request.headers.get("x-forwarded-proto", "")
+    scheme = proto.split(",")[0].strip() or request.url.scheme or "http"
+    fwd_host = request.headers.get("x-forwarded-host", "")
+    host = fwd_host.split(",")[0].strip() or request.headers.get("host", "localhost:8080")
+    return f"{scheme}://{host}"
+
+
+def saml_endpoints(request: Request) -> dict[str, str]:
+    """SP entity id + ACS/SLO/metadata URLs derived from the incoming request."""
+    base = external_base_url(request)
+    return {
+        "entity_id": f"{base}/auth/saml/metadata",
+        "acs": f"{base}/auth/saml/callback",
+        "slo": f"{base}/auth/saml/logout",
+        "metadata": f"{base}/auth/saml/metadata",
+    }
+
+
+def build_settings(cfg: dict, endpoints: dict[str, str]) -> dict:
+    """python3-saml settings dict built from a ``sso_config`` row."""
+    sign_requests = bool(cfg.get("sign_requests"))
+    sp: dict = {
+        "entityId": endpoints["entity_id"],
+        "assertionConsumerService": {
+            "url": endpoints["acs"],
+            "binding": BINDING_HTTP_POST,
+        },
+        "singleLogoutService": {
+            "url": endpoints["slo"],
+            "binding": BINDING_HTTP_REDIRECT,
+        },
+        "NameIDFormat": NAMEID_FORMAT_EMAIL,
+    }
+    if cfg.get("sp_private_key") and cfg.get("sp_certificate"):
+        sp["privateKey"] = cfg["sp_private_key"]
+        sp["x509cert"] = cfg["sp_certificate"]
+
+    idp: dict = {
+        "entityId": cfg.get("entity_id") or "",
+        "singleSignOnService": {
+            "url": cfg.get("sso_url") or "",
+            "binding": BINDING_HTTP_REDIRECT,
+        },
+        "x509cert": cfg.get("x509_certificate") or "",
+    }
+    if cfg.get("slo_url"):
+        idp["singleLogoutService"] = {"url": cfg["slo_url"], "binding": BINDING_HTTP_REDIRECT}
+
+    return {
+        "strict": True,
+        "debug": False,
+        "sp": sp,
+        "idp": idp,
+        "security": {
+            "authnRequestsSigned": sign_requests,
+            "logoutRequestSigned": sign_requests,
+            "logoutResponseSigned": False,
+            "wantMessagesSigned": False,
+            "wantAssertionsSigned": True,
+            "wantNameIdEncrypted": False,
+            "wantAssertionsEncrypted": False,
+            "wantXmlValidation": True,
+            "signatureAlgorithm": "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256",
+            "digestAlgorithm": "http://www.w3.org/2001/04/xmlenc#sha256",
+            "rejectDeprecatedAlgorithm": True,
+            # FlowDeck is self-hosted: LAN/homelab deploys commonly reach the
+            # SP through single-label hosts (http://flowdeck/, docker service
+            # names). python3-saml rejects those URLs unless this is on.
+            "allowSingleLabelDomains": True,
+        },
+    }
+
+
+def _request_data(request: Request, script_name: str, post_data: dict | None = None) -> dict:
+    """Flat request dict expected by ``OneLogin_Saml2_Auth``."""
+    https = "on" if external_base_url(request).startswith("https") else "off"
+    return {
+        "https": https,
+        "http_host": request.headers.get("host", "localhost:8080"),
+        "script_name": script_name,
+        "request_uri": request.url.path,
+        "query_string": str(request.url.query or ""),
+        "get_data": dict(request.query_params),
+        "post_data": post_data or {},
+    }
+
+
+def _auth(request: Request, cfg: dict, script_name: str, post_data: dict | None = None):
+    from onelogin.saml2.auth import OneLogin_Saml2_Auth
+
+    settings = build_settings(cfg, saml_endpoints(request))
+    try:
+        return OneLogin_Saml2_Auth(
+            _request_data(request, script_name, post_data=post_data), old_settings=settings
+        )
+    except Exception as err:  # malformed IdP/SP config (bad cert, missing URL…)
+        raise SAMLError(f"Invalid SAML configuration: {err}") from err
+
+
+def create_login(request: Request, cfg: dict, relay_state: str) -> tuple[str, str]:
+    """Build the AuthnRequest. Returns ``(redirect_url, authn_request_id)``."""
+    auth = _auth(request, cfg, "/auth/saml/login")
+    try:
+        url = auth.login(return_to=relay_state)
+    except Exception as err:
+        raise SAMLError(f"Could not build the SAML AuthnRequest: {err}") from err
+    request_id = auth.get_last_request_id() or ""
+    if not request_id:
+        raise SAMLError("AuthnRequest was built without an id")
+    return url, request_id
+
+
+def process_response(request: Request, cfg: dict, post_data: dict, request_id: str) -> SAMLIdentity:
+    """Validate the IdP's SAMLResponse and extract the identity.
+
+    ``request_id`` is the id of the AuthnRequest we issued (from the
+    single-use ``sso_requests`` row): the toolkit rejects any response whose
+    ``InResponseTo`` does not match it.
+    """
+    auth = _auth(request, cfg, "/auth/saml/callback", post_data=post_data)
+    try:
+        auth.process_response(request_id=request_id or None)
+    except Exception as err:
+        raise SAMLError(f"SAML response could not be processed: {err}") from err
+
+    errors = auth.get_errors()
+    if errors:
+        raise SAMLError(auth.get_last_error_reason() or f"SAML errors: {', '.join(errors)}")
+    if not auth.is_authenticated():
+        raise SAMLError("SAML response did not authenticate the user")
+
+    name_id = auth.get_nameid() or ""
+    if not name_id:
+        raise SAMLError("SAML assertion carries no NameID")
+    return SAMLIdentity(
+        name_id=name_id,
+        name_id_format=auth.get_nameid_format() or "",
+        session_index=auth.get_session_index() or "",
+        attributes=auth.get_attributes() or {},
+        friendly_attributes=auth.get_friendlyname_attributes() or {},
+    )
+
+
+def metadata_xml(request: Request, cfg: dict) -> str:
+    """SP metadata XML (for the IdP configuration screen)."""
+    from onelogin.saml2.settings import OneLogin_Saml2_Settings
+
+    settings = OneLogin_Saml2_Settings(
+        build_settings(cfg, saml_endpoints(request)), custom_base_path=None
+    )
+    try:
+        xml = settings.get_sp_metadata()
+    except Exception as err:
+        raise SAMLError(f"Could not build the SP metadata: {err}") from err
+    if isinstance(xml, bytes):
+        xml = xml.decode("utf-8")
+    return xml
+
+
+def build_logout_url(request: Request, cfg: dict, return_to: str, name_id: str, session_index: str) -> str:
+    """SP-initiated Single Logout (HTTP-Redirect LogoutRequest to the IdP)."""
+    auth = _auth(request, cfg, "/auth/saml/logout")
+    if not cfg.get("slo_url"):
+        raise SAMLError("The IdP has no Single Logout URL configured")
+    try:
+        return auth.logout(
+            return_to=return_to,
+            name_id=name_id or None,
+            session_index=session_index or None,
+        )
+    except Exception as err:
+        raise SAMLError(f"Could not build the SAML LogoutRequest: {err}") from err
+
+
+def process_slo_form(request: Request, cfg: dict, form: dict, query: dict) -> tuple[str | None, list[str]]:
+    """Process a LogoutRequest / LogoutResponse received from the IdP.
+
+    ``form`` holds the POSTed fields, ``query`` the GET parameters (the
+    HTTP-Redirect binding delivers LogoutRequest/LogoutResponse there).
+    Returns ``(redirect_url, errors)``.
+    """
+    from onelogin.saml2.auth import OneLogin_Saml2_Auth
+
+    settings = build_settings(cfg, saml_endpoints(request))
+    https = "on" if external_base_url(request).startswith("https") else "off"
+    post_data = {k: v for k, v in form.items() if k in ("SAMLRequest", "SAMLResponse", "RelayState")}
+    if not post_data:
+        post_data = {"SAMLResponse": query["SAMLResponse"]} if "SAMLResponse" in query else {}
+    req_data = {
+        "https": https,
+        "http_host": request.headers.get("host", "localhost:8080"),
+        "script_name": "/auth/saml/logout",
+        "request_uri": request.url.path,
+        "query_string": str(request.url.query or ""),
+        "get_data": dict(query),
+        "post_data": post_data,
+    }
+    auth = OneLogin_Saml2_Auth(req_data, old_settings=settings)
+    try:
+        url = auth.process_slo(keep_local_session=True)
+    except Exception as err:
+        raise SAMLError(f"SAML logout could not be processed: {err}") from err
+    return url, auth.get_errors()