diff --git a/scripts/audit-core-whitehat-e2-credentials-02.py b/scripts/audit-core-whitehat-e2-credentials-02.py new file mode 100755 index 0000000..6fbca9e --- /dev/null +++ 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()) diff --git a/tests/test_audit_core_whitehat_e2_credentials_02.py b/tests/test_audit_core_whitehat_e2_credentials_02.py new file mode 100644 index 0000000..57e9963 --- /dev/null +++ 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()