467 lines
19 KiB
Python
467 lines
19 KiB
Python
|
#!/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())
|