Complete identity smoke path: id_token claims, registration entry, cutover docs

Prefer verified KeyCape id_token claims when /userinfo returns 401; soft-fail
userinfo. Add CSOC-WP-0003 registration entry (disabled until NetKingdom URL),
AAL step-up hooks, smoke/cutover evidence for tegwick OIDC without MFA.
This commit is contained in:
tegwick 2026-08-09 22:42:51 +02:00
parent 3bc16b581b
commit 29a9ff735e
14 changed files with 513 additions and 41 deletions

View file

@ -2,14 +2,17 @@
from __future__ import annotations
import logging
import secrets
from typing import Any
from urllib.parse import urlencode
import httpx
from authlib.integrations.httpx_client import OAuth2Client
from authlib.jose import JsonWebKey, jwt
from django.conf import settings
logger = logging.getLogger(__name__)
class OIDCConfigurationError(RuntimeError):
pass
@ -52,17 +55,18 @@ def _oauth_client() -> OAuth2Client:
return OAuth2Client(**kwargs)
def build_authorization_url(*, state: str, code_verifier: str) -> str:
def build_authorization_url(
*, state: str, code_verifier: str, acr_values: str | None = None
) -> str:
if not oidc_configured():
raise OIDCConfigurationError("OIDC is not enabled/configured")
doc = discovery_document()
auth_endpoint = doc["authorization_endpoint"]
client = _oauth_client()
uri, _ = client.create_authorization_url(
auth_endpoint,
state=state,
code_verifier=code_verifier,
)
parameters = {"state": state, "code_verifier": code_verifier}
if acr_values:
parameters["acr_values"] = acr_values
uri, _ = client.create_authorization_url(auth_endpoint, **parameters)
return uri
@ -80,17 +84,54 @@ def exchange_code(code: str, *, code_verifier: str) -> dict[str, Any]:
def fetch_userinfo(access_token: str) -> dict[str, Any]:
"""Best-effort userinfo. KeyCape may 401 for some subjects; id_token is enough."""
if not access_token:
return {}
doc = discovery_document()
userinfo_endpoint = doc.get("userinfo_endpoint")
if not userinfo_endpoint:
return {}
resp = httpx.get(
userinfo_endpoint,
headers={"Authorization": f"Bearer {access_token}"},
timeout=15.0,
try:
resp = httpx.get(
userinfo_endpoint,
headers={"Authorization": f"Bearer {access_token}"},
timeout=15.0,
)
if resp.status_code >= 400:
logger.warning(
"OIDC userinfo returned %s; continuing with id_token claims",
resp.status_code,
)
return {}
return resp.json()
except Exception:
logger.exception("OIDC userinfo request failed; continuing with id_token claims")
return {}
def decode_id_token(id_token: str) -> dict[str, Any]:
"""Verify id_token with issuer JWKS and return claims."""
if not id_token or id_token.count(".") != 2:
raise OIDCConfigurationError("token response missing a usable id_token")
doc = discovery_document()
jwks_uri = doc.get("jwks_uri")
if not jwks_uri:
raise OIDCConfigurationError("issuer discovery missing jwks_uri")
jwks = httpx.get(jwks_uri, timeout=15.0)
jwks.raise_for_status()
key_set = JsonWebKey.import_key_set(jwks.json())
claims = jwt.decode(
id_token,
key_set,
claims_options={
"iss": {"essential": True, "value": settings.OIDC_ISSUER.rstrip("/")},
"aud": {"essential": True, "value": settings.OIDC_CLIENT_ID},
"exp": {"essential": True},
"sub": {"essential": True},
},
)
resp.raise_for_status()
return resp.json()
claims.validate()
return dict(claims)
def new_pkce_pair() -> tuple[str, str]:
@ -101,10 +142,16 @@ def new_pkce_pair() -> tuple[str, str]:
def claims_from_token_response(token: dict[str, Any], userinfo: dict[str, Any]) -> dict[str, Any]:
"""Merge id_token claims (if present as dict) with userinfo."""
"""Prefer verified id_token claims; overlay optional userinfo."""
claims: dict[str, Any] = {}
# authlib may leave id_token as JWT string; userinfo is preferred when available
claims.update(userinfo or {})
if not claims.get("sub") and isinstance(token.get("userinfo"), dict):
id_token = token.get("id_token")
if isinstance(id_token, str) and id_token:
claims.update(decode_id_token(id_token))
elif isinstance(token.get("userinfo"), dict):
claims.update(token["userinfo"])
# userinfo is optional enrichment (KeyCape may 401 for some subjects)
if userinfo:
claims.update(userinfo)
if not claims.get("sub"):
raise OIDCConfigurationError("OIDC response has no subject claim")
return claims