Add attended Whitehat E2 credential projection

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

Makefile

@@ -466,5 +466,14 @@ help: ## Show this help
 	  /^[a-zA-Z_-]+:.*?##/ { printf "  \033[36m%-22s\033[0m %s\n", $$1, $$2 } \
 	  /^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) }' $(MAKEFILE_LIST)
 
-.PHONY: apps-pg-isolation-probe
+audit-whitehat-e2-preflight: ## Value-safe live preflight for the approved audit-core E2 window
+	python3 scripts/audit-core-whitehat-e2-credentials.py preflight
+
+audit-whitehat-e2-project: ## Project two exact mount-only credentials during the approved window
+	python3 scripts/audit-core-whitehat-e2-credentials.py project --confirm WH-ENG-20260822-AUDIT-E2-01:attended
+
+audit-whitehat-e2-cleanup: ## Remove the exact engagement credentials and reload audit-core
+	python3 scripts/audit-core-whitehat-e2-credentials.py cleanup --confirm WH-ENG-20260822-AUDIT-E2-01:attended
+
+.PHONY: audit-whitehat-e2-preflight audit-whitehat-e2-project audit-whitehat-e2-cleanup apps-pg-isolation-probe
 .PHONY: db-deploy db-status db-shell db-logs forgejo-db-deploy forgejo-db-status forgejo-db-shell apps-pg-deploy apps-pg-status apps-pg-shell apps-pg-logs net-kingdom-pg-inter-hub-networkpolicy-deploy pg-deploy pg-status pg-pgpool-check valkey-deploy valkey-status openbao-repo openbao-dry-run openbao-overlay-apply openbao-verify-login-overlay openbao-deploy openbao-status openbao-verify openbao-verify-post-unseal openbao-configure-initial openbao-configure-ssh openbao-verify-ssh openbao-verify-authenticated openbao-configure-external-secrets-issue-core openbao-configure-external-secrets-activity-core openbao-configure-external-secrets-forgejo openbao-validate-restore-evidence openbao-validate-emergency-evidence credential-grants-validate credential-change-applier-dry-run credential-change-applier-apply-plan credential-change-applier-apply credential-change-runbook credential-change-record-evidence credential-change-lifecycle-plan credential-change-lifecycle-event credential-change-import-inventory agent-high-risk-boundary-check openbao-credential-change-appliers-dry-run openbao-configure-credential-change-appliers openbao-token-grants-dry-run openbao-configure-token-grants openbao-verify-token-grants-dry-run openbao-verify-token-grants openbao-verify-token-grants-smoke credential-helper-dry-run credential-tests credential-exec-ops-warden-smoke argocd-bootstrap-dry-run argocd-bootstrap-deploy argocd-repo-apply argocd-status backup forgejo-backup forgejo-backup-dry-run forgejo-backup-status forgejo-package-prune forgejo-package-prune-dry-run reef-deployables apps-pg-evidence consumption-preflight help

docs/audit-core-whitehat-e2-credential-projection.md

@@ -0,0 +1,60 @@
+# Audit-core Whitehat E2 credential projection
+
+This is the attended custody procedure for
+`WH-ENG-20260822-AUDIT-E2-01`. It creates two ordinary, tenant-scoped audit-core
+sender identities and projects only `token-a` and `token-b` into the Whitehat
+runner mount. It never returns credential values.
+
+## Safety gates
+
+- `project` is hard-gated to 18:00Z–18:03Z on 2026-08-22. This leaves the
+  remainder of the approved 18:00Z–18:15Z window for the bounded probe and
+  cleanup.
+- Both audit-core identities carry `expires_at: 2026-08-22T18:15:00Z`, checked
+  by the receiver on every authentication.
+- OpenBao stores the two values at separate engagement paths. The ephemeral ESO
+  policy can read only those paths; it cannot read the full sender registry.
+- The generated Kubernetes Secret has exactly `token-a` and `token-b` and is
+  consumed only through the read-only volume declared by Whitehat.
+- Cleanup removes identities by exact name, preserving every unrelated sender.
+  It deletes the two KV paths, ESO role/policy, ClusterSecretStore,
+  ExternalSecret, and mounted Secret, then reloads audit-core if required.
+
+## Attended sequence
+
+Keep the operator terminal attached from projection through cleanup. The
+platform-admin token file must remain outside Git with mode `0600`.
+
+```bash
+python3 scripts/audit-core-whitehat-e2-credentials.py preflight
+
+# Run only after 18:00Z and no later than 18:03Z.
+python3 scripts/audit-core-whitehat-e2-credentials.py project \
+  --confirm WH-ENG-20260822-AUDIT-E2-01:attended
+```
+
+After projection succeeds, Whitehat may apply its approved runner contract and
+execute only the invocation recorded in
+`whitehat-security/engagements/runtime/WH-ENG-20260822-AUDIT-E2-01-invocation.md`.
+Stop immediately on an engagement abort threshold.
+
+Whether the probe succeeds, fails, or aborts, clean up before 18:15Z:
+
+```bash
+python3 scripts/audit-core-whitehat-e2-credentials.py cleanup \
+  --confirm WH-ENG-20260822-AUDIT-E2-01:attended
+```
+
+The command output is a value-safe JSON receipt. Record that receipt, the
+audit-core rollout/readiness result, the Whitehat runner result, and the
+absence of the mounted Secret. Never record a bearer, Secret data, sender
+registry payload, or OpenBao response body.
+
+## Abort path
+
+If `project` fails after changing custody, it invokes the same exact cleanup
+automatically. If the process itself is interrupted, run `cleanup` immediately;
+it is idempotent and is allowed before or after the projection window. If
+audit-core does not return 1/1 Ready after registry cleanup, do not re-add the
+temporary identities—keep the Whitehat Secret absent and escalate the target
+recovery separately.

