AUDIT-WP-0018-T04: founder-run seed of the nk-bounded-executor bearer; T05 live activation waits

- scripts/seed-nk-bounded-executor-sender.py (+ silent attended wrapper): one attended process mints once, writes CCR-2026-0063 custody (platform/workloads/net-kingdom/bounded-executor/audit-sender, cas 0) and appends the sender to the audit-core registry (cas), syncs audit-core-senders and compares by hash, restarts the receiver, probes the live scope without writing an event, and checks sso/nk-bind-executor-audit by hash when it exists. Receipt only; no value printed or stored. - Refuses unless CCR-2026-0063 is approved by its three owners, cluster is railiance01, receiver image and live scope ConfigMap match, the ingress policy exists and the coding-agent boundary covers the custody path. - tests/test_seed_nk_bounded_executor_sender.py: 16 synthetic tests, including the probe run against the real receiver over HTTP. - docs/bounded-executor-receipts.md: 2026-11-04 live state and founder runbook (single-object ingress apply, preflight, seed, verify). - Read-only check: ingress policy not live; other nine policies match. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 523928@bnt-lap001 Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350
2026-11-04 01:39:36 +01:00 · 2026-11-04 01:39:36 +01:00 · 876c3f6647
commit 876c3f6647
parent 06d0029f1e
6 changed files with 855 additions and 1 deletions
--- a/tests/test_seed_nk_bounded_executor_sender.py
+++ b/tests/test_seed_nk_bounded_executor_sender.py
@ -0,0 +1,250 @@
+"""AUDIT-WP-0018-T04. Seed script for the nk-bounded-executor bearer, on fixtures.
+
+Uses a fake OpenBao and obviously fake token strings. No real credential, no
+cluster.
+"""
+import copy
+import importlib.util
+import json
+import sys
+from pathlib import Path
+import subprocess
+import threading
+from wsgiref.simple_server import WSGIRequestHandler, make_server
+
+import pytest
+
+from audit_core.ingestion import IngestionApplication
+from audit_core.senders import SenderRegistry
+from audit_core.sqlite_backend import SQLiteAuditBackend
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts" / "seed-nk-bounded-executor-sender.py"
+SPEC = importlib.util.spec_from_file_location("seed_nk_bounded_executor_sender", SCRIPT)
+SEED = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = SEED
+SPEC.loader.exec_module(SEED)
+
+OTHER = [
+    {"name": "tenant-engine", "tokens": ["fake-tenant-engine-not-a-secret"],
+     "sources": ["tenant-engine"], "tenants": ["*"]},
+    {"name": "operator", "tokens": ["fake-operator-not-a-secret"], "sources": ["*"],
+     "tenants": ["*"], "may_write": False, "may_read": True},
+]
+FAKE = "f" * 64  # matches the minted shape, obviously not a secret
+
+
+class FakeBao:
+    """KV v2 with cas enforcement for the registry and the custody path."""
+
+    def __init__(self, rows=None, custody=None, extra_fields=None):
+        self.kv = {
+            SEED.REGISTRY: [1, dict({"senders.json": json.dumps(rows if rows is not None else OTHER)},
+                                    **(extra_fields or {}))],
+        }
+        if custody is not None:
+            self.kv[SEED.CUSTODY_DATA] = [1, custody]
+        self.writes = []
+
+    def __call__(self, args, *, payload=None):
+        if args[0] == "read":
+            path = args[-1]
+            if path not in self.kv:
+                raise SEED.NotFound()
+            version, data = self.kv[path]
+            return {"data": {"metadata": {"version": version}, "data": copy.deepcopy(data)}}
+        assert args[0] == "write"
+        path = args[-2]
+        current = self.kv.get(path, [0, None])[0]
+        assert payload["options"]["cas"] == current, "cas must match the version read"
+        self.writes.append(path)
+        self.kv[path] = [current + 1, copy.deepcopy(payload["data"])]
+        return {"data": {"version": current + 1}}
+
+
+class _Quiet(WSGIRequestHandler):
+    def log_message(self, *args):
+        pass
+
+
+def run_seed(fake):
+    receipt = {}
+    token = SEED.seed(SEED.scope_entry(), receipt, lambda: None, call=fake)
+    return token, receipt
+
+
+def registry_rows(fake):
+    return json.loads(fake.kv[SEED.REGISTRY][1]["senders.json"])
+
+
+def test_first_run_mints_once_into_custody_and_registry():
+    fake = FakeBao(extra_fields={"unrelated": "kept"})
+    token, receipt = run_seed(fake)
+    assert SEED.TOKEN_SHAPE.fullmatch(token)
+    assert fake.writes == [SEED.CUSTODY_DATA, SEED.REGISTRY]
+    assert fake.kv[SEED.CUSTODY_DATA][1] == {"AUDIT_TOKEN": token, "CUSTODY_REQUEST": "CCR-2026-0063"}
+    rows = registry_rows(fake)
+    assert rows[:2] == OTHER
+    assert rows[2] == SEED.desired(token, SEED.scope_entry())
+    assert fake.kv[SEED.REGISTRY][1]["unrelated"] == "kept"
+    assert receipt["custody"] == "created" and receipt["registry"] == "appended"
+    assert receipt["token_sha256_12"] == SEED.fingerprint(token)
+    assert token not in json.dumps(receipt)
+
+
+def test_rerun_reuses_and_writes_nothing():
+    fake = FakeBao()
+    token, _ = run_seed(fake)
+    fake.writes.clear()
+    again, receipt = run_seed(fake)
+    assert again == token and fake.writes == []
+    assert receipt["custody"] == "reused" and receipt["registry"] == "already_present"
+
+
+def test_resume_after_custody_written_but_registry_not():
+    fake = FakeBao(custody={"AUDIT_TOKEN": FAKE, "CUSTODY_REQUEST": "CCR-2026-0063"})
+    token, receipt = run_seed(fake)
+    assert token == FAKE and fake.writes == [SEED.REGISTRY]
+    assert receipt["custody"] == "reused" and receipt["registry"] == "appended"
+
+
+@pytest.mark.parametrize("custody, code", [
+    ({"AUDIT_TOKEN": FAKE, "CUSTODY_REQUEST": "CCR-2026-0099"}, "custody_provenance_mismatch"),
+    ({"AUDIT_TOKEN": FAKE}, "custody_provenance_mismatch"),
+    ({"AUDIT_TOKEN": "short", "CUSTODY_REQUEST": "CCR-2026-0063"}, "custody_token_shape"),
+])
+def test_foreign_or_malformed_custody_refuses(custody, code):
+    fake = FakeBao(custody=custody)
+    with pytest.raises(SEED.Refused, match=code):
+        run_seed(fake)
+    assert fake.writes == []
+
+
+def test_registered_sender_without_custody_refuses():
+    rows = OTHER + [{"name": "nk-bounded-executor", "tokens": [FAKE]}]
+    fake = FakeBao(rows=rows)
+    with pytest.raises(SEED.Refused, match="registry_sender_without_custody"):
+        run_seed(fake)
+    assert fake.writes == []
+
+
+def test_drifted_registry_entry_refuses():
+    entry = SEED.desired(FAKE, SEED.scope_entry())
+    entry["tenants"] = ["*"]
+    fake = FakeBao(rows=OTHER + [entry],
+                   custody={"AUDIT_TOKEN": FAKE, "CUSTODY_REQUEST": "CCR-2026-0063"})
+    with pytest.raises(SEED.Refused, match="registered_sender_drift"):
+        run_seed(fake)
+    assert fake.writes == []
+
+
+def test_custody_token_already_used_by_another_sender_refuses():
+    rows = OTHER + [{"name": "someone", "tokens": [FAKE]}]
+    fake = FakeBao(rows=rows, custody={"AUDIT_TOKEN": FAKE, "CUSTODY_REQUEST": "CCR-2026-0063"})
+    with pytest.raises(SEED.Refused, match="sender_token_collision"):
+        run_seed(fake)
+
+
+def test_empty_or_duplicate_registry_refuses():
+    with pytest.raises(SEED.Refused, match="registry_must_preserve_existing_senders"):
+        run_seed(FakeBao(rows=[]))
+    with pytest.raises(SEED.Refused, match="registry_identity_invalid_or_duplicate"):
+        run_seed(FakeBao(rows=OTHER + [OTHER[0]]))
+
+
+def test_concurrent_registry_removal_is_caught_on_readback():
+    class Lossy(FakeBao):
+        def __call__(self, args, *, payload=None):
+            result = super().__call__(args, payload=payload)
+            if args[0] == "write" and args[-2] == SEED.REGISTRY:
+                rows = registry_rows(self)[1:]  # someone else dropped tenant-engine
+                self.kv[SEED.REGISTRY][1]["senders.json"] = json.dumps(rows)
+            return result
+
+    with pytest.raises(SEED.Refused, match="registry_readback_mismatch"):
+        run_seed(Lossy())
+
+
+# --- contract checks ----------------------------------------------------------
+
+def approved_ccr():
+    doc, _ = SEED.load_ccr()
+    doc = copy.deepcopy(doc)
+    doc["status"] = "approved"
+    doc["review"]["comments"] = [
+        {"reviewer": owner, "decision": "approved"} for owner in SEED.CCR_APPROVERS
+    ]
+    return doc
+
+
+@pytest.mark.skipif(not SEED.CCR_FILE.exists(), reason="railiance-platform checkout absent")
+def test_ccr_contract_as_drafted_matches_and_refuses_until_approved():
+    pytest.importorskip("yaml")
+    SEED.check_ccr(approved_ccr())
+    proposed, _ = SEED.load_ccr()
+    with pytest.raises(SEED.Refused, match="approved_ccr_required"):
+        SEED.check_ccr(proposed)
+    partial = approved_ccr()
+    partial["review"]["comments"] = partial["review"]["comments"][:2]
+    with pytest.raises(SEED.Refused, match="named_owner_reviews_required"):
+        SEED.check_ccr(partial)
+    moved = approved_ccr()
+    moved["openbao"]["kv_path"] = "platform/workloads/net-kingdom/other"
+    with pytest.raises(SEED.Refused, match="ccr_custody_contract_mismatch"):
+        SEED.check_ccr(moved)
+
+
+def test_wrong_confirmation_refuses_before_any_call(tmp_path, monkeypatch):
+    monkeypatch.setattr(SEED, "bao", lambda *a, **k: pytest.fail("bao called"))
+    receipt = tmp_path / "r.json"
+    rc = SEED.main(["--kubeconfig", "/nonexistent", "--confirm", "yes", "--receipt", str(receipt)])
+    assert rc == 1
+    doc = json.loads(receipt.read_text())
+    assert doc["error"] == "exact_confirmation_required" and doc["credential_values_emitted"] is False
+    # Receipt is never overwritten: a second run needs a fresh path.
+    with pytest.raises(FileExistsError):
+        SEED.main(["--kubeconfig", "/nonexistent", "--confirm", SEED.CONFIRM, "--receipt", str(receipt)])
+
+
+def test_wrapper_is_silent_and_pins_the_confirmation():
+    text = (ROOT / "scripts" / "attended-seed-nk-bounded-executor-sender.sh").read_text()
+    assert ">/dev/null 2>&1" in text and SEED.CONFIRM in text
+
+
+# --- the seeded entry and the live probe, against the real receiver -----------
+
+def test_seeded_entry_yields_the_declared_identity_with_the_live_overlay():
+    entry = SEED.desired(FAKE, SEED.scope_entry())
+    reg = SenderRegistry.from_env({
+        "AUDIT_CORE_SENDERS": json.dumps(OTHER + [entry]),
+        "AUDIT_CORE_SENDERS_SCOPE_PATH": str(ROOT / "deploy" / "senders-scope.json"),
+    })
+    identity = reg.authenticate(f"Bearer {FAKE}")
+    assert identity.name == "nk-bounded-executor"
+    assert identity.sources == frozenset({"net-kingdom.bounded-executor"})
+    assert identity.tenants == frozenset({"platform"})
+    assert identity.may_write and not identity.may_read
+    assert identity.secret_policy == "reject" and identity.is_load_bearing
+
+
+def test_probe_expectations_match_the_receiver(tmp_path):
+    """Runs the probe source itself against the real receiver over HTTP."""
+    entry = SEED.desired(FAKE, SEED.scope_entry())
+    reg = SenderRegistry.from_env({"AUDIT_CORE_SENDERS": json.dumps(OTHER + [entry])})
+    app = IngestionApplication(SQLiteAuditBackend(str(tmp_path / "p.db")), reg)
+    server = make_server("127.0.0.1", 0, app, handler_class=_Quiet)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        source = SEED.RECEIVER_PROBE.replace("127.0.0.1:8080", f"127.0.0.1:{server.server_port}")
+
+        def run(token):
+            proc = subprocess.run([sys.executable, "-c", source], input=token.encode(),
+                                  capture_output=True, check=True, timeout=30)
+            return json.loads(proc.stdout)
+
+        assert run(FAKE) == SEED.PROBE_EXPECTED
+        # An unknown bearer gets 401 everywhere, so the expected pattern proves registration.
+        assert set(run("fake-unknown").values()) == {401}
+    finally:
+        server.shutdown()