prowler/tests/providers/kubernetes/kubernetes_provider_test.py

from argparse import Namespace
from unittest.mock import patch

from kubernetes.config.config_exception import ConfigException

from kubernetes import client
from prowler.config.config import (
    default_config_file_path,
    default_fixer_config_file_path,
    load_and_validate_config_file,
)
from prowler.providers.kubernetes.exceptions.exceptions import (
    KubernetesSetUpSessionError,
)
from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
from prowler.providers.kubernetes.models import (
    KubernetesIdentityInfo,
    KubernetesSession,
)
from tests.providers.kubernetes.kubernetes_fixtures import KUBERNETES_CONFIG


def mock_set_kubernetes_credentials(*_):
    return ("apiclient", "context")


def mock_get_context_user_roles(*_):
    return []


class TestKubernetesProvider:
    def test_kubernetes_provider_no_namespaces(
        self,
    ):
        context = {
            "name": "test-context",
            "context": {
                "user": "test-user",
                "cluster": "test-cluster",
            },
        }
        with (
            patch(
                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.setup_session",
                return_value=KubernetesSession(
                    api_client=client.ApiClient,
                    context=context,
                ),
            ),
            patch(
                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
                return_value=["namespace-1"],
            ),
        ):
            arguments = Namespace()
            arguments.kubeconfig_file = "dummy_path"
            arguments.context = None
            arguments.only_logs = False
            arguments.namespace = None
            fixer_config = load_and_validate_config_file(
                "kubernetes", default_fixer_config_file_path
            )

            # Instantiate the KubernetesProvider with mocked arguments
            kubernetes_provider = KubernetesProvider(
                arguments.kubeconfig_file,
                arguments.context,
                arguments.namespace,
                config_path=default_config_file_path,
                fixer_config=fixer_config,
            )
            assert isinstance(kubernetes_provider.session, KubernetesSession)
            assert kubernetes_provider.session.api_client is not None
            assert kubernetes_provider.session.context == {
                "name": "test-context",
                "context": {"cluster": "test-cluster", "user": "test-user"},
            }
            assert kubernetes_provider.namespaces == ["namespace-1"]
            assert isinstance(kubernetes_provider.identity, KubernetesIdentityInfo)
            assert kubernetes_provider.identity.context == "test-context"
            assert kubernetes_provider.identity.cluster == "test-cluster"
            assert kubernetes_provider.identity.user == "test-user"

            assert kubernetes_provider.audit_config == KUBERNETES_CONFIG

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_test_connection_with_kubeconfig_content(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
    ):
        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        connection = KubernetesProvider.test_connection(
            kubeconfig_file=None,
            kubeconfig_content=kubeconfig_content,
            provider_id="example-context",
            raise_on_exception=False,
        )

        assert connection.is_connected
        assert connection.error is None

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config")
    def test_kubernetes_test_connection_with_kubeconfig_file(
        self, mock_load_kube_config, mock_list_kube_config_contexts, mock_list_namespace
    ):
        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_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        connection = KubernetesProvider.test_connection(
            kubeconfig_file="dummy_kubeconfig_path",
            kubeconfig_content="",
            provider_id="test-context",
            raise_on_exception=False,
        )

        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="",
            kubeconfig_content="",
            namespace="test-namespace",
            provider_id="test-context",
            raise_on_exception=False,
        )

        assert connection.is_connected
        assert connection.error is None

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_test_connection_with_kubeconfig_content_invalid_provider_id(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
    ):
        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = {
            "apiVersion": "v1",
            "clusters": [
                {
                    "cluster": {
                        "server": "https://kubernetes.example.com",
                    },
                    "name": "example-cluster",
                }
            ],
            "contexts": [
                {
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                    "name": "example-context",
                }
            ],
            "current-context": "example-context",
            "kind": "Config",
            "preferences": {},
            "users": [
                {
                    "name": "example-user",
                    "user": {
                        "token": "EXAMPLE_TOKEN",
                    },
                }
            ],
        }

        connection = KubernetesProvider.test_connection(
            kubeconfig_file=None,
            kubeconfig_content=kubeconfig_content,
            provider_id="example-context-invalid",
            raise_on_exception=False,
        )

        assert not connection.is_connected
        assert connection.error is not None
        assert isinstance(connection.error, KubernetesSetUpSessionError)

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_test_connection_with_kubeconfig_content_valid_provider_id(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
    ):
        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        connection = KubernetesProvider.test_connection(
            kubeconfig_file=None,
            kubeconfig_content=kubeconfig_content,
            provider_id="example-context",
            raise_on_exception=False,
        )

        assert connection.is_connected
        assert connection.error is None

    def test_kubernetes_provider_incluster_with_env_var(self, monkeypatch):
        monkeypatch.setenv("CLUSTER_NAME", "env-cluster-name")

        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.client.ApiClient"),
            patch(
                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
                return_value=["default"],
            ),
        ):
            session = KubernetesProvider.setup_session(
                kubeconfig_file=None,
                kubeconfig_content=None,
                context=None,
                cluster_name=None,
            )
            assert isinstance(session, KubernetesSession)
            assert session.context["context"]["cluster"] == "env-cluster-name"

    def test_kubernetes_provider_incluster_with_cli_flag(self):
        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.client.ApiClient"),
            patch(
                "prowler.providers.kubernetes.kubernetes_provider.KubernetesProvider.get_all_namespaces",
                return_value=["default"],
            ),
        ):
            session = KubernetesProvider.setup_session(
                kubeconfig_file=None,
                kubeconfig_content=None,
                context=None,
                cluster_name="cli-cluster-name",
            )
            assert isinstance(session, KubernetesSession)
            assert session.context["context"]["cluster"] == "cli-cluster-name"

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_provider_proxy_from_env(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
        monkeypatch,
    ):

        monkeypatch.setenv("HTTPS_PROXY", "http://my.internal.proxy:8888")

        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        session = KubernetesProvider.setup_session(
            kubeconfig_content=kubeconfig_content,
            context="example-context",
        )

        assert isinstance(session, KubernetesSession)
        assert isinstance(session.api_client, client.ApiClient)
        assert isinstance(session.api_client.configuration, client.Configuration)
        assert session.api_client.configuration.verify_ssl
        assert session.api_client.configuration.proxy == "http://my.internal.proxy:8888"

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_provider_disable_tls_verification(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
        monkeypatch,
    ):
        monkeypatch.setenv("K8S_SKIP_TLS_VERIFY", "true")

        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        session = KubernetesProvider.setup_session(
            kubeconfig_content=kubeconfig_content,
            context="example-context",
        )

        assert isinstance(session, KubernetesSession)
        assert isinstance(session.api_client, client.ApiClient)
        assert isinstance(session.api_client.configuration, client.Configuration)
        assert session.api_client.configuration.verify_ssl is False
        assert session.api_client.configuration.proxy is None

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_provider_kubeconfig_content(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
    ):
        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        session = KubernetesProvider.setup_session(
            kubeconfig_content=kubeconfig_content,
            context="example-context",
        )

        assert isinstance(session, KubernetesSession)
        assert isinstance(session.api_client, client.ApiClient)

        assert session.context == {
            "name": "example-context",
            "context": {
                "cluster": "example-cluster",
                "user": "example-user",
            },
        }

    @patch(
        "prowler.providers.kubernetes.kubernetes_provider.client.CoreV1Api.list_namespace"
    )
    @patch("kubernetes.config.list_kube_config_contexts")
    @patch("kubernetes.config.load_kube_config_from_dict")
    def test_kubernetes_provider_kubeconfig_content_proxy_settings(
        self,
        mock_load_kube_config_from_dict,
        mock_list_kube_config_contexts,
        mock_list_namespace,
        monkeypatch,
    ):
        monkeypatch.setenv("HTTPS_PROXY", "http://my.internal.proxy:8888")
        monkeypatch.setenv("K8S_SKIP_TLS_VERIFY", "true")

        mock_load_kube_config_from_dict.return_value = None
        mock_list_kube_config_contexts.return_value = (
            [
                {
                    "name": "example-context",
                    "context": {
                        "cluster": "example-cluster",
                        "user": "example-user",
                    },
                }
            ],
            None,
        )
        mock_list_namespace.return_value.items = [
            client.V1Namespace(metadata=client.V1ObjectMeta(name="namespace-1")),
        ]

        kubeconfig_content = '{"apiVersion": "v1", "clusters": [{"cluster": {"server": "https://kubernetes.example.com"}, "name": "example-cluster"}], "contexts": [{"context": {"cluster": "example-cluster", "user": "example-user"}, "name": "example-context"}], "current-context": "example-context", "kind": "Config", "preferences": {}, "users": [{"name": "example-user", "user": {"token": "EXAMPLE_TOKEN"}}]}'

        session = KubernetesProvider.setup_session(
            kubeconfig_content=kubeconfig_content,
            context="example-context",
        )

        assert isinstance(session, KubernetesSession)
        assert isinstance(session.api_client, client.ApiClient)

        assert session.context == {
            "name": "example-context",
            "context": {
                "cluster": "example-cluster",
                "user": "example-user",
            },
        }

        assert session.api_client.configuration.proxy == "http://my.internal.proxy:8888"
        assert session.api_client.configuration.verify_ssl is False

    def test_set_proxy_settings_no_proxy_no_tls_skip(self):
        """Test set_proxy_settings with no environment variables set."""
        with patch.dict("os.environ", {}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert hasattr(config, "proxy")
            assert config.proxy is None
            assert hasattr(config, "verify_ssl")
            assert config.verify_ssl is True

    def test_set_proxy_settings_with_https_proxy_uppercase(self):
        """Test set_proxy_settings with HTTPS_PROXY environment variable."""
        proxy_url = "http://proxy.example.com:8080"
        with patch.dict("os.environ", {"HTTPS_PROXY": proxy_url}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy == proxy_url
            assert config.verify_ssl is True

    def test_set_proxy_settings_with_https_proxy_lowercase(self):
        """Test set_proxy_settings with https_proxy environment variable."""
        proxy_url = "http://proxy.example.com:3128"
        with patch.dict("os.environ", {"https_proxy": proxy_url}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy == proxy_url
            assert config.verify_ssl is True

    def test_set_proxy_settings_uppercase_proxy_takes_precedence(self):
        """Test that HTTPS_PROXY takes precedence over https_proxy."""
        uppercase_proxy = "http://uppercase.proxy.com:8080"
        lowercase_proxy = "http://lowercase.proxy.com:3128"
        with patch.dict(
            "os.environ",
            {"HTTPS_PROXY": uppercase_proxy, "https_proxy": lowercase_proxy},
            clear=True,
        ):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy == uppercase_proxy
            assert config.verify_ssl is True

    def test_set_proxy_settings_with_tls_skip_true(self):
        """Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to true."""
        with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "true"}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy is None
            assert config.verify_ssl is False

    def test_set_proxy_settings_with_tls_skip_true_uppercase(self):
        """Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to TRUE."""
        with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "TRUE"}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy is None
            assert config.verify_ssl is False

    def test_set_proxy_settings_with_tls_skip_false(self):
        """Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to false."""
        with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "false"}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy is None
            assert config.verify_ssl is True

    def test_set_proxy_settings_with_tls_skip_invalid_value(self):
        """Test set_proxy_settings with K8S_SKIP_TLS_VERIFY set to invalid value."""
        with patch.dict("os.environ", {"K8S_SKIP_TLS_VERIFY": "invalid"}, clear=True):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy is None
            assert config.verify_ssl is True

    def test_set_proxy_settings_with_both_proxy_and_tls_skip(self):
        """Test set_proxy_settings with both proxy and TLS skip settings."""
        proxy_url = "http://secure.proxy.com:8080"
        with patch.dict(
            "os.environ",
            {"HTTPS_PROXY": proxy_url, "K8S_SKIP_TLS_VERIFY": "true"},
            clear=True,
        ):
            config = KubernetesProvider.set_proxy_settings()

            # Verify it's a Configuration instance from kubernetes.client
            from kubernetes.client import Configuration

            assert isinstance(config, Configuration)

            assert config.proxy == proxy_url
            assert config.verify_ssl is False