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

NK-WP-0056-T09: prepare synthetic journey identity for founder run
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 
lane.py setup/verify/teardown (attended; CCR-2026-0057 sha256-bound):
generates password and TOTP seed in memory, KV CAS=0 with coding agents
fenced, LLDAP user syn-journey with no groups, privacyIDEA TOTP token with
enrollment confirmation, live journey as proof. SecurityCase for the public
KeyCape client nk-synthetic-journey; syn-journey declared SERVICE (planned);
proposed syn-<check> naming row in the SI standard; founder runbook.

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 12:42:36 +01:00
parent 22e50ef04b
commit ddde021bd0
10 changed files with 800 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -22,4 +22,4 @@ jobs:
run: |
set -eu
cd identity-source/sso-mfa/k8s/synthetic-journey
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest test_journey -v
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest test_journey test_lane -v

2
WORK-RECORDS.md
View file

@ -343,7 +343,7 @@
| 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 | done | — | 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-T09 | wait | — | 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 |

1
canon/standards/service-identity-privilege_v0.1.md
View file

@ -243,6 +243,7 @@ design to attribute it is STATE-WP-0100 (§10).
| Service-admin account | `<short>-admin` | the short form belongs to one service: `pi-admin` (privacyIDEA), `ld-admin` (LLDAP only) |
| Break-glass | `platform-root` | the king credential |
| Human | the person's own login | `admin`, `tegwick` |
| Synthetic sign-in principal | `syn-<check>` | proposed 2026-10-31 (NK-WP-0056-T09): a `SERVICE` that signs in like a person to prove a journey (password and MFA), with no groups or rights and credentials only in OpenBao; never `svc-` (bind accounts must not sign in, SI-3) |
| Workload, service, agent | the token subject form in use | for example `service:secrets-engine`; IAM Profile v0.3 governs the token |
The name `admin` is the human platform administrator's name wherever it appears (decided 2026-10-24); no

23
security-cases/nk-synthetic-journey.yaml Normal file
View file

@ -0,0 +1,23 @@
# KeyCape client for the synthetic MFA journey (NK-WP-0056-T09). Proposed: no acceptance yet.
# Public PKCE client; the redirect URI is never served: the runner reads the code from the
# Location header. mfa_required pins MFA, and the runner also requests acr urn:netkingdom:aal2.
# access_groups is required by the type; synthetic-journey has no members and no service maps it.
apiVersion: netkingdom.io/security-case/v0.1
kind: SecurityCase
spec:
id: nk-synthetic-journey
type: oidc-browser-relying-party@1.1
status: proposed
owner:
repo: net-kingdom
person: founder
parameters:
client_id: nk-synthetic-journey
display_name: NetKingdom synthetic journey
redirect_uris: https://journey.invalid/callback
allowed_scopes: openid,profile,email,groups
client_type: public
mfa_required: "true"
access_groups: synthetic-journey
consuming_service: synthetic-journey
consumer_owner_repo: net-kingdom

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

@ -100,6 +100,14 @@ It needs Docker, openssl and uv, and agents may run it. `--debug-logs` prints co
which can contain values, to the terminal only. Results:
`docs/evidence/2026-10-31-nk56-t08-synthetic-stack/`.
## Live identity (T09)
`lane.py` (setup / verify / teardown) is the attended, founder-run lane for the live identity
`syn-journey` under CCR-2026-0057. It generates the password and TOTP seed in memory, holds them in
OpenBao with coding agents fenced, creates the user with no groups, enrolls the token, and proves
the result with one live journey. The order of steps is in `T09-RUNBOOK.md`; the KeyCape client is
`security-cases/nk-synthetic-journey.yaml`. Tests: `python3 -m unittest test_lane` (fakes only).
## Not yet proven
- Live: the synthetic user, its token, the journey client and the OpenBao-held secrets (T09), and an

96
sso-mfa/k8s/synthetic-journey/T09-RUNBOOK.md Normal file
View file

