diff --git a/.forgejo/workflows/identity-journeys.yaml b/.forgejo/workflows/identity-journeys.yaml index 0e7a4eb..0991fff 100644 --- a/.forgejo/workflows/identity-journeys.yaml +++ b/.forgejo/workflows/identity-journeys.yaml @@ -2,7 +2,7 @@ name: Identity provider journey acceptance on: push: branches: [main] - paths: ["identity-provisioner/**", ".forgejo/workflows/identity-journeys.yaml"] + paths: ["identity-provisioner/**", "sso-mfa/k8s/synthetic-journey/**", ".forgejo/workflows/identity-journeys.yaml"] workflow_dispatch: jobs: provider: @@ -18,3 +18,8 @@ jobs: tar xzf identity-source.tar.gz -C identity-source --strip-components=1 cd identity-source PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=identity-provisioner python3 -m unittest discover -s identity-provisioner/tests -v + - name: Run synthetic MFA journey runner tests (NK-WP-0056-T05) + run: | + set -eu + cd identity-source/sso-mfa/k8s/synthetic-journey + PYTHONDONTWRITEBYTECODE=1 python3 -m unittest test_journey -v diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index ae6e3f7..baeb927 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -339,9 +339,12 @@ | task | NK-WP-0056-T02 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | | task | NK-WP-0056-T03 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | | task | NK-WP-0056-T04 | done | — | workplans/NK-WP-0056-bounded-executor-path.md | -| task | NK-WP-0056-T05 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | +| task | NK-WP-0056-T05 | progress | — | workplans/NK-WP-0056-bounded-executor-path.md | | task | NK-WP-0056-T06 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | | task | NK-WP-0056-T07 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | +| task | NK-WP-0056-T08 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | +| task | NK-WP-0056-T09 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | +| task | NK-WP-0056-T10 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md | | intake | NK-IN-0001 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md | | intake | NK-IN-0002 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md | | intake | NET-IN-0001 | open | — | intakes/intakes.md | diff --git a/sso-mfa/k8s/synthetic-journey/README.md b/sso-mfa/k8s/synthetic-journey/README.md new file mode 100644 index 0000000..eaba26a --- /dev/null +++ b/sso-mfa/k8s/synthetic-journey/README.md @@ -0,0 +1,84 @@ +# Synthetic MFA journey (NK-WP-0056-T05) + +`journey.py` signs a synthetic user in through KeyCape, Authelia and privacyIDEA over plain HTTP, +with no browser and no person, and writes a value-free receipt. It is the check a bounded executor +runs between consumers during a bind rotation (NK-WP-0056-T06): a failing journey means "revert". + +Python 3.11+ stdlib only, so it runs in the bare CI image and in a minimal executor pod. + +## What it proves + +| Step | Request | Bind on the path | +| --- | --- | --- | +| `discovery` | KeyCape discovery; issuer and endpoints on the issuer origin | | +| `keycape_authorize` | `/authorize`, PKCE S256, `acr_values=urn:netkingdom:aal2`, `prompt=login`; must redirect to Authelia `/api/oidc/authorization` | | +| `authelia_first_factor` | Authelia `/api/firstfactor` with the user's password | `svc-authelia` | +| `authelia_authorization` | the upstream request with the Authelia session; must return a code to KeyCape's callback | | +| `keycape_callback` | KeyCape callback; must render the OTP challenge with our state | `svc-keycape` | +| `keycape_otp` | TOTP code posted to KeyCape, which checks it with privacyIDEA; must redirect to the journey client with a code and our state | `svc-privacyidea` | +| `keycape_token` | code exchange with the PKCE verifier | | +| `id_token` | RS256 signature against KeyCape's JWKS (≥ 2048-bit), `iss`, `aud`, `nonce`, `exp`, `preferred_username`, `assurance` = aal2 with `otp` | | + +Requesting aal2 makes the OTP challenge mandatory whatever the user's enrollment state. A sign-in +without a challenge is a failure (`mfa_not_challenged`). `svc-provisioner` is not on this path. + +The run stops at the first failed step and exits 1. The receipt names the step, a fixed reason +(`first_factor_rejected`, `otp_rejected`, `signature_invalid`, …), the HTTP status and the bind on +that step. It never holds the password, the seed, a code, a cookie or a token, and never echoes a +response body. + +## Running + +```bash +python3 journey.py --config journey.json \ + --password-file /dev/fd/3 --otp-seed-file /dev/fd/4 [--receipt receipt.json] 3<… 4<… +``` + +Secrets come only from files or descriptors. The config is non-secret and refuses `password`, +`seed`, `otp_seed` and `client_secret` keys: + +```json +{ + "issuer": "https://kc.coulomb.social", + "client_id": "nk-synthetic-journey", + "redirect_uri": "https://journey.invalid/callback", + "username": "syn-journey", + "otp": {"digits": 6, "period": 30, "algorithm": "sha1", "min_remaining_seconds": 5}, + "host_map": { + "https://kc.coulomb.social": "http://keycape.sso.svc.cluster.local:8080", + "https://auth.coulomb.social": "http://authelia.sso.svc.cluster.local:9091" + }, + "timeout_seconds": 10, + "deadline_seconds": 120 +} +``` + +`redirect_uri` is never fetched: the runner reads the code from the `Location` header. `host_map` +lets an in-cluster executor reach the services directly while sending the public `Host` and +`X-Forwarded-Proto`/`-Host`, so they see the URLs they issued. Cookies are kept per public origin. +`--client-secret-file` exists only for a confidential client. The journey client is meant to be +public with PKCE. + +Exit codes: `0` pass, `1` journey failed (fail closed), `2` usage or config error. + +## Tests + +```bash +cd sso-mfa/k8s/synthetic-journey && python3 -m unittest test_journey -v +``` + +In-process fakes of KeyCape and Authelia, with privacyIDEA's check behind KeyCape, each with a bind +switch. The tests cover the healthy pass, each broken bind failing at its step, wrong password and +seed, no OTP challenge, a tampered signature, an aal1 token, a state mismatch, the OTP window wait, +the deadline, an unreachable service, CLI exit codes, public-to-internal host mapping, the RFC +4226/6238 vectors, and that no secret value reaches the receipt. They run in the identity-provider +journey acceptance workflow (`.forgejo/workflows/identity-journeys.yaml`). + +## Not yet proven + +- Against the real services: the synthetic stack on the pinned images is NK-WP-0056-T08. Authelia's + first-factor API is exercised the same way by `authelia/tests/probe_claims.py` and + `lldap/bind-split-rehearsal.py`; KeyCape's flow follows `src/internal/server/oidc/authorize.go` + as of key-cape `e2b5fa1`. +- Live: the synthetic user, its token, the journey client and the OpenBao-held secrets (T09), and an + unattended run by the executor identity (T10). diff --git a/sso-mfa/k8s/synthetic-journey/journey.py b/sso-mfa/k8s/synthetic-journey/journey.py new file mode 100644 index 0000000..3e7ddf0 --- /dev/null +++ b/sso-mfa/k8s/synthetic-journey/journey.py @@ -0,0 +1,573 @@ +#!/usr/bin/env python3 +"""Synthetic MFA sign-in journey through KeyCape, Authelia and privacyIDEA (NK-WP-0056-T05). + +Drives the browser journey over plain HTTP, without a browser and without a person: + + 1. discovery KeyCape /.well-known/openid-configuration (issuer must match the config) + 2. keycape_authorize KeyCape /authorize with PKCE S256 and acr_values=urn:netkingdom:aal2; + must redirect to Authelia's OIDC authorization endpoint + 3. authelia_first_factor Authelia /api/firstfactor (Authelia binds LDAP as svc-authelia) + 4. authelia_authorization the upstream authorization request; must return a code to KeyCape + 5. keycape_callback KeyCape /authorize/callback; must render the OTP challenge + (KeyCape looks the user up, bound as svc-keycape) + 6. keycape_otp POST the TOTP code; KeyCape validates it through privacyIDEA + (privacyIDEA resolves the user, bound as svc-privacyidea); + must redirect to the journey client with a code and the original state + 7. keycape_token code exchange with the PKCE verifier + 8. id_token RS256 signature against KeyCape's JWKS, iss/aud/nonce/exp, + preferred_username and assurance (aal2, mfa, otp) + +The run stops at the first failed step and exits 1 (fail closed). The receipt is value-free: it never +contains the password, the seed, a code, a cookie or a token, and names failures with a fixed reason +vocabulary instead of echoing response bodies. + +Secrets are read from files (or /dev/fd/N), never from argv or the environment. Everything else comes +from a JSON config, see README.md. Exit codes: 0 pass, 1 journey failed, 2 usage or config error. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +import html +import http.client +import json +import re +import secrets +import struct +import sys +import time +import uuid +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import parse_qs, urlencode, urljoin, urlsplit + +SCHEMA = "netkingdom.synthetic-journey.receipt/v1" +AAL2_ACR = "urn:netkingdom:aal2" +# Which service bind each step depends on. A step can fail for other reasons too; the mapping says +# which bind is on the path, so a broken bind must surface no later than this step. +BIND_ON_PATH = { + "authelia_first_factor": "svc-authelia", + "keycape_callback": "svc-keycape", + "keycape_otp": "svc-privacyidea", +} +SHA256_DIGEST_INFO = bytes.fromhex("3031300d060960864801650304020105000420") +USERNAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") +DEFAULTS = { + "scope": "openid profile", + "acr_values": AAL2_ACR, + "timeout_seconds": 10, + "deadline_seconds": 120, + "otp": {"digits": 6, "period": 30, "algorithm": "sha1", "min_remaining_seconds": 5}, + "host_map": {}, +} + + +class ConfigError(Exception): + pass + + +class JourneyFailure(Exception): + def __init__(self, step: str, reason: str, status: int | None = None): + super().__init__(f"{step}: {reason}") + self.step = step + self.reason = reason + self.status = status + + +# -------------------------------------------------------------------------------------------------- +# Configuration and secrets +# -------------------------------------------------------------------------------------------------- + + +def origin(url: str) -> str: + parts = urlsplit(url) + if parts.scheme not in ("http", "https") or not parts.netloc: + raise ConfigError(f"not an http(s) URL: {url!r}") + return f"{parts.scheme}://{parts.netloc}" + + +def load_config(path: str) -> dict: + try: + raw = json.loads(Path(path).read_text()) + except (OSError, ValueError) as exc: + raise ConfigError(f"config unreadable: {exc.__class__.__name__}") from None + if not isinstance(raw, dict): + raise ConfigError("config must be a JSON object") + cfg = {**DEFAULTS, **raw, "otp": {**DEFAULTS["otp"], **raw.get("otp", {})}} + for key in ("issuer", "client_id", "redirect_uri", "username"): + if not isinstance(cfg.get(key), str) or not cfg[key]: + raise ConfigError(f"config needs {key}") + for forbidden in ("password", "otp_seed", "seed", "client_secret"): + if forbidden in raw or forbidden in raw.get("otp", {}): + raise ConfigError(f"config must not carry {forbidden}; use the *-file options") + if not USERNAME.fullmatch(cfg["username"]): + raise ConfigError("username has unexpected characters") + origin(cfg["issuer"]) + origin(cfg["redirect_uri"]) + if AAL2_ACR not in cfg["acr_values"].split(): + raise ConfigError("acr_values must request urn:netkingdom:aal2: the journey proves MFA") + otp = cfg["otp"] + if otp["algorithm"] not in ("sha1", "sha256", "sha512") or otp["digits"] not in (6, 7, 8): + raise ConfigError("unsupported otp algorithm or digits") + if not 0 <= otp["min_remaining_seconds"] < otp["period"]: + raise ConfigError("otp.min_remaining_seconds must be below otp.period") + if not isinstance(cfg["host_map"], dict): + raise ConfigError("host_map must be an object") + cfg["host_map"] = {origin(k): origin(v) for k, v in cfg["host_map"].items()} + return cfg + + +def read_secret(path: str, what: str) -> str: + try: + value = Path(path).read_text().strip() + except OSError: + raise ConfigError(f"{what} file unreadable") from None + if not value or "\n" in value: + raise ConfigError(f"{what} file must hold exactly one non-empty line") + return value + + +def decode_seed(seed: str) -> bytes: + clean = re.sub(r"[\s-]", "", seed).upper() + try: + key = base64.b32decode(clean + "=" * (-len(clean) % 8)) + except ValueError: + raise ConfigError("otp seed is not base32") from None + if len(key) < 10: + raise ConfigError("otp seed is shorter than 80 bits") + return key + + +# -------------------------------------------------------------------------------------------------- +# Primitives: TOTP, PKCE, RS256 +# -------------------------------------------------------------------------------------------------- + + +def hotp(key: bytes, counter: int, digits: int = 6, algorithm: str = "sha1") -> str: + mac = hmac.new(key, struct.pack(">Q", counter), algorithm).digest() + offset = mac[-1] & 0x0F + value = struct.unpack(">I", mac[offset : offset + 4])[0] & 0x7FFFFFFF + return str(value % 10**digits).zfill(digits) + + +def b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def b64url_decode(value: str) -> bytes: + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + +def pkce_pair() -> tuple[str, str]: + verifier = secrets.token_urlsafe(48) + return verifier, b64url(hashlib.sha256(verifier.encode()).digest()) + + +def rs256_valid(signing_input: bytes, signature: bytes, jwk: dict) -> bool: + """RSASSA-PKCS1-v1_5 with SHA-256, stdlib only. Requires a modulus of at least 2048 bits.""" + try: + n = int.from_bytes(b64url_decode(jwk["n"]), "big") + e = int.from_bytes(b64url_decode(jwk["e"]), "big") + except (KeyError, TypeError, ValueError): + return False + k = (n.bit_length() + 7) // 8 + if n.bit_length() < 2048 or len(signature) != k: + return False + s = int.from_bytes(signature, "big") + if s >= n: + return False + em = pow(s, e, n).to_bytes(k, "big") + tail = SHA256_DIGEST_INFO + hashlib.sha256(signing_input).digest() + expected = b"\x00\x01" + b"\xff" * (k - 3 - len(tail)) + b"\x00" + tail + return hmac.compare_digest(em, expected) + + +# -------------------------------------------------------------------------------------------------- +# HTTP without redirect following, with an explicit cookie store per public origin +# -------------------------------------------------------------------------------------------------- + + +class Response: + def __init__(self, status: int, headers: list[tuple[str, str]], body: bytes): + self.status = status + self.headers = headers + self.body = body + + def header(self, name: str) -> str: + for key, value in self.headers: + if key.lower() == name.lower(): + return value + return "" + + def json(self): + return json.loads(self.body) + + +class Client: + """Talks to public URLs, optionally connecting to a mapped internal origin. + + When a public origin is mapped, the request goes to the internal origin with the public Host and + X-Forwarded-Proto/-Host, so services behind an ingress see the URL they issued. Cookies are kept + per public origin and only sent back to that origin. + """ + + def __init__(self, host_map: dict[str, str], timeout: float, deadline: float, clock=time.monotonic): + self.host_map = host_map + self.timeout = timeout + self.deadline = deadline + self.clock = clock + self.cookies: dict[str, dict[str, str]] = {} + + def request(self, step: str, method: str, url: str, *, form=None, json_body=None, headers=None): + if self.clock() > self.deadline: + raise JourneyFailure(step, "deadline_exceeded") + public = urlsplit(url) + pub_origin = f"{public.scheme}://{public.netloc}" + target = urlsplit(self.host_map.get(pub_origin, pub_origin)) + path = public.path or "/" + if public.query: + path += "?" + public.query + head = {"Host": public.netloc, "Accept": "application/json, text/html"} + if pub_origin in self.host_map: + head["X-Forwarded-Proto"] = public.scheme + head["X-Forwarded-Host"] = public.netloc + jar = self.cookies.get(pub_origin) + if jar: + head["Cookie"] = "; ".join(f"{k}={v}" for k, v in jar.items()) + body = None + if form is not None: + body = urlencode(form).encode() + head["Content-Type"] = "application/x-www-form-urlencoded" + elif json_body is not None: + body = json.dumps(json_body).encode() + head["Content-Type"] = "application/json" + head.update(headers or {}) + conn_cls = http.client.HTTPSConnection if target.scheme == "https" else http.client.HTTPConnection + remaining = max(0.1, min(self.timeout, self.deadline - self.clock())) + conn = conn_cls(target.netloc, timeout=remaining) + try: + conn.request(method, path, body=body, headers=head) + raw = conn.getresponse() + response = Response(raw.status, raw.getheaders(), raw.read(1 << 20)) + except (OSError, http.client.HTTPException): + raise JourneyFailure(step, "unreachable") from None + finally: + conn.close() + for key, value in response.headers: + if key.lower() == "set-cookie": + self._store_cookie(pub_origin, value) + return response + + def _store_cookie(self, pub_origin: str, header: str) -> None: + first, *attrs = [part.strip() for part in header.split(";")] + name, sep, value = first.partition("=") + if not sep or not name: + return + expired = any(a.lower().replace(" ", "") in ("max-age=0",) for a in attrs) or value == "" + jar = self.cookies.setdefault(pub_origin, {}) + if expired: + jar.pop(name, None) + else: + jar[name] = value + + +# -------------------------------------------------------------------------------------------------- +# The journey +# -------------------------------------------------------------------------------------------------- + + +def location(step: str, response: Response, base: str) -> str: + if response.status not in (301, 302, 303, 307, 308): + raise JourneyFailure(step, "expected_redirect", response.status) + value = response.header("Location") + if not value: + raise JourneyFailure(step, "redirect_without_location", response.status) + return urljoin(base, value) + + +def same_endpoint(url: str, expected: str) -> bool: + a, b = urlsplit(url), urlsplit(expected) + return (a.scheme, a.netloc, a.path) == (b.scheme, b.netloc, b.path) + + +def single(query: dict, name: str) -> str | None: + values = query.get(name) + return values[0] if values and len(values) == 1 else None + + +class Journey: + def __init__(self, cfg: dict, password: str, seed: bytes, client_secret: str | None = None, + clock=time.monotonic, wall=time.time, sleep=time.sleep): + self.cfg = cfg + self.password = password + self.seed = seed + self.client_secret = client_secret + self.wall = wall + self.sleep = sleep + self.clock = clock + self.client = Client(cfg["host_map"], cfg["timeout_seconds"], clock() + cfg["deadline_seconds"], clock) + self.steps: list[dict] = [] + self.claims_summary: dict = {} + + def step(self, name: str, fn): + started = self.clock() + try: + result = fn() + except JourneyFailure as failure: + self.steps.append({"name": name, "ok": False, "reason": failure.reason, "status": failure.status, + "elapsed_ms": int((self.clock() - started) * 1000)}) + raise + self.steps.append({"name": name, "ok": True, "elapsed_ms": int((self.clock() - started) * 1000)}) + return result + + def run(self) -> None: + cfg = self.cfg + issuer = cfg["issuer"].rstrip("/") + meta = self.step("discovery", lambda: self.discovery(issuer)) + verifier, challenge = pkce_pair() + state, nonce = secrets.token_urlsafe(24), secrets.token_urlsafe(24) + upstream = self.step("keycape_authorize", lambda: self.authorize(meta, challenge, state, nonce)) + self.step("authelia_first_factor", lambda: self.first_factor(upstream)) + callback = self.step("authelia_authorization", lambda: self.upstream_authorization(upstream)) + self.step("keycape_callback", lambda: self.keycape_callback(callback, state)) + code = self.step("keycape_otp", lambda: self.submit_otp(issuer, state)) + id_token = self.step("keycape_token", lambda: self.token(meta, code, verifier)) + self.step("id_token", lambda: self.verify_id_token(meta, id_token, nonce)) + + # 1 + def discovery(self, issuer: str) -> dict: + r = self.client.request("discovery", "GET", issuer + "/.well-known/openid-configuration") + if r.status != 200: + raise JourneyFailure("discovery", "unexpected_status", r.status) + try: + meta = r.json() + except ValueError: + raise JourneyFailure("discovery", "not_json", r.status) from None + if meta.get("issuer", "").rstrip("/") != issuer: + raise JourneyFailure("discovery", "issuer_mismatch") + for key in ("authorization_endpoint", "token_endpoint", "jwks_uri"): + if not isinstance(meta.get(key), str) or origin(meta[key]) != origin(issuer): + raise JourneyFailure("discovery", f"{key}_off_issuer") + return meta + + # 2 + def authorize(self, meta: dict, challenge: str, state: str, nonce: str) -> str: + cfg = self.cfg + query = urlencode({ + "response_type": "code", "client_id": cfg["client_id"], "redirect_uri": cfg["redirect_uri"], + "scope": cfg["scope"], "state": state, "nonce": nonce, "code_challenge": challenge, + "code_challenge_method": "S256", "acr_values": cfg["acr_values"], "prompt": "login", + }) + url = meta["authorization_endpoint"] + "?" + query + r = self.client.request("keycape_authorize", "GET", url) + upstream = location("keycape_authorize", r, url) + if origin(upstream) == origin(cfg["issuer"]): + raise JourneyFailure("keycape_authorize", "did_not_leave_keycape") + if not urlsplit(upstream).path.endswith("/api/oidc/authorization"): + raise JourneyFailure("keycape_authorize", "upstream_not_authelia_authorization") + return upstream + + # 3 + def first_factor(self, upstream: str) -> None: + url = origin(upstream) + "/api/firstfactor" + r = self.client.request("authelia_first_factor", "POST", url, json_body={ + "username": self.cfg["username"], "password": self.password, "keepMeLoggedIn": False, + }) + if r.status != 200: + raise JourneyFailure("authelia_first_factor", "first_factor_rejected", r.status) + try: + ok = r.json().get("status") == "OK" + except (ValueError, AttributeError): + ok = False + if not ok: + raise JourneyFailure("authelia_first_factor", "first_factor_not_ok", r.status) + + # 4 + def upstream_authorization(self, upstream: str) -> str: + r = self.client.request("authelia_authorization", "GET", upstream) + callback = location("authelia_authorization", r, upstream) + expected = self.cfg["issuer"].rstrip("/") + "/authorize/callback" + if not same_endpoint(callback, expected): + raise JourneyFailure("authelia_authorization", "upstream_did_not_issue_code") + query = parse_qs(urlsplit(callback).query) + if "error" in query or not single(query, "code"): + raise JourneyFailure("authelia_authorization", "upstream_did_not_issue_code") + return callback + + # 5 + def keycape_callback(self, callback: str, state: str) -> None: + r = self.client.request("keycape_callback", "GET", callback) + if r.status in (301, 302, 303, 307, 308): + target = urljoin(callback, r.header("Location")) + if same_endpoint(target, self.cfg["redirect_uri"]): + # Signed in without an OTP challenge despite aal2: never acceptable here. + raise JourneyFailure("keycape_callback", "mfa_not_challenged", r.status) + raise JourneyFailure("keycape_callback", "unexpected_redirect", r.status) + if r.status != 200: + raise JourneyFailure("keycape_callback", "callback_rejected", r.status) + page = r.body.decode("utf-8", "replace") + form = re.search(r']*action="/authorize/callback"[^>]*>(.*?)', page, re.S | re.I) + if not form or not re.search(r'name="mfa_token"', form.group(1)): + raise JourneyFailure("keycape_callback", "mfa_challenge_missing", r.status) + hidden = re.search(r'name="state"\s+value="([^"]*)"', form.group(1)) + if not hidden or html.unescape(hidden.group(1)) != state: + raise JourneyFailure("keycape_callback", "challenge_state_mismatch", r.status) + + # 6 + def submit_otp(self, issuer: str, state: str) -> str: + otp = self.cfg["otp"] + period = otp["period"] + now = self.wall() + remaining = period - (now % period) + if remaining < otp["min_remaining_seconds"]: + self.sleep(remaining + 0.5) + now = self.wall() + code = hotp(self.seed, int(now // period), otp["digits"], otp["algorithm"]) + url = issuer + "/authorize/callback" + r = self.client.request("keycape_otp", "POST", url, form={"state": state, "mfa_token": code}) + if r.status == 401: + raise JourneyFailure("keycape_otp", "otp_rejected", r.status) + target = location("keycape_otp", r, url) + if not same_endpoint(target, self.cfg["redirect_uri"]): + raise JourneyFailure("keycape_otp", "unexpected_redirect", r.status) + query = parse_qs(urlsplit(target).query) + if "error" in query: + raise JourneyFailure("keycape_otp", "authorization_error", r.status) + if single(query, "state") != state: + raise JourneyFailure("keycape_otp", "state_mismatch", r.status) + code_value = single(query, "code") + if not code_value: + raise JourneyFailure("keycape_otp", "no_authorization_code", r.status) + return code_value + + # 7 + def token(self, meta: dict, code: str, verifier: str) -> str: + form = {"grant_type": "authorization_code", "code": code, "redirect_uri": self.cfg["redirect_uri"], + "code_verifier": verifier, "client_id": self.cfg["client_id"]} + headers = {} + if self.client_secret is not None: + basic = base64.b64encode(f"{self.cfg['client_id']}:{self.client_secret}".encode()).decode() + headers["Authorization"] = "Basic " + basic + r = self.client.request("keycape_token", "POST", meta["token_endpoint"], form=form, headers=headers) + if r.status != 200: + raise JourneyFailure("keycape_token", "token_rejected", r.status) + try: + id_token = r.json().get("id_token") + except (ValueError, AttributeError): + id_token = None + if not isinstance(id_token, str): + raise JourneyFailure("keycape_token", "no_id_token", r.status) + return id_token + + # 8 + def verify_id_token(self, meta: dict, id_token: str, nonce: str) -> None: + step = "id_token" + try: + head_b64, payload_b64, sig_b64 = id_token.split(".") + header = json.loads(b64url_decode(head_b64)) + claims = json.loads(b64url_decode(payload_b64)) + signature = b64url_decode(sig_b64) + except ValueError: + raise JourneyFailure(step, "malformed") from None + if header.get("alg") != "RS256": + raise JourneyFailure(step, "unexpected_alg") + r = self.client.request(step, "GET", meta["jwks_uri"]) + if r.status != 200: + raise JourneyFailure(step, "jwks_unavailable", r.status) + try: + keys = [k for k in r.json().get("keys", []) if k.get("kty") == "RSA"] + except (ValueError, AttributeError): + raise JourneyFailure(step, "jwks_unavailable", r.status) from None + if header.get("kid") is not None: + keys = [k for k in keys if k.get("kid") == header["kid"]] + signing_input = f"{head_b64}.{payload_b64}".encode() + if not any(rs256_valid(signing_input, signature, k) for k in keys): + raise JourneyFailure(step, "signature_invalid") + cfg = self.cfg + aud = claims.get("aud") + audiences = aud if isinstance(aud, list) else [aud] + assurance = claims.get("assurance") if isinstance(claims.get("assurance"), dict) else {} + methods = assurance.get("methods") if isinstance(assurance.get("methods"), list) else [] + checks = [ + (str(claims.get("iss", "")).rstrip("/") == cfg["issuer"].rstrip("/"), "issuer_mismatch"), + (cfg["client_id"] in audiences, "audience_mismatch"), + (claims.get("nonce") == nonce, "nonce_mismatch"), + (isinstance(claims.get("exp"), (int, float)) and claims["exp"] > self.wall(), "expired"), + (claims.get("preferred_username") == cfg["username"], "username_mismatch"), + (assurance.get("level") == "aal2" and assurance.get("mfa") is True, "not_aal2"), + ("otp" in methods, "otp_not_in_methods"), + ] + self.claims_summary = { + "signature_verified": True, + "kid": header.get("kid"), + "assurance_level": assurance.get("level"), + "assurance_methods": [m for m in methods if isinstance(m, str)], + "assurance_source": assurance.get("source"), + "preferred_username_matches": claims.get("preferred_username") == cfg["username"], + } + for ok, reason in checks: + if not ok: + raise JourneyFailure(step, reason) + + +# -------------------------------------------------------------------------------------------------- +# Receipt and CLI +# -------------------------------------------------------------------------------------------------- + + +def utc_now() -> str: + return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + + +def run_journey(cfg: dict, password: str, seed: bytes, client_secret: str | None = None, **kw) -> dict: + receipt = { + "schema": SCHEMA, + "run_id": str(uuid.uuid4()), + "started_at": utc_now(), + "runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "target": {"issuer": cfg["issuer"], "client_id": cfg["client_id"], "username": cfg["username"], + "acr_values": cfg["acr_values"], "host_mapped": sorted(cfg["host_map"])}, + "binds_on_path": BIND_ON_PATH, + } + journey = Journey(cfg, password, seed, client_secret, **kw) + try: + journey.run() + receipt.update(result="pass", failed_step=None, reason=None) + except JourneyFailure as failure: + receipt.update(result="fail", failed_step=failure.step, reason=failure.reason, + bind_on_failed_step=BIND_ON_PATH.get(failure.step)) + receipt["steps"] = journey.steps + receipt["claims"] = journey.claims_summary + receipt["finished_at"] = utc_now() + return receipt + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--config", required=True, help="non-secret JSON config") + parser.add_argument("--password-file", required=True, help="file or /dev/fd/N with the user's password") + parser.add_argument("--otp-seed-file", required=True, help="file or /dev/fd/N with the base32 TOTP seed") + parser.add_argument("--client-secret-file", help="only for a confidential journey client") + parser.add_argument("--receipt", help="write the receipt here as well as to stdout") + args = parser.parse_args(argv) + try: + cfg = load_config(args.config) + password = read_secret(args.password_file, "password") + seed = decode_seed(read_secret(args.otp_seed_file, "otp seed")) + client_secret = read_secret(args.client_secret_file, "client secret") if args.client_secret_file else None + except ConfigError as exc: + print(json.dumps({"schema": SCHEMA, "result": "error", "reason": str(exc)}), file=sys.stderr) + return 2 + receipt = run_journey(cfg, password, seed, client_secret) + text = json.dumps(receipt, indent=2, sort_keys=True) + if args.receipt: + Path(args.receipt).write_text(text + "\n") + print(text) + return 0 if receipt["result"] == "pass" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/sso-mfa/k8s/synthetic-journey/test_journey.py b/sso-mfa/k8s/synthetic-journey/test_journey.py new file mode 100644 index 0000000..cd38fdc --- /dev/null +++ b/sso-mfa/k8s/synthetic-journey/test_journey.py @@ -0,0 +1,469 @@ +"""Tests for the synthetic MFA journey runner (NK-WP-0056-T05). Stdlib only, synthetic values only. + +The fakes model the three consumer binds: a broken svc-authelia bind makes Authelia refuse every first +factor, a broken svc-keycape bind makes KeyCape's user lookup fail at the callback, and a broken +svc-privacyidea bind makes privacyIDEA's resolver (and so KeyCape's OTP check) fail. +""" + +from __future__ import annotations + +import base64 +import contextlib +import hashlib +import importlib.util +import io +import json +import random +import secrets +import tempfile +import threading +import time +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, urlencode, urlsplit + +HERE = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("journey", HERE / "journey.py") +journey = importlib.util.module_from_spec(spec) +spec.loader.exec_module(journey) + +PASSWORD = "synthetic-only-journey-password" +SEED_B32 = "JBSWY3DPEHPK3PXPJBSWY3DPEHPK3PXP" +SEED = base64.b32decode(SEED_B32) +USER = "syn-journey" +CLIENT = "nk-synthetic-journey" +REDIRECT = "http://127.0.0.1:9/journey/callback" # never served: the runner only reads the Location + + +# ------------------------------------------------------------------------------------------------ +# A throwaway RSA key, generated in pure Python so the test needs no third-party package +# ------------------------------------------------------------------------------------------------ + + +def _probable_prime(bits: int, rng: random.Random) -> int: + small = [p for p in range(3, 2000, 2) if all(p % q for q in range(3, int(p**0.5) + 1, 2))] + while True: + n = rng.getrandbits(bits) | (1 << (bits - 1)) | (1 << (bits - 2)) | 1 + if any(n % p == 0 for p in small): + continue + d, r = n - 1, 0 + while d % 2 == 0: + d //= 2 + r += 1 + for _ in range(24): + x = pow(rng.randrange(2, n - 1), d, n) + if x in (1, n - 1): + continue + for _ in range(r - 1): + x = pow(x, 2, n) + if x == n - 1: + break + else: + break + else: + return n + + +class RSAKey: + def __init__(self, bits=2048, seed=56): + rng = random.Random(seed) + e = 65537 + while True: + p, q = _probable_prime(bits // 2, rng), _probable_prime(bits // 2, rng) + phi = (p - 1) * (q - 1) + if p != q and phi % e and (p * q).bit_length() == bits: + break + self.n, self.e, self.d = p * q, e, pow(e, -1, phi) + + def jwk(self, kid="journey-test"): + k = (self.n.bit_length() + 7) // 8 + return {"kty": "RSA", "kid": kid, "alg": "RS256", "use": "sig", + "n": journey.b64url(self.n.to_bytes(k, "big")), "e": journey.b64url(self.e.to_bytes(3, "big"))} + + def sign(self, data: bytes) -> bytes: + k = (self.n.bit_length() + 7) // 8 + tail = journey.SHA256_DIGEST_INFO + hashlib.sha256(data).digest() + em = b"\x00\x01" + b"\xff" * (k - 3 - len(tail)) + b"\x00" + tail + return pow(int.from_bytes(em, "big"), self.d, self.n).to_bytes(k, "big") + + +KEY: RSAKey = None # generated once in setUpModule + + +def setUpModule(): + global KEY + KEY = RSAKey() + + +# ------------------------------------------------------------------------------------------------ +# Fakes +# ------------------------------------------------------------------------------------------------ + + +class World: + """Shared state of the fake stack, with switches for every failure the tests provoke.""" + + def __init__(self): + self.binds = {"svc-authelia": True, "svc-keycape": True, "svc-privacyidea": True} + self.kc_public = "" # public KeyCape origin, set once servers run + self.authelia_public = "" + self.pending = {} # KeyCape state -> request + self.upstream_codes = {} # Authelia code -> username + self.sessions = {} # Authelia session cookie -> username + self.kc_codes = {} # KeyCape code -> (pending, username) + self.skip_mfa = False + self.token_level = "aal2" + self.tamper_signature = False + self.return_wrong_state = False + self.seen_hosts = [] + self.issued = [] # every cookie, code and token value handed out + + +def base_url(handler) -> str: + proto = handler.headers.get("X-Forwarded-Proto", "http") + return f"{proto}://{handler.headers['Host']}" + + +class Base(BaseHTTPRequestHandler): + world: World + + def send(self, status, body=b"", headers=None, ctype="application/json"): + if isinstance(body, (dict, list)): + body = json.dumps(body).encode() + elif isinstance(body, str): + body = body.encode() + self.send_response(status) + for key, value in (headers or {}).items(): + self.send_header(key, value) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def body(self): + return self.rfile.read(int(self.headers.get("Content-Length", "0"))) + + def log_message(self, *args): + return + + +class FakeAuthelia(Base): + def do_POST(self): + self.world.seen_hosts.append(self.headers["Host"]) + if self.path != "/api/firstfactor": + return self.send(404, {"status": "KO"}) + payload = json.loads(self.body()) + if not self.world.binds["svc-authelia"]: + # Authelia cannot bind LDAP: every first factor fails. Reflect the password to prove + # the runner never copies a response body into the receipt. + return self.send(401, {"status": "KO", "message": "Authentication failed", "echo": payload.get("password")}) + if payload.get("username") != USER or payload.get("password") != PASSWORD: + return self.send(401, {"status": "KO", "message": "Authentication failed"}) + session = secrets.token_hex(16) + self.world.sessions[session] = USER + self.world.issued.append(session) + self.send(200, {"status": "OK"}, {"Set-Cookie": f"authelia_session={session}; Path=/; HttpOnly"}) + + def do_GET(self): + self.world.seen_hosts.append(self.headers["Host"]) + url = urlsplit(self.path) + if url.path != "/api/oidc/authorization": + return self.send(404) + q = {k: v[0] for k, v in parse_qs(url.query).items()} + cookie = self.headers.get("Cookie", "") + session = dict(p.strip().split("=", 1) for p in cookie.split(";") if "=" in p).get("authelia_session") + user = self.world.sessions.get(session) + if not user: + return self.send(302, headers={"Location": base_url(self) + "/?rd=" + "x"}) + code = secrets.token_urlsafe(16) + self.world.upstream_codes[code] = user + self.world.issued.append(code) + self.send(302, headers={"Location": q["redirect_uri"] + "?" + urlencode({"code": code, "state": q["state"]})}) + + +class FakeKeyCape(Base): + def do_GET(self): + self.world.seen_hosts.append(self.headers["Host"]) + url = urlsplit(self.path) + q = {k: v[0] for k, v in parse_qs(url.query).items()} + base = base_url(self) + if url.path == "/.well-known/openid-configuration": + return self.send(200, {"issuer": base, "authorization_endpoint": base + "/authorize", + "token_endpoint": base + "/token", "jwks_uri": base + "/jwks"}) + if url.path == "/jwks": + return self.send(200, {"keys": [KEY.jwk()]}) + if url.path == "/authorize": + if q.get("code_challenge_method") != "S256" or q.get("client_id") != CLIENT: + return self.send(400, "bad request", ctype="text/html") + self.world.pending[q["state"]] = {**q, "user": None} + upstream = self.world.authelia_public + "/api/oidc/authorization?" + urlencode({ + "client_id": "keycape", "redirect_uri": base + "/authorize/callback", + "response_type": "code", "state": q["state"]}) + return self.send(302, headers={"Location": upstream}) + if url.path == "/authorize/callback": + pending = self.world.pending.get(q.get("state")) + user = self.world.upstream_codes.pop(q.get("code"), None) + if not pending or not user: + return self.send(400, "unknown or expired state", ctype="text/html") + if not self.world.binds["svc-keycape"]: + return self.send(401, "authentication failed", ctype="text/html") + if self.world.skip_mfa: + return self.redirect_with_code(pending, user) + pending["user"] = user + form = ('
' + f'' + '' + '
') + return self.send(200, f"{form}", ctype="text/html") + self.send(404) + + def do_POST(self): + self.world.seen_hosts.append(self.headers["Host"]) + form = {k: v[0] for k, v in parse_qs(self.body().decode()).items()} + if self.path == "/authorize/callback": + pending = self.world.pending.pop(form.get("state"), None) + if not pending or not pending["user"]: + return self.send(400, "mfa challenge not active", ctype="text/html") + if not self.world.binds["svc-privacyidea"] or not self.privacyidea_check(form.get("mfa_token", "")): + return self.send(401, "MFA validation failed", ctype="text/html") + return self.redirect_with_code(pending, pending["user"]) + if self.path == "/token": + entry = self.world.kc_codes.pop(form.get("code"), None) + if not entry: + return self.send(400, {"error": "invalid_grant"}) + pending, user = entry + verifier = form.get("code_verifier", "") + if journey.b64url(hashlib.sha256(verifier.encode()).digest()) != pending["code_challenge"]: + return self.send(400, {"error": "invalid_grant"}) + return self.send(200, {"id_token": self.id_token(pending, user), "token_type": "Bearer"}) + self.send(404) + + def privacyidea_check(self, code: str) -> bool: + counter = int(time.time() // 30) + return any(journey.hotp(SEED, c) == code for c in (counter - 1, counter, counter + 1)) + + def redirect_with_code(self, pending, user): + code = secrets.token_urlsafe(16) + self.world.kc_codes[code] = (pending, user) + self.world.issued.append(code) + state = "not-the-state" if self.world.return_wrong_state else pending["state"] + return self.send(302, headers={"Location": pending["redirect_uri"] + "?" + urlencode({"code": code, "state": state})}) + + def id_token(self, pending, user): + level = self.world.token_level + claims = {"iss": base_url(self), "aud": [CLIENT], "sub": "opaque", "nonce": pending["nonce"], + "exp": int(time.time()) + 300, "iat": int(time.time()), "preferred_username": user, + "assurance": {"level": level, "methods": ["pwd", "otp"] if level == "aal2" else ["pwd"], + "mfa": level == "aal2", "source": "key-cape", "at": int(time.time())}} + head = journey.b64url(json.dumps({"alg": "RS256", "kid": "journey-test"}).encode()) + payload = journey.b64url(json.dumps(claims).encode()) + signature = KEY.sign(f"{head}.{payload}".encode()) + if self.world.tamper_signature: + signature = bytes([signature[0] ^ 1]) + signature[1:] + token = f"{head}.{payload}.{journey.b64url(signature)}" + self.world.issued.append(token) + return token + + +def serve(handler_cls, world): + handler = type(handler_cls.__name__, (handler_cls,), {"world": world}) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start() + return server + + +# ------------------------------------------------------------------------------------------------ +# Tests +# ------------------------------------------------------------------------------------------------ + + +class JourneyTests(unittest.TestCase): + mapped = False + + def setUp(self): + self.world = World() + self.kc = serve(FakeKeyCape, self.world) + self.authelia = serve(FakeAuthelia, self.world) + kc_local = f"http://127.0.0.1:{self.kc.server_port}" + authelia_local = f"http://127.0.0.1:{self.authelia.server_port}" + if self.mapped: + self.world.kc_public, self.world.authelia_public = "https://kc.example.test", "https://auth.example.test" + host_map = {self.world.kc_public: kc_local, self.world.authelia_public: authelia_local} + else: + self.world.kc_public, self.world.authelia_public = kc_local, authelia_local + host_map = {} + self.cfg = journey.load_config(self.write_config({ + "issuer": self.world.kc_public, "client_id": CLIENT, "redirect_uri": REDIRECT, + "username": USER, "host_map": host_map, "deadline_seconds": 30, + })) + + def tearDown(self): + for server in (self.kc, self.authelia): + server.shutdown() + server.server_close() + + def write_config(self, data): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + path = Path(tmp.name) / "config.json" + path.write_text(json.dumps(data)) + return str(path) + + def run_journey(self, password=PASSWORD, seed=SEED, **kw): + return journey.run_journey(self.cfg, password, seed, **kw) + + def assert_no_secret(self, receipt): + text = json.dumps(receipt) + for value in [PASSWORD, SEED_B32, SEED.hex(), *self.world.issued]: + self.assertNotIn(value, text) + counter = int(time.time() // 30) + for c in (counter - 1, counter, counter + 1): + self.assertNotIn(f'"{journey.hotp(SEED, c)}"', text) + + def assert_failed_at(self, receipt, step, reason, bind=None): + self.assertEqual(receipt["result"], "fail") + self.assertEqual((receipt["failed_step"], receipt["reason"]), (step, reason)) + self.assertEqual(receipt.get("bind_on_failed_step"), bind) + self.assertEqual(receipt["steps"][-1]["name"], step) + self.assertFalse(receipt["steps"][-1]["ok"]) + self.assertTrue(all(s["ok"] for s in receipt["steps"][:-1])) + self.assert_no_secret(receipt) + + def test_healthy_stack_passes_with_aal2(self): + receipt = self.run_journey() + self.assertEqual(receipt["result"], "pass", receipt) + self.assertEqual([s["name"] for s in receipt["steps"]], [ + "discovery", "keycape_authorize", "authelia_first_factor", "authelia_authorization", + "keycape_callback", "keycape_otp", "keycape_token", "id_token"]) + self.assertEqual(receipt["claims"]["assurance_level"], "aal2") + self.assertIn("otp", receipt["claims"]["assurance_methods"]) + self.assertTrue(receipt["claims"]["signature_verified"]) + self.assert_no_secret(receipt) + + def test_each_broken_bind_fails_closed_at_its_step(self): + expected = { + "svc-authelia": ("authelia_first_factor", "first_factor_rejected"), + "svc-keycape": ("keycape_callback", "callback_rejected"), + "svc-privacyidea": ("keycape_otp", "otp_rejected"), + } + for bind, (step, reason) in expected.items(): + with self.subTest(bind=bind): + self.world.binds = {b: b != bind for b in self.world.binds} + self.assert_failed_at(self.run_journey(), step, reason, bind) + + def test_wrong_password_and_wrong_seed_fail(self): + self.assert_failed_at(self.run_journey(password="not-it"), "authelia_first_factor", + "first_factor_rejected", "svc-authelia") + self.assert_failed_at(self.run_journey(seed=b"0123456789abcdef0123"), "keycape_otp", + "otp_rejected", "svc-privacyidea") + + def test_sign_in_without_otp_challenge_is_a_failure(self): + self.world.skip_mfa = True + self.assert_failed_at(self.run_journey(), "keycape_callback", "mfa_not_challenged", "svc-keycape") + + def test_token_checks(self): + self.world.tamper_signature = True + self.assert_failed_at(self.run_journey(), "id_token", "signature_invalid") + self.world.tamper_signature = False + self.world.token_level = "aal1" + self.assert_failed_at(self.run_journey(), "id_token", "not_aal2") + + def test_state_mismatch_after_otp(self): + self.world.return_wrong_state = True + self.assert_failed_at(self.run_journey(), "keycape_otp", "state_mismatch", "svc-privacyidea") + + def test_waits_out_a_closing_otp_window(self): + slept = [] + fake_now = [1_000_000 * 30 + 28.0] # two seconds before the window closes + j = journey.Journey(self.cfg, PASSWORD, SEED, wall=lambda: fake_now[0], + sleep=lambda s: (slept.append(s), fake_now.__setitem__(0, fake_now[0] + s))) + with self.assertRaises(journey.JourneyFailure): + j.run() # the fake privacyIDEA checks real time, so the synthetic clock's code is rejected + self.assertEqual(len(slept), 1) + self.assertGreaterEqual(slept[0], 2.0) + + def test_deadline_exceeded(self): + ticks = iter([0, 0] + [1000] * 50) + j = journey.Journey(self.cfg, PASSWORD, SEED, clock=lambda: next(ticks)) + with self.assertRaises(journey.JourneyFailure) as caught: + j.run() + self.assertEqual(caught.exception.reason, "deadline_exceeded") + + def test_unreachable_keycape(self): + self.kc.shutdown() + self.kc.server_close() + self.kc = serve(FakeKeyCape, World()) # keeps tearDown simple; the configured port is gone + receipt = self.run_journey() + self.assertEqual((receipt["failed_step"], receipt["reason"]), ("discovery", "unreachable")) + + def test_cli_exit_codes_and_receipt_file(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + d = Path(tmp.name) + (d / "pw").write_text(PASSWORD + "\n") + (d / "seed").write_text(SEED_B32 + "\n") + cfg_path = self.write_config({"issuer": self.world.kc_public, "client_id": CLIENT, + "redirect_uri": REDIRECT, "username": USER, + "host_map": self.cfg["host_map"]}) + args = ["--config", cfg_path, "--password-file", str(d / "pw"), "--otp-seed-file", str(d / "seed"), + "--receipt", str(d / "receipt.json")] + with contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(journey.main(args), 0) + self.assertEqual(json.loads((d / "receipt.json").read_text())["result"], "pass") + self.world.binds["svc-keycape"] = False + self.assertEqual(journey.main(args), 1) + with contextlib.redirect_stderr(io.StringIO()) as err: + (d / "seed").write_text("not base32 !\n") + self.assertEqual(journey.main(args), 2) + self.assertNotIn("not base32 !", err.getvalue()) + + +class MappedJourneyTests(JourneyTests): + """Same journey through public origins mapped to internal ones (executor inside the cluster).""" + + mapped = True + + def test_public_host_reaches_the_services(self): + self.assertEqual(self.run_journey()["result"], "pass") + self.assertIn("kc.example.test", self.world.seen_hosts) + self.assertIn("auth.example.test", self.world.seen_hosts) + + +class ConfigAndPrimitiveTests(unittest.TestCase): + def write(self, data): + path = Path(tempfile.mkdtemp()) / "c.json" + path.write_text(json.dumps(data)) + return str(path) + + def test_config_refuses_secrets_and_weaker_acr(self): + good = {"issuer": "https://kc.example.test", "client_id": CLIENT, "redirect_uri": REDIRECT, "username": USER} + journey.load_config(self.write(good)) + for bad in ({**good, "password": "x"}, {**good, "otp": {"seed": "x"}}, {**good, "acr_values": "aal1"}, + {**good, "username": "bad user"}, {k: v for k, v in good.items() if k != "issuer"}): + with self.subTest(bad=bad), self.assertRaises(journey.ConfigError): + journey.load_config(self.write(bad)) + + def test_hotp_rfc4226_vectors(self): + key = b"12345678901234567890" + expected = ["755224", "287082", "359152", "969429", "338314"] + self.assertEqual([journey.hotp(key, i) for i in range(5)], expected) + + def test_totp_rfc6238_sha1_vector(self): + self.assertEqual(journey.hotp(b"12345678901234567890", 59 // 30, 8), "94287082") + + def test_rs256_rejects_short_modulus(self): + small = RSAKey(bits=1024, seed=7) + data = b"payload" + self.assertFalse(journey.rs256_valid(data, small.sign(data), small.jwk())) + self.assertTrue(journey.rs256_valid(data, KEY.sign(data), KEY.jwk())) + self.assertFalse(journey.rs256_valid(b"other", KEY.sign(data), KEY.jwk())) + + def test_seed_validation(self): + with self.assertRaises(journey.ConfigError): + journey.decode_seed("AAAA") + self.assertEqual(journey.decode_seed(SEED_B32.lower()), SEED) + + +if __name__ == "__main__": + unittest.main() diff --git a/workplans/NK-WP-0056-bounded-executor-path.md b/workplans/NK-WP-0056-bounded-executor-path.md index ac54629..9cb9a63 100644 --- a/workplans/NK-WP-0056-bounded-executor-path.md +++ b/workplans/NK-WP-0056-bounded-executor-path.md @@ -9,7 +9,7 @@ flavor: design owner: claude topic_slug: netkingdom created: "2026-10-30" -updated: "2026-10-30" +updated: "2026-10-31" related: - NK-WP-0054 - NK-WP-0040 @@ -241,7 +241,7 @@ No bind rotation needs the custody copy decrypted. The custody copy keeps the cu ```task id: NK-WP-0056-T05 -status: todo +status: progress priority: medium assignee: claude-code depends_on: [NK-WP-0056-T04] @@ -252,8 +252,77 @@ A synthetic test user with an OTP seed held in OpenBao, and a journey that signs and KeyCape with MFA and resolves through privacyIDEA. Reuse the identity-provider journey acceptance. The executor runs it as the check between consumers and reverts on failure. -**Done when:** the journey runs unattended, and fails closed against a deliberately broken bind on a -synthetic directory. +**2026-10-31, founder decision: split into T05 (runner), T08 (synthetic stack on real pinned images), +T09 (synthetic user and seed in OpenBao) and T10 (unattended live run).** T05 is now the runner only: +a stdlib journey that drives KeyCape `/authorize` (PKCE, `acr_values=urn:netkingdom:aal2`), Authelia +`/api/firstfactor` and its OIDC authorization, the KeyCape callback and OTP form (KeyCape validates the +code through privacyIDEA), and the KeyCape token exchange, then verifies the ID token's signature and +`assurance` (aal2, `otp`). Secrets come from files or descriptors, never argv; the receipt is value-free +and names the first failed step. All three consumer binds are on the path (`svc-authelia` at first +factor, `svc-keycape` at the callback, `svc-privacyidea` at the OTP check); `svc-provisioner` is not. + +**Done when:** the runner and its tests (in-process fakes per bind, fail-closed at the named step, no +secret in the receipt) pass in the identity-provider journey acceptance workflow. + +**2026-10-31, implemented:** `sso-mfa/k8s/synthetic-journey/journey.py` (stdlib; RS256 verified against +KeyCape's JWKS without third-party packages; `host_map` for an in-cluster executor) with 26 tests in +`test_journey.py`. Each broken bind fails closed at its step (`svc-authelia` → `authelia_first_factor`, +`svc-keycape` → `keycape_callback`, `svc-privacyidea` → `keycape_otp`), and a sign-in without an OTP +challenge, an aal1 token or a tampered signature fails. Green locally and in the workflow's pinned +`python:3.12-slim` image with the network off; the job is added to `identity-journeys.yaml`. Done +once the workflow run on `main` is green. + +## Synthetic identity stack on pinned images + +```task +id: NK-WP-0056-T08 +status: todo +priority: medium +assignee: claude-code +depends_on: [NK-WP-0056-T05] +``` + +A throwaway stack on the pinned production images (LLDAP, Authelia, privacyIDEA, KeyCape), published on +127.0.0.1 only, with generated values and production-shaped binds (`svc-authelia`, `svc-keycape`, +`svc-privacyidea`), a synthetic user with a TOTP token, and the T05 runner against it. Extends the +`bind-split-rehearsal.py` pattern. + +**Done when:** the journey passes on the healthy stack and fails closed, at the expected step, for a +deliberately broken bind of each of the three consumers, with value-free receipts. + +## Synthetic user and seed in OpenBao + +```task +id: NK-WP-0056-T09 +status: todo +priority: medium +assignee: claude-code +needs_human: true +intervention_note: "Founder runs the live setup (LLDAP user, privacyIDEA token, OpenBao KV) under a CCR." +depends_on: [NK-WP-0056-T08] +``` + +An unprivileged live user (for example `syn-journey`, no groups), its own privacyIDEA TOTP token, and +password and seed held in OpenBao with coding agents fenced, a dedicated KeyCape public client for the +journey, prepared as `bind-lane.py`-style setup/verify helpers under a CCR. Founder-run. + +**Done when:** setup and verify pass live and one founder-started journey run passes with a receipt. + +## Unattended live journey + +```task +id: NK-WP-0056-T10 +status: todo +priority: medium +assignee: claude-code +depends_on: [NK-WP-0056-T03, NK-WP-0056-T09] +``` + +The executor identity (T03) reads the journey secrets from OpenBao itself and runs the journey as the +check between consumers; failure triggers the revert path. + +**Done when:** the journey runs unattended against live with no person involved, and fails closed +against a deliberately broken bind on a synthetic directory (T08). ## Pilot: unattended bind rotation under one admission @@ -264,7 +333,7 @@ priority: high assignee: claude-code needs_human: true intervention_note: "Founder signs the admission for the single live pilot rotation." -depends_on: [NK-WP-0056-T01, NK-WP-0056-T03, NK-WP-0056-T04, NK-WP-0056-T05] +depends_on: [NK-WP-0056-T01, NK-WP-0056-T03, NK-WP-0056-T04, NK-WP-0056-T05, NK-WP-0056-T10] state_hub_task_id: "6c3939f8-23a9-5f70-a3b1-94f4b291706b" ```