fix(checks_loader): solve issue related with checks from compliance (#5601)

prowler/lib/check/checks_loader.py

@@ -1,21 +1,22 @@
 from colorama import Fore, Style
 
 from prowler.lib.check.check import parse_checks_from_file
+from prowler.lib.check.compliance_models import Compliance
 from prowler.lib.check.models import CheckMetadata, Severity
 from prowler.lib.logger import logger
 
 
 # Generate the list of checks to execute
 def load_checks_to_execute(
-    bulk_checks_metadata: dict,
-    bulk_compliance_frameworks: dict,
-    checks_file: str,
-    check_list: list,
-    service_list: list,
-    severities: list,
-    compliance_frameworks: list,
-    categories: set,
     provider: str,
+    bulk_checks_metadata: dict = None,
+    bulk_compliance_frameworks: dict = None,
+    checks_file: str = None,
+    check_list: list = None,
+    service_list: list = None,
+    severities: list = None,
+    compliance_frameworks: list = None,
+    categories: set = None,
 ) -> set:
     """Generate the list of checks to execute based on the cloud provider and the input arguments given"""
     try:
@@ -25,6 +26,8 @@ def load_checks_to_execute(
         check_categories = {}
         check_severities = {severity.value: [] for severity in Severity}
 
+        if not bulk_checks_metadata:
+            bulk_checks_metadata = CheckMetadata.get_bulk(provider=provider)
         # First, loop over the bulk_checks_metadata to extract the needed subsets
         for check, metadata in bulk_checks_metadata.items():
             try:
@@ -85,10 +88,12 @@ def load_checks_to_execute(
 
         # Handle if there are compliance frameworks passed using --compliance
         elif compliance_frameworks:
+            if not bulk_compliance_frameworks:
+                bulk_compliance_frameworks = Compliance.get_bulk(provider=provider)
             for compliance_framework in compliance_frameworks:
                 checks_to_execute.update(
                     CheckMetadata.list(
-                        bulk_checks_metadata=bulk_compliance_frameworks,
+                        bulk_compliance_frameworks=bulk_compliance_frameworks,
                         compliance_framework=compliance_framework,
                     )
                 )

prowler/lib/check/models.py

@@ -187,11 +187,11 @@ class CheckMetadata(BaseModel):
         Returns:
             set: A set of checks.
         """
-        checks_from_provider = {}
-        checks_from_severity = {}
-        checks_from_category = {}
-        checks_from_service = {}
-        checks_from_compliance_framework = {}
+        checks_from_provider = set()
+        checks_from_severity = set()
+        checks_from_category = set()
+        checks_from_service = set()
+        checks_from_compliance_framework = set()
         # If the bulk checks metadata is not provided, get it
         if not bulk_checks_metadata:
             bulk_checks_metadata = {}
@@ -223,7 +223,7 @@ class CheckMetadata(BaseModel):
                 available_providers = [p.value for p in Provider]
                 for provider in available_providers:
                     bulk_compliance_frameworks = Compliance.get_bulk(provider=provider)
-            checks_from_compliance_framework = set(
+            checks_from_compliance_framework = (
                 CheckMetadata.list_by_compliance_framework(
                     bulk_compliance_frameworks=bulk_compliance_frameworks,
                     compliance_framework=compliance_framework,

tests/lib/check/check_loader_test.py

@@ -4,6 +4,7 @@ from prowler.lib.check.checks_loader import (
     load_checks_to_execute,
     update_checks_to_execute_with_aliases,
 )
+from prowler.lib.check.compliance_models import Compliance, Compliance_Requirement
 from prowler.lib.check.models import CheckMetadata, Code, Recommendation, Remediation
 
 S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME = "s3_bucket_level_public_access_block"
@@ -55,105 +56,49 @@ class TestCheckLoader:
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = None
-        service_list = None
-        severities = None
-        compliance_frameworks = None
-        categories = None
 
-        with patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list",
-            return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
-        ):
-            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
-            )
+        assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metatada,
+            provider=self.provider,
+        )
 
     def test_load_checks_to_execute_with_check_list(self):
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
         check_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME]
-        service_list = None
-        severities = None
-        compliance_frameworks = None
-        categories = None
 
         assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-            bulk_checks_metatada,
-            bulk_compliance_frameworks,
-            checks_file,
-            check_list,
-            service_list,
-            severities,
-            compliance_frameworks,
-            categories,
-            self.provider,
+            bulk_checks_metadata=bulk_checks_metatada,
+            check_list=check_list,
+            provider=self.provider,
         )
 
     def test_load_checks_to_execute_with_severities(self):
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
-        service_list = None
         severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
-        compliance_frameworks = None
-        categories = None
 
         assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-            bulk_checks_metatada,
-            bulk_compliance_frameworks,
-            checks_file,
-            check_list,
-            service_list,
-            severities,
-            compliance_frameworks,
-            categories,
-            self.provider,
+            bulk_checks_metadata=bulk_checks_metatada,
+            severities=severities,
+            provider=self.provider,
         )
 
     def test_load_checks_to_execute_with_severities_and_services(self):
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
         service_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_SERVICE]
         severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
-        compliance_frameworks = None
-        categories = None
 
-        with patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list_by_service",
-            return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
-        ):
-            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
-            )
+        assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metatada,
+            service_list=service_list,
+            severities=severities,
+            provider=self.provider,
+        )
 
     def test_load_checks_to_execute_with_severities_and_services_not_within_severity(
         self,
@@ -161,32 +106,15 @@ class TestCheckLoader:
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
         service_list = ["ec2"]
         severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
-        compliance_frameworks = None
-        categories = None
 
-        with patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list_by_severity",
-            return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
-        ), patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list_by_service",
-            return_value={"ec2_ami_public"},
-        ):
-            assert set() == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
-            )
+        assert set() == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metatada,
+            service_list=service_list,
+            severities=severities,
+            provider=self.provider,
+        )
 
     def test_load_checks_to_execute_with_checks_file(
         self,
@@ -194,28 +122,15 @@ class TestCheckLoader:
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
         checks_file = "path/to/test_file"
-        check_list = []
-        service_list = []
-        severities = []
-        compliance_frameworks = None
-        categories = None
-
         with patch(
             "prowler.lib.check.checks_loader.parse_checks_from_file",
             return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
         ):
             assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
+                bulk_checks_metadata=bulk_checks_metatada,
+                checks_file=checks_file,
+                provider=self.provider,
             )
 
     def test_load_checks_to_execute_with_service_list(
@@ -224,59 +139,40 @@ class TestCheckLoader:
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
         service_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_SERVICE]
-        severities = []
-        compliance_frameworks = None
-        categories = None
 
-        with patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list_by_service",
-            return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
-        ):
-            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
-            )
+        assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+            bulk_checks_metadata=bulk_checks_metatada,
+            service_list=service_list,
+            provider=self.provider,
+        )
 
     def test_load_checks_to_execute_with_compliance_frameworks(
         self,
     ):
-        bulk_checks_metatada = {
-            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
+        bulk_compliance_frameworks = {
+            "soc2_aws": Compliance(
+                Framework="SOC2",
+                Provider="aws",
+                Version="2.0",
+                Description="This CIS Benchmark is the product of a community consensus process and consists of secure configuration guidelines developed for Azuee Platform",
+                Requirements=[
+                    Compliance_Requirement(
+                        Checks=[S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME],
+                        Id="",
+                        Description="",
+                        Attributes=[],
+                    )
+                ],
+            ),
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
-        service_list = []
-        severities = []
-        compliance_frameworks = ["test-compliance-framework"]
-        categories = None
+        compliance_frameworks = ["soc2_aws"]
 
-        with patch(
-            "prowler.lib.check.checks_loader.CheckMetadata.list",
-            return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
-        ):
-            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-                bulk_checks_metatada,
-                bulk_compliance_frameworks,
-                checks_file,
-                check_list,
-                service_list,
-                severities,
-                compliance_frameworks,
-                categories,
-                self.provider,
-            )
+        assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+            bulk_compliance_frameworks=bulk_compliance_frameworks,
+            compliance_frameworks=compliance_frameworks,
+            provider=self.provider,
+        )
 
     def test_load_checks_to_execute_with_categories(
         self,
@@ -284,26 +180,61 @@ class TestCheckLoader:
         bulk_checks_metatada = {
             S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
         }
-        bulk_compliance_frameworks = None
-        checks_file = None
-        check_list = []
-        service_list = []
-        severities = []
-        compliance_frameworks = []
         categories = {"internet-exposed"}
 
         assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
-            bulk_checks_metatada,
-            bulk_compliance_frameworks,
-            checks_file,
-            check_list,
-            service_list,
-            severities,
-            compliance_frameworks,
-            categories,
-            self.provider,
+            bulk_checks_metadata=bulk_checks_metatada,
+            categories=categories,
+            provider=self.provider,
         )
 
+    def test_load_checks_to_execute_no_bulk_checks_metadata(self):
+        bulk_checks_metatada = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
+        }
+        with patch(
+            "prowler.lib.check.checks_loader.CheckMetadata.get_bulk",
+            return_value=bulk_checks_metatada,
+        ):
+            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+                provider=self.provider,
+            )
+
+    def test_load_checks_to_execute_no_bulk_compliance_frameworks(self):
+        bulk_compliance_frameworks = {
+            "soc2_aws": Compliance(
+                Framework="SOC2",
+                Provider="aws",
+                Version="2.0",
+                Description="This CIS Benchmark is the product of a community consensus process and consists of secure configuration guidelines developed for Azuee Platform",
+                Requirements=[
+                    Compliance_Requirement(
+                        Checks=[S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME],
+                        Id="",
+                        Description="",
+                        Attributes=[],
+                    )
+                ],
+            ),
+        }
+
+        compliance_frameworks = ["soc2_aws"]
+
+        bulk_checks_metatada = {
+            S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_metadata()
+        }
+        with patch(
+            "prowler.lib.check.checks_loader.CheckMetadata.get_bulk",
+            return_value=bulk_checks_metatada,
+        ), patch(
+            "prowler.lib.check.checks_loader.Compliance.get_bulk",
+            return_value=bulk_compliance_frameworks,
+        ):
+            assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
+                compliance_frameworks=compliance_frameworks,
+                provider=self.provider,
+            )
+
     def test_update_checks_to_execute_with_aliases(self):
         checks_to_execute = {"renamed_check"}
         check_aliases = {"renamed_check": ["check_name"]}