Inspect backup client delivery and record gated refresh requirements

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29

scripts/inspect_backup_delivery.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Read-only backup delivery inspection; never reads a Kubernetes Secret."""
+import argparse
+import hashlib
+import json
+import re
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def capture(args):
+    result = subprocess.run(args, capture_output=True, timeout=30, check=False)
+    if result.returncode:
+        raise RuntimeError("inspection_command_failed")
+    return result.stdout
+
+
+def inspect(platform_root, command=capture):
+    def resource(kind, name=None, namespace=None):
+        args = ["kubectl"]
+        if namespace:
+            args += ["-n", namespace]
+        args += ["get", kind]
+        if name:
+            args += [name]
+        return json.loads(command(args + ["-o", "json"]))
+    nodes = resource("nodes")
+    if not any(a.get("address") == "92.205.62.239" for n in nodes["items"]
+               for a in n.get("status", {}).get("addresses", [])):
+        raise RuntimeError("wrong_cluster")
+    es = resource("externalsecret", "actcore-backup-offsite", "activity-core")
+    deployment = resource("deployment", "actcore-worker", "activity-core")
+    spec = deployment["spec"]["template"]["spec"]
+    workers = [c for c in spec["containers"] if any(
+        m["mountPath"] == "/opt/railiance-platform" for m in c.get("volumeMounts", []))]
+    if len(workers) != 1:
+        raise RuntimeError("ambiguous_worker")
+    volumes = [v for v in spec["volumes"] if v["name"] == "backup-verified"]
+    if len(volumes) != 1:
+        raise RuntimeError("missing_bundle")
+    name = volumes[0]["configMap"]["name"]
+    if not re.fullmatch(r"backup-verified-[a-f0-9]{12}", name):
+        raise RuntimeError("unexpected_bundle_name")
+    cm = resource("configmap", name, "activity-core")
+    common = platform_root / "lib/railiance-backup-common.sh"
+    digest = hashlib.sha256(common.read_bytes()).hexdigest()
+    bundle_digest = hashlib.sha256(cm["data"]["lib__railiance-backup-common.sh"].encode()).hexdigest()
+    mounted = {}
+    for lane, path in [("forgejo", "/opt/railiance-backup-verified/lib/railiance-backup-common.sh"),
+                       ("cnpg", "/opt/railiance-platform/lib/railiance-backup-common.sh")]:
+        output = command(["kubectl", "-n", "activity-core", "exec", "deployment/actcore-worker",
+                          "-c", workers[0]["name"], "--", "sha256sum", path])
+        actual = output.split()[0].decode()
+        if not re.fullmatch(r"[a-f0-9]{64}", actual):
+            raise RuntimeError("invalid_digest_output")
+        mounted[lane] = actual == digest
+    return {"read_only": True, "secret_objects_read": False, "inspected_at": datetime.now(timezone.utc).isoformat(),
+            "source_common_sha256": digest, "source_public_dav_header_present": b"X-Requested-With" in common.read_bytes(),
+            "bundle": name, "bundle_common_matches_source": bundle_digest == digest,
+            "mounted_common_matches_source": mounted,
+            "externalsecret_ready": any(c.get("type") == "Ready" and c.get("status") == "True"
+                                        for c in es.get("status", {}).get("conditions", [])),
+            "externalsecret_refresh_time": es.get("status", {}).get("refreshTime"),
+            "refresh_interval": es["spec"].get("refreshInterval"),
+            "worker_secret_env_requires_new_process": any(e.get("secretRef", {}).get("name") == "actcore-runtime-secret"
+                                                         for e in workers[0].get("envFrom", [])),
+            "credential_currentness_proved": False, "provider_roundtrip_proved": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--platform-root", type=Path, required=True)
+    args = parser.parse_args()
+    try:
+        receipt = inspect(args.platform_root)
+    except Exception:
+        receipt = {"read_only": True, "ok": False, "error_code": "backup_delivery_inspection_failed"}
+    print(json.dumps(receipt, sort_keys=True))
+    return 0 if receipt.get("ok", True) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())