feat(kubernetes): add Kubernetes provider (#3226)

Co-authored-by: Pepe Fagoaga <pepe@verica.io>

tests/config/config_test.py

@@ -54,7 +54,7 @@ config_aws = {
 
 class Test_Config:
     def test_get_aws_available_regions(self):
-        assert len(get_aws_available_regions()) == 32
+        assert len(get_aws_available_regions()) == 33
 
     @mock.patch(
         "prowler.config.config.requests.get", new=mock_prowler_get_latest_release
@@ -179,6 +179,13 @@ class Test_Config:
 
         assert load_and_validate_config_file(provider, config_test_file) is None
 
+    def test_load_and_validate_config_file_kubernetes(self):
+        path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
+        config_test_file = f"{path}/fixtures/config.yaml"
+        provider = "kubernetes"
+
+        assert load_and_validate_config_file(provider, config_test_file) is None
+
     def test_load_and_validate_config_file_azure(self):
         path = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
         config_test_file = f"{path}/fixtures/config.yaml"

tests/config/fixtures/config.yaml

@@ -59,3 +59,6 @@ azure:
 
 # GCP Configuration
 gcp:
+
+# Kubernetes Configuration
+kubernetes:

tests/lib/check/custom_checks_metadata_test.py

@@ -21,6 +21,7 @@ CUSTOM_CHECKS_METADATA_FIXTURE_FILE_NOT_VALID = f"{os.path.dirname(os.path.realp
 AWS_PROVIDER = "aws"
 AZURE_PROVIDER = "azure"
 GCP_PROVIDER = "gcp"
+KUBERNETES_PROVIDER = "kubernetes"
 
 S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME = "s3_bucket_level_public_access_block"
 S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY = "medium"
@@ -81,6 +82,11 @@ class TestCustomChecksMetadata:
             GCP_PROVIDER, CUSTOM_CHECKS_METADATA_FIXTURE_FILE
         ) == {"Checks": {"bigquery_dataset_cmk_encryption": {"Severity": "low"}}}
 
+    def test_parse_custom_checks_metadata_file_for_kubernetes(self):
+        assert parse_custom_checks_metadata_file(
+            KUBERNETES_PROVIDER, CUSTOM_CHECKS_METADATA_FIXTURE_FILE
+        ) == {"Checks": {"bigquery_dataset_cmk_encryption": {"Severity": "low"}}}
+
     def test_parse_custom_checks_metadata_file_for_aws_validation_error(self, caplog):
         caplog.set_level(logging.CRITICAL)
 

tests/lib/check/fixtures/custom_checks_metadata_example.yaml

@@ -13,3 +13,7 @@ CustomChecksMetadata:
     Checks:
       bigquery_dataset_cmk_encryption:
         Severity: low
+  kubernetes:
+    Checks:
+      apiserver_anonymous_requests:
+        Severity: low

tests/lib/cli/parser_test.py

@@ -11,11 +11,11 @@ prowler_command = "prowler"
 
 # capsys
 # https://docs.pytest.org/en/7.1.x/how-to/capture-stdout-stderr.html
-prowler_default_usage_error = "usage: prowler [-h] [-v] {aws,azure,gcp} ..."
+prowler_default_usage_error = "usage: prowler [-h] [-v] {aws,azure,gcp,kubernetes} ..."
 
 
 def mock_get_available_providers():
-    return ["aws", "azure", "gcp"]
+    return ["aws", "azure", "gcp", "kubernetes"]
 
 
 class Test_Parser:
@@ -153,6 +153,41 @@ class Test_Parser:
         assert not parsed.list_categories
         assert not parsed.credentials_file
 
+    def test_default_parser_no_arguments_kubernetes(self):
+        provider = "kubernetes"
+        command = [prowler_command, provider]
+        parsed = self.parser.parse(command)
+        assert parsed.provider == provider
+        assert not parsed.quiet
+        assert len(parsed.output_modes) == 4
+        assert "csv" in parsed.output_modes
+        assert "html" in parsed.output_modes
+        assert "json" in parsed.output_modes
+        assert not parsed.output_filename
+        assert "output" in parsed.output_directory
+        assert not parsed.verbose
+        assert not parsed.no_banner
+        assert not parsed.slack
+        assert not parsed.unix_timestamp
+        assert parsed.log_level == "CRITICAL"
+        assert not parsed.log_file
+        assert not parsed.only_logs
+        assert not parsed.checks
+        assert not parsed.checks_file
+        assert not parsed.checks_folder
+        assert not parsed.services
+        assert not parsed.severity
+        assert not parsed.compliance
+        assert len(parsed.categories) == 0
+        assert not parsed.excluded_checks
+        assert not parsed.excluded_services
+        assert not parsed.list_checks
+        assert not parsed.list_services
+        assert not parsed.list_compliance
+        assert not parsed.list_compliance_requirements
+        assert not parsed.list_categories
+        assert not parsed.credentials_file
+
     def test_root_parser_version_short(self):
         command = [prowler_command, "-v"]
         with pytest.raises(SystemExit) as wrapped_exit:
@@ -194,15 +229,18 @@ class Test_Parser:
     def test_root_parser_azure_provider(self):
         command = [prowler_command, "azure"]
         parsed = self.parser.parse(command)
-        print(parsed)
         assert parsed.provider == "azure"
 
     def test_root_parser_gcp_provider(self):
         command = [prowler_command, "gcp"]
         parsed = self.parser.parse(command)
-        print(parsed)
         assert parsed.provider == "gcp"
 
+    def test_root_parser_kubernetes_provider(self):
+        command = [prowler_command, "kubernetes"]
+        parsed = self.parser.parse(command)
+        assert parsed.provider == "kubernetes"
+
     def test_root_parser_quiet_short(self):
         command = [prowler_command, "-q"]
         parsed = self.parser.parse(command)
@@ -1096,6 +1134,22 @@ class Test_Parser:
         assert parsed.project_ids[0] == project_1
         assert parsed.project_ids[1] == project_2
 
+    def test_parser_kubernetes_auth_kubeconfig_file(self):
+        argument = "--kubeconfig-file"
+        file = "config"
+        command = [prowler_command, "kubernetes", argument, file]
+        parsed = self.parser.parse(command)
+        assert parsed.provider == "kubernetes"
+        assert parsed.kubeconfig_file == file
+
+    def test_parser_kubernetes_auth_context(self):
+        argument = "--context"
+        context = "default"
+        command = [prowler_command, "kubernetes", argument, context]
+        parsed = self.parser.parse(command)
+        assert parsed.provider == "kubernetes"
+        assert parsed.context == context
+
     def test_validate_azure_region_valid_regions(self):
         expected_regions = [
             "AzureChinaCloud",

tests/providers/common/audit_info_test.py

@@ -22,6 +22,8 @@ from prowler.providers.common.audit_info import (
 from prowler.providers.common.models import Audit_Metadata
 from prowler.providers.gcp.gcp_provider import GCP_Provider
 from prowler.providers.gcp.lib.audit_info.models import GCP_Audit_Info
+from prowler.providers.kubernetes.kubernetes_provider import Kubernetes_Provider
+from prowler.providers.kubernetes.lib.audit_info.models import Kubernetes_Audit_Info
 
 EXAMPLE_AMI_ID = "ami-12c6146b"
 AWS_ACCOUNT_NUMBER = "123456789012"
@@ -93,6 +95,14 @@ def mock_get_project_ids(*_):
     return ["project"]
 
 
+def mock_set_kubernetes_credentials(*_):
+    return ("apiclient", "context")
+
+
+def mock_get_context_user_roles(*_):
+    return []
+
+
 class Test_Set_Audit_Info:
     # Mocked Audit Info
     def set_mocked_audit_info(self):
@@ -278,6 +288,30 @@ class Test_Set_Audit_Info:
         audit_info = set_provider_audit_info(provider, arguments)
         assert isinstance(audit_info, GCP_Audit_Info)
 
+    @patch.object(
+        Kubernetes_Provider, "__set_credentials__", new=mock_set_kubernetes_credentials
+    )
+    @patch.object(
+        Kubernetes_Provider, "get_context_user_roles", new=mock_get_context_user_roles
+    )
+    def test_set_audit_info_kubernetes(self):
+        provider = "kubernetes"
+        arguments = {
+            "profile": None,
+            "role": None,
+            "session_duration": None,
+            "external_id": None,
+            "regions": None,
+            "organizations_role": None,
+            "subscriptions": None,
+            "context": "default",
+            "kubeconfig_file": "config",
+            "config_file": default_config_file_path,
+        }
+
+        audit_info = set_provider_audit_info(provider, arguments)
+        assert isinstance(audit_info, Kubernetes_Audit_Info)
+
     @mock_resourcegroupstaggingapi
     @mock_ec2
     def test_get_tagged_resources(self):

tests/providers/common/common_outputs_test.py

@@ -16,10 +16,12 @@ from prowler.providers.common.outputs import (
     Aws_Output_Options,
     Azure_Output_Options,
     Gcp_Output_Options,
+    Kubernetes_Output_Options,
     get_provider_output_model,
     set_provider_output_options,
 )
 from prowler.providers.gcp.lib.audit_info.models import GCP_Audit_Info
+from prowler.providers.kubernetes.lib.audit_info.models import Kubernetes_Audit_Info
 
 AWS_ACCOUNT_NUMBER = "012345678912"
 DATETIME = "20230101120000"
@@ -51,6 +53,20 @@ class Test_Common_Output_Options:
         )
         return audit_info
 
+    # Mocked Kusbernete Audit Info
+    def set_mocked_kubernetes_audit_info(self):
+        audit_info = Kubernetes_Audit_Info(
+            api_client=None,
+            context={
+                "name": "test-context",
+                "context": {"cluster": "test-cluster", "user": "XXXXXXXXX"},
+            },
+            audit_resources=None,
+            audit_metadata=None,
+            audit_config=None,
+        )
+        return audit_info
+
     # Mocked AWS Audit Info
     def set_mocked_aws_audit_info(self):
         audit_info = AWS_Audit_Info(
@@ -147,6 +163,37 @@ class Test_Common_Output_Options:
         # Delete testing directory
         rmdir(arguments.output_directory)
 
+    def test_set_provider_output_options_kubernetes(self):
+        #  Set the cloud provider
+        provider = "kubernetes"
+        # Set the arguments passed
+        arguments = Namespace()
+        arguments.quiet = True
+        arguments.output_modes = ["html", "csv", "json"]
+        arguments.output_directory = "output_test_directory"
+        arguments.verbose = True
+        arguments.output_filename = "output_test_filename"
+        arguments.only_logs = False
+        arguments.unix_timestamp = False
+
+        audit_info = self.set_mocked_kubernetes_audit_info()
+        mutelist_file = ""
+        bulk_checks_metadata = {}
+        output_options = set_provider_output_options(
+            provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
+        )
+        assert isinstance(output_options, Kubernetes_Output_Options)
+        assert output_options.is_quiet
+        assert output_options.output_modes == ["html", "csv", "json"]
+        assert output_options.output_directory == arguments.output_directory
+        assert output_options.mutelist_file == ""
+        assert output_options.bulk_checks_metadata == {}
+        assert output_options.verbose
+        assert output_options.output_filename == arguments.output_filename
+
+        # Delete testing directory
+        rmdir(arguments.output_directory)
+
     def test_set_provider_output_options_aws_no_output_filename(self):
         #  Set the cloud provider
         provider = "aws"
@@ -362,7 +409,7 @@ class Test_Common_Output_Options:
         )
 
     def test_gcp_get_assessment_summary(self):
-        # Mock Azure Audit Info
+        # Mock GCP Audit Info
         audit_info = self.set_mocked_gcp_audit_info()
         profile = "default"
         assert (
@@ -395,11 +442,54 @@ class Test_Common_Output_Options:
             """
         )
 
+    def test_kubernetes_get_assessment_summary(self):
+        # Mock Kubernetes Audit Info
+        audit_info = self.set_mocked_kubernetes_audit_info()
+        assert (
+            get_assessment_summary(audit_info)
+            == """
+            <div class="col-md-2">
+                <div class="card">
+                    <div class="card-header">
+                        Kubernetes Assessment Summary
+                    </div>
+                    <ul class="list-group list-group-flush">
+                        <li class="list-group-item">
+                            <b>Kubernetes Context:</b> """
+            + audit_info.context["name"]
+            + """
+                        </li>
+                    </ul>
+                </div>
+            </div>
+            <div class="col-md-4">
+                <div class="card">
+                    <div class="card-header">
+                        Kubernetes Credentials
+                    </div>
+                    <ul class="list-group list-group-flush">
+                        <li class="list-group-item">
+                            <b>Kubernetes Cluster:</b> """
+            + audit_info.context["context"]["cluster"]
+            + """
+                        </li>
+                        <li class="list-group-item">
+                            <b>Kubernetes User:</b> """
+            + audit_info.context["context"]["user"]
+            + """
+                        </li>
+                    </ul>
+                </div>
+            </div>
+            """
+        )
+
     def test_get_provider_output_model(self):
         audit_info_class_names = [
             "AWS_Audit_Info",
             "GCP_Audit_Info",
             "Azure_Audit_Info",
+            "Kubernetes_Audit_Info",
         ]
         for class_name in audit_info_class_names:
             provider_prefix = class_name.split("_", 1)[0].lower().capitalize()