#!/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())