From 14bab496b585d34230a04a4cc864fe48a73dbf39 Mon Sep 17 00:00:00 2001 From: Sergio Garcia <38561120+sergargar@users.noreply.github.com> Date: Mon, 19 Feb 2024 13:50:42 +0100 Subject: [PATCH] chore(tests): add kubernetes provider tests (#3265) --- .../kubernetes/kubernetes_provider_new.py | 4 +- .../providers/kubernetes/audit_info_utils.py | 21 +++ .../kubernetes/kubernetes_provider_test.py | 172 ++++++++++++++++++ .../kubernetes/lib/service/service_test.py | 13 ++ 4 files changed, 208 insertions(+), 2 deletions(-) create mode 100644 tests/providers/kubernetes/audit_info_utils.py create mode 100644 tests/providers/kubernetes/kubernetes_provider_test.py create mode 100644 tests/providers/kubernetes/lib/service/service_test.py diff --git a/prowler/providers/kubernetes/kubernetes_provider_new.py b/prowler/providers/kubernetes/kubernetes_provider_new.py index ea5552b24c..55f43645c7 100644 --- a/prowler/providers/kubernetes/kubernetes_provider_new.py +++ b/prowler/providers/kubernetes/kubernetes_provider_new.py @@ -1,5 +1,6 @@ import os import sys +from argparse import Namespace from typing import Any, Optional from colorama import Fore, Style @@ -17,9 +18,8 @@ class KubernetesProvider(Provider): audit_metadata: Optional[Any] audit_config: Optional[dict] - def __init__(self, arguments: dict): + def __init__(self, arguments: Namespace): logger.info("Instantiating Kubernetes Provider ...") - self.api_client, self.context = self.setup_session( arguments.kubeconfig_file, arguments.context ) diff --git a/tests/providers/kubernetes/audit_info_utils.py b/tests/providers/kubernetes/audit_info_utils.py new file mode 100644 index 0000000000..20175516b5 --- /dev/null +++ b/tests/providers/kubernetes/audit_info_utils.py @@ -0,0 +1,21 @@ +from kubernetes import client + +from prowler.providers.common.models import Audit_Metadata +from prowler.providers.kubernetes.lib.audit_info.models import Kubernetes_Audit_Info + + +# Mocked Kubernetes Audit Info +def set_mocked_audit_info(): + audit_info = Kubernetes_Audit_Info( + api_client=client.ApiClient, + context=None, + audit_config=None, + audit_resources=[], + audit_metadata=Audit_Metadata( + services_scanned=0, + expected_checks=[], + completed_checks=0, + audit_progress=0, + ), + ) + return audit_info diff --git a/tests/providers/kubernetes/kubernetes_provider_test.py b/tests/providers/kubernetes/kubernetes_provider_test.py new file mode 100644 index 0000000000..40f981e187 --- /dev/null +++ b/tests/providers/kubernetes/kubernetes_provider_test.py @@ -0,0 +1,172 @@ +import io +import sys +from argparse import Namespace +from unittest.mock import MagicMock, patch + +from prowler.providers.kubernetes.kubernetes_provider_new import KubernetesProvider + + +class TestKubernetesProvider: + @patch("kubernetes.client.RbacAuthorizationV1Api") + @patch("kubernetes.client.ApiClient") + @patch("kubernetes.config.load_kube_config") + @patch("kubernetes.config.load_incluster_config") + @patch("kubernetes.config.list_kube_config_contexts") + def test_setup_session( + self, + mock_list_kube_config_contexts, + mock_load_incluster_config, + mock_load_kube_config, + mock_api_client, + mock_rbac_api, + ): + # Mocking the return value of list_kube_config_contexts + mock_list_kube_config_contexts.return_value = ( + [ + { + "name": "context_name", + "context": {"cluster": "test-cluster", "user": "test-user"}, + } + ], + 0, + ) + + # Create a Namespace object for arguments + args = Namespace(kubeconfig_file="dummy_path", context=None, only_logs=False) + + # Instantiate the KubernetesProvider with mocked arguments + provider = KubernetesProvider(args) + + # Assert that an ApiClient has been created + assert provider.api_client is not None + + # Assert that the context has been correctly set + assert provider.context == { + "name": "context_name", + "context": {"cluster": "test-cluster", "user": "test-user"}, + } + + @patch("kubernetes.client.RbacAuthorizationV1Api") + @patch("kubernetes.config.list_kube_config_contexts") + @patch("kubernetes.config.load_incluster_config") + @patch("kubernetes.config.load_kube_config") + def test_get_context_user_roles( + self, + mock_load_kube_config, + mock_load_incluster_config, + mock_list_kube_config_contexts, + mock_rbac_api, + ): + mock_list_kube_config_contexts.return_value = ( + [ + { + "name": "context_name", + "context": {"cluster": "test-cluster", "user": "test-user"}, + } + ], + 0, + ) + + # Mock the RbacAuthorizationV1Api methods + cluster_role_binding = MagicMock() + role_binding = MagicMock() + cluster_role_binding.list_cluster_role_binding.return_value = MagicMock( + items=[] + ) + role_binding.list_role_binding_for_all_namespaces.return_value = MagicMock( + items=[] + ) + + mock_rbac_api.return_value = MagicMock( + list_cluster_role_binding=cluster_role_binding.list_cluster_role_binding, + list_role_binding_for_all_namespaces=role_binding.list_role_binding_for_all_namespaces, + ) + + args = Namespace(kubeconfig_file=None, context=None, only_logs=False) + provider = KubernetesProvider(args) + + roles = provider.get_context_user_roles() + + assert isinstance(roles, list) + + @patch("kubernetes.client.RbacAuthorizationV1Api") + @patch("kubernetes.client.ApiClient") + @patch("kubernetes.config.load_kube_config") + @patch("kubernetes.config.load_incluster_config") + @patch("kubernetes.config.list_kube_config_contexts") + @patch("sys.stdout", new_callable=MagicMock) + def test_print_credentials( + self, + mock_stdout, + mock_list_kube_config_contexts, + mock_load_incluster_config, + mock_load_kube_config, + mock_api_client, + mock_rbac_api, + ): + mock_list_kube_config_contexts.return_value = ( + [ + { + "name": "context_name", + "context": {"cluster": "test-cluster", "user": "test-user"}, + } + ], + 0, + ) + + args = Namespace(kubeconfig_file=None, context=None, only_logs=False) + provider = KubernetesProvider(args) + provider.context = { + "context": {"cluster": "test-cluster", "user": "test-user"}, + "namespace": "default", + } + provider.get_context_user_roles = MagicMock(return_value=["ClusterRole: admin"]) + + # Capture print output + captured_output = io.StringIO() + sys.stdout = captured_output + + provider.print_credentials() + + # Reset standard output + sys.stdout = sys.__stdout__ + + output = captured_output.getvalue() + assert "[test-cluster]" in output + assert "[test-user]" in output + assert "[default]" in output + assert "[ClusterRole: admin]" in output + + @patch("kubernetes.client.RbacAuthorizationV1Api") + @patch("kubernetes.config.list_kube_config_contexts") + @patch("kubernetes.config.load_incluster_config") + @patch("kubernetes.config.load_kube_config") + def test_search_and_save_roles( + self, + mock_load_kube_config, + mock_load_incluster_config, + mock_list_kube_config_contexts, + mock_rbac_api, + ): + mock_list_kube_config_contexts.return_value = ( + [ + { + "name": "context_name", + "context": {"cluster": "test-cluster", "user": "test-user"}, + } + ], + 0, + ) + mock_rbac_api.return_value.list_cluster_role_binding.return_value = MagicMock( + items=[] + ) + mock_rbac_api.return_value.list_role_binding_for_all_namespaces.return_value = ( + MagicMock(items=[]) + ) + + args = Namespace(kubeconfig_file=None, context=None, only_logs=False) + provider = KubernetesProvider(args) + provider.context = {"context": {"user": "test-user"}} + + roles = provider.search_and_save_roles([], [], "test-user", "ClusterRole") + assert isinstance(roles, list) diff --git a/tests/providers/kubernetes/lib/service/service_test.py b/tests/providers/kubernetes/lib/service/service_test.py new file mode 100644 index 0000000000..f3552c5dc4 --- /dev/null +++ b/tests/providers/kubernetes/lib/service/service_test.py @@ -0,0 +1,13 @@ +from kubernetes import client + +from prowler.providers.kubernetes.lib.service.service import KubernetesService +from tests.providers.kubernetes.audit_info_utils import set_mocked_audit_info + + +class Test_KubernetesService: + def test_KubernetesService_init(self): + audit_info = set_mocked_audit_info() + service = KubernetesService(audit_info) + + assert service.context is None + assert service.api_client == client.ApiClient