diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 1692a7530..5546220cd 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -41,6 +41,7 @@ jobs: - 'talos/**' - 'scripts/validate-naming.py' - 'scripts/validate-embedded-json.py' + - 'scripts/generate-kubescape-exceptions.py' - '.github/workflows/ci.yaml' talos: - 'talos/**' @@ -152,25 +153,32 @@ jobs: # FAILS if a change drops the posture below the threshold, so new # findings must be fixed (or justified) before merge. # - # The threshold (85) is a regression FLOOR, not the live score. The - # Kubescape compliance score is **environment-dependent** — the same - # binary on the same manifests scores higher locally (macOS, ≈94%) than on - # the Linux CI runner (≈87%), a gap that isn't the render, the framework - # cache, or the merge content — and the absolute value also shifts with the - # ksail render, so it can't be reproduced exactly offline. CI is the source - # of truth; the gate sits a few points under the observed CI value to catch - # real regressions without flaking on that variance. (NB the floor may need - # re-baselining after a ksail version bump.) + # The scan consumes the platform's justified exceptions: the + # ClusterSecurityException CRs (k8s/bases/infrastructure/ + # cluster-security-exceptions/ — the single source of truth, also + # consumed in-cluster by the kubescape-operator) are converted at scan + # time into Kubescape's native exceptions format by + # scripts/generate-kubescape-exceptions.py (#2264). With runtime- + # enforced and except-only findings suppressed, the threshold gates the + # residual REAL posture instead of sitting ~15 points under it. # - # It is not higher because the residual findings are either runtime- - # enforced (Kyverno securityContext/limits, CiliumNetworkPolicy) or - # genuinely unfixable — notably C-0002, the KubeVirt operator's pods/exec - # RBAC, which it needs to manage VMs and can only be excepted. Native scan - # exceptions have now shipped (devantler-tech/ksail#5369); wiring the - # platform's ClusterSecurityException CRs into the scan and ratcheting the - # threshold toward 100 is tracked in #2264. Until then, ratchet up as - # genuine gaps are fixed; never lower it. - run: ksail workload scan --framework nsa --compliance-threshold 85 + # The threshold is a regression FLOOR, not the live score. The + # Kubescape compliance score has historically been environment- + # dependent (Linux CI runner vs macOS) and shifts with the ksail + # render, so CI is the source of truth: the observed CI score with + # exceptions applied is 98.87% (2026-07-11, ksail 7.165.2 — matching + # the local value on that date), and the gate sits a few points under + # it to catch real regressions without flaking on run-to-run + # variance. (NB the floor may need re-baselining after a ksail + # version bump.) + # + # Known un-excepted residual: C-0012 on the local-dev MinIO Deployment + # (deliberate dev-only credential — tracked for a root-cause fix, not + # an exception). Ratchet the floor up as genuine gaps close; never + # lower it. + run: | + python3 scripts/generate-kubescape-exceptions.py -o /tmp/kubescape-exceptions.json + ksail workload scan --framework nsa --exceptions /tmp/kubescape-exceptions.json --compliance-threshold 95 naming: name: 🏷️ Validate Naming Conventions diff --git a/AGENTS.md b/AGENTS.md index 62cf1ee5c..b3fc5f534 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -95,12 +95,12 @@ kubectl apply --dry-run=client -f `flux check` and other cluster-dependent checks require a running cluster — they are **not** part of static validation and should not be run during maintenance. -CI runs **static manifest validation** on PRs that touch k8s-related paths (`k8s/**`, `ksail*.yaml`, `.sops.yaml`, `talos*/**`, or `ci.yaml`) — the `validate` job in `.github/workflows/ci.yaml` first json-parses every registered embedded-JSON ConfigMap key via [`scripts/validate-embedded-json.py`](scripts/validate-embedded-json.py) (keys listed in the script's `REGISTERED_KEYS` or ending in `.json` — schema validation treats such blobs as opaque strings, so a stray comma would otherwise ship silently; run it locally when touching one), then runs `ksail workload validate` for both the local and prod overlays plus a Kubescape `ksail workload scan --framework nsa --compliance-threshold 85`. It is fast, needs no secrets (so it runs on fork PRs too), and starts no cluster. PRs touching `talos/**` or `talos-local/**` additionally run the `validate-talos` job: it renders the machine config with every patch applied (placeholder values stand in for env-expanded secrets like `${WG_SERVER_PRIVATE_KEY}`) and `talosctl validate`s the result, so a broken patch or an empty env expansion fails the PR event instead of the merge group's deploy (#2477). There is **no longer a full-cluster system test**: the local Docker cluster is a thin manual test-bed (see [Local Development Cluster](#local-development-cluster)), not a CI prod stand-in. +CI runs **static manifest validation** on PRs that touch k8s-related paths (`k8s/**`, `ksail*.yaml`, `.sops.yaml`, `talos*/**`, the validation scripts `scripts/validate-naming.py` / `scripts/validate-embedded-json.py` / `scripts/generate-kubescape-exceptions.py`, or `ci.yaml` — the authoritative list is the `k8s` filter in `.github/workflows/ci.yaml`) — the `validate` job in `.github/workflows/ci.yaml` first json-parses every registered embedded-JSON ConfigMap key via [`scripts/validate-embedded-json.py`](scripts/validate-embedded-json.py) (keys listed in the script's `REGISTERED_KEYS` or ending in `.json` — schema validation treats such blobs as opaque strings, so a stray comma would otherwise ship silently; run it locally when touching one), then runs `ksail workload validate` for both the local and prod overlays plus a Kubescape scan (`scripts/generate-kubescape-exceptions.py` converts the `ClusterSecurityException` CRs into Kubescape's exceptions format, then `ksail workload scan --framework nsa --exceptions --compliance-threshold ` gates on the score — the exact floor lives in `ci.yaml`). It is fast, needs no secrets (so it runs on fork PRs too), and starts no cluster. PRs touching `talos/**` or `talos-local/**` additionally run the `validate-talos` job: it renders the machine config with every patch applied (placeholder values stand in for env-expanded secrets like `${WG_SERVER_PRIVATE_KEY}`) and `talosctl validate`s the result, so a broken patch or an empty env expansion fails the PR event instead of the merge group's deploy (#2477). There is **no longer a full-cluster system test**: the local Docker cluster is a thin manual test-bed (see [Local Development Cluster](#local-development-cluster)), not a CI prod stand-in. The scan is a **hard gate**: it fails the PR if the NSA compliance score drops below the threshold, so new findings must be fixed or justified before merge. Two non-obvious limits: -- **ksail is Renovate-managed** (the Setup step, grouped `ksail` with the deploy pins). It was previously frozen at 7.65.0 because 7.66.x parallelised the in-process Helm render and made it racy — `ksail workload validate` threw varying YAML parse errors and the scan score swung run-to-run. That race is resolved upstream ([devantler-tech/ksail#5371](https://github.com/devantler-tech/ksail/issues/5371), closed), so the pin is lifted back onto the latest release. Tripwire: if `validate`/`scan` swing run-to-run again, re-pin to a known-good version and reopen #5371. -- **The threshold (85) is a regression floor, not the actual score.** The Kubescape compliance score is **environment-dependent**: the same ksail binary on the same manifests reports ≈**87%** on the Linux CI runner but ≈**94%** locally (macOS) — a gap that is *not* the render mode, the framework cache, or PR-merge content (all ruled out) — and the absolute value also shifts with the ksail render, so **CI is the source of truth and the score can't be reproduced exactly offline** (re-baseline the floor after a ksail bump). It is below 100 because the residual findings are either runtime-enforced (Kyverno securityContext/limits mutation, `CiliumNetworkPolicy`) — invisible to a static scan — or genuinely unfixable, notably **C-0002** (the KubeVirt operator's `pods/exec` RBAC, which it needs to manage VMs and can only be excepted). The platform documents these as kubescape `ClusterSecurityException` CRs (`k8s/bases/infrastructure/cluster-security-exceptions/`); native scan exceptions have now shipped ([ksail#5369](https://github.com/devantler-tech/ksail/issues/5369)), and wiring them in to ratchet the threshold toward 100 is tracked in [#2264](https://github.com/devantler-tech/platform/issues/2264). Until then, **ratchet up** as genuine gaps are fixed; never lower it. +- **ksail is Renovate-managed** (the Setup step, grouped `ksail` with the deploy pins). It was previously frozen at 7.65.0 because 7.66.x parallelised the in-process Helm render and made it racy — two distinct symptoms of the same regression: `ksail workload validate` non-deterministically corrupted the render with varying YAML parse errors ([devantler-tech/ksail#5362](https://github.com/devantler-tech/ksail/issues/5362), closed — contained since KSail v7.163.1 by the [ksail#5978](https://github.com/devantler-tech/ksail/issues/5978) stream-splitting fix, which is what let the temporary `--skip-helm-render` workaround be removed), and the scan's compliance score swung run-to-run ([devantler-tech/ksail#5371](https://github.com/devantler-tech/ksail/issues/5371), closed). Both are resolved upstream, so the pin is lifted back onto the latest release. Tripwire (kept in sync with the comments in `.github/workflows/ci.yaml`): if `validate` output or the `scan` score varies run-to-run again, re-add `--skip-helm-render` and reopen ksail#5362 (or re-pin to a known-good version, reopening #5371 if only the score swings). +- **The threshold is a regression floor, not the actual score — and the scan runs WITH the platform's justified exceptions applied.** The `ClusterSecurityException` CRs (`k8s/bases/infrastructure/cluster-security-exceptions/` — the single source of truth, consumed in-cluster by the kubescape-operator) are converted at scan time into Kubescape's native exceptions format by [`scripts/generate-kubescape-exceptions.py`](scripts/generate-kubescape-exceptions.py) (fail-closed: an unrecognised CR shape aborts the scan step rather than silently dropping or widening an exception; unit tests in `scripts/tests/`), so runtime-enforced (Kyverno mutation, `CiliumNetworkPolicy`) and except-only findings (e.g. **C-0002**, the KubeVirt operator's `pods/exec` RBAC) no longer depress the score and the floor gates the residual REAL posture (#2264). The score has historically been **environment-dependent** (Linux CI runner vs macOS — a gap that is *not* the render mode, the framework cache, or PR-merge content, all ruled out) and shifts with the ksail render, so **CI is the source of truth** (re-baseline the floor after a ksail bump); the observed CI reference with exceptions applied is **≈98.9%** (2026-07-11, ksail 7.165.2), with the floor a few points under it. A new justified exception is added as a CSE CR (kind+name-scoped, minimal — see the existing CRs' conventions), never by lowering the floor: **ratchet up** as genuine gaps close; never lower it. ## Local Development Cluster diff --git a/k8s/bases/infrastructure/cluster-security-exceptions/exec-into-container-rbac.yaml b/k8s/bases/infrastructure/cluster-security-exceptions/exec-into-container-rbac.yaml index 12360c2f0..a5f7a38f8 100644 --- a/k8s/bases/infrastructure/cluster-security-exceptions/exec-into-container-rbac.yaml +++ b/k8s/bases/infrastructure/cluster-security-exceptions/exec-into-container-rbac.yaml @@ -29,6 +29,9 @@ # - Flux — kustomize-controller and helm-controller (both bound via # cluster-reconciler-flux-system) and flux-operator hold cluster-admin; # arbitrary-manifest reconciliation is the GitOps trust model. +# - KubeVirt — the kubevirt-operator ClusterRole explicitly grants +# `pods/exec` (vendored upstream bundle); virt-operator needs it to manage +# VM lifecycles. AGENTS.md documents this finding as except-only. # # NOT covered (intentionally): the two tenant Flux-impersonation SAs # (wedding-app, ascoachingogvaner) inherit pods/exec only via the built-in `edit` @@ -41,11 +44,13 @@ metadata: spec: reason: >- Velero (all-or-nothing backup/restore RBAC + cluster-admin), CloudNativePG - (operator execs into Postgres pods — explicit pods/exec grant), and the Flux + (operator execs into Postgres pods — explicit pods/exec grant), the Flux GitOps reconcilers kustomize-controller, helm-controller (both via the cluster-reconciler-flux-system binding), and flux-operator (cluster-admin - trust model) legitimately hold pods/exec by design. The specific RBAC bindings are - matched by kind+name so C-0002 still catches any new exec grant. + trust model), and KubeVirt's virt-operator (explicit pods/exec grant to + manage VM lifecycles) legitimately hold pods/exec by design. The specific + RBAC bindings are matched by kind+name so C-0002 still catches any new + exec grant. posture: - controlID: C-0002 action: ignore @@ -73,3 +78,10 @@ spec: - apiGroup: rbac.authorization.k8s.io kind: ClusterRoleBinding name: ^flux-operator$ + # KubeVirt — virt-operator manages VM lifecycles via an explicit + # pods/exec grant in the kubevirt-operator ClusterRole (vendored + # upstream operator bundle), bound by this CRB. Documented as + # except-only in AGENTS.md's scan-threshold rationale. + - apiGroup: rbac.authorization.k8s.io + kind: ClusterRoleBinding + name: ^kubevirt-operator$ diff --git a/scripts/generate-kubescape-exceptions.py b/scripts/generate-kubescape-exceptions.py new file mode 100644 index 000000000..69bfa5226 --- /dev/null +++ b/scripts/generate-kubescape-exceptions.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +"""Generate a Kubescape exceptions file from the ClusterSecurityException CRs. + +The platform documents every justified posture finding as a +`ClusterSecurityException` CR in +`k8s/bases/infrastructure/cluster-security-exceptions/` — that directory is the +single source of truth for what is excepted and why. The in-cluster +kubescape-operator consumes the CRs directly, but the offline CI scan +(`ksail workload scan --exceptions `) takes Kubescape's native format: a +JSON array of PostureExceptionPolicy objects. This script derives that file +from the CRs at scan time, so CI and the cluster can never disagree about the +exception set. + +Fail-closed by design: any CR shape this converter does not recognise (an +unknown `spec.match` key, a posture action other than `ignore`, a +namespaceSelector that isn't the `kubernetes.io/metadata.name In [...]` +expression) aborts with a non-zero exit instead of silently dropping or +widening an exception. + +Requires PyYAML (preinstalled on GitHub-hosted runners; on macOS the system +`/usr/bin/python3` ships it). Run from anywhere: + + python3 scripts/generate-kubescape-exceptions.py -o /tmp/exceptions.json +""" + +import argparse +import json +import os +import re +import sys + +try: + import yaml +except ImportError: # pragma: no cover + sys.exit( + "PyYAML is required (preinstalled on GitHub runners; locally try " + "/usr/bin/python3 or `python3 -m pip install pyyaml`)." + ) + +DEFAULT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "k8s", + "bases", + "infrastructure", + "cluster-security-exceptions", +) +NAMESPACE_NAME_KEY = "kubernetes.io/metadata.name" + + +def anchor(value, path, name): + """Anchor a plain value into an exact-match regex; keep explicit regexes. + + CR authors write resource `name` fields as anchored regexes already + (`^velero-server$`) but plain `kind`/controlID values; Kubescape treats + every designator attribute and controlID as a regex, so an unanchored + plain value would substring-match (C-0002 would also match C-0020). + A value anchored on only one end (`^foo` or `foo$`) is still + substring-matchable at the open end, so it fails closed instead of + passing through unescaped. + """ + if value.startswith("^") and value.endswith("$"): + return value + if value.startswith("^") or value.endswith("$"): + fail(path, name, f"partially anchored regex value {value!r}") + return f"^{re.escape(value)}$" + + +def fail(path, name, message): + """Abort the run with a non-zero exit naming the offending CR.""" + sys.exit(f"{path}: ClusterSecurityException {name!r}: {message}") + + +def convert_namespace_selector(selector, path, name): + """Map a namespaceSelector to a single namespace-regex designator.""" + unknown = set(selector) - {"matchExpressions"} + if unknown: + fail(path, name, f"unsupported namespaceSelector keys {sorted(unknown)}") + expressions = selector.get("matchExpressions") or [] + if len(expressions) != 1: + fail(path, name, "expected exactly one namespaceSelector matchExpression") + expr = expressions[0] + if expr.get("key") != NAMESPACE_NAME_KEY or expr.get("operator") != "In": + fail( + path, + name, + f"only `{NAMESPACE_NAME_KEY} In [...]` matchExpressions are supported", + ) + values = expr.get("values") or [] + if not values: + fail(path, name, "namespaceSelector matchExpression has no values") + pattern = "^(" + "|".join(re.escape(v) for v in values) + ")$" + return [{"designatorType": "Attributes", "attributes": {"namespace": pattern}}] + + +def convert_resources(resources, path, name): + """Map `match.resources` entries to Attributes designators. + + `apiGroup` is intentionally dropped: PostureExceptionPolicy designator + attributes have no apiGroup field, and the anchored kind+name pair is + what scopes the exception (same mapping the in-cluster operator applies). + """ + designators = [] + for entry in resources: + # The CRD's match.resources[] schema allows exactly apiGroup, kind and + # name — a namespace key would be dropped in-cluster, so accepting it + # here would let the CI exception diverge from what the operator + # applies. Fail closed on it like any other unknown key. + unknown = set(entry) - {"apiGroup", "kind", "name"} + if unknown: + fail(path, name, f"unsupported match.resources keys {sorted(unknown)}") + if "kind" not in entry: + fail(path, name, "match.resources entry without a kind") + attributes = {"kind": anchor(entry["kind"], path, name)} + if "name" in entry: + attributes["name"] = anchor(entry["name"], path, name) + designators.append( + {"designatorType": "Attributes", "attributes": attributes} + ) + return designators + + +def resolve_match(match, path, name): + """Map `spec.match` to designators (resources / namespaceSelector / all).""" + unknown = set(match) - {"resources", "namespaceSelector"} + if unknown: + fail(path, name, f"unsupported match keys {sorted(unknown)}") + if "resources" in match and "namespaceSelector" in match: + fail(path, name, "both match.resources and match.namespaceSelector set") + if "resources" in match: + if not match["resources"]: + fail(path, name, "match.resources is empty") + return convert_resources(match["resources"], path, name) + if "namespaceSelector" in match: + if not match["namespaceSelector"]: + fail(path, name, "match.namespaceSelector is empty") + return convert_namespace_selector(match["namespaceSelector"], path, name) + # No match => the exception applies cluster-wide for its controls. + return [{"designatorType": "Attributes", "attributes": {"namespace": ".*"}}] + + +def convert_document(doc, path): + """Convert one ClusterSecurityException document; None for other kinds.""" + if not isinstance(doc, dict) or doc.get("kind") != "ClusterSecurityException": + return None + name = (doc.get("metadata") or {}).get("name") + if not name: + fail(path, "", "missing metadata.name") + spec = doc.get("spec") or {} + + posture = spec.get("posture") or [] + if not posture: + fail(path, name, "spec.posture is empty") + policies = [] + for control in posture: + action = control.get("action") + if action != "ignore": + fail(path, name, f"unsupported posture action {action!r}") + control_id = control.get("controlID") + if not control_id: + fail(path, name, "posture entry without a controlID") + policies.append({"controlID": anchor(control_id, path, name)}) + + match = spec.get("match") + if match is not None and (not isinstance(match, dict) or not match): + # Fail closed: an explicit-but-malformed match ([], "", false, {}) + # must never be coerced into the cluster-wide default. + fail(path, name, f"spec.match must be a non-empty mapping, got {match!r}") + resources = resolve_match(match or {}, path, name) + + policy = { + "name": name, + "policyType": "postureExceptionPolicy", + "actions": ["alertOnly"], + "resources": resources, + "posturePolicies": policies, + } + if spec.get("reason"): + policy["reason"] = " ".join(str(spec["reason"]).split()) + return policy + + +def generate(directory): + """Convert every CSE document under `directory` into sorted policies.""" + policies = [] + seen = set() + for filename in sorted(os.listdir(directory)): + if not filename.endswith((".yaml", ".yml")): + continue + path = os.path.join(directory, filename) + with open(path, encoding="utf-8") as handle: + documents = list(yaml.safe_load_all(handle)) + for doc in documents: + policy = convert_document(doc, path) + if policy is None: + continue + if policy["name"] in seen: + fail(path, policy["name"], "duplicate exception name") + seen.add(policy["name"]) + policies.append(policy) + if not policies: + sys.exit(f"{directory}: no ClusterSecurityException documents found") + return sorted(policies, key=lambda p: p["name"]) + + +def main(): + """Parse CLI arguments and write the generated exceptions file.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "directory", + nargs="?", + default=DEFAULT_DIR, + help="directory holding the ClusterSecurityException CRs", + ) + parser.add_argument( + "-o", + "--output", + help="output file (stdout if omitted)", + ) + args = parser.parse_args() + + policies = generate(args.directory) + rendered = json.dumps(policies, indent=2) + "\n" + if args.output: + with open(args.output, "w", encoding="utf-8") as handle: + handle.write(rendered) + print( + f"wrote {len(policies)} exception policies to {args.output}", + file=sys.stderr, + ) + else: + sys.stdout.write(rendered) + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/test_generate_kubescape_exceptions.py b/scripts/tests/test_generate_kubescape_exceptions.py new file mode 100644 index 000000000..2014b2957 --- /dev/null +++ b/scripts/tests/test_generate_kubescape_exceptions.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +"""Unit tests for scripts/generate-kubescape-exceptions.py. + +Requires PyYAML (like the module under test); no cluster, no network. Run +from anywhere: + + python3 -m unittest scripts/tests/test_generate_kubescape_exceptions.py -v + +The module under test is loaded by file path (its filename is hyphenated, +matching this repo's script-naming convention, so it cannot be imported with +a normal `import` statement). +""" + +import importlib.util +import os +import tempfile +import textwrap +import unittest + +MODULE_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "generate-kubescape-exceptions.py", +) +spec = importlib.util.spec_from_file_location( + "generate_kubescape_exceptions", MODULE_PATH +) +mod = importlib.util.module_from_spec(spec) +spec.loader.exec_module(mod) + + +def cse(name, posture, match=None, reason="documented reason"): + """Build a minimal ClusterSecurityException document for tests.""" + doc = { + "apiVersion": "kubescape.io/v1beta1", + "kind": "ClusterSecurityException", + "metadata": {"name": name}, + "spec": {"reason": reason, "posture": posture}, + } + if match is not None: + doc["spec"]["match"] = match + return doc + + +IGNORE = [{"controlID": "C-0002", "action": "ignore"}] + + +class ConvertDocumentTests(unittest.TestCase): + """Behaviour of convert_document for each supported/rejected CR shape.""" + + def test_non_cse_documents_are_skipped(self): + """Non-CSE documents (and empty docs) convert to None, not errors.""" + self.assertIsNone( + mod.convert_document({"kind": "Kustomization", "resources": []}, "f") + ) + self.assertIsNone(mod.convert_document(None, "f")) + + def test_cluster_wide_exception_gets_match_all_namespace(self): + """A CSE without match applies cluster-wide via a `.*` namespace.""" + policy = mod.convert_document(cse("all", IGNORE), "f") + self.assertEqual( + policy["resources"], + [{"designatorType": "Attributes", "attributes": {"namespace": ".*"}}], + ) + + def test_policy_shape_and_control_anchoring(self): + """The generated policy carries the exact Kubescape policy shape.""" + policy = mod.convert_document(cse("shape", IGNORE), "f") + self.assertEqual(policy["name"], "shape") + self.assertEqual(policy["policyType"], "postureExceptionPolicy") + self.assertEqual(policy["actions"], ["alertOnly"]) + self.assertEqual(policy["reason"], "documented reason") + (control,) = policy["posturePolicies"] + # Anchored exact-match regex so C-0002 cannot substring-match C-0020. + self.assertTrue(control["controlID"].startswith("^C")) + self.assertTrue(control["controlID"].endswith("0002$")) + + def test_reason_block_scalar_is_flattened_to_one_line(self): + """Multi-line reason scalars are flattened to one line.""" + policy = mod.convert_document( + cse("flat", IGNORE, reason="line one\nline two"), "f" + ) + self.assertEqual(policy["reason"], "line one line two") + + def test_resources_map_to_attribute_designators(self): + """match.resources entries become anchored Attributes designators.""" + match = { + "resources": [ + { + "apiGroup": "rbac.authorization.k8s.io", + "kind": "ClusterRoleBinding", + "name": "^velero-server$", + }, + {"apiGroup": "batch", "kind": "CronJob"}, + ] + } + policy = mod.convert_document(cse("res", IGNORE, match), "f") + self.assertEqual( + policy["resources"], + [ + { + "designatorType": "Attributes", + "attributes": { + "kind": "^ClusterRoleBinding$", + "name": "^velero-server$", + }, + }, + {"designatorType": "Attributes", "attributes": {"kind": "^CronJob$"}}, + ], + ) + + def test_namespace_selector_maps_to_alternation(self): + """A metadata.name In [...] selector becomes a namespace alternation.""" + match = { + "namespaceSelector": { + "matchExpressions": [ + { + "key": "kubernetes.io/metadata.name", + "operator": "In", + "values": ["kube-system", "velero"], + } + ] + } + } + policy = mod.convert_document(cse("ns", IGNORE, match), "f") + self.assertEqual( + policy["resources"], + [ + { + "designatorType": "Attributes", + "attributes": {"namespace": "^(kube\\-system|velero)$"}, + } + ], + ) + + def test_unknown_match_key_fails_closed(self): + """An unsupported match key aborts instead of being dropped.""" + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, {"labelSelector": {}}), "f") + + def test_match_labels_selector_fails_closed(self): + """matchLabels namespace selectors are unsupported and abort.""" + match = {"namespaceSelector": {"matchLabels": {"team": "x"}}} + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match), "f") + + def test_non_ignore_action_fails_closed(self): + """A posture action other than `ignore` aborts.""" + posture = [{"controlID": "C-0002", "action": "alert"}] + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", posture), "f") + + def test_resource_without_kind_fails_closed(self): + """A match.resources entry without a kind aborts.""" + match = {"resources": [{"name": "^x$"}]} + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match), "f") + + def test_resource_namespace_key_fails_closed(self): + """The CRD's match.resources[] has no namespace field, so a namespace + key aborts instead of generating a CI exception the in-cluster + operator would not apply.""" + match = {"resources": [{"kind": "Job", "namespace": "velero"}]} + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match), "f") + + def test_malformed_match_fails_closed(self): + """An explicit-but-malformed match must not widen to cluster-wide. + + `match: []` / `""` / `false` / `{}` are falsey but explicit — coercing + them into the no-match default would silently grant the cluster-wide + `.*` namespace designator. + """ + for malformed in ([], "", False, {}, 0, "resources"): + with self.subTest(match=malformed): + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match=malformed), "f") + + def test_both_match_shapes_fails_closed(self): + """Setting both match.resources and match.namespaceSelector aborts.""" + match = { + "resources": [{"kind": "Job"}], + "namespaceSelector": {"matchExpressions": []}, + } + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match), "f") + + def test_empty_resources_fails_closed(self): + """An explicit but empty match.resources aborts instead of widening + the exception to cluster-wide scope.""" + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, {"resources": []}), "f") + + def test_empty_namespace_selector_fails_closed(self): + """An explicit but empty match.namespaceSelector aborts instead of + widening the exception to cluster-wide scope.""" + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, {"namespaceSelector": {}}), "f") + + def test_partially_anchored_value_fails_closed(self): + """A value anchored on only one end is ambiguous and aborts.""" + for value in ("^half-anchored", "half-anchored$"): + match = {"resources": [{"kind": "Job", "name": value}]} + with self.assertRaises(SystemExit): + mod.convert_document(cse("bad", IGNORE, match), "f") + + def test_fully_anchored_value_passes_through_unescaped(self): + """A fully `^...$`-anchored value is kept as the author's regex.""" + match = {"resources": [{"kind": "Job", "name": "^(a|b)-server$"}]} + policy = mod.convert_document(cse("ok", IGNORE, match), "f") + self.assertEqual( + policy["resources"][0]["attributes"]["name"], "^(a|b)-server$" + ) + + +class GenerateTests(unittest.TestCase): + """Behaviour of generate() over a directory of CR files.""" + + def write(self, directory, filename, text): + """Write a dedented YAML fixture file into the test directory.""" + with open(os.path.join(directory, filename), "w", encoding="utf-8") as f: + f.write(textwrap.dedent(text)) + + def test_generates_sorted_policies_and_skips_kustomization(self): + """Policies come back name-sorted; kustomization.yaml is skipped.""" + with tempfile.TemporaryDirectory() as directory: + self.write( + directory, + "kustomization.yaml", + """\ + apiVersion: kustomize.config.k8s.io/v1beta1 + kind: Kustomization + resources: + - b.yaml + """, + ) + self.write( + directory, + "b.yaml", + """\ + apiVersion: kubescape.io/v1beta1 + kind: ClusterSecurityException + metadata: + name: zeta + spec: + reason: >- + multi-line + reason text + posture: + - controlID: C-0057 + action: ignore + """, + ) + self.write( + directory, + "a.yaml", + """\ + apiVersion: kubescape.io/v1beta1 + kind: ClusterSecurityException + metadata: + name: alpha + spec: + posture: + - controlID: C-0002 + action: ignore + match: + resources: + - apiGroup: rbac.authorization.k8s.io + kind: ClusterRoleBinding + name: ^kubevirt-operator$ + """, + ) + policies = mod.generate(directory) + self.assertEqual([p["name"] for p in policies], ["alpha", "zeta"]) + self.assertEqual(policies[1]["reason"], "multi-line reason text") + + def test_duplicate_names_fail_closed(self): + """Two CSEs with the same name abort the run.""" + with tempfile.TemporaryDirectory() as directory: + for filename in ("a.yaml", "b.yaml"): + self.write( + directory, + filename, + """\ + apiVersion: kubescape.io/v1beta1 + kind: ClusterSecurityException + metadata: + name: same + spec: + posture: + - controlID: C-0002 + action: ignore + """, + ) + with self.assertRaises(SystemExit): + mod.generate(directory) + + def test_empty_directory_fails_closed(self): + """A directory without any CSE documents aborts the run.""" + with tempfile.TemporaryDirectory() as directory: + with self.assertRaises(SystemExit): + mod.generate(directory) + + def test_real_repo_crs_convert_cleanly(self): + """The live CSE directory must always convert — the CI invariant.""" + policies = mod.generate(mod.DEFAULT_DIR) + names = {p["name"] for p in policies} + self.assertIn("exec-into-container-rbac", names) + exec_policy = next(p for p in policies if p["name"] == "exec-into-container-rbac") + self.assertIn( + { + "designatorType": "Attributes", + "attributes": {"kind": "^ClusterRoleBinding$", "name": "^kubevirt-operator$"}, + }, + exec_policy["resources"], + ) + for policy in policies: + self.assertTrue(policy["resources"]) + self.assertTrue(policy["posturePolicies"]) + + +if __name__ == "__main__": + unittest.main()