home-network-k3s-impl/conformance/tests/l0-static/least-privilege-rbac/check.sh

#!/usr/bin/env bash
# =============================================================================
# L0 static check (offline, no cluster) — least-privilege-rbac
# Serves: R-SEC-02 ("The platform MUST follow least-privilege for workload
#         identities and network reach." — spec docs/00-vision-and-scope.md;
#         detail in docs/07 "least privilege ... least network reach").
#
# The requirement has TWO halves and this check covers the STATIC half of each,
# at the layer docs/14 puts them ("L0 Static/policy | Pre-apply, in CI").
#
# GIVEN the repository as git tracks it,
# WHEN  every tracked RBAC object and every least-network-reach policy is inspected,
# THEN — workload identity —
#       (1) no in-repo Role/ClusterRole grants a wildcard apiGroup, resource or verb,
#           and none grants escalate/bind/impersonate (the privilege-escalation verbs);
#       (2) no (Cluster)RoleBinding binds a ServiceAccount — a WORKLOAD identity — to
#           `cluster-admin` or to any in-repo role holding wildcard rules. Human/Group
#           subjects are out of scope: R-SEC-02 constrains workload identities, and the
#           operator's own admin path is an OIDC group binding, deliberately;
#       (3) every ServiceAccount subject is namespaced (no cluster-wide ambiguity);
#       (4) the conformance runner's own SA — this workstream's PRIVILEGED TEST
#           IDENTITY, whose least-privilege authoring docs/14 explicitly delegates to
#           WS-TEST — is bound only to its own scoped ClusterRole, carries no wildcard
#           and no privilege-escalation verb, and grants nothing over RBAC objects
#           themselves (no rbac.authorization.k8s.io rules);
#     — network reach —
#       (5) the policies that implement least network reach exist, and each is Enforce
#           (or carries the Flux token that substitutes to Enforce): the default-deny
#           baseline generator, the baseline SHAPE validator, the host-namespace/
#           privilege restriction, and the egress-class label requirement;
#       (6) each of those policies is actually EXERCISED by the L0 Kyverno suite with a
#           NEGATIVE row (result: fail) — a policy nobody proves rejects anything is a
#           declaration, not a gate.
#
# WHAT THIS DOES *NOT* PROVE (honest gap, cluster layer — R-SEC-02 is NOT fully covered):
#   - RUNTIME least privilege of workload identities: that the API server actually
#     denies the conformance SA (or any workload SA) the verbs it was not granted.
#     That needs a live cluster (L2) and no test asserts it today.
#   - RUNTIME least network reach: that pod-to-pod traffic is actually denied. The
#     behaviour is exercised by tests/l2-behavioral/{default-deny,deny-undeclared},
#     but those are tagged R-NET-01/R-NET-02 and are cluster-gated (skipped in Phase 0);
#     nothing proves it under an R-SEC-02 tag.
#   - Helm-chart-generated RBAC (e.g. what a HelmRelease's chart creates in-cluster) is
#     not in the repo, so it is outside this static scan entirely.
# =============================================================================
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT="$(cd "${DIR}/../../../.." && pwd)"        # repo root
PYTHON="${PYTHON:-python3}"

cd "${ROOT}"
"${PYTHON}" - "${ROOT}" <<'PY'
import os, re, subprocess, sys
from pathlib import Path

try:
    import yaml
except ImportError:                                  # pragma: no cover
    print("check FAIL: PyYAML is required (same dependency as catalog/render.py)",
          file=sys.stderr)
    raise SystemExit(1)

ROOT = Path(sys.argv[1])
FAILURES: list[str] = []
CHECKS = 0


def check(cond, msg):
    global CHECKS
    CHECKS += 1
    if not cond:
        FAILURES.append(msg)


def tracked() -> list[str]:
    out = subprocess.run(["git", "-C", str(ROOT), "ls-files", "-z"],
                         capture_output=True, text=True, check=True).stdout
    return [p for p in out.split("\0") if p]


def docs_of(rel: str):
    try:
        return [d for d in yaml.safe_load_all((ROOT / rel).read_text(errors="replace"))
                if isinstance(d, dict)]
    except Exception:
        return []


FILES = tracked()
YAML_FILES = [f for f in FILES if re.search(r"\.(ya?ml)(\.example)?$", f)]
check(len(YAML_FILES) > 50, f"only {len(YAML_FILES)} YAML files seen — not a real checkout")

ESCALATION_VERBS = {"escalate", "bind", "impersonate"}
PRIVILEGED_ROLES = {"cluster-admin", "admin", "edit"}

roles: dict[tuple[str, str], dict] = {}   # (kind, name) -> {"file":…, "wildcard":bool, …}
bindings: list[dict] = []

