#!/usr/bin/env python3 """Attended seed of the nk-bounded-executor sender bearer (AUDIT-WP-0018-T04). NetKingdom NK-WP-0056-T15. railiance-platform CCR-2026-0063 holds the custody contract. One process mints one bearer and writes it to the two places that need it: custody: platform/data/workloads/net-kingdom/bounded-executor/audit-sender fields AUDIT_TOKEN, CUSTODY_REQUEST=CCR-2026-0063 (cas 0) registry: platform/data/workloads/audit-core/senders one appended entry "nk-bounded-executor" (cas on the version read) It then syncs ExternalSecret audit-core/audit-core-senders, restarts the receiver, and proves without writing an event that the live receiver knows the bearer with the exact registered scope. If railiance-platform's ExternalSecret sso/bounded-executor-tokens exists (manifests/nk-bind-executor-eso.yaml), the script syncs it and compares key audit-core by hash. This script is silent. The wrapper discards stdout and stderr, so `warden access --exec` fails closed on any output. The token is never printed, never passed on a command line and never written to the receipt. The receipt holds versions, booleans and a 12-hex SHA-256 fingerprint. Re-running is safe. Existing custody that this request wrote (version 1, CUSTODY_REQUEST matching) is reused. A registry entry that equals the desired entry is not rewritten. Anything else refuses and changes nothing: a sender without custody, a drifted entry, another request's custody, a token collision or a concurrent registry write. CCR-2026-0063 names this script as its custody procedure (railiance-platform f3b0c4f). It is the only seeder; railiance-platform's nk_bind_executor_custody.py fences the coding-agent boundary first and verifies afterwards. Founder-run only. The exact command is in docs/bounded-executor-receipts.md, section "Founder runbook". The script does not create OpenBao policies or roles, the coding-agent boundary, the ClusterSecretStore or the ExternalSecret in sso. Those belong to railiance-platform (credential-change applier, nk_bind_executor_custody.py fence, manifests/nk-bind-executor-eso.yaml). """ from __future__ import annotations import argparse import copy import hashlib import json import os import re import secrets import subprocess import time from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SENDER = "nk-bounded-executor" SOURCE = "net-kingdom.bounded-executor" TENANT = "platform" CCR_ID = "CCR-2026-0063" CCR_FILE = Path("/home/worsch/railiance-platform/credential-change-requests/" "CCR-2026-0063-nk-bind-executor-audit-sender.yaml") CCR_APPROVERS = ("platform-owner", "audit-core-owner", "net-kingdom-owner") CONFIRM = "SEED nk-bounded-executor CCR-2026-0063" REGISTRY = "platform/data/workloads/audit-core/senders" CUSTODY_KV = "platform/workloads/net-kingdom/bounded-executor/audit-sender" CUSTODY_DATA = "platform/data/workloads/net-kingdom/bounded-executor/audit-sender" CUSTODY_META = "platform/metadata/workloads/net-kingdom/bounded-executor/audit-sender" BOUNDARY = "sys/policies/acl/agent-high-risk-boundary" CLUSTER_UID = "a553c742-0115-43d4-99a4-a5ca56fe0786" # railiance01 kube-system SSO_SECRET = "bounded-executor-tokens" # ExternalSecret and Secret name (CCR-2026-0063 delivery) SSO_KEY = "audit-core" # the key the rapp run Job mounts TOKEN_SHAPE = re.compile(r"[A-Za-z0-9_-]{64}") # Runs inside the receiver pod. The bearer arrives on stdin and leaves only as # HTTP status codes. No request writes an event. RECEIVER_PROBE = r'''import json,sys,urllib.request,urllib.error tok=sys.stdin.read().strip() def get(p,auth=True): r=urllib.request.Request("http://127.0.0.1:8080"+p) if auth: r.add_header("Authorization","Bearer "+tok) try: with urllib.request.urlopen(r,timeout=10) as x: return x.status except urllib.error.HTTPError as e: return e.code w="since=2026-01-01T00:00:00%2B00:00&until=2026-01-02T00:00:00%2B00:00" print(json.dumps({ "read":get("/v1/events/probe-not-an-id"), "own_tenant":get("/v1/reconciliation?source=net-kingdom.bounded-executor&tenant=platform&"+w), "other_tenant":get("/v1/reconciliation?source=net-kingdom.bounded-executor&tenant=tenant:platform&"+w), "other_source":get("/v1/reconciliation?source=tenant-engine&tenant=platform&"+w)})) ''' PROBE_EXPECTED = {"read": 403, "own_tenant": 200, "other_tenant": 403, "other_source": 403} class Refused(Exception): """A named refusal. The message is a reason code, never a value.""" class NotFound(Exception): """The OpenBao path holds no value.""" def require(condition: bool, code: str) -> None: if not condition: raise Refused(code) def fingerprint(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest()[:12] # --- contract --------------------------------------------------------------- def scope_entry(path: Path = ROOT / "deploy" / "senders-scope.json") -> dict: rows = json.loads(path.read_text()) entry = next((r for r in rows if r.get("name") == SENDER), None) require(entry is not None, "scope_entry_missing") require(entry.get("sources") == [SOURCE] and entry.get("tenants") == [TENANT] and entry.get("may_write") is True and entry.get("may_read") is False and entry.get("secret_policy") == "reject" and entry.get("evidence_kind") == "load-bearing", "scope_entry_drift") return entry def desired(token: str, scope: dict) -> dict: """Registry entry: exactly the declared scope plus the bearer.""" entry = {"name": SENDER, "tokens": [token]} entry.update({k: copy.deepcopy(v) for k, v in scope.items() if k != "name"}) return entry def check_ccr(doc: dict) -> None: require(doc.get("id") == CCR_ID, "ccr_identity_mismatch") require(doc.get("status") in {"approved", "applied", "verified", "active"}, "approved_ccr_required") comments = (doc.get("review") or {}).get("comments") or [] for owner in CCR_APPROVERS: require(any(isinstance(c, dict) and c.get("decision") == "approved" and owner in str(c.get("reviewer", "")) for c in comments), "named_owner_reviews_required") ob = doc.get("openbao") or {} require(ob.get("kv_path") == CUSTODY_KV and ob.get("fields") == ["AUDIT_TOKEN", "CUSTODY_REQUEST"] and ob.get("metadata_read") is False, "ccr_custody_contract_mismatch") auth = ob.get("auth") or {} require(auth.get("audience") == "openbao" and auth.get("token_explicit_max_ttl") == "15m" and ob.get("token_self_lifecycle") is True, "ccr_reader_shape_mismatch") delivery = doc.get("delivery") or {} target = str(delivery.get("target", "")) require(delivery.get("namespace") == "sso" and f"sso/{SSO_SECRET}" in target and f"key {SSO_KEY}" in target and "seed-nk-bounded-executor-sender" in str(delivery.get("custody_procedure", "")), "ccr_delivery_mismatch") def load_ccr(path: Path = CCR_FILE) -> tuple[dict, str]: import yaml # system PyYAML; only needed for the CCR file raw = path.read_bytes() return yaml.safe_load(raw), hashlib.sha256(raw).hexdigest() # --- process boundaries ----------------------------------------------------- def bao(args: list[str], *, payload: dict | None = None) -> dict: env = os.environ.copy() for name in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN"): env.pop(name, None) proc = subprocess.run(["bao", *args], input=None if payload is None else json.dumps(payload).encode(), capture_output=True, env=env, check=False, timeout=60) if proc.returncode != 0: err = proc.stderr.decode(errors="replace").lower() # inspected, never emitted if "no value found" in err or "404" in err: raise NotFound() raise Refused("bao_failed") return json.loads(proc.stdout.decode() or "{}") class Kube: def __init__(self, kubeconfig: str): self.base = ["kubectl", "--kubeconfig", kubeconfig, "--request-timeout=30s"] def run(self, args: list[str], *, stdin: bytes | None = None) -> subprocess.CompletedProcess: return subprocess.run([*self.base, *args], input=stdin, capture_output=True, check=False, timeout=90) def json(self, args: list[str], code: str = "kubectl_failed") -> dict: proc = self.run([*args, "-o", "json"]) require(proc.returncode == 0, code) return json.loads(proc.stdout.decode()) def exists(self, args: list[str]) -> bool: return self.run([*args, "-o", "name"]).returncode == 0 # --- OpenBao ---------------------------------------------------------------- def snapshot(call=bao) -> tuple[int, dict, list, set]: raw = call(["read", "-format=json", REGISTRY])["data"] version = raw["metadata"]["version"] require(type(version) is int and version > 0, "registry_version_missing") body = raw["data"] require(isinstance(body, dict) and isinstance(body.get("senders.json"), str), "registry_shape_invalid") try: rows = json.loads(body["senders.json"]) except ValueError: raise Refused("registry_shape_invalid") from None require(isinstance(rows, list) and bool(rows), "registry_must_preserve_existing_senders") names: set = set() tokens: set = set() for row in rows: require(isinstance(row, dict) and isinstance(row.get("name"), str) and row["name"] and row["name"] not in names, "registry_identity_invalid_or_duplicate") names.add(row["name"]) values = row.get("tokens") or ([row["token"]] if row.get("token") else []) require(isinstance(values, list) and bool(values), "registry_tokens_invalid") for value in values: require(isinstance(value, str) and value and value not in tokens, "registry_token_collision") tokens.add(value) return version, body, rows, tokens def read_custody(call=bao) -> dict | None: try: return call(["read", "-format=json", CUSTODY_DATA])["data"] except NotFound: return None def seed(scope: dict, receipt: dict, save, call=bao) -> str: """Mint or reuse the bearer, then merge it into the registry. Returns the token.""" version, body, rows, used = snapshot(call) originals = copy.deepcopy(rows) found = next((r for r in rows if r["name"] == SENDER), None) receipt.update(registry_version_before=version, other_senders=len(rows) - (found is not None)) existing = read_custody(call) if existing is None: require(found is None, "registry_sender_without_custody") token = secrets.token_urlsafe(48) require(TOKEN_SHAPE.fullmatch(token) is not None and token not in used, "sender_token_collision") result = call(["write", "-format=json", CUSTODY_DATA, "-"], payload={"options": {"cas": 0}, "data": {"AUDIT_TOKEN": token, "CUSTODY_REQUEST": CCR_ID}}) require((result.get("data") or {}).get("version") == 1, "initial_custody_version_mismatch") receipt.update(custody="created", custody_version=1) else: data = existing.get("data") or {} require(existing["metadata"]["version"] == 1 and set(data) == {"AUDIT_TOKEN", "CUSTODY_REQUEST"} and data["CUSTODY_REQUEST"] == CCR_ID, "custody_provenance_mismatch") token = data["AUDIT_TOKEN"] require(isinstance(token, str) and TOKEN_SHAPE.fullmatch(token) is not None, "custody_token_shape") if found is None: require(token not in used, "sender_token_collision") receipt.update(custody="reused", custody_version=1) receipt["token_sha256_12"] = fingerprint(token) want = desired(token, scope) if found is not None: require(found == want, "registered_sender_drift") receipt["registry"] = "already_present" else: receipt["phase"] = "registry_cas" save() updated = [*rows, want] payload = dict(body, **{"senders.json": json.dumps(updated, separators=(",", ":"))}) call(["write", "-format=json", REGISTRY, "-"], payload={"options": {"cas": version}, "data": payload}) receipt["registry"] = "appended" after_version, after_body, after_rows, _ = snapshot(call) require(all(r in after_rows for r in originals) and want in after_rows and len(after_rows) == len(originals) + (found is None) and all(after_body.get(k) == v for k, v in body.items() if k != "senders.json"), "registry_readback_mismatch") receipt.update(registry_version_after=after_version, unchanged_other_senders=True) save() return token # --- Kubernetes ------------------------------------------------------------- def secret_entry_sha(kube: Kube) -> str | None: import base64 secret = kube.json(["-n", "audit-core", "get", "secret", "audit-core-senders"]) raw = (secret.get("data") or {}).get("senders.json") if not raw: return None rows = json.loads(base64.b64decode(raw)) row = next((r for r in rows if isinstance(r, dict) and r.get("name") == SENDER), None) if not row or not row.get("tokens"): return None return hashlib.sha256(row["tokens"][-1].encode()).hexdigest() def wait_for(check, timeout: int, interval: int = 3) -> bool: deadline = time.time() + timeout while time.time() < deadline: try: if check(): return True except Exception: pass time.sleep(interval) return False def receiver_pod(kube: Kube, image: str) -> str | None: dep = kube.json(["-n", "audit-core", "get", "deployment", "audit-core"]) if dep["spec"]["template"]["spec"]["containers"][0]["image"] != image: return None if dep.get("status", {}).get("readyReplicas") != 1 or \ dep["status"].get("observedGeneration", 0) < dep["metadata"]["generation"]: return None pods = kube.json(["-n", "audit-core", "get", "pods", "-l", "app.kubernetes.io/name=audit-core,app.kubernetes.io/component=receiver"])["items"] if len(pods) != 1 or pods[0]["metadata"].get("deletionTimestamp"): return None statuses = pods[0].get("status", {}).get("containerStatuses") or [] if not statuses or not all(s.get("ready") for s in statuses) \ or pods[0]["spec"]["containers"][0]["image"] != image: return None return pods[0]["metadata"]["name"] def live_scope_matches(kube: Kube, scope: dict) -> bool: cm = kube.json(["-n", "audit-core", "get", "configmap", "audit-core-senders-scope"], "scope_configmap_unreadable") rows = json.loads((cm.get("data") or {}).get("senders-scope.json") or "[]") return next((r for r in rows if r.get("name") == SENDER), None) == scope def expected_image() -> str: text = (ROOT / "deploy" / "audit-core.yaml").read_text() match = re.search(r"image:\s*(forgejo\.coulomb\.social/coulomb/audit-core@sha256:[0-9a-f]{64})", text) require(match is not None, "receiver_digest_required") return match.group(1) def probe(kube: Kube, pod: str, token: str) -> dict: proc = kube.run(["-n", "audit-core", "exec", "-i", pod, "-c", "audit-core", "--", "python", "-c", RECEIVER_PROBE], stdin=token.encode()) require(proc.returncode == 0, "receiver_probe_failed") return json.loads(proc.stdout.decode()) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--kubeconfig", required=True) parser.add_argument("--confirm", required=True) parser.add_argument("--receipt", type=Path, required=True) args = parser.parse_args(argv) receipt: dict = {"schema": "audit-core.sender-seed.v1", "sender": SENDER, "ccr": CCR_ID, "custody_path": CUSTODY_KV, "status": "refused", "credential_values_emitted": False, "started_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat()} fd = os.open(args.receipt, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) def save() -> None: os.lseek(fd, 0, os.SEEK_SET) os.ftruncate(fd, 0) os.write(fd, (json.dumps(receipt, indent=2) + "\n").encode()) os.fsync(fd) try: save() require(args.confirm == CONFIRM, "exact_confirmation_required") scope = scope_entry() ccr, ccr_sha = load_ccr() check_ccr(ccr) receipt["ccr_sha256"] = ccr_sha require(Path.home().parent.name == ".warden-attended-login", "attended_warden_envelope_required") require(not any(os.environ.get(n) for n in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")), "token_env_must_be_unset") identity = bao(["token", "lookup", "-format=json"])["data"] policies = identity.get("policies") or [] require("platform-admin" in policies and "root" not in policies, "attended_platform_admin_required") kube = Kube(args.kubeconfig) ns = kube.json(["get", "namespace", "kube-system"], "not_railiance01") require(ns.get("metadata", {}).get("uid") == CLUSTER_UID, "not_railiance01") image = expected_image() require(receiver_pod(kube, image) is not None, "receiver_not_current") require(live_scope_matches(kube, scope), "live_scope_configmap_mismatch") require(kube.exists(["-n", "audit-core", "get", "networkpolicy", "audit-core-nk-bounded-executor-ingress"]), "ingress_policy_not_applied") boundary = bao(["read", "-format=json", BOUNDARY])["data"]["policy"] require(all(f'"{p}"' in boundary for p in (CUSTODY_DATA, CUSTODY_META)), "coding_agent_boundary_missing") receipt.update(phase="preflight_ok", receiver_image=image) save() token = seed(scope, receipt, save) want = hashlib.sha256(token.encode()).hexdigest() receipt["phase"] = "registry_sync" save() require(kube.run(["-n", "audit-core", "annotate", "externalsecret", "audit-core-senders", f"force-sync={int(time.time())}", "--overwrite"]).returncode == 0, "externalsecret_annotate_failed") require(wait_for(lambda: secret_entry_sha(kube) == want, 120), "registry_secret_sync_timeout") receipt["registry_secret_hash_match"] = True receipt["phase"] = "receiver_reload" save() require(kube.run(["-n", "audit-core", "rollout", "restart", "deployment/audit-core"]).returncode == 0, "receiver_restart_failed") time.sleep(5) holder: dict = {} require(wait_for(lambda: holder.update(pod=receiver_pod(kube, image)) or holder["pod"], 240, 5), "receiver_not_ready_after_restart") result = probe(kube, holder["pod"], token) receipt["receiver_probe"] = result require(result == PROBE_EXPECTED, "receiver_probe_mismatch") receipt["phase"] = "sso_delivery" save() if kube.exists(["-n", "sso", "get", "externalsecret", SSO_SECRET]): kube.run(["-n", "sso", "annotate", "externalsecret", SSO_SECRET, f"force-sync={int(time.time())}", "--overwrite"]) def sso_sha() -> bool: import base64 secret = kube.json(["-n", "sso", "get", "secret", SSO_SECRET]) value = base64.b64decode((secret.get("data") or {})[SSO_KEY]).decode() return hashlib.sha256(value.encode()).hexdigest() == want require(wait_for(sso_sha, 120), "sso_secret_hash_mismatch") receipt.update(sso_secret_hash_match=True, status="seeded_and_delivered") else: receipt.update(sso_secret_hash_match=None, status="seeded_pending_sso_delivery") receipt["phase"] = "done" receipt["finished_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat() save() return 0 except Exception as exc: receipt["status"] = "refused" receipt["error"] = str(exc) if isinstance(exc, Refused) else "contained_operation_failed" save() return 1 finally: os.close(fd) if __name__ == "__main__": raise SystemExit(main())