feat(security): reject secret last-applied annotations (CUST-WP-0073)

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

scripts/secret_annotation_maintenance.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Bounded CUST-WP-0073-T03 maintenance. Never emit kubectl output/errors.
+
+Run on railiance01 through the supervised admin path. Default is inspection;
+--clean removes only the last-applied annotation, leaving Secret data untouched.
+"""
+import argparse
+import json
+import re
+import subprocess
+from datetime import datetime, timezone
+
+KEY = "kubectl.kubernetes.io/last-applied-configuration"
+PRESENCE = ('{{ $found := false }}{{ range $key, $_ := .metadata.annotations }}'
+            '{{ if eq $key "' + KEY + '" }}{{ $found = true }}{{ end }}{{ end }}'
+            '{{ if $found }}HAS-ANNOTATION{{ else }}clean{{ end }}')
+NAME = re.compile(r"^[a-z0-9][a-z0-9.-]*$")
+
+
+def run(args):
+    # Even a template error can contain the entire Secret. Never forward it.
+    result = subprocess.run(["kubectl", *args], capture_output=True, text=True, timeout=30)
+    if result.returncode:
+        raise RuntimeError("kubectl operation failed; output suppressed")
+    return result.stdout.strip()
+
+
+def inspect(namespace, name):
+    value = run(["-n", namespace, "get", "secret", name, "-o", "go-template=" + PRESENCE])
+    if value not in ("clean", "HAS-ANNOTATION"):
+        raise RuntimeError("unexpected presence result; output suppressed")
+    return value == "HAS-ANNOTATION"
+
+
+def maintain(clean=False):
+    # Custom columns use fixed universally-present identity fields; no annotation
+    # or data output. Validate before using any returned text as an argument.
+    identities = run(["get", "secrets", "-A", "--no-headers", "-o",
+                      "custom-columns=NAMESPACE:.metadata.namespace,NAME:.metadata.name"])
+    rows = []
+    for line in identities.splitlines():
+        pair = line.split()
+        if len(pair) != 2 or not all(NAME.fullmatch(value) for value in pair):
+            raise RuntimeError("invalid Secret identity output; suppressed")
+        rows.append(pair)
+    report = {"captured_at": datetime.now(timezone.utc).isoformat(),
+              "mode": "clean" if clean else "inspect", "checked": 0,
+              "annotated": [], "cleaned": [], "complete": False}
+    try:
+        for namespace, name in rows:
+            report["checked"] += 1
+            if not inspect(namespace, name):
+                continue
+            identity = namespace + "/" + name
+            report["annotated"].append(identity)
+            if clean:
+                # A single JSON patch operation cannot modify credential data.
+                patch = [{"op": "remove", "path": "/metadata/annotations/" + KEY.replace("/", "~1")}]
+                run(["-n", namespace, "patch", "secret", name, "--type=json",
+                     "-p", json.dumps(patch)])
+                if inspect(namespace, name):
+                    raise RuntimeError("annotation still present")
+                report["cleaned"].append(identity)
+        report["complete"] = True
+    except (RuntimeError, subprocess.SubprocessError, OSError):
+        report["error"] = "maintenance incomplete; raw output suppressed; inspect before retry"
+    return report
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--clean", action="store_true")
+    args = parser.parse_args()
+    try:
+        report = maintain(args.clean)
+    except (RuntimeError, subprocess.SubprocessError, OSError):
+        report = {"complete": False, "error": "inventory failed; raw output suppressed"}
+    print(json.dumps(report, indent=2))
+    return 0 if report["complete"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())