#!/usr/bin/env python3 # /// script # dependencies = ["PyYAML>=6,<7", "jsonschema>=4,<5"] # /// """Synthetic pilot: the executor rotates the svc-authelia bind on the T08 stack (NK-WP-0056-T12). Agent-runnable, local Docker only, generated values, everything removed afterwards. Starts the T08 identity stack (LLDAP, privacyIDEA, Authelia, KeyCape on the pinned images) and an OpenBao dev server on the platform's pinned 2.5.4 image (KV v2 at `platform`, `cas_required` on the bind). The executor gets its own token with exactly CCR-2026-0058's policy file plus a journey-read policy, never the root token. Not on this stack: approval-engine (an in-process carrier with its consume semantics stands in), audit-core (an in-memory sink), and Kubernetes. Stand-ins replace the ExternalSecret delivery (re-project the KV value into Authelia's configuration) and the rollout (restart the Authelia container). The real Kubernetes adapters (operations.KubeDelivery, KubeRestart) are covered by unit tests and exercised live in T06. Runs: rotation one admission, the bind rotates; afterwards the old value is refused and the journey passes revert a second admission whose first delivery projects a wrong value while reporting success: Authelia does not become ready after the restart (definite failure), the executor reverts in declared order, the previous value is back and the journey passes again Writes a value-free report. Exit 0 when both runs end as expected. Run: uv run bounded-executor/synthetic_pilot.py [--report FILE] """ from __future__ import annotations import argparse import base64 import copy import datetime as dt import hashlib import importlib.util import json import pathlib import secrets import sys import tempfile import yaml HERE = pathlib.Path(__file__).resolve().parent REPO = HERE.parent def _load(name, path): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module X = _load("executor", HERE / "executor.py") OPS = _load("operations", HERE / "operations.py") stack_mod = _load("stack", REPO / "sso-mfa/k8s/synthetic-journey/stack.py") bea = X.bea OPENBAO_IMAGE = "quay.io/openbao/openbao@sha256:436eaf9778cad75507ff70ea26ace30dcbe15606e619ac3823495663d7f7c115" # 2.5.4 BIND_KV = "platform/workloads/net-kingdom/directory-binds/svc-authelia" JOURNEY_KV = "platform/workloads/net-kingdom/synthetic-journey/syn-journey" POLICY_FILE = REPO.parent / "railiance-platform/openbao/policies/net-kingdom-bind-executor-svc-authelia-write.hcl" JOURNEY_POLICY = 'path "platform/data/workloads/net-kingdom/synthetic-journey/syn-journey" { capabilities = ["read"] }\n' EXECUTOR = "service:nk-bind-executor" APPROVER_SUB = "synthetic-admin-sub" class OpenBaoDev: def __init__(self, network: str, tag: str): self.name = f"nk56-executor-{tag}-openbao" self.root = secrets.token_hex(16) stack_mod.docker("run", "--detach", "--pull=never", "--name", self.name, "--network", network, "--publish", "127.0.0.1::8200", "--cap-add", "IPC_LOCK", "--env", "BAO_DEV_ROOT_TOKEN_ID", OPENBAO_IMAGE, "server", "-dev", "-dev-listen-address=0.0.0.0:8200", env={"BAO_DEV_ROOT_TOKEN_ID": self.root}) self.addr = f"http://127.0.0.1:{stack_mod.port(self.name, 8200)}" stack_mod.wait_http(self.addr + "/v1/sys/health", "openbao") self.admin = OPS.OpenBao(self.addr, token=self.root) def call(self, method, path, body=None): return self.admin._call(method, path, body) def setup(self, bind_value: str, journey_password: str, seed: bytes) -> str: self.call("POST", "/v1/sys/mounts/platform", {"type": "kv", "options": {"version": "2"}}) self.admin.write(BIND_KV, {OPS.FIELD: bind_value}, cas=0) self.call("POST", "/v1/platform/metadata/workloads/net-kingdom/directory-binds/svc-authelia", {"cas_required": True}) self.admin.write(JOURNEY_KV, {"PASSWORD": journey_password, "TOTP_SEED": base64.b32encode(seed).decode()}, cas=0) self.call("PUT", "/v1/sys/policies/acl/bind-executor-write", {"policy": POLICY_FILE.read_text()}) self.call("PUT", "/v1/sys/policies/acl/journey-read", {"policy": JOURNEY_POLICY}) tokens = [] for policy in ("bind-executor-write", "journey-read"): # one token per lane, as the live roles tokens.append(self.call("POST", "/v1/auth/token/create", {"policies": [policy], "no_default_policy": True, "ttl": "15m"})["auth"]["client_token"]) return tuple(tokens) def remove(self): stack_mod.docker("rm", "--force", self.name, check=False) class Delivery: """Stand-in for the ExternalSecret: projects the KV value into Authelia's configuration file.""" def __init__(self, stack, bao_admin): self.stack, self.bao, self.corrupt = stack, bao_admin, 0 # number of next deliveries to corrupt def force_sync(self) -> dt.datetime: at = OPS.now_utc() value = self.bao.read(BIND_KV)[0][OPS.FIELD] if self.corrupt: self.corrupt -= 1 value = secrets.token_urlsafe(32) # a one-off wrong projection that still reports success path = self.stack.workdir / "authelia.yml" config = yaml.safe_load(path.read_text()) config["authentication_backend"]["ldap"]["password"] = value path.write_text(yaml.safe_dump(config)) # same inode: the bind mount sees it self.synced_at = OPS.now_utc() return at def synced(self, since: dt.datetime) -> bool: return getattr(self, "synced_at", None) is not None and self.synced_at >= since class Restart: """Stand-in for the rollout: restart the Authelia container and wait for it to answer.""" def __init__(self, stack): self.stack = stack def restart(self): stack_mod.docker("restart", self.stack.names["authelia"]) # docker assigns a new ephemeral host port on restart (an artifact of this stand-in) self.stack.ports["authelia"] = stack_mod.port(self.stack.names["authelia"], 9091) try: stack_mod.wait_http(f"http://127.0.0.1:{self.stack.ports['authelia']}/api/health", "authelia", {"Host": "auth.journey.test", "X-Forwarded-Proto": "https"}, attempts=60) except stack_mod.SetupError: raise OPS.RolloutFailed("authelia did not become ready") from None class MemoryCarrier: """approval-engine's approval object and consume semantics, in memory.""" def __init__(self): self.approvals = {} def approve(self, approval_id, admission): spec = admission["spec"] self.approvals[approval_id] = { "id": approval_id, "status": "approved", "binding": {"action": "bounded-execution", "actor": EXECUTOR, "principal": "admin", "purpose": spec["purpose"], "target": {"admission_id": spec["id"], "admission_digest": bea.admission_digest(admission)}, "human_control": True, "pdp_path": False}, "validity": {"not_before": spec["window"]["not_before"], "expires_at": spec["window"]["not_after"]}, "entries": [{"subject_id": APPROVER_SUB, "assurance": "aal2", "principal_type": "human", "approved_at": OPS.now_utc().isoformat()}]} def get(self, approval_id): return copy.deepcopy(self.approvals[approval_id]) def consume(self, approval_id, request_digest, decision_id): a = self.approvals[approval_id] if a["status"] != "approved": raise RuntimeError("409") a["status"] = "consumed" a["consumption"] = {"request_digest": request_digest, "decision_id": decision_id, "consumed_at": OPS.now_utc().isoformat()} return {"approval_id": approval_id, "status": "consumed", "request_digest": request_digest, "decision_id": decision_id, "consumed_at": a["consumption"]["consumed_at"], "idempotent": False} def admission(admission_id: str, attempts: int) -> dict: """The pilot example, pinned to the current files, with a window around now.""" doc = bea.load(REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml") spec = doc["spec"] spec["id"] = admission_id now = dt.datetime.now(dt.UTC).replace(microsecond=0) spec["window"] = {"not_before": (now - dt.timedelta(minutes=5)).isoformat().replace("+00:00", "Z"), "not_after": (now + dt.timedelta(hours=2)).isoformat().replace("+00:00", "Z")} spec["budget"]["attempts"] = attempts files = [] for f in spec["plan"]["source"]["files"]: files.append({"path": f["path"], "sha256": hashlib.sha256((REPO / f["path"]).read_bytes()).hexdigest()}) spec["plan"]["source"]["files"] = files pins = {f["path"]: f["sha256"] for f in files} for c in spec["checks"]: c["sha256"] = pins[c["ref"]] spec["plan_digest"] = bea.plan_digest(doc) report = bea.evaluate(doc, json.loads(bea.SCHEMA.read_text())) if not report["valid"]: raise SystemExit(f"synthetic admission invalid: {report['findings']}") return doc def lldap_command(): return ["docker", "run", "--rm", "--pull=never", "--network", "host", "--env", "LLDAP_USER_PASSWORD", "--entrypoint", "/app/lldap_set_password", stack_mod.LLDAP_IMAGE] def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) parser.add_argument("--report") args = parser.parse_args(argv) report = {"schema": "netkingdom.bounded-executor.synthetic-pilot/v1", "started_at": OPS.now_utc().isoformat(), "images": {"openbao": OPENBAO_IMAGE}, "policy_file_sha256": hashlib.sha256(POLICY_FILE.read_bytes()).hexdigest(), "runs": []} ok = False with tempfile.TemporaryDirectory(prefix="nk56-executor-") as tmp: stack = stack_mod.Stack(pathlib.Path(tmp), pi_cache_timeout=120) bao = None try: stack.up() bao = OpenBaoDev(stack.network, stack.tag) executor_token, journey_token = bao.setup(stack.pw["svc-authelia"], stack.pw[stack_mod.USER], stack.seed) executor_bao = OPS.OpenBao(bao.addr, token=executor_token) journey_bao = OPS.OpenBao(bao.addr, token=journey_token) directory = OPS.Directory(stack.lldap, lldap_command()) delivery = Delivery(stack, bao.admin) journey_state = {} operations, checks = OPS.pilot_registry( bao=executor_bao, journey_bao=journey_bao, directory=directory, delivery=delivery, restart=Restart(stack), bind_kv=BIND_KV, bind_user="svc-authelia", journey_kv=JOURNEY_KV, journey_config=stack.journey_config, journey_state=journey_state) carrier, events, alerts = MemoryCarrier(), [], [] audit = type("Sink", (), {"emit": lambda _s, e: events.append(e)})() cfg = X.Config(executor_identity=EXECUTOR, runtime_principal="synthetic:executor-token", tenant="tenant:platform", environment="TEST", approver_subjects={"admin": APPROVER_SUB}) deps = X.Deps(carrier=carrier, operations=operations, checks=checks, trust=lambda s: True, audit=audit, alert=lambda kind, info: alerts.append(kind)) def run(admission_id, attempts, corrupt): doc = admission(admission_id, attempts) approval = "apr-" + admission_id carrier.approve(approval, doc) before_value, before_version = bao.admin.read(BIND_KV) delivery.corrupt = 1 if corrupt else 0 result = X.Run(doc, approval, cfg, deps).execute() delivery.corrupt = 0 after_value, after_version = bao.admin.read(BIND_KV) changed = after_value != before_value old_refused = directory.login("svc-authelia", before_value[OPS.FIELD]) is None new_works = directory.login("svc-authelia", after_value[OPS.FIELD]) is not None receipt = result.get("receipt", {}).get("data", {}) return {"admission": admission_id, "outcome": result["outcome"], "reason": result.get("reason"), "kv_version": [before_version, after_version], "kv_value_changed": changed, "previous_value_refused_by_lldap": old_refused, "current_kv_value_signs_in": new_works, "steps": [(s["id"], s["kind"], s["outcome"], [c["result"] for c in s.get("checks", [])], s.get("error_type")) for s in receipt.get("steps", [])], "reverted": receipt.get("reverted"), "pdp": receipt.get("pdp"), "events": [e["type"] for e in events if e["correlation_id"] == result.get("run_id")]} rotation = run("BEA-2026-9001", 1, corrupt=False) rotation["met"] = (rotation["outcome"] == "completed" and rotation["kv_value_changed"] and rotation["previous_value_refused_by_lldap"] and rotation["current_kv_value_signs_in"]) report["runs"].append(rotation) revert = run("BEA-2026-9002", 1, corrupt=True) revert["met"] = (revert["outcome"] == "reverted" and not revert["kv_value_changed"] and revert["current_kv_value_signs_in"] and any(s[0] == "restart" and s[2] in ("check_failed", "failed") for s in revert["steps"]) and all(s[2] == "done" for s in revert["steps"] if s[1] == "revert")) report["runs"].append(revert) report["alerts"] = alerts secrets_in_report = [stack.pw["svc-authelia"], executor_token, journey_token, bao.root] text = json.dumps(report) report["no_values_in_report"] = not any(v in text for v in secrets_in_report) ok = rotation["met"] and revert["met"] and report["no_values_in_report"] finally: if bao: bao.remove() stack.down() report["all_met"] = ok report["finished_at"] = OPS.now_utc().isoformat() text = json.dumps(report, indent=2, default=str) if args.report: pathlib.Path(args.report).write_text(text + "\n") print(text) return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())