Advance supervised agent records and close verified Secret annotation guard

docs/changes/CUST-WP-0073/prove_secret_annotation_guard.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Positive/negative admission proof using only a uniquely named synthetic Secret."""
+import copy
+import json
+import subprocess
+import uuid
+from datetime import datetime, timezone
+
+KEY = "kubectl.kubernetes.io/last-applied-configuration"
+
+
+def run(args, obj=None):
+    return subprocess.run(["kubectl", "-n", "whitehat", *args],
+                          input=json.dumps(obj) if obj is not None else None,
+                          capture_output=True, text=True, timeout=30)
+
+
+def denied(result):
+    # Do not accept connectivity/RBAC failures as admission-policy success.
+    return result.returncode != 0 and "Secret last-applied annotations are forbidden" in result.stderr
+
+
+def main():
+    name = "cust-0073-proof-" + uuid.uuid4().hex[:12]
+    obj = {"apiVersion": "v1", "kind": "Secret", "metadata": {"name": name},
+           "type": "Opaque", "data": {"fixture": "c3ludGhldGlj"}}
+    report = {"captured_at": datetime.now(timezone.utc).isoformat(), "namespace": "whitehat",
+              "fixture": name, "synthetic_only": True, "checks": {}}
+    created = False
+    try:
+        result = run(["create", "--field-manager=cust-0073-proof", "-f", "-"], obj)
+        created = result.returncode == 0
+        report["checks"]["clean_create_allowed"] = created
+        if not created:
+            raise RuntimeError("synthetic create failed")
+        for label, value in [("empty", ""), ("populated", "synthetic")]:
+            annotated = copy.deepcopy(obj)
+            annotated["metadata"]["name"] = name + "-denied"
+            annotated["metadata"]["annotations"] = {KEY: value}
+            report["checks"][label + "_annotated_create_denied"] = denied(run(["create", "--dry-run=server", "-f", "-"], annotated))
+            patch = {"metadata": {"annotations": {KEY: value}}}
+            report["checks"][label + "_annotated_update_denied"] = denied(run(["patch", "secret", name, "--dry-run=server", "--type=merge", "-p", json.dumps(patch)]))
+        report["checks"]["client_apply_denied"] = denied(run(["apply", "--dry-run=server", "-f", "-"], obj))
+        report["checks"]["clean_server_apply_allowed"] = run(["apply", "--server-side", "--field-manager=cust-0073-proof", "-f", "-"], obj).returncode == 0
+        report["checks"]["clean_update_allowed"] = run(["patch", "secret", name, "--type=merge", "-p", json.dumps({"data": {"fixture": "c3ludGhldGljLXVwZGF0ZQ=="}})]).returncode == 0
+    except (RuntimeError, subprocess.SubprocessError, OSError):
+        report["error"] = "proof incomplete; raw output suppressed"
+    finally:
+        if created:
+            try:
+                report["checks"]["fixture_removed"] = run(["delete", "secret", name, "--wait=true"]).returncode == 0
+            except (subprocess.SubprocessError, OSError):
+                report["checks"]["fixture_removed"] = False
+    report["passed"] = "error" not in report and len(report["checks"]) == 9 and all(report["checks"].values())
+    print(json.dumps(report, indent=2))
+    return 0 if report["passed"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())