add attended custody for audit-core E2 retry

Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02991-be07-7bb3-8b6d-e9701b5621de
2026-08-22 20:49:11 +02:00 · 2026-08-22 20:49:11 +02:00 · 891d90e6c9
commit 891d90e6c9
parent a557208a4a
2 changed files with 517 additions and 0 deletions
--- a/scripts/audit-core-whitehat-e2-credentials-02.py
+++ b/scripts/audit-core-whitehat-e2-credentials-02.py
@ -0,0 +1,466 @@
+#!/usr/bin/env python3
+"""Attended, value-safe credentials for WH-ENG-20260822-AUDIT-E2-02."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import secrets
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+
+ENGAGEMENT = "WH-ENG-20260822-AUDIT-E2-02"
+START = datetime(2026, 8, 22, 19, 15, tzinfo=UTC)
+LATEST_PROJECT = datetime(2026, 8, 22, 19, 18, tzinfo=UTC)
+EXPIRES = datetime(2026, 8, 22, 19, 30, tzinfo=UTC)
+EXPIRES_TEXT = "2026-08-22T19:30:00Z"
+EXPECTED_IMAGE = (
+    "forgejo.coulomb.social/coulomb/audit-core@"
+    "sha256:c2fe39a0185b99be3fc0cb14d2de69772b8e66e20490097c9d11d90cc39719a6"
+)
+AUTHORITY_PATH = "platform/workloads/audit-core/senders"
+TOKEN_PATHS = {
+    "token-a": f"platform/engagements/{ENGAGEMENT}/audit-core/token-a",
+    "token-b": f"platform/engagements/{ENGAGEMENT}/audit-core/token-b",
+}
+IDENTITIES = {
+    "token-a": {
+        "name": "whitehat-e2-a-20260822-02",
+        "tenant": "tenant:trial:whitehat-a-20260822-02",
+    },
+    "token-b": {
+        "name": "whitehat-e2-b-20260822-02",
+        "tenant": "tenant:trial:whitehat-b-20260822-02",
+    },
+}
+POLICY = "external-secrets-whitehat-audit-e2"
+ROLE = "external-secrets-whitehat-audit-e2"
+STORE = "openbao-whitehat-audit-e2"
+SECRET = "whitehat-e2-audit-credentials"
+CONFIRM = f"{ENGAGEMENT}:attended"
+ROOT = Path(__file__).resolve().parents[1]
+POLICY_FILE = ROOT / "openbao" / "policies" / "external-secrets-whitehat-audit-e2.hcl"
+MANIFEST_FILE = ROOT / "manifests" / "whitehat-audit-e2-projection.yaml"
+
+
+class ProcedureError(RuntimeError):
+    pass
+
+
+def utc_now() -> datetime:
+    return datetime.now(UTC)
+
+
+def safe_run(
+    command: list[str],
+    *,
+    label: str,
+    env: dict[str, str] | None = None,
+    input_text: str | None = None,
+    allow_missing: bool = False,
+) -> subprocess.CompletedProcess[str]:
+    completed = subprocess.run(
+        command,
+        text=True,
+        input=input_text,
+        capture_output=True,
+        env=env,
+        check=False,
+    )
+    if completed.returncode != 0 and not allow_missing:
+        # stdout/stderr may contain secret material for Bao and Kubernetes
+        # reads, so deliberately do not include either in the exception.
+        raise ProcedureError(f"{label} failed (exit {completed.returncode})")
+    return completed
+
+
+def secure_unlink(path: Path) -> None:
+    if not path.exists():
+        return
+    size = path.stat().st_size
+    with path.open("r+b", buffering=0) as handle:
+        handle.write(b"\0" * size)
+        handle.flush()
+        os.fsync(handle.fileno())
+    path.unlink()
+
+
+class Operator:
+    def __init__(self, remote: str, token_file: Path) -> None:
+        mode = stat.S_IMODE(token_file.stat().st_mode) if token_file.exists() else None
+        if mode != 0o600:
+            raise ProcedureError(f"OpenBao token file must exist with mode 0600: {token_file}")
+        token = token_file.read_text(encoding="utf-8").splitlines()[0].strip()
+        if not token:
+            raise ProcedureError("OpenBao token file is empty")
+        self.remote = remote
+        self.bao_env = dict(os.environ, BAO_ADDR="https://bao.coulomb.social", BAO_TOKEN=token)
+
+    def bao(self, args: list[str], *, label: str, input_text: str | None = None, allow_missing: bool = False) -> subprocess.CompletedProcess[str]:
+        return safe_run(["bao", *args], label=label, env=self.bao_env, input_text=input_text, allow_missing=allow_missing)
+
+    def kubectl(self, args: list[str], *, label: str, input_text: str | None = None, allow_missing: bool = False) -> subprocess.CompletedProcess[str]:
+        return safe_run(
+            ["ssh", "-o", "BatchMode=yes", self.remote, "kubectl", *args],
+            label=label,
+            input_text=input_text,
+            allow_missing=allow_missing,
+        )
+
+    def registry(self) -> list[dict[str, Any]]:
+        result = self.bao(
+            ["kv", "get", "-field=senders.json", AUTHORITY_PATH],
+            label="read sender registry",
+        )
+        try:
+            payload = json.loads(result.stdout)
+        except json.JSONDecodeError as exc:
+            raise ProcedureError("sender registry is not valid JSON") from exc
+        if not isinstance(payload, list) or not payload:
+            raise ProcedureError("sender registry must be a non-empty list")
+        return payload
+
+    def bao_exists(self, args: list[str], *, label: str) -> bool:
+        result = self.bao(args, label=label, allow_missing=True)
+        return result.returncode == 0
+
+    def kubectl_exists(self, args: list[str], *, label: str) -> bool:
+        result = self.kubectl(args, label=label, allow_missing=True)
+        return result.returncode == 0 and bool(result.stdout.strip())
+
+    def write_registry(self, registry: list[dict[str, Any]]) -> None:
+        with tempfile.TemporaryDirectory(prefix="railiance-audit-registry-") as temp:
+            root = Path(temp)
+            os.chmod(root, 0o700)
+            document = root / "senders.json"
+            document.write_text(json.dumps(registry, separators=(",", ":")), encoding="utf-8")
+            os.chmod(document, 0o600)
+            try:
+                self.bao(
+                    ["kv", "patch", AUTHORITY_PATH, f"senders.json=@{document}"],
+                    label="write sender registry",
+                )
+            finally:
+                secure_unlink(document)
+
+
+def temporary_names() -> set[str]:
+    return {entry["name"] for entry in IDENTITIES.values()}
+
+
+def add_temporary_identities(registry: list[dict[str, Any]], tokens: dict[str, str]) -> list[dict[str, Any]]:
+    present = {str(entry.get("name")) for entry in registry if isinstance(entry, dict)}
+    overlap = present & temporary_names()
+    if overlap:
+        raise ProcedureError(f"temporary identities already exist: {sorted(overlap)}")
+    updated = list(registry)
+    for handle, identity in IDENTITIES.items():
+        updated.append({
+            "name": identity["name"],
+            "tokens": [tokens[handle]],
+            "sources": ["whitehat-security"],
+            "tenants": [identity["tenant"]],
+            "may_write": True,
+            "may_read": True,
+            "expires_at": EXPIRES_TEXT,
+        })
+    return updated
+
+
+def remove_temporary_identities(registry: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int]:
+    names = temporary_names()
+    updated = [entry for entry in registry if not isinstance(entry, dict) or entry.get("name") not in names]
+    return updated, len(registry) - len(updated)
+
+
+def projection_manifest() -> str:
+    rendered = MANIFEST_FILE.read_text(encoding="utf-8")
+    if "data:" not in rendered or "token-a" not in rendered or "token-b" not in rendered:
+        raise ProcedureError("projection manifest is incomplete")
+    return rendered
+
+
+def assert_confirm(value: str | None) -> None:
+    if value != CONFIRM:
+        raise ProcedureError(f"live command requires --confirm {CONFIRM}")
+
+
+def time_state(now: datetime) -> str:
+    if now < START:
+        return "before-window"
+    if now > LATEST_PROJECT:
+        return "projection-cutoff-passed"
+    return "projection-window-open"
+
+
+def live_preflight(operator: Operator) -> dict[str, Any]:
+    status_result = operator.bao(["status", "-format=json"], label="read OpenBao status")
+    try:
+        bao_status = json.loads(status_result.stdout)
+    except json.JSONDecodeError as exc:
+        raise ProcedureError("OpenBao status is not valid JSON") from exc
+    if not bao_status.get("initialized") or bao_status.get("sealed"):
+        raise ProcedureError("OpenBao is not initialized and unsealed")
+    image = operator.kubectl(
+        ["-n", "audit-core", "get", "deploy", "audit-core", "-o", "jsonpath={.spec.template.spec.containers[0].image}"],
+        label="read audit-core image",
+    ).stdout.strip()
+    if image != EXPECTED_IMAGE:
+        raise ProcedureError("audit-core is not on the engagement-approved image digest")
+    ready = operator.kubectl(
+        ["-n", "audit-core", "get", "deploy", "audit-core", "-o", "jsonpath={.status.readyReplicas}/{.status.replicas}"],
+        label="read audit-core readiness",
+    ).stdout.strip()
+    if ready != "1/1":
+        raise ProcedureError(f"audit-core is not 1/1 Ready (observed {ready or 'unknown'})")
+    operator.kubectl(["get", "ns", "whitehat", "-o", "name"], label="verify whitehat namespace")
+    operator.kubectl(["-n", "external-secrets", "get", "sa", "external-secrets", "-o", "name"], label="verify ESO service account")
+    registry = operator.registry()
+    existing = sorted(
+        temporary_names()
+        & {str(item.get("name")) for item in registry if isinstance(item, dict)}
+    )
+    exact_paths_present = sorted(
+        handle
+        for handle, path in TOKEN_PATHS.items()
+        if operator.bao_exists(
+            ["kv", "metadata", "get", "-format=json", path],
+            label=f"check exact {handle} metadata",
+        )
+    )
+    projection_resources = []
+    for label, args in (
+        (f"ClusterSecretStore/{STORE}", ["get", "clustersecretstore", STORE, "-o", "name"]),
+        (f"ExternalSecret/whitehat/{SECRET}", ["-n", "whitehat", "get", "externalsecret", SECRET, "-o", "name"]),
+        (f"Secret/whitehat/{SECRET}", ["-n", "whitehat", "get", "secret", SECRET, "-o", "name"]),
+    ):
+        if operator.kubectl_exists(args, label=f"check {label}"):
+            projection_resources.append(label)
+    return {
+        "engagement_id": ENGAGEMENT,
+        "time_state": time_state(utc_now()),
+        "target_image_matches": True,
+        "target_ready": ready,
+        "openbao_initialized": True,
+        "openbao_sealed": False,
+        "temporary_identities_present": existing,
+        "exact_token_paths_present": exact_paths_present,
+        "projection_resources_present": projection_resources,
+        "secret_values_observed": False,
+    }
+
+
+def write_exact_tokens(operator: Operator, tokens: dict[str, str]) -> None:
+    with tempfile.TemporaryDirectory(prefix="railiance-whitehat-e2-") as temp:
+        root = Path(temp)
+        os.chmod(root, 0o700)
+        files: list[Path] = []
+        try:
+            for handle, value in tokens.items():
+                path = root / handle
+                path.write_text(value, encoding="utf-8")
+                os.chmod(path, 0o600)
+                files.append(path)
+                operator.bao(
+                    [
+                        "kv", "put", TOKEN_PATHS[handle], f"token=@{path}",
+                        f"engagement_id={ENGAGEMENT}", f"expires_at={EXPIRES_TEXT}",
+                    ],
+                    label=f"write exact {handle} path",
+                )
+        finally:
+            for path in files:
+                secure_unlink(path)
+
+
+def configure_projection(operator: Operator) -> None:
+    operator.bao(["policy", "write", POLICY, str(POLICY_FILE)], label="write exact ESO policy")
+    operator.bao(
+        [
+            "write", f"auth/kubernetes/role/{ROLE}",
+            "bound_service_account_names=external-secrets",
+            "bound_service_account_namespaces=external-secrets",
+            f"policies={POLICY}", "ttl=5m", "max_ttl=15m",
+        ],
+        label="write exact ESO Kubernetes role",
+    )
+    operator.kubectl(["apply", "-f", "-"], label="apply projection manifest", input_text=projection_manifest())
+    operator.kubectl(["wait", "--for=condition=Ready", f"clustersecretstore/{STORE}", "--timeout=90s"], label="wait for exact store")
+    operator.kubectl(["-n", "whitehat", "wait", "--for=condition=Ready", f"externalsecret/{SECRET}", "--timeout=90s"], label="wait for exact ExternalSecret")
+    keys = operator.kubectl(
+        ["-n", "whitehat", "get", "secret", SECRET, "-o", "go-template={{range $k, $_ := .data}}{{$k}}{{\"\\n\"}}{{end}}"],
+        label="verify projected key names",
+    ).stdout.splitlines()
+    if sorted(keys) != ["token-a", "token-b"]:
+        raise ProcedureError(f"projected Secret has unexpected key names: {sorted(keys)}")
+
+
+def force_sender_sync_and_restart(operator: Operator) -> None:
+    before = operator.kubectl(
+        ["-n", "audit-core", "get", "secret", "audit-core-senders", "-o", "jsonpath={.metadata.resourceVersion}"],
+        label="read sender Secret resource version",
+    ).stdout.strip()
+    stamp = str(int(time.time()))
+    operator.kubectl(
+        ["-n", "audit-core", "annotate", "externalsecret", "audit-core-senders", f"force-sync={stamp}", "--overwrite"],
+        label="force sender registry sync",
+    )
+    deadline = time.monotonic() + 120
+    while time.monotonic() < deadline:
+        current = operator.kubectl(
+            ["-n", "audit-core", "get", "secret", "audit-core-senders", "-o", "jsonpath={.metadata.resourceVersion}"],
+            label="poll sender Secret resource version",
+        ).stdout.strip()
+        if current and current != before:
+            break
+        time.sleep(2)
+    else:
+        raise ProcedureError("audit-core sender Secret did not refresh within 120s")
+    operator.kubectl(["-n", "audit-core", "rollout", "restart", "deploy/audit-core"], label="restart audit-core registry reader")
+    operator.kubectl(["-n", "audit-core", "rollout", "status", "deploy/audit-core", "--timeout=120s"], label="wait for audit-core rollout")
+
+
+def delete_projection(operator: Operator) -> None:
+    operator.kubectl(["-n", "whitehat", "delete", "externalsecret", SECRET, "--ignore-not-found"], label="delete exact ExternalSecret")
+    operator.kubectl(["-n", "whitehat", "delete", "secret", SECRET, "--ignore-not-found"], label="delete mounted Secret")
+    operator.kubectl(["delete", "clustersecretstore", STORE, "--ignore-not-found"], label="delete exact store")
+    for handle, path in TOKEN_PATHS.items():
+        if operator.bao_exists(["kv", "metadata", "get", "-format=json", path], label=f"check exact {handle} metadata before delete"):
+            operator.bao(["kv", "metadata", "delete", path], label=f"delete exact {handle} KV path")
+    if operator.bao_exists(["read", f"auth/kubernetes/role/{ROLE}"], label="check exact ESO role before delete"):
+        operator.bao(["delete", f"auth/kubernetes/role/{ROLE}"], label="delete exact ESO role")
+    if operator.bao_exists(["policy", "read", POLICY], label="check exact ESO policy before delete"):
+        operator.bao(["policy", "delete", POLICY], label="delete exact ESO policy")
+
+
+def verify_cleanup_absent(operator: Operator) -> None:
+    # Recheck both control planes after deletion. A connectivity failure is
+    # caught by the status/namespace reads before absence is interpreted.
+    operator.bao(["status", "-format=json"], label="verify OpenBao after cleanup")
+    operator.kubectl(["get", "ns", "whitehat", "-o", "name"], label="verify Kubernetes after cleanup")
+    for handle, path in TOKEN_PATHS.items():
+        if operator.bao_exists(["kv", "metadata", "get", "-format=json", path], label=f"verify exact {handle} path absence"):
+            raise ProcedureError(f"exact {handle} KV path remains after cleanup")
+    if operator.bao_exists(["read", f"auth/kubernetes/role/{ROLE}"], label="verify exact ESO role absence"):
+        raise ProcedureError("exact ESO role remains after cleanup")
+    if operator.bao_exists(["policy", "read", POLICY], label="verify exact ESO policy absence"):
+        raise ProcedureError("exact ESO policy remains after cleanup")
+    resources = (
+        ["-n", "whitehat", "get", "externalsecret", SECRET, "-o", "name"],
+        ["-n", "whitehat", "get", "secret", SECRET, "-o", "name"],
+        ["get", "clustersecretstore", STORE, "-o", "name"],
+    )
+    if any(operator.kubectl_exists(args, label="verify projection resource absence") for args in resources):
+        raise ProcedureError("a projection resource remains after cleanup")
+
+
+def cleanup(operator: Operator) -> dict[str, Any]:
+    registry = operator.registry()
+    updated, removed = remove_temporary_identities(registry)
+    if removed:
+        operator.write_registry(updated)
+        force_sender_sync_and_restart(operator)
+    delete_projection(operator)
+    remaining = sorted(
+        temporary_names()
+        & {str(item.get("name")) for item in operator.registry() if isinstance(item, dict)}
+    )
+    if remaining:
+        raise ProcedureError(f"temporary identities remain after cleanup: {remaining}")
+    verify_cleanup_absent(operator)
+    ready = operator.kubectl(
+        ["-n", "audit-core", "get", "deploy", "audit-core", "-o", "jsonpath={.status.readyReplicas}/{.status.replicas}"],
+        label="verify audit-core readiness after cleanup",
+    ).stdout.strip()
+    if ready != "1/1":
+        raise ProcedureError(f"audit-core is not 1/1 Ready after cleanup (observed {ready or 'unknown'})")
+    return {
+        "engagement_id": ENGAGEMENT,
+        "cleanup_completed_at": utc_now().replace(microsecond=0).isoformat().replace("+00:00", "Z"),
+        "temporary_identities_removed": removed,
+        "mounted_secret_absent": True,
+        "exact_kv_paths_deleted": sorted(TOKEN_PATHS.values()),
+        "target_ready": True,
+        "secret_values_observed": False,
+    }
+
+
+def project(operator: Operator) -> dict[str, Any]:
+    now = utc_now()
+    if not START <= now <= LATEST_PROJECT:
+        raise ProcedureError(
+            f"projection permitted only {START.isoformat()} through {LATEST_PROJECT.isoformat()}; "
+            f"current time is {now.isoformat()}"
+        )
+    preflight = live_preflight(operator)
+    if (
+        preflight["temporary_identities_present"]
+        or preflight["exact_token_paths_present"]
+        or preflight["projection_resources_present"]
+    ):
+        raise ProcedureError("cleanup is required before a new projection")
+    tokens = {"token-a": secrets.token_urlsafe(48), "token-b": secrets.token_urlsafe(48)}
+    if tokens["token-a"] == tokens["token-b"]:
+        raise ProcedureError("token generator returned duplicate values")
+    registry_written = False
+    try:
+        write_exact_tokens(operator, tokens)
+        operator.write_registry(add_temporary_identities(operator.registry(), tokens))
+        registry_written = True
+        configure_projection(operator)
+        force_sender_sync_and_restart(operator)
+    except Exception:
+        if registry_written:
+            cleanup(operator)
+        else:
+            delete_projection(operator)
+        raise
+    finally:
+        tokens.clear()
+    return {
+        "engagement_id": ENGAGEMENT,
+        "projected_at": utc_now().replace(microsecond=0).isoformat().replace("+00:00", "Z"),
+        "expires_at": EXPIRES_TEXT,
+        "identities": sorted(temporary_names()),
+        "mounted_secret": f"whitehat/{SECRET}",
+        "mounted_keys": ["token-a", "token-b"],
+        "target_image_matches": True,
+        "target_ready": True,
+        "secret_values_observed": False,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("command", choices=["preflight", "project", "cleanup"])
+    parser.add_argument("--remote", default="railiance01")
+    parser.add_argument("--token-file", type=Path, default=Path.home() / ".local/openbao/platform-admin.token")
+    parser.add_argument("--confirm")
+    args = parser.parse_args()
+    try:
+        operator = Operator(args.remote, args.token_file)
+        if args.command == "preflight":
+            result = live_preflight(operator)
+        elif args.command == "project":
+            assert_confirm(args.confirm)
+            result = project(operator)
+        else:
+            assert_confirm(args.confirm)
+            result = cleanup(operator)
+    except (OSError, IndexError, ProcedureError) as exc:
+        print(f"credential procedure failed: {exc}", file=sys.stderr)
+        return 1
+    print(json.dumps(result, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/tests/test_audit_core_whitehat_e2_credentials_02.py
+++ b/tests/test_audit_core_whitehat_e2_credentials_02.py
@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import importlib.util
+import unittest
+from datetime import UTC, datetime
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "audit_core_whitehat_e2_credentials",
+    ROOT / "scripts" / "audit-core-whitehat-e2-credentials-02.py",
+)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+class CredentialProcedureTests(unittest.TestCase):
+    def test_add_and_remove_only_named_temporary_identities(self) -> None:
+        original = [{"name": "user-engine", "tokens": ["existing"]}]
+        updated = module.add_temporary_identities(original, {"token-a": "a", "token-b": "b"})
+        self.assertEqual(3, len(updated))
+        cleaned, removed = module.remove_temporary_identities(updated)
+        self.assertEqual(2, removed)
+        self.assertEqual(original, cleaned)
+
+    def test_temporary_identities_are_exactly_scoped_and_expiring(self) -> None:
+        updated = module.add_temporary_identities(
+            [{"name": "existing"}], {"token-a": "a", "token-b": "b"}
+        )
+        temporary = updated[1:]
+        self.assertEqual({"whitehat-security"}, {item["sources"][0] for item in temporary})
+        self.assertTrue(all(item["may_read"] and item["may_write"] for item in temporary))
+        self.assertTrue(all(item["expires_at"] == module.EXPIRES_TEXT for item in temporary))
+        self.assertEqual(2, len({item["tenants"][0] for item in temporary}))
+
+    def test_projection_window_is_fail_closed(self) -> None:
+        self.assertEqual("before-window", module.time_state(datetime(2026, 8, 22, 19, 14, tzinfo=UTC)))
+        self.assertEqual("projection-window-open", module.time_state(datetime(2026, 8, 22, 19, 16, tzinfo=UTC)))
+        self.assertEqual("projection-cutoff-passed", module.time_state(datetime(2026, 8, 22, 19, 19, tzinfo=UTC)))
+
+    def test_manifest_projects_only_two_exact_keys(self) -> None:
+        manifest = module.projection_manifest()
+        self.assertEqual(2, manifest.count("secretKey: token-"))
+        self.assertNotIn("workloads/audit-core/senders", manifest)
+        self.assertIn("conditions:\n    - namespaces:\n        - whitehat", manifest)
+
+
+if __name__ == "__main__":
+    unittest.main()