3f9e4535d1
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
429 lines
18 KiB
Python
Executable file
429 lines
18 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Value-safe preflight for RAILIANCE-WP-0024 recovery exercises.
|
|
|
|
This helper is deliberately read-only. It never reads Kubernetes Secret data,
|
|
database usernames/passwords, OpenBao lease payloads, or unseal material. Live
|
|
lease revocation and a host reboot remain separate, attended actions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
EXPECTED_IMAGE = (
|
|
"forgejo.coulomb.social/coulomb/audit-core@"
|
|
"sha256:c2fe39a0185b99be3fc0cb14d2de69772b8e66e20490097c9d11d90cc39719a6"
|
|
)
|
|
EXPECTED_DB_KEYS = {"username", "password", "host", "port", "dbname"}
|
|
|
|
|
|
class PreflightError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class Remote:
|
|
def __init__(self, host: str) -> None:
|
|
self.host = host
|
|
|
|
def run(self, command: list[str], *, label: str) -> str:
|
|
remote_command = " ".join(shlex.quote(part) for part in command)
|
|
completed = subprocess.run(
|
|
["ssh", "-o", "BatchMode=yes", self.host, remote_command],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if completed.returncode != 0:
|
|
# Remote output can include provider or application detail that is
|
|
# inappropriate for durable evidence. Report only the failed step.
|
|
raise PreflightError(f"{label} failed (exit {completed.returncode})")
|
|
return completed.stdout.strip()
|
|
|
|
def kubectl(self, args: list[str], *, label: str) -> str:
|
|
return self.run(["kubectl", *args], label=label)
|
|
|
|
def kubectl_json(self, args: list[str], *, label: str) -> dict[str, Any]:
|
|
raw = self.kubectl([*args, "-o", "json"], label=label)
|
|
try:
|
|
value = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise PreflightError(f"{label} returned invalid JSON") from exc
|
|
if not isinstance(value, dict):
|
|
raise PreflightError(f"{label} did not return an object")
|
|
return value
|
|
|
|
|
|
def condition(resource: dict[str, Any], kind: str) -> str | None:
|
|
for item in resource.get("status", {}).get("conditions", []):
|
|
if item.get("type") == kind:
|
|
return item.get("status")
|
|
return None
|
|
|
|
|
|
def rfc3339(value: str) -> datetime:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(UTC)
|
|
|
|
|
|
def latest_completed_backup(backups: dict[str, Any], now: datetime) -> dict[str, Any]:
|
|
completed = [
|
|
item
|
|
for item in backups.get("items", [])
|
|
if item.get("status", {}).get("phase") == "completed"
|
|
and item.get("status", {}).get("stoppedAt")
|
|
]
|
|
if not completed:
|
|
raise PreflightError("no completed platform-pg backup is visible")
|
|
latest = max(completed, key=lambda item: item["status"]["stoppedAt"])
|
|
stopped = rfc3339(latest["status"]["stoppedAt"])
|
|
return {
|
|
"name": latest["metadata"]["name"],
|
|
"stopped_at": stopped.isoformat().replace("+00:00", "Z"),
|
|
"age_hours": round((now - stopped).total_seconds() / 3600, 2),
|
|
"method": latest.get("spec", {}).get("method"),
|
|
}
|
|
|
|
|
|
def resource_condition(resource: dict[str, Any]) -> dict[str, Any]:
|
|
conditions = resource.get("status", {}).get("conditions", [])
|
|
current = conditions[0] if conditions else {}
|
|
return {
|
|
"ready": current.get("status") == "True",
|
|
"reason": current.get("reason"),
|
|
}
|
|
|
|
|
|
def secret_key_names(description: str) -> set[str]:
|
|
"""Extract names from ``kubectl describe secret`` without reading data."""
|
|
keys: set[str] = set()
|
|
for line in description.splitlines():
|
|
match = re.match(r"^([^\s:]+):\s+\d+ bytes$", line.strip())
|
|
if match:
|
|
keys.add(match.group(1))
|
|
return keys
|
|
|
|
|
|
def validate_snapshot_receipt(
|
|
path: Path,
|
|
*,
|
|
live_openbao: dict[str, Any],
|
|
now: datetime,
|
|
max_age_hours: float,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
receipt = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise PreflightError("OpenBao snapshot receipt is unavailable or invalid JSON") from exc
|
|
if not isinstance(receipt, dict) or receipt.get("receipt_version") != 1:
|
|
raise PreflightError("OpenBao snapshot receipt has an unsupported shape/version")
|
|
required_strings = (
|
|
"receipt_id", "created_at", "operator", "source_cluster",
|
|
"source_namespace", "source_pod", "cluster_id", "snapshot_sha256",
|
|
"encrypted_snapshot_sha256", "encrypted_location_ref",
|
|
)
|
|
for key in required_strings:
|
|
if not isinstance(receipt.get(key), str) or not receipt[key].strip():
|
|
raise PreflightError(f"OpenBao snapshot receipt is missing {key}")
|
|
required_true = (
|
|
"snapshot_created", "source_initialized", "source_unsealed",
|
|
"snapshot_encrypted", "encrypted_copy_off_host", "encryption_verified",
|
|
"hash_verified", "no_secret_material_recorded",
|
|
)
|
|
for key in required_true:
|
|
if receipt.get(key) is not True:
|
|
raise PreflightError(f"OpenBao snapshot receipt requires {key}=true")
|
|
if (
|
|
receipt["source_cluster"] != "railiance01"
|
|
or receipt["source_namespace"] != "openbao"
|
|
or receipt["source_pod"] != "openbao-0"
|
|
or receipt["cluster_id"] != live_openbao.get("cluster_id")
|
|
):
|
|
raise PreflightError("OpenBao snapshot receipt does not match the live source")
|
|
try:
|
|
created = rfc3339(receipt["created_at"])
|
|
snapshot_index = int(receipt["raft_applied_index"])
|
|
live_index = int(live_openbao["raft_applied_index"])
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise PreflightError("OpenBao snapshot receipt has invalid time/index metadata") from exc
|
|
age_hours = (now - created).total_seconds() / 3600
|
|
if age_hours < -(5 / 60) or age_hours > max_age_hours:
|
|
raise PreflightError("OpenBao snapshot receipt is outside the permitted age")
|
|
if snapshot_index <= 0 or snapshot_index > live_index:
|
|
raise PreflightError("OpenBao snapshot receipt has an impossible Raft index")
|
|
digest_pattern = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
digests = (receipt["snapshot_sha256"], receipt["encrypted_snapshot_sha256"])
|
|
if any(not digest_pattern.fullmatch(value) for value in digests):
|
|
raise PreflightError("OpenBao snapshot receipt has an invalid digest")
|
|
if any(len(set(value.removeprefix("sha256:"))) <= 1 for value in digests):
|
|
raise PreflightError("OpenBao snapshot receipt contains a placeholder digest")
|
|
if digests[0] == digests[1]:
|
|
raise PreflightError("plain and encrypted snapshot digests must differ")
|
|
encoded = json.dumps(receipt, sort_keys=True)
|
|
for marker in (
|
|
"BEGIN PRIVATE KEY", "BEGIN OPENSSH PRIVATE KEY", "AGE-SECRET-KEY-1",
|
|
"OPENBAO_ROOT_TOKEN", "VAULT_TOKEN", "hvs.", "<", "YYYY-MM-DD",
|
|
):
|
|
if marker in encoded:
|
|
raise PreflightError("OpenBao snapshot receipt contains forbidden material/placeholder")
|
|
return {
|
|
"receipt_id": receipt["receipt_id"],
|
|
"created_at": created.isoformat().replace("+00:00", "Z"),
|
|
"age_hours": round(age_hours, 2),
|
|
"cluster_id_matches": True,
|
|
"raft_applied_index": snapshot_index,
|
|
"encrypted_copy_off_host": True,
|
|
"verified": True,
|
|
"secret_values_observed": False,
|
|
}
|
|
|
|
|
|
def common_state(remote: Remote, now: datetime) -> dict[str, Any]:
|
|
node_list = remote.kubectl_json(["get", "nodes"], label="read node state")
|
|
nodes = node_list.get("items", [])
|
|
if len(nodes) != 1:
|
|
raise PreflightError(f"expected one railiance01 node, observed {len(nodes)}")
|
|
node = nodes[0]
|
|
|
|
cluster = remote.kubectl_json(
|
|
["-n", "databases", "get", "cluster", "platform-pg"],
|
|
label="read platform-pg state",
|
|
)
|
|
deployment = remote.kubectl_json(
|
|
["-n", "audit-core", "get", "deployment", "audit-core"],
|
|
label="read audit-core deployment",
|
|
)
|
|
pod_list = remote.kubectl_json(
|
|
["-n", "audit-core", "get", "pods", "-l", "app.kubernetes.io/name=audit-core"],
|
|
label="read audit-core pod metadata",
|
|
)
|
|
pods = pod_list.get("items", [])
|
|
if len(pods) != 1:
|
|
raise PreflightError(f"expected one audit-core pod, observed {len(pods)}")
|
|
pod = pods[0]
|
|
container_statuses = pod.get("status", {}).get("containerStatuses", [])
|
|
restart_count = sum(int(item.get("restartCount", 0)) for item in container_statuses)
|
|
|
|
stores: dict[str, Any] = {}
|
|
for name in ("openbao-audit-core", "openbao-audit-core-database"):
|
|
resource = remote.kubectl_json(
|
|
["get", "clustersecretstore", name], label=f"read {name} state"
|
|
)
|
|
stores[name] = resource_condition(resource)
|
|
|
|
external_secrets: dict[str, Any] = {}
|
|
for name in ("audit-core-database", "audit-core-senders"):
|
|
resource = remote.kubectl_json(
|
|
["-n", "audit-core", "get", "externalsecret", name],
|
|
label=f"read {name} state",
|
|
)
|
|
external_secrets[name] = resource_condition(resource)
|
|
|
|
bao_raw = remote.kubectl(
|
|
["-n", "openbao", "exec", "openbao-0", "--", "bao", "status", "-format=json"],
|
|
label="read OpenBao seal state",
|
|
)
|
|
try:
|
|
bao_status = json.loads(bao_raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise PreflightError("OpenBao seal state returned invalid JSON") from exc
|
|
image = deployment["spec"]["template"]["spec"]["containers"][0]["image"]
|
|
state = {
|
|
"captured_at": now.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
|
|
"node": {
|
|
"name": node["metadata"]["name"],
|
|
"ready": condition(node, "Ready") == "True",
|
|
"k3s_active": remote.run(
|
|
["systemctl", "is-active", "k3s"], label="read k3s service state"
|
|
) == "active",
|
|
"uptime_seconds": float(
|
|
remote.run(["cat", "/proc/uptime"], label="read host uptime").split()[0]
|
|
),
|
|
},
|
|
"platform_pg": {
|
|
"ready_instances": cluster.get("status", {}).get("readyInstances", 0),
|
|
"instances": cluster.get("spec", {}).get("instances", 0),
|
|
"phase": cluster.get("status", {}).get("phase"),
|
|
"continuous_archiving": condition(cluster, "ContinuousArchiving") == "True",
|
|
"last_backup_succeeded": condition(cluster, "LastBackupSucceeded") == "True",
|
|
},
|
|
"openbao": {
|
|
"initialized": bool(bao_status.get("initialized")),
|
|
"sealed": bool(bao_status.get("sealed")),
|
|
"seal_type": bao_status.get("type"),
|
|
"threshold": bao_status.get("t"),
|
|
"shares": bao_status.get("n"),
|
|
"cluster_id": bao_status.get("cluster_id"),
|
|
"raft_applied_index": bao_status.get("raft_applied_index"),
|
|
},
|
|
"external_secret_stores": stores,
|
|
"external_secrets": external_secrets,
|
|
"audit_core": {
|
|
"image": image,
|
|
"image_matches_reviewed_digest": image == EXPECTED_IMAGE,
|
|
"ready_replicas": deployment.get("status", {}).get("readyReplicas", 0),
|
|
"replicas": deployment.get("spec", {}).get("replicas", 0),
|
|
"pod_uid": pod["metadata"]["uid"],
|
|
"restart_count": restart_count,
|
|
},
|
|
"secret_values_observed": False,
|
|
}
|
|
return state
|
|
|
|
|
|
def automated_common_pass(state: dict[str, Any]) -> bool:
|
|
return all(
|
|
(
|
|
state["node"]["ready"],
|
|
state["node"]["k3s_active"],
|
|
state["platform_pg"]["ready_instances"] == state["platform_pg"]["instances"] == 1,
|
|
state["platform_pg"]["continuous_archiving"],
|
|
state["platform_pg"]["last_backup_succeeded"],
|
|
state["openbao"]["initialized"],
|
|
not state["openbao"]["sealed"],
|
|
all(item["ready"] for item in state["external_secret_stores"].values()),
|
|
all(item["ready"] for item in state["external_secrets"].values()),
|
|
state["audit_core"]["image_matches_reviewed_digest"],
|
|
state["audit_core"]["ready_replicas"] == state["audit_core"]["replicas"] == 1,
|
|
)
|
|
)
|
|
|
|
|
|
def endpoint_status(remote: Remote, path: str) -> int:
|
|
code = remote.kubectl(
|
|
[
|
|
"-n", "audit-core", "exec", "deploy/audit-core", "--", "python", "-c",
|
|
"import urllib.request,urllib.error; "
|
|
f"u='http://127.0.0.1:8080{path}'; "
|
|
"\ntry:\n r=urllib.request.urlopen(u,timeout=3); print(r.status)"
|
|
"\nexcept urllib.error.HTTPError as e:\n print(e.code)",
|
|
],
|
|
label=f"probe audit-core {path}",
|
|
)
|
|
try:
|
|
return int(code)
|
|
except ValueError as exc:
|
|
raise PreflightError(f"audit-core {path} returned an invalid status") from exc
|
|
|
|
|
|
def database_lease_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
|
|
now = datetime.now(UTC)
|
|
state = common_state(remote, now)
|
|
secret_rv = remote.kubectl(
|
|
["-n", "audit-core", "get", "secret", "audit-core-database", "-o", "jsonpath={.metadata.resourceVersion}"],
|
|
label="read database Secret metadata",
|
|
)
|
|
description = remote.kubectl(
|
|
["-n", "audit-core", "describe", "secret", "audit-core-database"],
|
|
label="read database Secret key names",
|
|
)
|
|
keys = secret_key_names(description)
|
|
health_code = endpoint_status(remote, "/healthz")
|
|
ready_code = endpoint_status(remote, "/readyz")
|
|
gates = {
|
|
"approved_window_id_recorded": bool(args.approved_window_id),
|
|
"audit_core_owner_acknowledged": args.audit_core_owner_ack,
|
|
"rapp_postgres_owner_acknowledged": args.rapp_postgres_owner_ack,
|
|
"approved_synthetic_load_contract": bool(args.synthetic_load_id),
|
|
"attended_abort_operator_named": bool(args.abort_operator),
|
|
}
|
|
automated = automated_common_pass(state) and keys == EXPECTED_DB_KEYS and health_code == 200 and ready_code == 200
|
|
return {
|
|
"procedure": "audit-core-database-lease-recovery",
|
|
"preflight_only": True,
|
|
"automated_checks_passed": automated,
|
|
"ready_for_live_execution": automated and all(gates.values()),
|
|
"operator_gates": gates,
|
|
"baseline": {
|
|
"database_secret_resource_version": secret_rv,
|
|
"database_secret_keys": sorted(keys),
|
|
"health_status": health_code,
|
|
"readiness_status": ready_code,
|
|
"pod_uid": state["audit_core"]["pod_uid"],
|
|
"restart_count": state["audit_core"]["restart_count"],
|
|
},
|
|
"state": state,
|
|
"secret_values_observed": False,
|
|
}
|
|
|
|
|
|
def reboot_preflight(remote: Remote, args: argparse.Namespace) -> dict[str, Any]:
|
|
now = datetime.now(UTC)
|
|
state = common_state(remote, now)
|
|
backups = remote.kubectl_json(
|
|
["-n", "databases", "get", "backup", "-l", "cnpg.io/cluster=platform-pg"],
|
|
label="read platform-pg backup metadata",
|
|
)
|
|
latest = latest_completed_backup(backups, now)
|
|
snapshot_evidence = Path(args.openbao_snapshot_evidence).resolve() if args.openbao_snapshot_evidence else None
|
|
snapshot_receipt = None
|
|
if snapshot_evidence:
|
|
snapshot_receipt = validate_snapshot_receipt(
|
|
snapshot_evidence,
|
|
live_openbao=state["openbao"],
|
|
now=now,
|
|
max_age_hours=args.max_snapshot_age_hours,
|
|
)
|
|
gates = {
|
|
"approved_window_id_recorded": bool(args.approved_window_id),
|
|
"openbao_snapshot_evidence_present": bool(snapshot_receipt),
|
|
"unseal_quorum_attested": args.unseal_quorum_attested,
|
|
"provider_console_access_attested": args.provider_console_attested,
|
|
"host_cluster_platform_database_audit_owners_acknowledged": args.all_owners_ack,
|
|
"attended_abort_operator_named": bool(args.abort_operator),
|
|
}
|
|
automated = automated_common_pass(state) and latest["age_hours"] <= args.max_backup_age_hours
|
|
return {
|
|
"procedure": "railiance01-coordinated-reboot",
|
|
"preflight_only": True,
|
|
"automated_checks_passed": automated,
|
|
"ready_for_live_execution": automated and all(gates.values()),
|
|
"operator_gates": gates,
|
|
"latest_platform_pg_backup": latest,
|
|
"max_backup_age_hours": args.max_backup_age_hours,
|
|
"openbao_snapshot_receipt": snapshot_receipt,
|
|
"max_snapshot_age_hours": args.max_snapshot_age_hours,
|
|
"state": state,
|
|
"secret_values_observed": False,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("procedure", choices=["database-lease", "node-reboot"])
|
|
parser.add_argument("--remote", default="railiance01")
|
|
parser.add_argument("--approved-window-id")
|
|
parser.add_argument("--abort-operator")
|
|
parser.add_argument("--audit-core-owner-ack", action="store_true")
|
|
parser.add_argument("--rapp-postgres-owner-ack", action="store_true")
|
|
parser.add_argument("--synthetic-load-id")
|
|
parser.add_argument("--openbao-snapshot-evidence")
|
|
parser.add_argument("--unseal-quorum-attested", action="store_true")
|
|
parser.add_argument("--provider-console-attested", action="store_true")
|
|
parser.add_argument("--all-owners-ack", action="store_true")
|
|
parser.add_argument("--max-backup-age-hours", type=float, default=26.0)
|
|
parser.add_argument("--max-snapshot-age-hours", type=float, default=24.0)
|
|
args = parser.parse_args()
|
|
try:
|
|
remote = Remote(args.remote)
|
|
result = (
|
|
database_lease_preflight(remote, args)
|
|
if args.procedure == "database-lease"
|
|
else reboot_preflight(remote, args)
|
|
)
|
|
except (OSError, ValueError, KeyError, PreflightError) as exc:
|
|
print(f"recovery preflight failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|