diff --git a/prowler/providers/kubernetes/kubernetes_provider.py b/prowler/providers/kubernetes/kubernetes_provider.py index c9967d412d..53ebfa8139 100644 --- a/prowler/providers/kubernetes/kubernetes_provider.py +++ b/prowler/providers/kubernetes/kubernetes_provider.py @@ -207,6 +207,7 @@ def setup_session( def test_connection( kubeconfig_file: str = "~/.kube/config", kubeconfig_content: dict = None, + namespace: str = None, input_context: str = "", raise_on_exception: bool = True, ) -> Connection: @@ -216,6 +217,7 @@ def test_connection( Args: kubeconfig_file (str): Path to the kubeconfig file. kubeconfig_content (dict): Content of the kubeconfig file. + namespace (str): Namespace name. input_context (str): Context name. raise_on_exception (bool): Whether to raise an exception on error. Returns: @@ -225,7 +227,12 @@ def test_connection( KubernetesProvider.setup_session( kubeconfig_file, kubeconfig_content, input_context ) - client.CoreV1Api().list_namespace(timeout_seconds=2, _request_timeout=2) + if namespace: + client.CoreV1Api().list_namespaced_pod( + namespace, timeout_seconds=2, _request_timeout=2 + ) + else: + client.CoreV1Api().list_namespace(timeout_seconds=2, _request_timeout=2) return Connection(is_connected=True) except KubernetesSetUpSessionError as setup_session_error: logger.critical( diff --git a/tests/providers/kubernetes/kubernetes_provider_test.py b/tests/providers/kubernetes/kubernetes_provider_test.py index cc485b0745..cbffd29526 100644 --- a/tests/providers/kubernetes/kubernetes_provider_test.py +++ b/tests/providers/kubernetes/kubernetes_provider_test.py @@ -182,3 +182,42 @@ def test_kubernetes_test_connection_with_kubeconfig_file( assert connection.is_connected assert connection.error is None + + @patch( + "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespaced_pod" + ) + @patch("kubernetes.config.list_kube_config_contexts") + @patch("kubernetes.config.load_kube_config") + def test_kubernetes_test_connection_with_namespace_input( + self, + mock_load_kube_config, + mock_list_kube_config_contexts, + mock_list_namespaced_pod, + ): + mock_load_kube_config.return_value = None + mock_list_kube_config_contexts.return_value = ( + [ + { + "name": "test-context", + "context": { + "cluster": "test-cluster", + "user": "test-user", + }, + } + ], + None, + ) + mock_list_namespaced_pod.return_value.items = [ + client.V1Pod(metadata=client.V1ObjectMeta(name="pod-1")), + ] + + connection = KubernetesProvider.test_connection( + kubeconfig_file="dummy_kubeconfig_path", + kubeconfig_content={}, + namespace="test-namespace", + input_context="test-context", + raise_on_exception=False, + ) + + assert connection.is_connected + assert connection.error is None