railiance-platform/tests/test_custodian_allocation_review.py

"""Offline guard for the CUST-WP-0071-T05 weekly allocation assessment addon.

Asserts the founder's 2026-10-21 boundary: read-only get/list RBAC limited to
the runner's reads, Monday 08:00 Europe/Berlin, Forbid, suspended, retained
PVC that ArgoCD never deletes, not referenced by any live Application, and
vendored sources matching vendor.lock.json.
"""
from pathlib import Path
import subprocess
import sys

import yaml

ROOT = Path(__file__).resolve().parents[1]
ADDON = ROOT / "argocd/platform-addons/custodian-allocation-review"
APPS = ROOT / "argocd/railiance01/applications"
DRAFT = ROOT / "argocd/railiance01/drafts/custodian-allocation-review.application.yaml"
EXPECTED_RULES = {
    ("", "nodes"), ("", "pods"), ("", "persistentvolumeclaims"), ("", "persistentvolumes"),
    ("storage.k8s.io", "storageclasses"), ("metrics.k8s.io", "nodes"), ("metrics.k8s.io", "pods"),
}


def docs():
    kust = yaml.safe_load((ADDON / "kustomization.yaml").read_text())
    out = []
    for resource in kust["resources"]:
        out.extend(d for d in yaml.safe_load_all((ADDON / resource).read_text()) if d)
    return out


def one(kind):
    found = [d for d in docs() if d["kind"] == kind]
    assert len(found) == 1, kind
    return found[0]


def test_rbac_is_read_only_and_exact():
    role = one("ClusterRole")
    seen = set()
    for rule in role["rules"]:
        assert set(rule["verbs"]) == {"get", "list"}
        assert "*" not in rule["apiGroups"] and "*" not in rule["resources"]
        assert "resourceNames" not in rule and "nonResourceURLs" not in rule
        seen |= {(g, r) for g in rule["apiGroups"] for r in rule["resources"]}
    assert seen == EXPECTED_RULES
    binding = one("ClusterRoleBinding")
    assert binding["roleRef"]["name"] == role["metadata"]["name"]
    assert binding["subjects"] == [{"kind": "ServiceAccount", "name": "custodian-allocation-review",
                                    "namespace": "custodian-assessment"}]
    assert not [d for d in docs() if d["kind"] in {"Role", "RoleBinding", "Secret"}]


def test_cronjob_schedule_suspended_and_hardened():
    spec = one("CronJob")["spec"]
    assert spec["schedule"] == "0 8 * * 1" and spec["timeZone"] == "Europe/Berlin"
    assert spec["suspend"] is True and spec["concurrencyPolicy"] == "Forbid"
    pod = spec["jobTemplate"]["spec"]["template"]["spec"]
    assert pod["serviceAccountName"] == "custodian-allocation-review"
    assert pod["securityContext"]["runAsNonRoot"] is True
    (container,) = pod["containers"]
    assert "@sha256:" in container["image"]
    assert container["securityContext"]["readOnlyRootFilesystem"] is True
    assert container["securityContext"]["allowPrivilegeEscalation"] is False
    assert container["resources"]["requests"]["cpu"] == "20m"
    assert not any("secret" in v for v in pod["volumes"])
    claim = next(v for v in pod["volumes"] if "persistentVolumeClaim" in v)["persistentVolumeClaim"]["claimName"]
    pvc = one("PersistentVolumeClaim")
    assert pvc["metadata"]["name"] == claim
    assert pvc["metadata"]["annotations"]["argocd.argoproj.io/sync-options"] == "Prune=false,Delete=false"


def test_not_registered_with_any_live_application():
    for path in APPS.glob("*.y*ml"):
        assert "custodian-allocation-review" not in path.read_text(), path
    draft = yaml.safe_load(DRAFT.read_text())
    assert "automated" not in draft["spec"].get("syncPolicy", {})
    assert "finalizers" not in draft["metadata"]


def test_vendored_sources_match_lock():
    result = subprocess.run([sys.executable, str(ROOT / "scripts/vendor_custodian_allocation_review.py"), "--check"],
                            capture_output=True, text=True)
    assert result.returncode == 0, result.stderr


def test_default_deny_egress_allows_only_dns_api_prometheus_and_hub():
    policies = {d["metadata"]["name"]: d["spec"] for d in docs() if d["kind"] == "NetworkPolicy"}
    deny = policies.pop("custodian-assessment-default-deny")
    assert deny["podSelector"] == {} and set(deny["policyTypes"]) == {"Ingress", "Egress"}
    assert "ingress" not in deny and "egress" not in deny
    (egress,) = policies.values()
    assert egress["policyTypes"] == ["Egress"] and "ingress" not in egress
    allowed = set()
    for rule in egress["egress"]:
        assert rule.get("to") and rule.get("ports"), "every egress rule names peers and ports"
        for peer in rule["to"]:
            if "ipBlock" in peer:
                assert "except" not in peer["ipBlock"]
                key = peer["ipBlock"]["cidr"]
            else:
                key = (peer["namespaceSelector"]["matchLabels"]["kubernetes.io/metadata.name"],
                       tuple(sorted(peer["podSelector"]["matchLabels"].items())))
            allowed |= {(key, p["protocol"], p["port"]) for p in rule["ports"]}
    assert allowed == {
        (("kube-system", (("k8s-app", "kube-dns"),)), "UDP", 53),
        (("kube-system", (("k8s-app", "kube-dns"),)), "TCP", 53),
        ("10.43.0.1/32", "TCP", 443),
        ("92.205.62.239/32", "TCP", 6443),
        (("telemetry", (("app.kubernetes.io/name", "prometheus"),)), "TCP", 9090),
        (("state-hub", (("app", "state-hub"),)), "TCP", 8000),
    }
    env = {e["name"]: e["value"] for e in one("CronJob")["spec"]["jobTemplate"]["spec"]["template"]["spec"]["containers"][0]["env"]}
    assert env["STATE_HUB_URL"] == "http://state-hub.state-hub.svc:8000"
    assert env["PROMETHEUS_URL"].startswith("http://telemetry-prometheus.telemetry.svc:9090")


def test_platform_addons_project_admits_the_addon_kinds():
    project = yaml.safe_load((ROOT / "argocd/railiance01/bootstrap/02-railiance-platform-addons-project.yaml").read_text())
    allowed = {(k["group"], k["kind"]) for k in project["spec"]["namespaceResourceWhitelist"]}
    allowed |= {(k["group"], k["kind"]) for k in project["spec"]["clusterResourceWhitelist"]}
    for doc in docs():
        group = doc["apiVersion"].rpartition("/")[0]
        assert (group, doc["kind"]) in allowed, doc["kind"]
    assert ("", "ConfigMap") in allowed  # generated source ConfigMap