manifests/whitehat-audit-e2-projection.yaml

@@ -0,0 +1,54 @@
+# Ephemeral, exact-path projection for WH-ENG-20260822-AUDIT-E2-01.
+# Apply only through scripts/audit-core-whitehat-e2-credentials.py during the
+# approved window. No value is present in this file.
+---
+apiVersion: external-secrets.io/v1
+kind: ClusterSecretStore
+metadata:
+  name: openbao-whitehat-audit-e2
+  labels:
+    app.kubernetes.io/part-of: railiance-platform
+    railiance-platform/engagement: WH-ENG-20260822-AUDIT-E2-01
+spec:
+  provider:
+    vault:
+      server: http://openbao.openbao.svc:8200
+      path: platform
+      version: v2
+      auth:
+        kubernetes:
+          mountPath: kubernetes
+          role: external-secrets-whitehat-audit-e2
+          serviceAccountRef:
+            name: external-secrets
+            namespace: external-secrets
+  conditions:
+    - namespaces:
+        - whitehat
+---
+apiVersion: external-secrets.io/v1
+kind: ExternalSecret
+metadata:
+  name: whitehat-e2-audit-credentials
+  namespace: whitehat
+  labels:
+    app.kubernetes.io/part-of: railiance-platform
+    railiance-platform/engagement: WH-ENG-20260822-AUDIT-E2-01
+spec:
+  refreshInterval: 1m
+  secretStoreRef:
+    kind: ClusterSecretStore
+    name: openbao-whitehat-audit-e2
+  target:
+    name: whitehat-e2-audit-credentials
+    creationPolicy: Owner
+    deletionPolicy: Delete
+  data:
+    - secretKey: token-a
+      remoteRef:
+        key: engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-a
+        property: token
+    - secretKey: token-b
+      remoteRef:
+        key: engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-b
+        property: token

openbao/policies/external-secrets-whitehat-audit-e2.hcl

@@ -0,0 +1,20 @@
+# Ephemeral read policy for WH-ENG-20260822-AUDIT-E2-01.
+#
+# The External Secrets Operator may read exactly two per-engagement values.
+# It cannot read the audit-core sender registry or any other engagement path.
+
+path "platform/data/engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-a" {
+  capabilities = ["read"]
+}
+
+path "platform/metadata/engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-a" {
+  capabilities = ["read"]
+}
+
+path "platform/data/engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-b" {
+  capabilities = ["read"]
+}
+
+path "platform/metadata/engagements/WH-ENG-20260822-AUDIT-E2-01/audit-core/token-b" {
+  capabilities = ["read"]
+}

scripts/audit-core-whitehat-e2-credentials.py

@@ -0,0 +1,466 @@
+#!/usr/bin/env python3
+"""Attended, value-safe credentials for WH-ENG-20260822-AUDIT-E2-01."""
+
+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-01"
+START = datetime(2026, 8, 22, 18, 0, tzinfo=UTC)
+LATEST_PROJECT = datetime(2026, 8, 22, 18, 3, tzinfo=UTC)
+EXPIRES = datetime(2026, 8, 22, 18, 15, tzinfo=UTC)
+EXPIRES_TEXT = "2026-08-22T18:15: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",
+        "tenant": "tenant:trial:whitehat-a-20260822",
+    },
+    "token-b": {
+        "name": "whitehat-e2-b-20260822",
+        "tenant": "tenant:trial:whitehat-b-20260822",
+    },
+}
+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())

tests/test_audit_core_whitehat_e2_credentials.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.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, 17, 59, tzinfo=UTC)))
+        self.assertEqual("projection-window-open", module.time_state(datetime(2026, 8, 22, 18, 1, tzinfo=UTC)))
+        self.assertEqual("projection-cutoff-passed", module.time_state(datetime(2026, 8, 22, 18, 4, 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()

workplans/RAILIANCE-WP-0024-audit-core-custody-and-recovery-coordination.md

@@ -72,6 +72,17 @@ both identities must expire no later than 18:15Z and cleanup/reload must be
 attended. No unattended scheduler or background credential job is authorized
 by this record.
 
+**Procedure implementation (2026-08-22):**
+`scripts/audit-core-whitehat-e2-credentials.py` now provides fail-closed
+`preflight`, `project`, and idempotent `cleanup` commands. The projection uses
+two exact per-engagement OpenBao paths, an ephemeral least-privilege Kubernetes
+auth role, a Whitehat-only ClusterSecretStore, and an ExternalSecret that emits
+exactly `token-a` and `token-b`. Project is hard-gated to 18:00Z–18:03Z;
+cleanup remains callable after interruption and removes only the named
+temporary identities and exact engagement resources. Four unit tests, a live
+value-safe preflight, policy formatting, and Kubernetes server-side dry-run
+passed. Live projection and cleanup remain waiting for the attended window.
+
 ## T02 — Define the runtime database lease recovery exercise
 
 ```task