for rel in YAML_FILES:
    for d in docs_of(rel):
        kind = d.get("kind")
        name = (d.get("metadata") or {}).get("name")
        if kind in ("Role", "ClusterRole"):
            wildcard, escalating, rbac_over_rbac = False, set(), False
            for r in d.get("rules") or []:
                if not isinstance(r, dict):
                    continue
                for field in ("apiGroups", "resources", "verbs"):
                    if "*" in (r.get(field) or []):
                        wildcard = True
                escalating |= set(v for v in (r.get("verbs") or [])) & ESCALATION_VERBS
                if "rbac.authorization.k8s.io" in (r.get("apiGroups") or []):
                    rbac_over_rbac = True
            roles[(kind, name)] = {"file": rel, "wildcard": wildcard,
                                   "escalating": escalating, "rbac": rbac_over_rbac,
                                   "doc": d}
            # (1) no wildcards, no privilege-escalation verbs.
            check(not wildcard,
                  f"{rel}: {kind}/{name} grants a wildcard apiGroup/resource/verb — that is "
                  "not least-privilege (R-SEC-02)")
            check(not escalating,
                  f"{rel}: {kind}/{name} grants privilege-escalation verb(s) "
                  f"{sorted(escalating)} (R-SEC-02)")
        elif kind in ("RoleBinding", "ClusterRoleBinding"):
            bindings.append({"file": rel, "name": name, "doc": d})

check(len(roles) > 0, "no Role/ClusterRole found in the repo — the RBAC scan is not looking at it")
check(len(bindings) > 0, "no RoleBinding/ClusterRoleBinding found — the RBAC scan is not looking at it")

for b in bindings:
    d, rel, name = b["doc"], b["file"], b["name"]
    ref = d.get("roleRef") or {}
    ref_kind, ref_name = ref.get("kind"), ref.get("name")
    sa_subjects = [s for s in (d.get("subjects") or [])
                   if isinstance(s, dict) and s.get("kind") == "ServiceAccount"]
    local = roles.get((ref_kind, ref_name))
    ref_is_privileged = (ref_name in PRIVILEGED_ROLES) or bool(local and local["wildcard"])
    # (2) a WORKLOAD identity must never hold a privileged role.
    check(not (sa_subjects and ref_is_privileged),
          f"{rel}: {d.get('kind')}/{name} binds ServiceAccount(s) "
          f"{[s.get('name') for s in sa_subjects]} to the privileged role {ref_name!r} — "
          "workload identities must be least-privilege (R-SEC-02)")
    # (3) SA subjects must name their namespace (a ClusterRoleBinding without one is
    #     ambiguous and, in a RoleBinding, silently defaults to the binding's namespace).
    for s in sa_subjects:
        check(bool(s.get("namespace")) or d.get("kind") == "RoleBinding",
              f"{rel}: {d.get('kind')}/{name} has a ServiceAccount subject "
              f"{s.get('name')!r} with no namespace (R-SEC-02)")

# --------------------------------- (4) our own privileged test identity, in detail
RBAC_FILE = "conformance/incluster/rbac.yaml"
check(RBAC_FILE in FILES, f"{RBAC_FILE} is missing — the conformance runner has no authored RBAC")
runner_role = roles.get(("ClusterRole", "conformance-runner"))
check(runner_role is not None,
      f"{RBAC_FILE}: no ClusterRole/conformance-runner — the test identity's grants are undefined")
if runner_role:
    check(runner_role["file"] == RBAC_FILE,
          f"ClusterRole/conformance-runner is defined in {runner_role['file']}, not {RBAC_FILE}")
    check(not runner_role["wildcard"], "ClusterRole/conformance-runner uses a wildcard (R-SEC-02)")
    check(not runner_role["escalating"],
          "ClusterRole/conformance-runner grants a privilege-escalation verb (R-SEC-02)")
    check(not runner_role["rbac"],
          "ClusterRole/conformance-runner grants rights over RBAC objects themselves — a test "
          "identity that can rewrite RBAC is not least-privilege (R-SEC-02)")

runner_bindings = [b for b in bindings
                   if any(s.get("kind") == "ServiceAccount" and s.get("name") == "conformance-runner"
                          for s in (b["doc"].get("subjects") or []))]
check(len(runner_bindings) >= 1, "the conformance-runner ServiceAccount has no binding at all")
for b in runner_bindings:
    ref = b["doc"].get("roleRef") or {}
    check(ref.get("name") == "conformance-runner",
          f"{b['file']}: conformance-runner is bound to {ref.get('name')!r} instead of its own "
          "scoped ClusterRole (R-SEC-02)")
    for s in b["doc"].get("subjects") or []:
        if s.get("kind") == "ServiceAccount":
            check(s.get("namespace") == "conformance",
                  f"{b['file']}: conformance-runner subject is in namespace "
                  f"{s.get('namespace')!r}, expected 'conformance' (R-SEC-02)")

