vergabe-teilnahme/vergabe_teilnahme/apps/accounts/oidc.py

"""Bounded NetKingdom OIDC relying party for one isolated company instance."""

import base64
import hashlib
import json
import secrets
import time
from functools import lru_cache
from urllib.parse import urlencode, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

import jwt
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured


class LoginRejectedError(ValueError):
    pass


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def configuration():
    issuer = settings.NETKINGDOM_ISSUER
    callback = settings.NETKINGDOM_CALLBACK
    for value in (issuer, callback):
        parts = urlsplit(value)
        if (
            parts.scheme != "https"
            or not parts.hostname
            or parts.username
            or parts.password
            or parts.query
            or parts.fragment
            or parts.hostname in {"localhost", "127.0.0.1"}
        ):
            raise ImproperlyConfigured("NetKingdom requires fixed HTTPS issuer and callback URLs")
    if not settings.NETKINGDOM_CLIENT_ID or not settings.NETKINGDOM_TENANT:
        raise ImproperlyConfigured("NetKingdom client and company binding are required")
    expected = settings.APP_BASE_PATH + "/accounts/oidc/callback/"
    if urlsplit(callback).path != expected:
        raise ImproperlyConfigured("NetKingdom callback must match the fixed company path")
    if settings.SESSION_ENGINE == "django.contrib.sessions.backends.signed_cookies":
        raise ImproperlyConfigured("OIDC requires server-side sessions")
    return issuer, settings.NETKINGDOM_CLIENT_ID, callback


def read_json(url, data=None):
    request = Request(
        url,
        data=data,
        headers={
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
        },
    )
    with build_opener(NoRedirect).open(request, timeout=10) as response:
        body = response.read(262145)
    if len(body) > 262144:
        raise LoginRejectedError("Oversized provider response")
    result = json.loads(body)
    if not isinstance(result, dict):
        raise LoginRejectedError("Invalid provider response")
    return result


@lru_cache(maxsize=8)
def discovery(issuer, period):
    metadata = read_json(issuer.rstrip("/") + "/.well-known/openid-configuration")
    if metadata.get("issuer") != issuer:
        raise LoginRejectedError("Issuer mismatch")
    for field in ("authorization_endpoint", "token_endpoint", "jwks_uri"):
        parts = urlsplit(metadata.get(field, ""))
        if (
            parts.scheme != "https"
            or parts.netloc != urlsplit(issuer).netloc
            or parts.username
            or parts.password
            or parts.fragment
            or parts.query
        ):
            raise LoginRejectedError("Unapproved provider endpoint")
    if "S256" not in metadata.get(
        "code_challenge_methods_supported", []
    ) or "RS256" not in metadata.get("id_token_signing_alg_values_supported", []):
        raise LoginRejectedError("Provider does not support the admitted flow")
    return metadata


@lru_cache(maxsize=8)
def key_client(uri):
    return jwt.PyJWKClient(uri, timeout=10, lifespan=300)


def begin():
    issuer, client, callback = configuration()
    metadata = discovery(issuer, int(time.time() // 300))
    pending = {
        "state": secrets.token_urlsafe(32),
        "nonce": secrets.token_urlsafe(32),
        "verifier": secrets.token_urlsafe(64),
        "created": time.time(),
    }
    challenge = (
        base64.urlsafe_b64encode(
            hashlib.sha256(pending["verifier"].encode("ascii")).digest(),
        )
        .decode("ascii")
        .rstrip("=")
    )
    parameters = {
        "response_type": "code",
        "client_id": client,
        "redirect_uri": callback,
        "scope": "openid profile groups",
        "state": pending["state"],
        "nonce": pending["nonce"],
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "prompt": "login",
        "tenant_hint": settings.NETKINGDOM_TENANT,
    }
    return pending, metadata["authorization_endpoint"] + "?" + urlencode(parameters)


def verify_claims(claims):
    """Identity evidence plus the invited pilot's explicit company admission rule."""
    tenant = settings.NETKINGDOM_TENANT
    groups, roles = claims.get("groups"), claims.get("roles")
    if (
        claims.get("tenant") != tenant
        or claims.get("tenant_source") != "directory"
        or claims.get("principal_type") != "human"
        or not isinstance(groups, list)
        or not all(isinstance(g, str) for g in groups)
        or not isinstance(roles, list)
        or not all(isinstance(r, str) for r in roles)
        or tenant + ":users" not in groups
        or {"netkingdom-suspended", "net-kingdom-admins"} & set(groups)
        or {"platform-operator", "platform-root", "emergency"} & set(roles)
    ):
        raise LoginRejectedError("Company membership is not established")
    assurance = claims.get("assurance")
    if not isinstance(assurance, dict) or assurance.get("level") not in {"aal1", "aal2", "aal3"}:
        raise LoginRejectedError("Missing authentication assurance")
    if not isinstance(claims.get("sub"), str) or not 0 < len(claims["sub"]) <= 512:
        raise LoginRejectedError("Invalid subject")


def complete(pending, state, code):
    if (
        not isinstance(pending, dict)
        or not state
        or not code
        or len(code) > 8192
        or not secrets.compare_digest(pending.get("state", ""), state)
        or not 0 <= time.time() - pending.get("created", 0) <= 600
    ):
        raise LoginRejectedError("Invalid or expired sign-in")
    issuer, client, callback = configuration()
    metadata = discovery(issuer, int(time.time() // 300))
    tokens = read_json(
        metadata["token_endpoint"],
        urlencode(
            {
                "grant_type": "authorization_code",
                "client_id": client,
                "redirect_uri": callback,
                "code": code,
                "code_verifier": pending["verifier"],
            }
        ).encode("ascii"),
    )
    token = tokens.get("id_token")
    if not isinstance(token, str) or len(token) > 32768:
        raise LoginRejectedError("Missing ID token")
    key = key_client(metadata["jwks_uri"]).get_signing_key_from_jwt(token)
    claims = jwt.decode(
        token,
        key.key,
        algorithms=["RS256"],
        issuer=issuer,
        audience=client,
        options={"require": ["iss", "sub", "aud", "exp", "iat", "nonce"]},
    )
    audience = claims["aud"]
    if (
        (isinstance(audience, list) and len(audience) > 1 and claims.get("azp") != client)
        or ("azp" in claims and claims["azp"] != client)
        or not isinstance(claims["nonce"], str)
        or not secrets.compare_digest(claims["nonce"], pending["nonce"])
    ):
        raise LoginRejectedError("Token binding mismatch")
    if any(type(claims[field]) is not int for field in ("iat", "exp")):
        raise LoginRejectedError("Invalid token timestamps")
    if claims["iat"] < pending["created"] - 60 or claims["exp"] <= claims["iat"]:
        raise LoginRejectedError("Token predates this sign-in")
    verify_claims(claims)
    # Store neither bearer tokens nor a provider password in the product session.
    return {
        "issuer": issuer,
        "tenant": settings.NETKINGDOM_TENANT,
        "client": client,
        "subject": claims["sub"],
        "label": str(claims.get("preferred_username") or "Ihr Benutzerkonto")[:150],
        "expires": min(int(claims["exp"]), int(time.time()) + 300),
    }