feat(kubernetes): support HTTPS_PROXY and K8S_SKIP_TLS_VERIFY (#7720)

docs/tutorials/kubernetes/misc.md

@@ -21,3 +21,23 @@ To specify the namespace(s) to be scanned, use the `--namespace` flag followed b
 ```console
 prowler --namespace namespace1 namespace2
 ```
+
+## Proxy and TLS Verification
+
+If your Kubernetes cluster is only accessible via an internal proxy, Prowler will respect the `HTTPS_PROXY` or `https_proxy` environment variable:
+
+```console
+export HTTPS_PROXY=http://my.internal.proxy:8888
+prowler kubernetes ...
+```
+
+If you need to skip TLS verification for internal proxies, you can set the `K8S_SKIP_TLS_VERIFY` environment variable:
+
+```console
+export K8S_SKIP_TLS_VERIFY=true
+prowler kubernetes ...
+```
+
+This will allow Prowler to connect to the cluster even if the proxy uses a self-signed certificate.
+
+These environment variables are supported both when using an external `kubeconfig` and in in-cluster mode.

prowler/providers/kubernetes/kubernetes_provider.py

@@ -2,6 +2,7 @@ import os
 from typing import Union
 
 from colorama import Fore, Style
+from kubernetes.client import ApiClient, Configuration
 from kubernetes.client.exceptions import ApiException
 from kubernetes.config.config_exception import ConfigException
 from requests.exceptions import Timeout
@@ -286,9 +287,21 @@ class KubernetesProvider(Provider):
                             "user": "service-account-name",
                         },
                     }
-                    return KubernetesSession(
-                        api_client=client.ApiClient(), context=context
+                    # Ensure proxy settings are respected
+                    configuration = Configuration.get_default_copy()
+                    proxy = os.environ.get("HTTPS_PROXY") or os.environ.get(
+                        "https_proxy"
                     )
+                    if proxy:
+                        configuration.proxy = proxy
+                    # Prevent SSL verification issues with internal proxies
+                    if os.environ.get("K8S_SKIP_TLS_VERIFY", "false").lower() == "true":
+                        configuration.verify_ssl = False
+
+                    return KubernetesSession(
+                        api_client=ApiClient(configuration), context=context
+                    )
+
                 if context:
                     contexts = config.list_kube_config_contexts(
                         config_file=kubeconfig_file
@@ -302,7 +315,19 @@ class KubernetesProvider(Provider):
                     context = config.list_kube_config_contexts(
                         config_file=kubeconfig_file
                     )[1]
-            return KubernetesSession(api_client=client.ApiClient(), context=context)
+                # Ensure proxy settings are respected
+                configuration = Configuration.get_default_copy()
+                proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
+                if proxy:
+                    configuration.proxy = proxy
+
+                # Prevent SSL verification issues with internal proxies
+                if os.environ.get("K8S_SKIP_TLS_VERIFY", "false").lower() == "true":
+                    configuration.verify_ssl = False
+
+                return KubernetesSession(
+                    api_client=ApiClient(configuration), context=context
+                )
 
         except parser.ParserError as parser_error:
             logger.critical(

tests/providers/kubernetes/kubernetes_provider_test.py

@@ -1,5 +1,5 @@
 from argparse import Namespace
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from kubernetes.config.config_exception import ConfigException
 
@@ -353,3 +353,59 @@ class TestKubernetesProvider:
             )
             assert isinstance(session, KubernetesSession)
             assert session.context["context"]["cluster"] == "cli-cluster-name"
+
+    def test_kubernetes_provider_proxy_from_env(self, monkeypatch):
+        monkeypatch.setenv("HTTPS_PROXY", "http://my.internal.proxy:8888")
+
+        captured = {}
+
+        def fake_api_client(configuration):
+            captured["proxy"] = getattr(configuration, "proxy", None)
+            return MagicMock()
+
+        with (
+            patch(
+                "kubernetes.config.load_kube_config",
+                side_effect=ConfigException("No kubeconfig"),
+            ),
+            patch("kubernetes.config.load_incluster_config", return_value=None),
+            patch(
+                "prowler.providers.kubernetes.kubernetes_provider.ApiClient",
+                side_effect=fake_api_client,
+            ),
+            patch(
+                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
+                return_value=["default"],
+            ),
+        ):
+            KubernetesProvider.setup_session()
+
+        assert captured["proxy"] == "http://my.internal.proxy:8888"
+
+    def test_kubernetes_provider_disable_tls_verification(self, monkeypatch):
+        monkeypatch.setenv("K8S_SKIP_TLS_VERIFY", "true")
+
+        captured = {}
+
+        def fake_api_client(configuration):
+            captured["verify_ssl"] = getattr(configuration, "verify_ssl", True)
+            return MagicMock()
+
+        with (
+            patch(
+                "kubernetes.config.load_kube_config",
+                side_effect=ConfigException("No kubeconfig"),
+            ),
+            patch("kubernetes.config.load_incluster_config", return_value=None),
+            patch(
+                "prowler.providers.kubernetes.kubernetes_provider.ApiClient",
+                side_effect=fake_api_client,
+            ),
+            patch(
+                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
+                return_value=["default"],
+            ),
+        ):
+            KubernetesProvider.setup_session()
+
+        assert captured["verify_ssl"] is False