diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py index a459fa7bc4..8d5ac97ac4 100644 --- a/prowler/lib/check/checks_loader.py +++ b/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, ) ) diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index 1b92db8865..446440b6aa 100644 --- a/prowler/lib/check/models.py +++ b/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, diff --git a/tests/lib/check/check_loader_test.py b/tests/lib/check/check_loader_test.py index 56a9754f40..72080fc560 100644 --- a/tests/lib/check/check_loader_test.py +++ b/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"]}