diff --git a/isvtest/src/isvtest/validations/k8s_storage.py b/isvtest/src/isvtest/validations/k8s_storage.py index e67c4e08..94feb1e3 100644 --- a/isvtest/src/isvtest/validations/k8s_storage.py +++ b/isvtest/src/isvtest/validations/k8s_storage.py @@ -1018,9 +1018,10 @@ class K8sCsiTenantScopedCredentialsCheck(BaseValidation): Secrets they reference via ``PersistentVolume.spec.csi.*SecretRef`` and via CSI controller/node pod specs (``envFrom``, ``env.valueFrom``, Secret volume mounts). Skipped when no ``CSIDriver`` objects exist. - * ``secrets-not-cross-namespace`` - every discovered Secret lives in - ``csi_driver_namespaces`` (or ``allowed_workload_namespaces``), never - in ``default`` or an unlisted workload namespace. + * ``secrets-not-cross-namespace`` - every discovered Secret lives in a + namespace where a CSI controller/node pod was actually found, or in + ``csi_driver_namespaces``/``allowed_workload_namespaces``, never in + ``default`` or an unlisted workload namespace. * ``no-shared-cluster-markers`` - no discovered Secret carries any of ``forbidden_labels`` or an annotation like ``csi.nvidia.com/shared=true``. @@ -1033,9 +1034,20 @@ class K8sCsiTenantScopedCredentialsCheck(BaseValidation): configMap, projected, downwardAPI, serviceAccountToken, csi). Any ``nfs``/``iscsi``/``persistentVolumeClaim`` volume fails this subtest. + CSI controller/node pods are discovered cluster-wide (``kubectl get + pods --all-namespaces``), filtered by the same sidecar-image heuristic + used elsewhere in this module, rather than by a configured namespace + list. Most CSI operators (Longhorn, Piraeus, Rook-Ceph, ...) do not + install into ``kube-system``; gating discovery on a namespace allowlist + let those drivers' controller pods go unseen entirely, silently + skipping ``serviceaccount-rbac-scoped`` and reporting a false pass + regardless of the ServiceAccount's actual RBAC. + Config keys (with defaults): - csi_driver_namespaces: Namespaces where CSI controller/node pods - live (default: ``["kube-system"]``). + csi_driver_namespaces: Extra namespaces where CSI Secrets are + permitted, on top of namespaces where a CSI pod was actually + discovered (default: ``["kube-system"]``, kept for backward + compatibility with existing provider configs). allowed_workload_namespaces: Extra namespaces where CSI Secrets are permitted (default: ``[]``). forbidden_labels: ``key=value`` label pairs whose presence on a CSI @@ -1057,8 +1069,6 @@ def run(self) -> None: forbidden_labels_raw = self.config.get("forbidden_labels") or ["shared-across-clusters=true"] forbidden_labels = _parse_label_pairs(forbidden_labels_raw) - permitted_namespaces = set(driver_namespaces) | set(allowed_workload_namespaces) - # Discover CSIDriver objects up front. If none exist we have nothing # to validate; the check is skipped so it is safe to enable on # clusters without any CSI driver installed. @@ -1078,13 +1088,26 @@ def run(self) -> None: self.set_passed("Skipped: no CSIDriver objects found") return + # Discover CSI controller/node pods cluster-wide rather than by a + # configured namespace allowlist - most CSI operators do not + # install into kube-system, and gating discovery on + # csi_driver_namespaces let their pods go unseen entirely (see + # class docstring). + all_pods = self._list_all_pods() + if all_pods is None: + self.set_failed("Failed to list pods across all namespaces") + return + pods_by_ns: dict[str, list[dict[str, Any]]] = {} - for ns in driver_namespaces: - pods = self._list_pods(ns) - if pods is None: - self.set_failed(f"Failed to list pods in namespace {ns!r}") - return - pods_by_ns[ns] = pods + for pod in all_pods: + if not _pod_has_csi_image(pod): + continue + ns = str((pod.get("metadata") or {}).get("namespace") or "") + if not ns: + continue + pods_by_ns.setdefault(ns, []).append(pod) + + permitted_namespaces = set(driver_namespaces) | set(allowed_workload_namespaces) | set(pods_by_ns.keys()) pvs = self._list_pvs() if pvs is None: @@ -1257,10 +1280,11 @@ def _list_csi_drivers(self) -> list[dict[str, Any]] | None: return None return _load_items(result.stdout) - def _list_pods(self, namespace: str) -> list[dict[str, Any]] | None: - result = self.run_command(f"{self._kubectl_base} get pods -n {shlex.quote(namespace)} -o json") + def _list_all_pods(self) -> list[dict[str, Any]] | None: + """List Pods across all namespaces, or ``None`` if the ``kubectl`` command fails.""" + result = self.run_command(f"{self._kubectl_base} get pods --all-namespaces -o json") if result.exit_code != 0: - self.log.error("kubectl get pods -n %s failed: %s", namespace, result.stderr.strip()) + self.log.error("kubectl get pods --all-namespaces failed: %s", result.stderr.strip()) return None return _load_items(result.stdout) diff --git a/isvtest/tests/test_k8s_storage.py b/isvtest/tests/test_k8s_storage.py index a4d97cd6..bbcc1e9b 100644 --- a/isvtest/tests/test_k8s_storage.py +++ b/isvtest/tests/test_k8s_storage.py @@ -1489,21 +1489,14 @@ def _router( cluster_roles = cluster_roles or [] def _route(cmd: str, timeout: int | None = None) -> CommandResult: + """Answer one kubectl invocation issued by the check under test.""" if fail_on and fail_on in cmd: return _fail(stderr="boom") if "get csidriver -o json" in cmd: return _ok(stdout=_items_json(csi_drivers)) - if "get pods -n " in cmd and "-o json" in cmd: - # Extract the namespace from the quoted `-n ''` fragment. - # shlex.quote renders most identifiers without quoting, so - # fall back to a simple split on whitespace. - parts = cmd.split() - ns = "" - for i, part in enumerate(parts): - if part == "-n" and i + 1 < len(parts): - ns = parts[i + 1].strip("'\"") - break - return _ok(stdout=_items_json(pods_by_ns.get(ns, []))) + if "get pods --all-namespaces -o json" in cmd: + all_pods = [pod for pods in pods_by_ns.values() for pod in pods] + return _ok(stdout=_items_json(all_pods)) if cmd.rstrip().endswith("get pv -o json"): return _ok(stdout=_items_json(pvs)) if "get secret " in cmd and "-o json" in cmd: @@ -1740,6 +1733,62 @@ def test_unrestricted_cluster_secret_grant_fails(self) -> None: assert not outcomes["serviceaccount-rbac-scoped"]["passed"] assert "csi-secret-reader" in outcomes["serviceaccount-rbac-scoped"]["message"] + def test_unrestricted_cluster_secret_grant_fails_outside_kube_system(self) -> None: + """Regression test: controller pods outside kube-system must still be caught. + + Reproduces the Longhorn/OpenNebula gap - Longhorn's CSI controller + pods run in ``longhorn-system``, not ``kube-system``, and its + ``longhorn-role`` ClusterRole grants unrestricted Secret access + identically to the ``csi-controller`` case above. With the old + ``csi_driver_namespaces``-gated pod discovery this would have gone + unseen and silently passed; cluster-wide discovery must catch it + without any provider config override. + """ + check = self._make({}) # No csi_driver_namespaces override - default config only. + csi_drivers = [{"kind": "CSIDriver", "metadata": {"name": "driver.longhorn.io"}}] + pods = [ + _pod( + name="longhorn-csi-plugin-controller", + namespace="longhorn-system", + images=["csi-provisioner:v4"], + service_account="longhorn-service-account", + ), + _pod( + name="longhorn-csi-plugin-node", + namespace="longhorn-system", + images=["csi-node-driver-registrar:v2"], + service_account="longhorn-service-account", + ), + ] + pvs: list[dict[str, Any]] = [] + crbs = [ + _crb( + name="longhorn-bind", + cluster_role="longhorn-role", + subject_namespace="longhorn-system", + subject_name="longhorn-service-account", + ) + ] + croles = [_cluster_role_secrets(name="longhorn-role", verbs=["get", "list", "watch"])] + with patch.object( + check, + "run_command", + side_effect=self._router( + csi_drivers=csi_drivers, + pods_by_ns={"longhorn-system": pods}, + pvs=pvs, + cluster_role_bindings=crbs, + cluster_roles=croles, + ), + ): + check.run() + + assert not check.passed + outcomes = {r["name"]: r for r in check._subtest_results} + assert not outcomes["serviceaccount-rbac-scoped"]["skipped"] + assert not outcomes["serviceaccount-rbac-scoped"]["passed"] + assert "longhorn-bind" in outcomes["serviceaccount-rbac-scoped"]["message"] + def test_node_plugin_with_persistent_volume_claim_fails(self) -> None: check = self._make({}) csi_drivers = [{"kind": "CSIDriver", "metadata": {"name": "x"}}]