coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

NK-WP-0056-T05: synthetic MFA journey runner; split T05 into T05/T08/T09/T10
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Identity provider journey acceptance / provider (push) Successful in 16s
Details

Browse source 
Stdlib runner that signs a synthetic user in through KeyCape (PKCE, aal2),
Authelia first factor and the privacyIDEA-backed OTP challenge, verifies the
ID token (RS256 via JWKS, assurance aal2/otp) and writes a value-free receipt
naming the first failed step and the bind on it. 26 tests with per-bind fakes
run in the identity-provider journey acceptance workflow.

Founder decision 2026-10-31: real pinned images for the synthetic stack (T08),
synthetic user and seed in OpenBao (T09), unattended live run (T10).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 994874@bnt-lap001
Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87
This commit is contained in:
tegwick 2026-10-31 11:50:30 +01:00
parent ad7d1cdf49
commit 0b0c6c9628
6 changed files with 1210 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

7
.forgejo/workflows/identity-journeys.yaml
View file

@ -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

5
WORK-RECORDS.md
View file

@ -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 |

84
sso-mfa/k8s/synthetic-journey/README.md Normal file
View file

@ -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).

573
sso-mfa/k8s/synthetic-journey/journey.py Normal file
View file

@ -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'<form[^>]*action="/authorize/callback"[^>]*>(.*?)</form>', 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())

469
sso-mfa/k8s/synthetic-journey/test_journey.py Normal file
View file

@ -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 = ('<form method="post" action="/authorize/callback" autocomplete="off">'
f'<input type="hidden" name="state" value="{q["state"]}">'
'<input id="mfa_token" name="mfa_token" type="text" required>'
'<button type="submit">Verify</button></form>')
return self.send(200, f"<html><body>{form}</body></html>", 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()

79
workplans/NK-WP-0056-bounded-executor-path.md
View file

@ -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"
```