1228cd8407
KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the bounded proof scripts; one attended script with go/no-go gates, typed ADOPT confirmation, prune/force-off SSA sync and post-checks; evaluator and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 527080@bnt-lap001 Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only drift check between child Application pins in Git and on railiance01.
|
|
|
|
`railiance-apps-root` syncs by hand from `argocd/railiance01/applications`. If a
|
|
release patches a child's live targetRevision without committing the same pin
|
|
here, the next root sync silently rolls that child back. Run this before any
|
|
root sync and after every release. Exit 1 on drift, 2 on a read failure.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import yaml
|
|
|
|
APPS_DIR = Path(__file__).resolve().parent.parent / "argocd/railiance01/applications"
|
|
|
|
|
|
def declared() -> dict[str, str]:
|
|
pins = {}
|
|
for path in sorted(APPS_DIR.glob("*.y*ml")):
|
|
for doc in yaml.safe_load_all(path.read_text()):
|
|
if doc and doc.get("kind") == "Application":
|
|
pins[doc["metadata"]["name"]] = doc["spec"]["source"]["targetRevision"]
|
|
return pins
|
|
|
|
|
|
def live() -> dict[str, str]:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import kube_route # railiance01 SSH lane; never the workstation kubeconfig
|
|
|
|
out = subprocess.run(
|
|
kube_route.ssh_argv(["kubectl", "-n", "argocd", "get", "applications.argoproj.io", "-o", "json"]),
|
|
capture_output=True, text=True, check=True,
|
|
).stdout
|
|
return {
|
|
item["metadata"]["name"]: item["spec"].get("source", {}).get("targetRevision", "")
|
|
for item in json.loads(out)["items"]
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
running = live()
|
|
except (subprocess.CalledProcessError, FileNotFoundError, ValueError) as exc:
|
|
print(f"ERROR: cannot read live Applications: {exc}", file=sys.stderr)
|
|
return 2
|
|
drift = 0
|
|
for name, pin in declared().items():
|
|
current = running.get(name)
|
|
if current == pin:
|
|
print(f"ok {name} {pin[:12]}")
|
|
else:
|
|
drift += 1
|
|
print(f"DRIFT {name} git={pin[:12]} live={(current or 'absent')[:12]}")
|
|
if drift:
|
|
print(f"{drift} child pin(s) differ: commit the live pin here (or revert live) "
|
|
"before syncing railiance-apps-root.", file=sys.stderr)
|
|
return 1 if drift else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|