567d2c671b
Namespace custodian-agents, ServiceAccount custodian-agent (no automount, no token Secret), curated read-only ClusterRole custodian-supervised-observer and its binding, per the founder decision of 2026-10-20. Offline pytest guard asserts no forbidden resource, no write verb, no wildcard. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 605711@bnt-lap001 Assistant-Session: 1f5e2256-181f-49ed-b7b1-a1c34bf37a61
110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
"""Offline guard for the custodian agent observer identity (RPF-WP-0062, CUST-WP-0073-T02).
|
|
|
|
Parses the GitOps manifests and asserts the founder's 2026-10-20 boundary:
|
|
read-only verbs, curated resources, no forbidden resources, no wildcards,
|
|
no static token Secret, no token automount, and a manual, pinned Application.
|
|
"""
|
|
from pathlib import Path
|
|
import re
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADDON = ROOT / "argocd/platform-addons/custodian-agents"
|
|
APP = ROOT / "argocd/railiance01/applications/custodian-agents.application.yaml"
|
|
|
|
READ_VERBS = {"get", "list", "watch"}
|
|
FORBIDDEN_RESOURCES = {
|
|
"secrets", "configmaps", "pods/log", "pods/exec", "pods/attach",
|
|
"pods/portforward", "serviceaccounts/token", "serviceaccounts", "nodes/proxy",
|
|
"services/proxy", "pods/proxy", "tokenreviews", "subjectaccessreviews",
|
|
"clusterroles", "clusterrolebindings", "roles", "rolebindings",
|
|
}
|
|
EXPECTED = {
|
|
"": {"namespaces", "nodes", "pods", "services", "endpoints", "events",
|
|
"persistentvolumeclaims"},
|
|
"apps": {"deployments", "statefulsets", "daemonsets", "replicasets"},
|
|
"discovery.k8s.io": {"endpointslices"},
|
|
"networking.k8s.io": {"ingresses"},
|
|
"events.k8s.io": {"events"},
|
|
"argoproj.io": {"applications", "appprojects"},
|
|
"postgresql.cnpg.io": {"clusters", "backups", "scheduledbackups"},
|
|
}
|
|
EXPECTED_NON_RESOURCE = {"/version", "/healthz", "/livez", "/readyz"}
|
|
|
|
|
|
def _docs():
|
|
kust = yaml.safe_load((ADDON / "kustomization.yaml").read_text())
|
|
docs = []
|
|
for res in kust["resources"]:
|
|
docs.extend(d for d in yaml.safe_load_all((ADDON / res).read_text()) if d)
|
|
return docs
|
|
|
|
|
|
def _one(kind):
|
|
found = [d for d in _docs() if d["kind"] == kind]
|
|
assert len(found) == 1, kind
|
|
return found[0]
|
|
|
|
|
|
def test_only_expected_kinds_no_secret():
|
|
kinds = sorted(d["kind"] for d in _docs())
|
|
assert kinds == ["ClusterRole", "ClusterRoleBinding", "Namespace", "ServiceAccount"]
|
|
for path in ADDON.iterdir():
|
|
assert "kind: Secret" not in path.read_text(), path
|
|
|
|
|
|
def test_service_account_names_and_no_automount():
|
|
sa = _one("ServiceAccount")
|
|
assert sa["metadata"] == {**sa["metadata"], "name": "custodian-agent",
|
|
"namespace": "custodian-agents"}
|
|
assert sa["automountServiceAccountToken"] is False
|
|
assert "secrets" not in sa
|
|
assert _one("Namespace")["metadata"]["name"] == "custodian-agents"
|
|
|
|
|
|
def test_cluster_role_read_only_curated_no_wildcards():
|
|
role = _one("ClusterRole")
|
|
assert role["metadata"]["name"] == "custodian-supervised-observer"
|
|
assert "aggregationRule" not in role
|
|
granted, non_resource = {}, set()
|
|
for rule in role["rules"]:
|
|
verbs = set(rule["verbs"])
|
|
assert verbs <= READ_VERBS, rule
|
|
values = [*rule["verbs"], *rule.get("apiGroups", []), *rule.get("resources", []),
|
|
*rule.get("nonResourceURLs", []), *rule.get("resourceNames", [])]
|
|
assert not any("*" in v for v in values), rule
|
|
if "nonResourceURLs" in rule:
|
|
assert verbs == {"get"} and not rule.get("resources"), rule
|
|
non_resource |= set(rule["nonResourceURLs"])
|
|
continue
|
|
assert verbs == READ_VERBS, rule
|
|
for group in rule["apiGroups"]:
|
|
for resource in rule["resources"]:
|
|
assert resource not in FORBIDDEN_RESOURCES, resource
|
|
assert "/" not in resource, f"no subresources: {resource}"
|
|
granted.setdefault(group, set()).add(resource)
|
|
assert granted == EXPECTED
|
|
assert non_resource == EXPECTED_NON_RESOURCE
|
|
|
|
|
|
def test_binding_targets_only_the_agent():
|
|
binding = _one("ClusterRoleBinding")
|
|
assert binding["roleRef"] == {"apiGroup": "rbac.authorization.k8s.io",
|
|
"kind": "ClusterRole",
|
|
"name": "custodian-supervised-observer"}
|
|
assert binding["subjects"] == [{"kind": "ServiceAccount", "name": "custodian-agent",
|
|
"namespace": "custodian-agents"}]
|
|
|
|
|
|
def test_application_manual_and_pinned():
|
|
if not APP.exists():
|
|
pytest.skip("Application not yet placed")
|
|
app = yaml.safe_load(APP.read_text())
|
|
spec = app["spec"]
|
|
assert spec["project"] == "railiance-platform-addons"
|
|
assert spec["source"]["path"] == "argocd/platform-addons/custodian-agents"
|
|
assert re.fullmatch(r"[0-9a-f]{40}", spec["source"]["targetRevision"])
|
|
assert "automated" not in spec.get("syncPolicy", {})
|
|
assert "finalizers" not in app["metadata"]
|