tegwick
94fda23546
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
122 lines
7.5 KiB
Python
122 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Render the optional outbox with caller auth on/off, without cluster access."""
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
import yaml
|
|
|
|
|
|
def render(*settings, succeeds=True):
|
|
command = ["helm", "template", "flex-auth", "charts/flex-auth", "-f", "values/stage2-canary.yaml"]
|
|
for setting in settings:
|
|
command += ["--set", setting]
|
|
result = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
if not succeeds:
|
|
assert result.returncode != 0, "invalid multi-writer configuration was accepted"
|
|
return []
|
|
assert result.returncode == 0, result.stderr
|
|
return list(yaml.safe_load_all(result.stdout))
|
|
|
|
|
|
def deployment(documents):
|
|
return next(d for d in documents if d["kind"] == "Deployment")
|
|
|
|
|
|
base = render()
|
|
assert not any(d["kind"] == "PersistentVolumeClaim" for d in base)
|
|
assert "--outbox-dir" not in deployment(base)["spec"]["template"]["spec"]["containers"][0]["args"]
|
|
|
|
enabled = ("outbox.enabled=true", "outbox.source=flex-auth.test")
|
|
for auth in ((), ("callerAuth.mode=enforce", "callerAuth.binding=test=system:serviceaccount:test:test")):
|
|
documents = render(*enabled, *auth)
|
|
pod = deployment(documents)["spec"]["template"]["spec"]
|
|
container = pod["containers"][0]
|
|
args = container["args"]
|
|
assert args[args.index("--outbox-dir") + 1] == "/var/lib/flex-auth/outbox"
|
|
assert args[args.index("--emission-source") + 1] == "flex-auth.test"
|
|
assert deployment(documents)["spec"]["strategy"]["type"] == "Recreate"
|
|
pvc = next(d for d in documents if d["kind"] == "PersistentVolumeClaim")
|
|
assert pvc["spec"]["accessModes"] == ["ReadWriteOnce"]
|
|
assert pvc["spec"]["resources"]["requests"]["storage"] == "1Gi"
|
|
assert any(v.get("persistentVolumeClaim", {}).get("claimName") == pvc["metadata"]["name"] for v in pod["volumes"])
|
|
assert any(v["name"] == "flex-auth-outbox" and not v.get("readOnly", False) for v in container["volumeMounts"])
|
|
if auth:
|
|
assert any(v["name"] == "flex-auth-reviewer" for v in pod["volumes"])
|
|
|
|
existing = render(*enabled, "outbox.existingClaim=retained-outbox")
|
|
assert not any(d["kind"] == "PersistentVolumeClaim" for d in existing)
|
|
pod = deployment(existing)["spec"]["template"]["spec"]
|
|
assert any(v.get("persistentVolumeClaim", {}).get("claimName") == "retained-outbox" for v in pod["volumes"])
|
|
render(*enabled, "replicaCount=2", succeeds=False)
|
|
render("outbox.enabled=true", succeeds=False)
|
|
print("optional outbox chart checks passed")
|
|
|
|
# Delivery rendering binds an existing admitted credential mount and narrow peers.
|
|
audit = ("auditDelivery.enabled=true", "auditDelivery.url=http://receiver.test:8000/v1/events",
|
|
"auditDelivery.secretName=approved-test-sender", "auditDelivery.secretKey=sender-token",
|
|
"auditDelivery.receiverNamespace=test-receiver", "auditDelivery.receiverPodName=test-audit",
|
|
"auditDelivery.receiverPort=8000", "auditDelivery.dnsNamespace=test-dns",
|
|
"auditDelivery.dnsPodLabel=test-dns")
|
|
for auth in ((), ("callerAuth.mode=enforce", "callerAuth.binding=test=system:serviceaccount:test:test")):
|
|
documents = render(*enabled, *audit, *auth)
|
|
pod = deployment(documents)["spec"]["template"]["spec"]
|
|
container = pod["containers"][0]
|
|
args = container["args"]
|
|
assert args[args.index("--audit-url")+1] == "http://receiver.test:8000/v1/events"
|
|
assert args[args.index("--audit-token-file")+1] == "/var/run/secrets/flex-auth-audit/token"
|
|
credential = next(v for v in pod["volumes"] if v["name"] == "flex-auth-audit")["secret"]
|
|
assert credential == {"secretName":"approved-test-sender", "defaultMode":288,
|
|
"items":[{"key":"sender-token", "path":"token"}]}
|
|
assert next(v for v in container["volumeMounts"] if v["name"] == "flex-auth-audit")["readOnly"] is True
|
|
assert not any(d["kind"] == "Secret" for d in documents)
|
|
network = next(d for d in documents if d["kind"] == "NetworkPolicy")["spec"]
|
|
receiver = next(rule for rule in network["egress"] if rule.get("ports") == [{"port":8000,"protocol":"TCP"}])
|
|
assert receiver["to"] == [{"namespaceSelector":{"matchLabels":{"kubernetes.io/metadata.name":"test-receiver"}},
|
|
"podSelector":{"matchLabels":{"app.kubernetes.io/name":"test-audit"}}}]
|
|
dns = next(rule for rule in network["egress"] if len(rule.get("ports",[])) == 2 and rule["ports"][0]["port"] == 53)
|
|
assert dns["to"][0]["namespaceSelector"]["matchLabels"]["kubernetes.io/metadata.name"] == "test-dns"
|
|
assert dns["to"][0]["podSelector"]["matchLabels"]["k8s-app"] == "test-dns"
|
|
render(*enabled, *audit, "args[0]=serve-readback", succeeds=False)
|
|
render(*audit, succeeds=False)
|
|
for invalid in ("auditDelivery.secretName=", "auditDelivery.secretKey=", "auditDelivery.receiverNamespace=",
|
|
"auditDelivery.receiverPodName=", "auditDelivery.receiverPort=0", "auditDelivery.dnsNamespace=",
|
|
"auditDelivery.dnsPodLabel=", "auditDelivery.url=http://user:pass@receiver/v1/events",
|
|
"auditDelivery.url=http://receiver/v1/events?x=1", "auditDelivery.mountPath=relative"):
|
|
render(*enabled, *audit, invalid, succeeds=False)
|
|
for document in base:
|
|
if document["kind"] == "Deployment":
|
|
assert "--audit-url" not in document["spec"]["template"]["spec"]["containers"][0]["args"]
|
|
print("inert scoped audit delivery chart checks passed")
|
|
|
|
# YAML/JSON numeric values must never be truncated into invalid Kubernetes ports.
|
|
# CLI --set integer paths alone do not cover float/bool/null decoder values.
|
|
for port in (8000.5, 1.5, 65535.5, True, False, None, 0, -1, 65536, "8000.5", "named-port"):
|
|
for encoding in ("yaml", "json"):
|
|
command = ["helm", "template", "flex-auth", "charts/flex-auth", "-f", "values/stage2-canary.yaml"]
|
|
for setting in (*enabled, *audit):
|
|
if not setting.startswith("auditDelivery.receiverPort="):
|
|
command += ["--set", setting]
|
|
if encoding == "json":
|
|
import json
|
|
command += ["--set-json", "auditDelivery.receiverPort=" + json.dumps(port)]
|
|
result = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
else:
|
|
# Later -f overrides existing YAML; no --set receiverPort override.
|
|
command = ["helm", "template", "flex-auth", "charts/flex-auth", "-f", "values/stage2-canary.yaml"]
|
|
for setting in (*enabled, *audit):
|
|
if not setting.startswith("auditDelivery.receiverPort="):
|
|
command += ["--set", setting]
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
overlay = Path(directory) / "port.yaml"
|
|
overlay.write_text(yaml.safe_dump({"auditDelivery":{"receiverPort":port}}))
|
|
command += ["-f", str(overlay)]
|
|
result = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
assert result.returncode != 0, f"{encoding} accepted non-integer/out-of-range port {port!r}"
|
|
assert "receiverPort" in result.stderr, result.stderr
|
|
for port in (1, 8000, 65535):
|
|
documents = render(*enabled, *audit, f"auditDelivery.receiverPort={port}")
|
|
policy = next(d for d in documents if d["kind"] == "NetworkPolicy")
|
|
scoped = next(rule for rule in policy["spec"]["egress"]
|
|
if rule.get("to", [{}])[0].get("podSelector",{}).get("matchLabels",{}).get("app.kubernetes.io/name") == "test-audit")
|
|
assert type(scoped["ports"][0]["port"]) is int and scoped["ports"][0]["port"] == port
|
|
print("receiver port integer/type boundary checks passed")
|