tegwick
a55b05bf7d
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
84 lines
4 KiB
Python
84 lines
4 KiB
Python
#!/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())
|