fix(security): hold annotation enforcement until ESO metadata is explicit

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e7fb-275d-7db0-b1df-39ed943427da

scripts/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())

scripts/secret_annotation_maintenance.py

@@ -43,11 +43,18 @@ def maintain(clean=False):
         if len(pair) != 2 or not all(NAME.fullmatch(value) for value in pair):
             raise RuntimeError("invalid Secret identity output; suppressed")
         rows.append(pair)
+    namespaces = run(["get", "namespaces", "-o", "name"]).splitlines()
+    if not all(value.startswith("namespace/") and NAME.fullmatch(value.split("/", 1)[1]) for value in namespaces):
+        raise RuntimeError("invalid namespace inventory; suppressed")
+    active_namespaces = {value.split("/", 1)[1] for value in namespaces}
     report = {"captured_at": datetime.now(timezone.utc).isoformat(),
               "mode": "clean" if clean else "inspect", "checked": 0,
-              "annotated": [], "cleaned": [], "complete": False}
+              "annotated": [], "cleaned": [], "orphaned_namespace": [], "complete": False}
     try:
         for namespace, name in rows:
+            if namespace not in active_namespaces:
+                report["orphaned_namespace"].append(namespace + "/" + name)
+                continue
             report["checked"] += 1
             if not inspect(namespace, name):
                 continue
@@ -61,7 +68,8 @@ def maintain(clean=False):
                 if inspect(namespace, name):
                     raise RuntimeError("annotation still present")
                 report["cleaned"].append(identity)
-        report["complete"] = True
+        report["active_namespace_scan_complete"] = True
+        report["complete"] = not report["orphaned_namespace"]
     except (RuntimeError, subprocess.SubprocessError, OSError):
         report["error"] = "maintenance incomplete; raw output suppressed; inspect before retry"
     return report