60 lines
3 KiB
Python
60 lines
3 KiB
Python
#!/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())
|