@ -0,0 +1,96 @@
# NK-WP-0056-T09: synthetic journey identity, founder runbook
Prepared 2026-10-31 by claude-code. Founder-run. Creates `syn-journey` (no groups) in LLDAP, its
TOTP token `SYNJOURNEY01` in privacyIDEA realm `coulomb`, its password and seed in OpenBao
(`platform/workloads/net-kingdom/synthetic-journey/syn-journey`, coding agents fenced, no workload
reader), and the public KeyCape client `nk-synthetic-journey`. Then one journey run proves it all.
What you decide by approving:
- **CCR-2026-0057** (railiance-platform): custody, delivery and lifecycle of the two values.
- **SecurityCase `nk-synthetic-journey`**: the KeyCape client (public, PKCE, MFA pinned, redirect
`https://journey.invalid/callback`, which is never served; access group `synthetic-journey`, which
has no members).
- **Identity route**: an attended, receipted owner trial by `admin`, not User Engine onboarding. This
is the route `docs/legacy-lldap-lifecycle-retirement.md` names for test identities. The name follows
the proposed `syn-<check>` convention (SI standard §5) and is declared `SERVICE`
(`tools/service-identity/declarations/lldap.yaml`, status `planned`).
Known open points (recorded, not blocking): KeyCape tokens say `principal_type: human` for every
user. KeyCape has no per-client group gate, so `syn-journey` can sign in to any client that admits
any authenticated user, but it holds no group, so no service grants it anything.
## 0. Shell
```bash
export KUBECONFIG=$HOME/.kube/config-railiance01
cd ~/railiance-platform && git pull --ff-only && git status
cd ~/net-kingdom && git pull --ff-only
```
## 1. Approve CCR-2026-0057 (in railiance-platform)
```bash
cd ~/railiance-platform
python3 scripts/credential-change.py render CCR-2026-0057 # read it first
python3 scripts/credential-change.py approve CCR-2026-0057 --reviewer platform-owner \
--comment "Approved as proposed for NK-WP-0056-T09."
python3 scripts/credential-change.py approve CCR-2026-0057 --reviewer net-kingdom-owner \
--comment "Approved as proposed for NK-WP-0056-T09."
git status # the CCR file changed; commit it (or tell claude-code to)
```
## 2. Register the KeyCape client
Tell claude-code that you accept the case. It then marks `security-cases/nk-synthetic-journey.yaml`
`accepted`, using your words as the record. Then:
```bash
cd ~/net-kingdom
C=$(mktemp --suffix=.yaml)
python3 sso-mfa/k8s/keycape/keycape-client.py client-from-case security-cases/nk-synthetic-journey.yaml > "$C"
python3 sso-mfa/k8s/keycape/keycape-client.py inspect --client-file "$C" # note uid and resourceVersion
# check-config needs key-cape c528daa or later; the deployed image (e2b5fa1) has it
B=$HOME/.local/share/key-cape/verified-bin/e2b5fa1; mkdir -p "$B"
cid=$(docker create forgejo.coulomb.social/coulomb/key-cape@sha256:9e082dca78f65a4af3e92d3de3705b99e49ca9b298d9562ad0d37511c4306f8d)
docker cp "$cid:/keycape" "$B/keycape"; docker rm "$cid"
python3 sso-mfa/k8s/keycape/keycape-client.py apply --client-file "$C" \
--case-acceptance "security-cases/nk-synthetic-journey.yaml (accepted <date>)" \
--check-config "$B/keycape" --expected-uid <uid> --expected-resource-version <rv>
kubectl -n sso rollout restart deployment/keycape
kubectl -n sso rollout status deployment/keycape --timeout=180s
```
## 3. Identity, token and custody
```bash
cd ~/net-kingdom
bao login -method=oidc -path=netkingdom role=platform-admin # fails silently later if skipped
D=$(sha256sum ~/railiance-platform/credential-change-requests/CCR-2026-0057-net-kingdom-synthetic-journey-identity.yaml | cut -d' ' -f1)
E=docs/evidence/$(date +%F)-nk56-t09-synthetic-identity; mkdir -p "$E"
python3 sso-mfa/k8s/synthetic-journey/lane.py setup --ccr-sha256 "$D" --receipt "$E/01-setup.json"
# prompts: password of the human admin account (LLDAP), then the privacyIDEA pi-admin password
python3 sso-mfa/k8s/synthetic-journey/lane.py verify --ccr-sha256 "$D" --receipt "$E/02-verify.json"
```
`setup` refuses before any change if the KV path, the LLDAP user or the token already exist. If it
stops part-way, `lane.py teardown` removes whatever it created: token, user (refused if the user
has gained a group) and KV. Then run `setup` again. Receipts hold no values.
`verify` passes when the KV password signs in to LLDAP and the journey reaches an aal2 token with
`otp` from `https://kc.coulomb.social`. If the journey fails, its receipt names the step:
| Failed step | Likely cause |
| --- | --- |
| `keycape_authorize` | client not registered, or KeyCape not restarted after step 2 |
| `authelia_first_factor` | the password did not land in LLDAP (setup receipt), or Authelia regulation (5 failures in 2 min ban for 10 min) |
| `keycape_otp` | token not enrolled or not active; wait 30 s and verify again (a code is accepted once per window) |
| `keycape_token` | KeyCape's user lookup or tenant resolution for a user with no groups |
## 4. Afterwards
Hand the two receipts to claude-code. It records the evidence, moves the CCR to `verified`, the
declaration to `active` and T09 to `done`.

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

