Implement attended root custody admission and crash-safe publisher

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29

scripts/root-grant-attended.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+set -euo pipefail
+exec python3 "$(dirname "$0")/root_grant_attended.py" "$@" >/dev/null 2>&1

scripts/root_grant_attended.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""Silent attended OpenBao admission/publisher; never emits server responses."""
+import argparse
+import datetime as dt
+import fcntl
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+
+from root_grant_publish import MAX_BYTES, RECORD_ID, prepare_payload
+from openbao_platform_admin_check import require_attended
+
+ROOT = Path(__file__).resolve().parents[1]
+PATH = "platform/data/custody/net-kingdom/platform-root"
+META = "platform/metadata/custody/net-kingdom/platform-root"
+ROLE_PATH = "auth/kubernetes/role/user-engine-root-custody-reader"
+POLICY = "user-engine-root-custody-reader"
+ROLE = {"bound_service_account_names": ["user-engine"],
+        "bound_service_account_namespaces": ["user-engine"], "audience": "openbao",
+        "token_policies": [POLICY], "token_no_default_policy": True,
+        "token_ttl": 60, "token_max_ttl": 60, "token_explicit_max_ttl": 60}
+
+
+class Refused(Exception):
+    pass
+
+
+def digest(document):
+    return hashlib.sha256(json.dumps(document, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest()
+
+
+def atomic_json(path, document):
+    path = Path(path)
+    if path.is_symlink():
+        raise Refused("unsafe_state_path")
+    with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as out:
+        temporary = Path(out.name)
+        os.chmod(temporary, 0o600)
+        out.write(json.dumps(document, sort_keys=True).encode())
+        out.flush()
+        os.fsync(out.fileno())
+    try:
+        os.replace(temporary, path)
+        fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+class Bao:
+    def call(self, command, payload=None, optional=False):
+        result = subprocess.run(["bao", *command], input=None if payload is None else json.dumps(payload).encode(),
+                                capture_output=True, timeout=15)
+        if result.returncode:
+            # Distinguish a documented missing-path message from auth/outage.
+            if optional and result.returncode == 2 and result.stderr.strip() == ("No value found at " + command[-1]).encode():
+                return None
+            raise Refused("openbao_request_failed")
+        if len(result.stdout) > MAX_BYTES * 2:
+            raise Refused("oversize_owner_response")
+        return json.loads(result.stdout) if result.stdout.strip() else {}
+
+    def read(self, path, optional=False):
+        response = self.call(["read", "-format=json", path], optional=optional)
+        return None if response is None else response["data"]
+
+    def write(self, path, payload):
+        return self.call(["write", "-format=json", path, "-"], payload)
+
+
+def check_baseline(actual, state, validator):
+    if state is None:
+        if actual is not None:
+            raise Refused("controlled_baseline_import_required")
+        return
+    previous = state["document"]
+    if validator.validate(previous) or previous["record_id"] != RECORD_ID or state["digest"] != digest(previous):
+        raise Refused("invalid_retained_baseline")
+    if actual is None or validator.validate(actual["data"]["document"]):
+        raise Refused("missing_or_invalid_source")
+    document = actual["data"]["document"]
+    if digest(document) != state["digest"] or actual["metadata"]["version"] != state["kv_version"]:
+        raise Refused("source_differs_from_retained_baseline")
+
+
+def retain_history(state_path, receipt):
+    history = state_path.parent / "history"
+    history.mkdir(mode=0o700, exist_ok=True)
+    target = history / (str(receipt["revision"]) + "-" + receipt["digest"] + ".json")
+    if target.exists():
+        if json.loads(target.read_text()) != receipt:
+            raise Refused("history_receipt_differs")
+        return
+    with target.open("x") as out:
+        os.chmod(target, 0o600)
+        json.dump(receipt, out, sort_keys=True)
+        out.flush()
+        os.fsync(out.fileno())
+    fd = os.open(history, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def publish(bao, candidate, state_path, validator, provenance):
+    state_path = Path(state_path)
+    if not isinstance(provenance, str) or not provenance.strip() or len(provenance) > 300 or validator.SECRET_VALUE.search(provenance):
+        raise Refused("nonsecret_human_provenance_required")
+    if state_path.is_symlink():
+        raise Refused("unsafe_state_path")
+    state = json.loads(state_path.read_text()) if state_path.exists() else None
+    actual = bao.read(PATH, optional=True)
+    # Pending state is fsynced before network mutation. Recover only exact readback;
+    # ambiguous network failures are never retried or cleared automatically.
+    if state is not None and state.get("pending"):
+        pending = state["pending"]
+        if actual is None or digest(actual["data"]["document"]) != pending["digest"] or actual["metadata"]["version"] != pending["expected_kv_version"]:
+            raise Refused("pending_publication_requires_owner_review")
+        if validator.validate(pending["document"]) or pending["document"]["record_id"] != RECORD_ID:
+            raise Refused("invalid_pending_document")
+        state = {key: value for key, value in pending.items() if key != "expected_kv_version"}
+        retain_history(state_path, state)
+        atomic_json(state_path, state)
+    check_baseline(actual, state, validator)
+    previous = None if state is None else state["document"]
+    version = 0 if actual is None else actual["metadata"]["version"]
+    if previous is not None and digest(candidate) == state["digest"]:
+        # Exact replay is an idempotent readback, not a new custody/KV revision.
+        return {**{key: value for key, value in state.items() if key != "document"}, "unchanged": True}
+    payload = prepare_payload(candidate, previous, version, validator)
+    now = dt.datetime.now(dt.timezone.utc)
+    times = [candidate["updated_at"]]
+    for entry in candidate["entries"]:
+        times.append(entry["granted"]["at"])
+        if entry.get("revoked"):
+            times.append(entry["revoked"]["at"])
+    if any(validator.parse_time(value) > now for value in times):
+        raise Refused("future_custody_action")
+    receipt = {"record_id": RECORD_ID, "revision": candidate["revision"], "digest": digest(candidate),
+               "kv_version": version + 1, "document": candidate, "provenance_ref": provenance,
+               "recorded_at": dt.datetime.now(dt.timezone.utc).isoformat()}
+    pending = dict(receipt, expected_kv_version=version + 1)
+    atomic_json(state_path, {**(state or {}), "pending": pending})
+    bao.write(PATH, payload)
+    returned = bao.read(PATH)
+    if returned["metadata"]["version"] != version + 1 or digest(returned["data"]["document"]) != receipt["digest"]:
+        raise Refused("publication_readback_failed")
+    # Independent immutable receipts precede advancing local state. They must be
+    # backed up off-host; restoring only this directory is not independent proof.
+    retain_history(state_path, receipt)
+    atomic_json(state_path, receipt)
+    return {key: value for key, value in receipt.items() if key != "document"}
+
+
+def provision(bao):
+    # Require existing Kubernetes auth; no auth mount or cluster reviewer mutation.
+    if bao.read("auth/kubernetes/config") is None:
+        raise Refused("kubernetes_auth_unavailable")
+    audit = bao.read("sys/audit")
+    if not audit:
+        raise Refused("audit_required")
+    policy = (ROOT / "openbao/root-grant/reader.hcl").read_text()
+    existing = bao.read("sys/policies/acl/" + POLICY, optional=True)
+    if existing is not None and existing.get("policy", existing.get("rules", "")).strip() != policy.strip():
+        raise Refused("existing_reader_policy_differs")
+    role = bao.read(ROLE_PATH, optional=True)
+    if role is not None and any(role.get(key) != value for key, value in ROLE.items()):
+        raise Refused("existing_reader_role_differs")
+    if existing is None:
+        bao.write("sys/policies/acl/" + POLICY, {"policy": policy})
+    if role is None:
+        bao.write(ROLE_PATH, ROLE)
+    fresh = bao.read(ROLE_PATH)
+    if any(fresh.get(key) != value for key, value in ROLE.items()):
+        raise Refused("role_readback_failed")
+    fresh_policy = bao.read("sys/policies/acl/" + POLICY)
+    if fresh_policy.get("policy", fresh_policy.get("rules", "")).strip() != policy.strip():
+        raise Refused("policy_readback_failed")
+    # Require CAS on this key without altering retention or destroying versions.
+    metadata = bao.read(META, optional=True)
+    if not metadata or not metadata.get("cas_required"):
+        bao.write(META, {"cas_required": True})
+    if bao.read(META).get("cas_required") is not True:
+        raise Refused("cas_required_readback_failed")
+    return {"reader_role": ROLE_PATH, "reader_namespace": "user-engine", "reader_service_account": "user-engine",
+            "audience": "openbao", "record_id": RECORD_ID, "cas_required": True, "audit_enabled": True}
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("action", choices=["provision", "publish", "restore-check"])
+    parser.add_argument("--receipt", type=Path, required=True)
+    parser.add_argument("--state", type=Path)
+    parser.add_argument("--candidate", type=Path)
+    parser.add_argument("--net-kingdom", type=Path)
+    parser.add_argument("--provenance-ref")
+    parser.add_argument("--validator-digest")
+    parser.add_argument("--schema-digest")
+    args = parser.parse_args()
+    try:
+        if args.receipt.exists() or args.receipt.is_symlink():
+            raise Refused("unique_run_receipt_required")
+        require_attended()
+        if os.getenv("BAO_ADDR") != "http://127.0.0.1:18200":
+            raise Refused("approved_operator_tunnel_required")
+        bao = Bao()
+        if args.action == "provision":
+            receipt = provision(bao)
+        else:
+            if not args.state or not args.net_kingdom or not args.validator_digest or not args.schema_digest:
+                raise Refused("pinned_source_and_state_required")
+            source = args.net_kingdom / "tools/custody-record/validate.py"
+            schema = args.net_kingdom / "canon/schemas/custody-record_v0.1.schema.json"
+            for path, expected in [(source, args.validator_digest), (schema, args.schema_digest)]:
+                if hashlib.sha256(path.read_bytes()).hexdigest() != expected:
+                    raise Refused("canon_pin_mismatch")
+            spec = importlib.util.spec_from_file_location("custody_validator", source)
+            validator = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(validator)
+            if not args.state.parent.is_dir():
+                raise Refused("owner_state_directory_required")
+            # Serialize this publisher, including reads and pending-journal recovery.
+            with (args.state.parent / "publisher.lock").open("a") as lock:
+                fcntl.flock(lock, fcntl.LOCK_EX)
+                if args.action == "restore-check":
+                    if not args.state.exists():
+                        raise Refused("retained_baseline_required")
+                    state = json.loads(args.state.read_text())
+                    if state.get("pending"):
+                        raise Refused("pending_publication_requires_owner_review")
+                    check_baseline(bao.read(PATH, optional=True), state, validator)
+                    receipt = {"record_id": RECORD_ID, "baseline_matches": True, "revision": state["document"]["revision"]}
+                else:
+                    if not args.candidate or not args.provenance_ref:
+                        raise Refused("candidate_and_human_provenance_required")
+                    candidate = json.loads(args.candidate.read_bytes()[:MAX_BYTES + 1])
+                    receipt = publish(bao, candidate, args.state, validator, args.provenance_ref)
+        atomic_json(args.receipt, {"ok": True, **receipt})
+        return 0
+    except Exception:
+        # CLI/process/validation errors may embed secrets. Never emit them.
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())