# --------------------- (5)/(6) least NETWORK REACH: the policies exist, Enforce, and
#                               are proven to reject something at L0.
# Validating policies: must Enforce AND be proven to REJECT something at L0.
REQUIRED_POLICIES = {
    "validate-default-deny-shape":   "a baseline that leaves a direction open is rejected",
    "restrict-host-namespaces":      "host namespaces / privileged containers are rejected",
    "require-egress-class-label":    "workloads must declare their egress class",
}
# Generating policies have no validationFailureAction and no negative row: the proof
# that they close network reach is that the L0 suite asserts the resource they GENERATE.
REQUIRED_GENERATE_POLICIES = {
    "require-default-deny-baseline": "every namespace gets a default-deny baseline",
}
policies: dict[str, dict] = {}
for rel in YAML_FILES:
    for d in docs_of(rel):
        if d.get("kind") in ("ClusterPolicy", "Policy"):
            policies[(d.get("metadata") or {}).get("name")] = {"file": rel, "doc": d}

for pname, why in REQUIRED_POLICIES.items():
    p = policies.get(pname)
    check(p is not None, f"policy {pname!r} is missing — nothing enforces: {why} (R-SEC-02)")
    if not p:
        continue
    action = str((p["doc"].get("spec") or {}).get("validationFailureAction", ""))
    check(action == "Enforce" or action.startswith("${"),
          f"{p['file']}: policy {pname!r} validationFailureAction is {action!r} — a least-reach "
          "policy in Audit mode blocks nothing (R-SEC-02)")

negatives: dict[str, int] = {}
suites = [f for f in FILES if os.path.basename(f) == "kyverno-test.yaml"]
check(len(suites) > 0, "no kyverno-test.yaml — the L0 policy gate is not exercised at all")
for suite in suites:
    for d in docs_of(suite):
        for r in d.get("results") or []:
            if isinstance(r, dict) and r.get("result") == "fail":
                negatives[r.get("policy")] = negatives.get(r.get("policy"), 0) + 1

for pname, why in REQUIRED_POLICIES.items():
    check(negatives.get(pname, 0) > 0,
          f"no L0 test row asserts that policy {pname!r} REJECTS a violating resource "
          f"(result: fail) — unproven gate for: {why} (R-SEC-02)")

# The generating half: the policy must exist, and the L0 suite must assert the resource
# it generates (a `generatedResource` row that passes) — otherwise nothing proves a new
# namespace actually arrives with its reach closed.
for pname, why in REQUIRED_GENERATE_POLICIES.items():
    p = policies.get(pname)
    check(p is not None, f"policy {pname!r} is missing — nothing enforces: {why} (R-SEC-02)")
    generated_rows = []
    for suite in suites:
        for d in docs_of(suite):
            for r in d.get("results") or []:
                if isinstance(r, dict) and r.get("policy") == pname \
                        and r.get("generatedResource") and r.get("result") == "pass":
                    generated_rows.append((suite, r.get("generatedResource")))
    check(len(generated_rows) > 0,
          f"no L0 test row asserts the resource generated by {pname!r} — unproven gate for: "
          f"{why} (R-SEC-02)")
    for suite, gen in generated_rows:
        gp = os.path.normpath(os.path.join(os.path.dirname(suite), str(gen)))
        check(gp in FILES, f"{suite}: generatedResource {gen!r} does not resolve to a tracked file")
        golden = "\n".join(l for l in (ROOT / gp).read_text(errors="replace").splitlines()
                           if not l.lstrip().startswith("#")) if gp in FILES else ""
        check("NetworkPolicy" in golden and re.search(r"policyTypes:", golden),
              f"{gp}: the asserted default-deny baseline is not a shaped NetworkPolicy (R-SEC-02)")

# The host-namespace policy must prove BOTH of its rules reject, not just one.
host_rules = set()
for suite in suites:
    for d in docs_of(suite):
        for r in d.get("results") or []:
            if isinstance(r, dict) and r.get("policy") == "restrict-host-namespaces" \
                    and r.get("result") == "fail":
                host_rules.add(r.get("rule"))
check({"disallow-host-namespaces", "disallow-privileged-containers"} <= host_rules,
      "the L0 suite does not prove BOTH restrict-host-namespaces rules reject a violating pod "
      f"(negative rows seen for: {sorted(host_rules)}) — R-SEC-02")

# --------------------------------------------------------------------------- verdict
if FAILURES:
    print(f"CHECK FAIL (least-privilege-rbac): {len(FAILURES)} of {CHECKS} assertions failed",
          file=sys.stderr)
    for f in FAILURES:
        print(f"  - {f}", file=sys.stderr)
    raise SystemExit(1)
print(f"check OK: least-privilege-rbac — {CHECKS} assertions over {len(roles)} role(s), "
      f"{len(bindings)} binding(s), {len(REQUIRED_POLICIES)} least-reach policies")
PY