@ -0,0 +1,388 @@
#!/usr/bin/env python3
"""Attended lane for the synthetic journey identity `syn-journey` (NK-WP-0056-T09).
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.
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`
(security-cases/nk-synthetic-journey.yaml) to be registered first.
teardown: delete the token, the LLDAP user and the KV entry (all versions). The fence and the
KeyCape client stay; remove the client with keycape-client.py --operation remove.
Values are never printed or written. Receipts hold identifiers, versions and booleans only.
Founder-run: it needs the attended platform-admin session, the `admin` password and the `pi-admin`
password (prompted). Test with test_lane.py; never against live from an agent session.
"""
from __future__ import annotations
import argparse
import base64
import getpass
import hashlib
import importlib.util
import json
import os
import secrets
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
REPO = HERE.parents[2] if len(HERE.parents) > 2 else HERE # repo root; only Live needs it
PLATFORM = Path(os.environ.get("NK_PLATFORM_ROOT", REPO.parent / "railiance-platform"))
USER = "syn-journey"
EMAIL = "syn-journey@synthetic.invalid"
DISPLAY = "Synthetic journey (NK-WP-0056)"
REALM = "coulomb"
SERIAL = "SYNJOURNEY01"
CLIENT = "nk-synthetic-journey"
REDIRECT = "https://journey.invalid/callback"
ISSUER = "https://kc.coulomb.social"
CCR_FILE = "CCR-2026-0057-net-kingdom-synthetic-journey-identity.yaml"
KV = "platform/workloads/net-kingdom/synthetic-journey/syn-journey"
KV_DATA = KV.replace("platform/", "platform/data/", 1)
KV_METADATA = KV.replace("platform/", "platform/metadata/", 1)
FIELDS = ("PASSWORD", "TOTP_SEED")
ATTENDED_AUTH = {"method": "attended-oidc", "mount": "netkingdom", "role": "platform-admin"}
PROCEDURE = "net-kingdom sso-mfa/k8s/synthetic-journey/lane.py setup"
BOUNDARY = "sys/policies/acl/agent-high-risk-boundary"
APPROVERS = {"platform-owner", "net-kingdom-owner"}
LLDAP = "https://lldap.coulomb.social"
PI = "https://pink.coulomb.social"
class Refused(Exception):
pass
def load(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
journey = load("journey", HERE / "journey.py")
# --------------------------------------------------------------------------------------------------
# Contract
# --------------------------------------------------------------------------------------------------
def contract(d, expected: str) -> dict:
path = PLATFORM / "credential-change-requests" / CCR_FILE
raw = path.read_bytes()
if hashlib.sha256(raw).hexdigest() != expected:
raise Refused("ccr_digest")
ccr, errors = d.validate_ccr(path)
if errors or ccr.get("status") not in {"approved", "applied", "verified", "active"}:
raise Refused("ccr_status")
approved = {r["reviewer"].removeprefix("User (").removesuffix(")")
for r in ccr.get("review", {}).get("comments", [])
if r.get("decision") == "approved" and (r.get("source_ref") or r.get("at"))}
if not APPROVERS <= approved:
raise Refused("ccr_approval")
a, delivery = ccr.get("openbao", {}), ccr.get("delivery", {})
if (ccr.get("request_type") != "attended-service-credential-delivery" or a.get("kv_path") != KV
or a.get("fields") != list(FIELDS) or a.get("auth") != ATTENDED_AUTH
or a.get("runtime_access") is not False or delivery.get("procedure") != PROCEDURE):
raise Refused("ccr_shape")
return ccr
# --------------------------------------------------------------------------------------------------
# Steps
# --------------------------------------------------------------------------------------------------
def fence_stanzas() -> list[str]:
return [f'path "{p}" {{ capabilities = ["deny"] }}' for p in (KV_DATA, KV_METADATA)]
def fence_present(d) -> bool:
current = d.boundary_policy()
return all(s in current for s in fence_stanzas())
def add_fence(d) -> None:
current = d.boundary_policy()
added = ""
for path, stanza in zip((KV_DATA, KV_METADATA), fence_stanzas()):
if f'"{path}"' not in current:
added += stanza + "\n"
elif stanza not in current:
raise Refused("fence_conflict")
if added:
if d.boundary_policy() != current:
raise Refused("fence_race")
d.write_boundary_policy(current + "\n" + added)
if d.boundary_policy() != current + "\n" + added:
raise Refused("fence_readback")
def lldap_user(d, token) -> dict | None:
users = d.lldap_gql(token, "{ users { id groups { displayName } } }")["users"]
match = [u for u in users if u["id"] == USER]
return match[0] if match else None
def pi_tokens(d) -> list[dict]:
value = d.pi_call(f"/token/?user={USER}&realm={REALM}")
tokens = value.get("tokens") if isinstance(value, dict) else None
if not isinstance(tokens, list):
raise Refused("pi_token_list")
by_serial = d.pi_call(f"/token/?serial={SERIAL}")
if not isinstance(by_serial, dict) or not isinstance(by_serial.get("tokens"), list):
raise Refused("pi_token_list")
return tokens + [t for t in by_serial["tokens"] if t not in tokens]
def current_code(seed_b32: str, wall) -> str:
key = journey.decode_seed(seed_b32)
return journey.hotp(key, int(wall() // 30))
def setup(d, expected: str, receipt: dict) -> None:
receipt["phase"] = "contract"
ccr = contract(d, expected)
receipt["phase"] = "identity"
d.attended_platform_admin()
receipt["phase"] = "absent"
if d.kv_read(KV_DATA) is not None:
raise Refused("kv_exists")
admin = d.lldap_admin_token()
if lldap_user(d, admin) is not None:
raise Refused("lldap_user_exists")
d.pi_login()
if pi_tokens(d):
raise Refused("pi_token_exists")
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}})
d.lldap_set_password(admin, USER, password)
created = lldap_user(d, admin)
if created is None or created["groups"]:
raise Refused("lldap_user_shape")
if not d.lldap_login(USER, password):
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})
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
or tokens[0].get("tokentype") != "totp" or tokens[0].get("rollout_state") not in ("", "enrolled", None)):
raise Refused("pi_token_shape")
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,
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)
def verify(d, expected: str, receipt: dict) -> None:
receipt["phase"] = "contract"
ccr = contract(d, expected)
receipt["phase"] = "identity"
d.attended_platform_admin()
receipt["phase"] = "kv"
entry = d.kv_read(KV_DATA)
if entry is None or set(entry["data"]) != set(FIELDS):
raise Refused("kv_shape")
receipt["kv_version"] = entry["metadata"]["version"]
receipt["phase"] = "fence_present"
if not fence_present(d):
raise Refused("fence_missing")
receipt["phase"] = "no_workload_role"
if d.workload_role_exists():
raise Refused("unexpected_workload_role")
receipt["phase"] = "lldap_login"
if not d.lldap_login(USER, entry["data"]["PASSWORD"]):
raise Refused("kv_password_not_current")
receipt["phase"] = "journey"
result = d.run_journey({"issuer": ISSUER, "client_id": CLIENT, "redirect_uri": REDIRECT, "username": USER},
entry["data"]["PASSWORD"], journey.decode_seed(entry["data"]["TOTP_SEED"]))
receipt["journey"] = result
if result.get("result") != "pass":
raise Refused("journey_failed")
receipt.update(phase="done", status="identity_verified", source_ccr=ccr["id"], ccr_sha256=expected,
kv_is_current=True, coding_agents_denied=True, workload_access=False, journey_passed=True)
def teardown(d, expected: str, receipt: dict) -> None:
receipt["phase"] = "contract"
contract(d, expected)
receipt["phase"] = "identity"
d.attended_platform_admin()
receipt["phase"] = "pi_token"
d.pi_login()
for token in pi_tokens(d):
if token.get("serial") != SERIAL:
raise Refused("unexpected_pi_token") # never delete what this lane did not create
d.pi_call(f"/token/{SERIAL}", method="DELETE")
receipt["phase"] = "lldap_user"
admin = d.lldap_admin_token()
user = lldap_user(d, admin)
if user is not None:
if user["groups"]:
raise Refused("lldap_user_has_groups")
d.lldap_gql(admin, "mutation($u: String!) { deleteUser(userId: $u) { ok } }", {"u": USER})
receipt["phase"] = "kv"
d.kv_destroy(KV_METADATA)
if d.kv_read(KV_DATA) is not None:
raise Refused("kv_still_present")
receipt.update(phase="done", status="identity_removed", lldap_user_removed=True, pi_token_removed=True,
kv_removed=True, keycape_client_removal_separate=True)
# --------------------------------------------------------------------------------------------------
# Live dependencies (loaded only when run for real)
# --------------------------------------------------------------------------------------------------
class Live:
def __init__(self, lldap_endpoint: str, pi_endpoint: str):
sys.path.insert(0, str(PLATFORM / "scripts"))
sys.path.insert(0, str(REPO / "sso-mfa/k8s/privacyidea"))
sys.path.insert(0, str(REPO / "sso-mfa/k8s/lldap"))
lane = load("bind_lane", REPO / "tools/service-binds/bind-lane.py")
self._bao, self._data = lane.bao, lane.data
self._attended = lane.attended_platform_admin
self._ccr = load("credential_change", PLATFORM / "scripts/credential-change.py")
self._split = load("bind_split", REPO / "sso-mfa/k8s/lldap/bind-split.py")
self._realm = load("realm_bootstrap", REPO / "sso-mfa/k8s/privacyidea/realm_bootstrap.py")
self._pi_api = load("pi_api", REPO / "sso-mfa/k8s/privacyidea/pi_api.py")
self.directory = self._split.Directory(lldap_endpoint)
self.lldap_endpoint = lldap_endpoint
self.pi_endpoint = pi_endpoint
self.pi = None
self.wall = __import__("time").time
def validate_ccr(self, path):
ccr, errors, _ = self._ccr.validate_ccr(path)
return ccr, errors
def attended_platform_admin(self):
self._attended()
def kv_read(self, path):
result = self._bao(["read", "-format=json", path], allow_failure=True)
if result.returncode != 0:
if b"No value found" in result.stdout + result.stderr:
return None
raise Refused("kv_read")
return self._data(result)["data"]
def kv_write_cas0(self, path, values):
return self._data(self._bao(["write", "-format=json", path, "-"],
payload={"options": {"cas": 0}, "data": values}))["data"]["version"]
def kv_destroy(self, metadata_path):
self._bao(["delete", metadata_path])
def boundary_policy(self):
return self._data(self._bao(["read", "-format=json", BOUNDARY]))["data"]["policy"]
def write_boundary_policy(self, text):
self._bao(["write", BOUNDARY, "-"], payload={"policy": text})
def workload_role_exists(self):
role = "auth/kubernetes/role/net-kingdom-synthetic-journey"
return self._bao(["read", "-format=json", role], allow_failure=True).returncode == 0
def lldap_admin_token(self):
token = self._split.admin_token(self.directory, None)
if not token:
raise Refused("lldap_admin_login")
return token
def lldap_gql(self, token, query, variables=None):
return self.directory.gql(token, query, variables)
def lldap_set_password(self, token, user, password):
self._split.set_password(self.lldap_endpoint, token, user, password)
def lldap_login(self, user, password):
return self.directory.login(user, password) is not None
def pi_login(self):
if self.pi is None:
self.pi = self._realm.Client(self.pi_endpoint, getpass.getpass("privacyIDEA pi-admin password: "))
def pi_call(self, path, payload=None, method=None):
self.pi_login()
if method is None:
return self.pi.call(path, payload)
status, body = self._pi_api.request(self.pi.url + path, payload, self.pi.token, method=method)
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("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 run_journey(self, cfg, password, seed):
path = Path(os.environ.get("TMPDIR", "/tmp")) / f"journey-{secrets.token_hex(4)}.json"
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
try:
with os.fdopen(fd, "w") as out:
json.dump(cfg, out)
return journey.run_journey(journey.load_config(str(path)), password, seed)
finally:
path.unlink()
def main(argv=None) -> int:
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("action", choices=["setup", "verify", "teardown"])
p.add_argument("--ccr-sha256", required=True)
p.add_argument("--receipt", required=True)
p.add_argument("--lldap", default=LLDAP, help="LLDAP web endpoint")
p.add_argument("--privacyidea", default=PI, help="privacyIDEA endpoint")
a = p.parse_args(argv)
fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
r = {"status": "failed", "action": a.action, "identity": USER, "values_recorded": False,
"source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
try:
d = Live(a.lldap, a.privacyidea)
{"setup": setup, "verify": verify, "teardown": teardown}[a.action](d, a.ccr_sha256, r)
except Exception as error: # fixed reasons only, never exception text from a dependency
r.update(status="failed", reason=str(error) if isinstance(error, Refused) else type(error).__name__)
finally:
with os.fdopen(fd, "w") as out:
json.dump(r, out, indent=2, sort_keys=True)
out.write("\n")
print(json.dumps({k: r[k] for k in ("status", "phase", "reason") if k in r}))
return int(r["status"] == "failed")
if __name__ == "__main__":
raise SystemExit(main())

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

@ -0,0 +1,260 @@
"""Tests for the attended syn-journey lane (NK-WP-0056-T09) against fakes. Stdlib only."""
from __future__ import annotations
import hashlib
import importlib.util
import json
import tempfile
import time
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location("lane", HERE / "lane.py")
lane = importlib.util.module_from_spec(spec)
spec.loader.exec_module(lane)
CCR = {
"id": "CCR-2026-0057", "status": "approved", "request_type": "attended-service-credential-delivery",
"review": {"comments": [{"reviewer": "platform-owner", "decision": "approved", "at": "2026-11-01"},
{"reviewer": "net-kingdom-owner", "decision": "approved", "at": "2026-11-01"}]},
"openbao": {"kv_path": lane.KV, "fields": ["PASSWORD", "TOTP_SEED"], "auth": lane.ATTENDED_AUTH,
"runtime_access": False},
"delivery": {"procedure": lane.PROCEDURE},
}
class Fake:
"""OpenBao, LLDAP and privacyIDEA as the lane sees them; records effects, holds values in memory."""
def __init__(self, ccr=None, verify_enrollment=True):
self.ccr = ccr or json.loads(json.dumps(CCR))
self.attended_ok = True
self.kv = None
self.kv_version = 0
self.policy = 'path "other" { capabilities = ["deny"] }\n'
self.users = {}
self.passwords = {}
self.tokens = {}
self.pi_logged_in = False
self.verify_enrollment = verify_enrollment
self.role = False
self.journey_result = "pass"
self.journey_args = None
self.effects = []
self.wall = time.time
def validate_ccr(self, path):
return self.ccr, []
def attended_platform_admin(self):
if not self.attended_ok:
raise lane.Refused("attended_platform_admin_required")
def kv_read(self, path):
assert path == lane.KV_DATA
if self.kv is None:
return None
return {"data": dict(self.kv), "metadata": {"version": self.kv_version}}
def kv_write_cas0(self, path, values):
assert path == lane.KV_DATA and self.kv is None
self.kv, self.kv_version = dict(values), 1
self.effects.append("kv_write")
return 1
def kv_destroy(self, path):
assert path == lane.KV_METADATA
self.kv = None
self.effects.append("kv_destroy")
def boundary_policy(self):
return self.policy
def write_boundary_policy(self, text):
self.policy = text
self.effects.append("fence")
def workload_role_exists(self):
return self.role
def lldap_admin_token(self):
return "admin-token"
def lldap_gql(self, token, query, variables=None):
assert token == "admin-token"
if "createUser" in query:
self.users[variables["u"]["id"]] = {"id": variables["u"]["id"], "groups": []}
self.effects.append("lldap_create")
return {"createUser": {"id": variables["u"]["id"]}}
if "deleteUser" in query:
self.users.pop(variables["u"])
self.effects.append("lldap_delete")
return {"deleteUser": {"ok": True}}
return {"users": list(self.users.values())}
def lldap_set_password(self, token, user, password):
self.passwords[user] = password
self.effects.append("lldap_password")
def lldap_login(self, user, password):
return user in self.users and self.passwords.get(user) == password
def pi_login(self):
self.pi_logged_in = True
def pi_call(self, path, payload=None, method=None):
assert self.pi_logged_in
if method == "DELETE":
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"]}
if path.startswith("/token/?"):
public = [{k: v for k, v in t.items() if not k.startswith("_")} for t in self.tokens.values()]
if "serial=" in path:
serial = path.split("serial=")[1]
return {"tokens": [t for t in public if t["serial"] == serial]}
return {"tokens": [t for t in public if t["username"] == lane.USER]}
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"}
class LaneTests(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
root = Path(tmp.name)
(root / "credential-change-requests").mkdir()
self.ccr_path = root / "credential-change-requests" / lane.CCR_FILE
self.ccr_path.write_text("id: CCR-2026-0057\n")
self.digest = hashlib.sha256(self.ccr_path.read_bytes()).hexdigest()
self._platform = lane.PLATFORM
lane.PLATFORM = root
self.addCleanup(setattr, lane, "PLATFORM", self._platform)
def run_action(self, fn, d, digest=None):
receipt = {"status": "failed"}
try:
fn(d, digest or self.digest, receipt)
except lane.Refused as error:
receipt["reason"] = str(error)
return receipt
def assert_value_free(self, receipt, d):
text = json.dumps(receipt)
for value in list(d.passwords.values()) + list((d.kv or {}).values()):
self.assertNotIn(value, text)
def test_setup_creates_exactly_the_identity(self):
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(set(d.kv), {"PASSWORD", "TOTP_SEED"})
self.assertEqual(d.passwords[lane.USER], d.kv["PASSWORD"])
self.assertEqual(d.users[lane.USER]["groups"], [])
self.assertEqual(bytes.fromhex(d.tokens[lane.SERIAL]["_otpkey"]), lane.journey.decode_seed(d.kv["TOTP_SEED"]))
self.assertTrue(lane.fence_present(d))
self.assertTrue(receipt["enrollment_confirmed_with_code"])
self.assertEqual(receipt["created"]["groups"], [])
self.assert_value_free(receipt, d)
def test_setup_without_verify_enrollment(self):
d = Fake(verify_enrollment=False)
receipt = self.run_action(lane.setup, d)
self.assertEqual(receipt["status"], "identity_prepared")
self.assertNotIn("pi_verify", d.effects)
def test_setup_refuses_before_any_effect(self):
cases = {
"ccr_digest": lambda d: None,
"attended_platform_admin_required": lambda d: setattr(d, "attended_ok", False),
"kv_exists": lambda d: setattr(d, "kv", {"PASSWORD": "x", "TOTP_SEED": "y"}),
"lldap_user_exists": lambda d: d.users.update({lane.USER: {"id": lane.USER, "groups": []}}),
"pi_token_exists": lambda d: d.tokens.update({lane.SERIAL: {"serial": lane.SERIAL, "username": lane.USER}}),
"ccr_status": lambda d: d.ccr.update(status="proposed"),
"ccr_approval": lambda d: d.ccr["review"]["comments"].pop(),
"ccr_shape": lambda d: d.ccr["openbao"].update(runtime_access=True),
}
for reason, mutate in cases.items():
with self.subTest(reason=reason):
d = Fake()
mutate(d)
digest = "0" * 64 if reason == "ccr_digest" else None
receipt = self.run_action(lane.setup, d, digest)
self.assertEqual(receipt.get("reason"), reason)
self.assertEqual(d.effects, [])
def test_fence_conflict_refuses(self):
d = Fake()
d.policy += f'path "{lane.KV_DATA}" {{ capabilities = ["read"] }}\n'
self.assertEqual(self.run_action(lane.setup, d)["reason"], "fence_conflict")
self.assertEqual(d.effects, [])
def test_verify_runs_the_journey_with_kv_values(self):
d = Fake()
self.run_action(lane.setup, d)
receipt = self.run_action(lane.verify, d)
self.assertEqual(receipt["status"], "identity_verified", receipt)
cfg, password, seed = d.journey_args
self.assertEqual((cfg["issuer"], cfg["client_id"], cfg["username"]), (lane.ISSUER, lane.CLIENT, lane.USER))
self.assertEqual(password, d.kv["PASSWORD"])
self.assertEqual(seed, lane.journey.decode_seed(d.kv["TOTP_SEED"]))
self.assert_value_free(receipt, d)
def test_verify_fails_closed(self):
for reason, mutate in {
"journey_failed": lambda d: setattr(d, "journey_result", "fail"),
"kv_password_not_current": lambda d: d.passwords.update({lane.USER: "rotated-elsewhere"}),
"unexpected_workload_role": lambda d: setattr(d, "role", True),
"fence_missing": lambda d: setattr(d, "policy", ""),
}.items():
with self.subTest(reason=reason):
d = Fake()
self.run_action(lane.setup, d)
mutate(d)
self.assertEqual(self.run_action(lane.verify, d).get("reason"), reason)
def test_teardown_removes_only_what_setup_created(self):
d = Fake()
self.run_action(lane.setup, d)
receipt = self.run_action(lane.teardown, d)
self.assertEqual(receipt["status"], "identity_removed", receipt)
self.assertEqual((d.kv, d.users, d.tokens), (None, {}, {}))
self.assertTrue(lane.fence_present(d)) # the fence stays
d2 = Fake()
self.run_action(lane.setup, d2)
d2.users[lane.USER]["groups"] = [{"displayName": "net-kingdom-admins"}]
self.assertEqual(self.run_action(lane.teardown, d2)["reason"], "lldap_user_has_groups")
def test_teardown_is_idempotent_after_partial_setup(self):
d = Fake()
d.kv, d.kv_version = {"PASSWORD": "p", "TOTP_SEED": "s"}, 1
receipt = self.run_action(lane.teardown, d)
self.assertEqual(receipt["status"], "identity_removed")
if __name__ == "__main__":
unittest.main()

11
tools/service-identity/declarations/lldap.yaml
View file

@ -84,6 +84,17 @@ identities:
owner: "NetKingdom directory and User Engine owners"
rotation: {path: "Secret rotation with the provisioner only", coupling: [identity-provisioner]}
status: active
- name: syn-journey
principal_type: SERVICE
purpose: "Synthetic MFA sign-in journey (NK-WP-0056-T09): signs in through KeyCape, Authelia and privacyIDEA so a bounded executor can check sign-in between consumers without a person; no groups, no rights; KeyCape tokens still say principal_type human"
grants: [{mode: USE, scope: sign-in, activation: STANDING, environments: [PRODUCTION], note: "no group: reaches no protected service"}]
interactive_login: true
second_factor: required
custody: [openbao-lease]
consumers: [{name: synthetic-journey, kind: service}]
owner: "NetKingdom directory and User Engine owners"
rotation: {path: "lane.py teardown then setup (CCR-2026-0057); password and TOTP seed together", coupling: [privacyidea]}
status: planned
exceptions:
- id: SI-X6
rules: [SI-3]

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

@ -305,7 +305,7 @@ or clear the resolver cache, and the attended `--rotate-bind` MFA proof has the
```task
id: NK-WP-0056-T09
status: todo
status: wait
priority: medium
assignee: claude-code
needs_human: true
@ -320,6 +320,16 @@ journey, prepared as `bind-lane.py`-style setup/verify helpers under a CCR. Foun
**Done when:** setup and verify pass live and one founder-started journey run passes with a receipt.
**2026-10-31, prepared for founder decision** (`sso-mfa/k8s/synthetic-journey/T09-RUNBOOK.md`):
CCR-2026-0057 (railiance-platform, proposed; `attended-service-credential-delivery`, KV
`platform/workloads/net-kingdom/synthetic-journey/syn-journey` fields `PASSWORD`, `TOTP_SEED`, no workload
reader), SecurityCase `security-cases/nk-synthetic-journey.yaml` (public PKCE client, MFA pinned), and
`lane.py` setup/verify/teardown with 8 fake-backed tests. Identity route: attended owner trial by `admin`
(the route the NK-WP-0049 retirement names for test identities), not User Engine onboarding; name
`syn-journey` under a proposed `syn-<check>` convention (SI §5), declared `SERVICE`, status `planned`.
Open, recorded: KeyCape marks every token `principal_type: human`; KeyCape has no per-client group gate.
Waiting on: CCR approval, case acceptance, founder run of the runbook.
## Unattended live journey
```task