diff --git a/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-setup-refused-pi-admin-login.json b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-setup-refused-pi-admin-login.json new file mode 100644 index 0000000..8e936b2 --- /dev/null +++ b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-setup-refused-pi-admin-login.json @@ -0,0 +1,9 @@ +{ + "action": "setup", + "identity": "syn-journey", + "phase": "absent", + "reason": "pi_admin_login", + "source_sha256": "1f921377d6722ec816d330434e777d58277b4e820ed8113f5e1d878752a42562", + "status": "failed", + "values_recorded": false +} diff --git a/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-verify-before-setup.json b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-verify-before-setup.json new file mode 100644 index 0000000..5199c30 --- /dev/null +++ b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-verify-before-setup.json @@ -0,0 +1,9 @@ +{ + "action": "verify", + "identity": "syn-journey", + "phase": "kv", + "reason": "kv_shape", + "source_sha256": "1f921377d6722ec816d330434e777d58277b4e820ed8113f5e1d878752a42562", + "status": "failed", + "values_recorded": false +} diff --git a/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/01-setup.json b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/01-setup.json new file mode 100644 index 0000000..d4f9f2b --- /dev/null +++ b/docs/evidence/2026-10-31-nk56-t09-synthetic-identity/01-setup.json @@ -0,0 +1,10 @@ +{ + "action": "setup", + "identity": "syn-journey", + "kv_version": 1, + "phase": "pi_token", + "reason": "pi_request:/token/init", + "source_sha256": "1f921377d6722ec816d330434e777d58277b4e820ed8113f5e1d878752a42562", + "status": "failed", + "values_recorded": false +} diff --git a/sso-mfa/k8s/synthetic-journey/lane.py b/sso-mfa/k8s/synthetic-journey/lane.py index e592ca5..9309031 100644 --- a/sso-mfa/k8s/synthetic-journey/lane.py +++ b/sso-mfa/k8s/synthetic-journey/lane.py @@ -3,10 +3,13 @@ setup: under an approved, sha256-bound CCR and an attended platform-admin OpenBao session: refuse unless the KV path, the LLDAP user and the privacyIDEA token are all absent; fence - coding agents from the KV path; generate a password and a TOTP seed in memory; write both - to KV (CAS=0); create `syn-journey` in LLDAP with no groups (as the human `admin`) and set - its password; enroll its TOTP token in realm `coulomb` (as `pi-admin`), confirming the - enrollment with a code when privacyIDEA's verify_enrollment policy asks for it. + coding agents from the KV path; generate a password in memory; create `syn-journey` in + LLDAP with no groups (as the human `admin`) and set its password; enroll its TOTP token + in realm `coulomb` (as `pi-admin`) with a key privacyIDEA generates (pi-admin's rights + include totp_force_server_generate, which silently replaces a supplied key), taking the + key from the enrollment response and confirming it with a code when the verify_enrollment + policy asks; then write password and seed to KV (CAS=0). A failure after the LLDAP step + leaves state for `teardown`. verify: KV holds exactly the two fields, the fence is present, no workload role exists, the KV password signs in to LLDAP, and the synthetic journey passes against the live issuer with the KV values (read in memory). Needs the KeyCape client `nk-synthetic-journey` @@ -28,6 +31,7 @@ import hashlib import importlib.util import json import os +import re import secrets import sys from pathlib import Path @@ -168,11 +172,7 @@ def setup(d, expected: str, receipt: dict) -> None: receipt["phase"] = "coding_agent_fence" add_fence(d) - receipt["phase"] = "kv" password = secrets.token_urlsafe(32) - seed_b32 = base64.b32encode(secrets.token_bytes(20)).decode() - receipt["kv_version"] = d.kv_write_cas0(KV_DATA, {"PASSWORD": password, "TOTP_SEED": seed_b32}) - receipt["phase"] = "lldap_user" d.lldap_gql(admin, "mutation($u: CreateUserInput!) { createUser(user: $u) { id } }", {"u": {"id": USER, "email": EMAIL, "displayName": DISPLAY}}) @@ -184,13 +184,15 @@ def setup(d, expected: str, receipt: dict) -> None: raise Refused("lldap_password_not_effective") receipt["phase"] = "pi_token" - seed_hex = journey.decode_seed(seed_b32).hex() - init = d.pi_call("/token/init", {"type": "totp", "serial": SERIAL, "otpkey": seed_hex, "genkey": 0, - "otplen": 6, "hashlib": "sha1", "timeStep": 30, "user": USER, - "realm": REALM, "description": "NK-WP-0056 synthetic journey"}) - if isinstance(init, dict) and init.get("rollout_state") == "verify" or d.pi_rollout_state(SERIAL) == "verify": - d.pi_call("/token/init", {"type": "totp", "serial": SERIAL, "verify": current_code(seed_b32, d.wall), - "user": USER, "realm": REALM}) + detail = d.pi_init({"type": "totp", "serial": SERIAL, "genkey": 1, "otplen": 6, "hashlib": "sha1", + "timeStep": 30, "user": USER, "realm": REALM, + "description": "NK-WP-0056 synthetic journey"}) + value = (detail.get("otpkey") or {}).get("value") if isinstance(detail, dict) else None + if not isinstance(value, str) or not re.fullmatch(r"seed://[0-9a-f]{40,128}", value): + raise Refused("pi_seed_missing") + seed_b32 = base64.b32encode(bytes.fromhex(value[len("seed://"):])).decode() + if detail.get("rollout_state") == "verify": + d.pi_init({"type": "totp", "serial": SERIAL, "verify": current_code(seed_b32, d.wall)}) receipt["enrollment_confirmed_with_code"] = True tokens = pi_tokens(d) if (len(tokens) != 1 or tokens[0].get("serial") != SERIAL or tokens[0].get("active") is not True @@ -198,10 +200,13 @@ def setup(d, expected: str, receipt: dict) -> None: or tokens[0].get("tokentype") != "totp" or tokens[0].get("rollout_state") not in ("", "enrolled", None)): raise Refused("pi_token_shape") + receipt["phase"] = "kv" + receipt["kv_version"] = d.kv_write_cas0(KV_DATA, {"PASSWORD": password, "TOTP_SEED": seed_b32}) + receipt["phase"] = "fence_present" if not fence_present(d): raise Refused("fence_readback") - receipt.update(phase="done", status="identity_prepared", source_ccr=ccr["id"], ccr_sha256=expected, + receipt.update(phase="done", status="identity_prepared", key_generated_by="privacyidea", source_ccr=ccr["id"], ccr_sha256=expected, created={"lldap_user": USER, "groups": [], "email": EMAIL, "pi_token_serial": SERIAL, "pi_realm": REALM, "kv_path": KV}, coding_agents_denied=True, workload_access=False, lldap_password_effective=True) @@ -356,9 +361,14 @@ class Live: raise Refused("pi_request") return result.get("value") - def pi_rollout_state(self, serial): - tokens = self.pi_call(f"/token/?serial={serial}").get("tokens", []) - return tokens[0].get("rollout_state") if len(tokens) == 1 else None + def pi_init(self, payload): + """POST /token/init and return its detail (holds the generated key: kept in memory only).""" + self.pi_login() + status, body = self._pi_api.request(self.pi.url + "/token/init", payload, self.pi.token) + result = body.get("result") if isinstance(body, dict) else None + if status != 200 or not isinstance(result, dict) or result.get("status") is not True: + raise Refused(f"pi_request:/token/init:{status}") + return body.get("detail") or {} def run_journey(self, cfg, password, seed): path = Path(os.environ.get("TMPDIR", "/tmp")) / f"journey-{secrets.token_hex(4)}.json" diff --git a/sso-mfa/k8s/synthetic-journey/test_lane.py b/sso-mfa/k8s/synthetic-journey/test_lane.py index cf8467f..dace477 100644 --- a/sso-mfa/k8s/synthetic-journey/test_lane.py +++ b/sso-mfa/k8s/synthetic-journey/test_lane.py @@ -5,6 +5,7 @@ from __future__ import annotations import hashlib import importlib.util import json +import secrets import tempfile import time import unittest @@ -110,20 +111,29 @@ class Fake: self.tokens.pop(path.rsplit("/", 1)[1]) self.effects.append("pi_delete") return True - if path == "/token/init": - if "verify" in payload: - token = self.tokens[payload["serial"]] - key = bytes.fromhex(token["_otpkey"]) - counter = int(time.time() // 30) - assert payload["verify"] in {lane.journey.hotp(key, c) for c in (counter - 1, counter, counter + 1)} - token["rollout_state"] = "enrolled" - self.effects.append("pi_verify") - return True - self.tokens[payload["serial"]] = { - "serial": payload["serial"], "tokentype": "totp", "active": True, "_otpkey": payload["otpkey"], - "username": payload["user"], "rollout_state": "verify" if self.verify_enrollment else ""} - self.effects.append("pi_init") - return {"rollout_state": self.tokens[payload["serial"]]["rollout_state"]} + return self.pi_tokens_call(path) + + def pi_init(self, payload): + assert self.pi_logged_in + if "verify" in payload: + token = self.tokens[payload["serial"]] + key = bytes.fromhex(token["_otpkey"]) + counter = int(time.time() // 30) + assert payload["verify"] in {lane.journey.hotp(key, c) for c in (counter - 1, counter, counter + 1)} + token["rollout_state"] = "enrolled" + self.effects.append("pi_verify") + return {"rollout_state": "enrolled"} + # As live: pi-admin's rights include totp_force_server_generate, so the server makes the key. + assert "otpkey" not in payload and payload.get("genkey") == 1 + key = secrets.token_bytes(20).hex() + self.tokens[payload["serial"]] = { + "serial": payload["serial"], "tokentype": "totp", "active": True, "_otpkey": key, + "username": payload["user"], "rollout_state": "verify" if self.verify_enrollment else ""} + self.effects.append("pi_init") + return {"rollout_state": self.tokens[payload["serial"]]["rollout_state"], + "otpkey": {"value": "seed://" + key}} + + def pi_tokens_call(self, path): if path.startswith("/token/?"): public = [{k: v for k, v in t.items() if not k.startswith("_")} for t in self.tokens.values()] if "user=" in path and lane.USER not in self.users: @@ -132,10 +142,6 @@ class Fake: return {"tokens": [t for t in public if t["serial"] == serial]} raise AssertionError(path) - def pi_rollout_state(self, serial): - token = self.tokens.get(serial) - return token["rollout_state"] if token else None - def run_journey(self, cfg, password, seed): self.journey_args = (cfg, password, seed) return {"result": self.journey_result, "failed_step": None if self.journey_result == "pass" else "keycape_otp"} @@ -171,11 +177,13 @@ class LaneTests(unittest.TestCase): d = Fake() receipt = self.run_action(lane.setup, d) self.assertEqual(receipt["status"], "identity_prepared", receipt) - self.assertEqual(d.effects, ["fence", "kv_write", "lldap_create", "lldap_password", "pi_init", "pi_verify"]) + self.assertEqual(d.effects, ["fence", "lldap_create", "lldap_password", "pi_init", "pi_verify", "kv_write"]) self.assertEqual(set(d.kv), {"PASSWORD", "TOTP_SEED"}) self.assertEqual(d.passwords[lane.USER], d.kv["PASSWORD"]) self.assertEqual(d.users[lane.USER]["groups"], []) + # The seed in KV is the key privacyIDEA generated, not one the lane chose. self.assertEqual(bytes.fromhex(d.tokens[lane.SERIAL]["_otpkey"]), lane.journey.decode_seed(d.kv["TOTP_SEED"])) + self.assertEqual(receipt["key_generated_by"], "privacyidea") self.assertTrue(lane.fence_present(d)) self.assertTrue(receipt["enrollment_confirmed_with_code"]) self.assertEqual(receipt["created"]["groups"], []) @@ -207,6 +215,15 @@ class LaneTests(unittest.TestCase): self.assertEqual(receipt.get("reason"), reason) self.assertEqual(d.effects, []) + def test_missing_generated_key_stops_before_kv(self): + d = Fake() + original = d.pi_init + d.pi_init = lambda payload: {k: v for k, v in original(payload).items() if k != "otpkey"} + receipt = self.run_action(lane.setup, d) + self.assertEqual(receipt["reason"], "pi_seed_missing") + self.assertIsNone(d.kv) + self.assertEqual(self.run_action(lane.teardown, d)["status"], "identity_removed") + def test_fence_conflict_refuses(self): d = Fake() d.policy += f'path "{lane.KV_DATA}" {{ capabilities = ["read"] }}\n'