fix(k8s_storage): discover CSI controller/node pods cluster-wide in K8sCsiTenantScopedCredentialsCheck

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)
 

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 '<ns>'` 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"}}]