diff --git a/prowler/config/config.py b/prowler/config/config.py index 8af991f593..ea764668da 100644 --- a/prowler/config/config.py +++ b/prowler/config/config.py @@ -1,6 +1,7 @@ import os import pathlib from datetime import datetime, timezone +from enum import Enum from os import getcwd import requests @@ -22,13 +23,20 @@ orange_color = "\033[38;5;208m" banner_color = "\033[1;92m" +class Provider(str, Enum): + AWS = "aws" + GCP = "gcp" + AZURE = "azure" + KUBERNETES = "kubernetes" + + # Compliance actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__))) def get_available_compliance_frameworks(provider=None): available_compliance_frameworks = [] - providers = ["aws", "gcp", "azure", "kubernetes"] + providers = [p.value for p in Provider] if provider: providers = [provider] for provider in providers: diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py index da11f231c8..16f04b2a13 100644 --- a/prowler/lib/check/check.py +++ b/prowler/lib/check/check.py @@ -1,4 +1,3 @@ -import functools import importlib import json import os @@ -293,32 +292,6 @@ def print_checks( print(message) -# Parse checks from compliance frameworks specification -def parse_checks_from_compliance_framework( - compliance_frameworks: list, bulk_compliance_frameworks: dict -) -> list: - """parse_checks_from_compliance_framework returns a set of checks from the given compliance_frameworks""" - checks_to_execute = set() - try: - for framework in compliance_frameworks: - # compliance_framework_json["Requirements"][*]["Checks"] - compliance_framework_checks_list = [ - requirement.Checks - for requirement in bulk_compliance_frameworks[framework].Requirements - ] - # Reduce nested list into a list - # Pythonic functional magic - compliance_framework_checks = functools.reduce( - lambda x, y: x + y, compliance_framework_checks_list - ) - # Then union this list of checks with the initial one - checks_to_execute = checks_to_execute.union(compliance_framework_checks) - except Exception as e: - logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}") - - return checks_to_execute - - # Import an input check using its path def import_check(check_path: str) -> ModuleType: lib = importlib.import_module(f"{check_path}") diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py index 00d93af4f6..a459fa7bc4 100644 --- a/prowler/lib/check/checks_loader.py +++ b/prowler/lib/check/checks_loader.py @@ -1,14 +1,7 @@ from colorama import Fore, Style -from prowler.lib.check.check import ( - parse_checks_from_compliance_framework, - parse_checks_from_file, -) -from prowler.lib.check.models import Severity -from prowler.lib.check.utils import ( - recover_checks_from_provider, - recover_checks_from_service, -) +from prowler.lib.check.check import parse_checks_from_file +from prowler.lib.check.models import CheckMetadata, Severity from prowler.lib.logger import logger @@ -29,8 +22,8 @@ def load_checks_to_execute( # Local subsets checks_to_execute = set() check_aliases = {} - check_severities = {severity.value: [] for severity in Severity} check_categories = {} + check_severities = {severity.value: [] for severity in Severity} # First, loop over the bulk_checks_metadata to extract the needed subsets for check, metadata in bulk_checks_metadata.items(): @@ -66,24 +59,39 @@ def load_checks_to_execute( checks_to_execute.update(check_severities[severity]) if service_list: - checks_to_execute = ( - recover_checks_from_service(service_list, provider) - & checks_to_execute - ) - + for service in service_list: + checks_to_execute = ( + set( + CheckMetadata.list( + bulk_checks_metadata=bulk_checks_metadata, + service=service, + ) + ) + & checks_to_execute + ) # Handle if there are checks passed using -C/--checks-file elif checks_file: checks_to_execute = parse_checks_from_file(checks_file, provider) # Handle if there are services passed using -s/--services elif service_list: - checks_to_execute = recover_checks_from_service(service_list, provider) + for service in service_list: + checks_to_execute.update( + CheckMetadata.list( + bulk_checks_metadata=bulk_checks_metadata, + service=service, + ) + ) # Handle if there are compliance frameworks passed using --compliance elif compliance_frameworks: - checks_to_execute = parse_checks_from_compliance_framework( - compliance_frameworks, bulk_compliance_frameworks - ) + for compliance_framework in compliance_frameworks: + checks_to_execute.update( + CheckMetadata.list( + bulk_checks_metadata=bulk_compliance_frameworks, + compliance_framework=compliance_framework, + ) + ) # Handle if there are categories passed using --categories elif categories: @@ -92,17 +100,13 @@ def load_checks_to_execute( # If there are no checks passed as argument else: - # Get all check modules to run with the specific provider - checks = recover_checks_from_provider(provider) - - for check_info in checks: - # Recover check name from import path (last part) - # Format: "providers.{provider}.services.{service}.{check_name}.{check_name}" - check_name = check_info[0] + # get all checks + for check_name in CheckMetadata.list( + bulk_checks_metadata=bulk_checks_metadata + ): checks_to_execute.add(check_name) - # Only execute threat detection checks if threat-detection category is set - if categories != [] and "threat-detection" not in categories: + if categories and categories != [] and "threat-detection" not in categories: for threat_detection_check in check_categories.get("threat-detection", []): checks_to_execute.discard(threat_detection_check) diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index dc10729a27..1b92db8865 100644 --- a/prowler/lib/check/models.py +++ b/prowler/lib/check/models.py @@ -1,12 +1,16 @@ +import functools import os import re import sys from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum +from typing import Set from pydantic import BaseModel, ValidationError, validator +from prowler.config.config import Provider +from prowler.lib.check.compliance_models import Compliance from prowler.lib.check.utils import recover_checks_from_provider from prowler.lib.logger import logger @@ -158,6 +162,214 @@ class CheckMetadata(BaseModel): return bulk_check_metadata + @staticmethod + def list( + bulk_checks_metadata: dict = None, + bulk_compliance_frameworks: dict = None, + provider: str = None, + severity: str = None, + category: str = None, + service: str = None, + compliance_framework: str = None, + ) -> Set["CheckMetadata"]: + """ + Returns a set of checks from the bulk checks metadata. + + Args: + provider (str): The provider of the checks. + bulk_checks_metadata (dict): The bulk checks metadata. + bulk_compliance_frameworks (dict): The bulk compliance frameworks. + severity (str): The severity of the checks. + category (str): The category of the checks. + service (str): The service of the checks. + compliance_framework (str): The compliance framework of the checks. + + Returns: + set: A set of checks. + """ + checks_from_provider = {} + checks_from_severity = {} + checks_from_category = {} + checks_from_service = {} + checks_from_compliance_framework = {} + # If the bulk checks metadata is not provided, get it + if not bulk_checks_metadata: + bulk_checks_metadata = {} + available_providers = [p.value for p in Provider] + for provider_name in available_providers: + bulk_checks_metadata.update(CheckMetadata.get_bulk(provider_name)) + if provider: + checks_from_provider = { + check_name + for check_name, check_metadata in bulk_checks_metadata.items() + if check_metadata.Provider == provider + } + if severity: + checks_from_severity = CheckMetadata.list_by_severity( + bulk_checks_metadata=bulk_checks_metadata, severity=severity + ) + if category: + checks_from_category = CheckMetadata.list_by_category( + bulk_checks_metadata=bulk_checks_metadata, category=category + ) + if service: + checks_from_service = CheckMetadata.list_by_service( + bulk_checks_metadata=bulk_checks_metadata, service=service + ) + if compliance_framework: + # Loaded here, as it is not always needed + if not bulk_compliance_frameworks: + bulk_compliance_frameworks = {} + 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( + CheckMetadata.list_by_compliance_framework( + bulk_compliance_frameworks=bulk_compliance_frameworks, + compliance_framework=compliance_framework, + ) + ) + + # Get all the checks: + checks = set(bulk_checks_metadata.keys()) + # Get the intersection of the checks + if len(checks_from_provider) > 0 or provider: + checks = checks & checks_from_provider + if len(checks_from_severity) > 0 or severity: + checks = checks & checks_from_severity + if len(checks_from_category) > 0 or category: + checks = checks & checks_from_category + if len(checks_from_service) > 0 or service: + checks = checks & checks_from_service + if len(checks_from_compliance_framework) > 0 or compliance_framework: + checks = checks & checks_from_compliance_framework + + return checks + + @staticmethod + def get(bulk_checks_metadata: dict, check_id: str) -> "CheckMetadata": + """ + Returns the check metadata from the bulk checks metadata. + + Args: + bulk_checks_metadata (dict): The bulk checks metadata. + check_id (str): The check ID. + + Returns: + CheckMetadata: The check metadata. + """ + + return bulk_checks_metadata.get(check_id, None) + + @staticmethod + def list_by_severity(bulk_checks_metadata: dict, severity: str = None) -> set: + """ + Returns a set of checks by severity from the bulk checks metadata. + + Args: + bulk_checks_metadata (dict): The bulk checks metadata. + severity (str): The severity. + + Returns: + set: A set of checks by severity. + """ + checks = set() + + if severity: + checks = { + check_name + for check_name, check_metadata in bulk_checks_metadata.items() + if check_metadata.Severity == severity + } + + return checks + + @staticmethod + def list_by_category(bulk_checks_metadata: dict, category: str = None) -> set: + """ + Returns a set of checks by category from the bulk checks metadata. + + Args: + bulk_checks_metadata (dict): The bulk checks metadata. + category (str): The category. + + Returns: + set: A set of checks by category. + """ + checks = set() + + if category: + checks = { + check_name + for check_name, check_metadata in bulk_checks_metadata.items() + if category in check_metadata.Categories + } + + return checks + + @staticmethod + def list_by_service(bulk_checks_metadata: dict, service: str = None) -> set: + """ + Returns a set of checks by service from the bulk checks metadata. + + Args: + bulk_checks_metadata (dict): The bulk checks metadata. + service (str): The service. + + Returns: + set: A set of checks by service. + """ + checks = set() + + if service: + if service == "lambda": + service = "awslambda" + checks = { + check_name + for check_name, check_metadata in bulk_checks_metadata.items() + if check_metadata.ServiceName == service + } + + return checks + + @staticmethod + def list_by_compliance_framework( + bulk_compliance_frameworks: dict, compliance_framework: str = None + ) -> set: + """ + Returns a set of checks by compliance framework from the bulk compliance frameworks. + + Args: + bulk_compliance_frameworks (dict): The bulk compliance frameworks. + compliance_framework (str): The compliance framework. + + Returns: + set: A set of checks by compliance framework. + """ + checks = set() + + if compliance_framework: + try: + checks_from_framework_list = [ + requirement.Checks + for requirement in bulk_compliance_frameworks[ + compliance_framework + ].Requirements + ] + # Reduce nested list into a list + # Pythonic functional magic + checks_from_framework = functools.reduce( + lambda x, y: x + y, checks_from_framework_list + ) + # Then union this list of checks with the initial one + checks = checks.union(checks_from_framework) + except Exception as e: + logger.error( + f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}" + ) + + return checks + class Check(ABC, CheckMetadata): """Prowler Check""" diff --git a/tests/lib/check/check_loader_test.py b/tests/lib/check/check_loader_test.py index 829004a201..56a9754f40 100644 --- a/tests/lib/check/check_loader_test.py +++ b/tests/lib/check/check_loader_test.py @@ -64,13 +64,8 @@ class TestCheckLoader: categories = None with patch( - "prowler.lib.check.checks_loader.recover_checks_from_provider", - return_value=[ - ( - f"{S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME}", - "path/to/{S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME}", - ) - ], + "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, @@ -145,7 +140,7 @@ class TestCheckLoader: categories = None with patch( - "prowler.lib.check.checks_loader.recover_checks_from_service", + "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( @@ -175,7 +170,10 @@ class TestCheckLoader: categories = None with patch( - "prowler.lib.check.checks_loader.recover_checks_from_service", + "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( @@ -235,7 +233,7 @@ class TestCheckLoader: categories = None with patch( - "prowler.lib.check.checks_loader.recover_checks_from_service", + "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( @@ -265,7 +263,7 @@ class TestCheckLoader: categories = None with patch( - "prowler.lib.check.checks_loader.parse_checks_from_compliance_framework", + "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( diff --git a/tests/lib/check/compliance_check_test.py b/tests/lib/check/compliance_check_test.py index 476ad80ce4..6994d00c29 100644 --- a/tests/lib/check/compliance_check_test.py +++ b/tests/lib/check/compliance_check_test.py @@ -10,98 +10,100 @@ from prowler.lib.check.compliance_models import ( ) from prowler.lib.check.models import CheckMetadata +custom_compliance_metadata = { + "framework1_aws": Compliance( + Framework="Framework1", + Provider="aws", + Version="1.0", + Description="Framework 1 Description", + Requirements=[ + Compliance_Requirement( + Id="1.1.1", + Description="description", + Attributes=[ + CIS_Requirement_Attribute( + Section="1. Identity", + Profile=CIS_Requirement_Attribute_Profile("Level 1"), + AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( + "Manual" + ), + Description="Description", + RationaleStatement="Rationale", + ImpactStatement="Impact", + RemediationProcedure="Remediation", + AuditProcedure="Audit", + AdditionalInformation="Additional", + References="References", + ) + ], + Checks=[ + "accessanalyzer_enabled", + "iam_user_mfa_enabled_console_access", + ], + ), + # Manual requirement + Compliance_Requirement( + Id="1.1.2", + Description="description", + Attributes=[ + CIS_Requirement_Attribute( + Section="1. Identity", + Profile=CIS_Requirement_Attribute_Profile("Level 1"), + AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( + "Manual" + ), + Description="Description", + RationaleStatement="Rationale", + ImpactStatement="Impact", + RemediationProcedure="Remediation", + AuditProcedure="Audit", + AdditionalInformation="Additional", + References="References", + ) + ], + Checks=[], + ), + ], + ), + "framework1_azure": Compliance( + Framework="Framework1", + Provider="azure", + Version="1.0", + Description="Framework 2 Description", + Requirements=[ + Compliance_Requirement( + Id="1.1.1", + Description="description", + Attributes=[ + CIS_Requirement_Attribute( + Section="1. Identity", + Profile=CIS_Requirement_Attribute_Profile("Level 1"), + AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( + "Manual" + ), + Description="Description", + RationaleStatement="Rationale", + ImpactStatement="Impact", + RemediationProcedure="Remediation", + AuditProcedure="Audit", + AdditionalInformation="Additional", + References="References", + ) + ], + Checks=[], + ) + ], + ), +} + class TestCompliance: - def get_custom_framework(self): - return { - "framework1_aws": Compliance( - Framework="Framework1", - Provider="aws", - Version="1.0", - Description="Framework 1 Description", - Requirements=[ - Compliance_Requirement( - Id="1.1.1", - Description="description", - Attributes=[ - CIS_Requirement_Attribute( - Section="1. Identity", - Profile=CIS_Requirement_Attribute_Profile("Level 1"), - AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( - "Manual" - ), - Description="Description", - RationaleStatement="Rationale", - ImpactStatement="Impact", - RemediationProcedure="Remediation", - AuditProcedure="Audit", - AdditionalInformation="Additional", - References="References", - ) - ], - Checks=["check1", "check2"], - ), - # Manual requirement - Compliance_Requirement( - Id="1.1.2", - Description="description", - Attributes=[ - CIS_Requirement_Attribute( - Section="1. Identity", - Profile=CIS_Requirement_Attribute_Profile("Level 1"), - AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( - "Manual" - ), - Description="Description", - RationaleStatement="Rationale", - ImpactStatement="Impact", - RemediationProcedure="Remediation", - AuditProcedure="Audit", - AdditionalInformation="Additional", - References="References", - ) - ], - Checks=[], - ), - ], - ), - "framework1_azure": Compliance( - Framework="Framework1", - Provider="azure", - Version="1.0", - Description="Framework 2 Description", - Requirements=[ - Compliance_Requirement( - Id="1.1.1", - Description="description", - Attributes=[ - CIS_Requirement_Attribute( - Section="1. Identity", - Profile=CIS_Requirement_Attribute_Profile("Level 1"), - AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus( - "Manual" - ), - Description="Description", - RationaleStatement="Rationale", - ImpactStatement="Impact", - RemediationProcedure="Remediation", - AuditProcedure="Audit", - AdditionalInformation="Additional", - References="References", - ) - ], - Checks=[], - ) - ], - ), - } - def get_custom_check_metadata(self): return { - "check1": CheckMetadata( + "accessanalyzer_enabled": CheckMetadata( Provider="aws", - CheckID="check1", + CheckID="accessanalyzer_enabled", CheckTitle="Check 1", CheckType=["type1"], ServiceName="service1", @@ -127,9 +129,9 @@ class TestCompliance: Notes="notes1", Compliance=[], ), - "check2": CheckMetadata( + "iam_user_mfa_enabled_console_access": CheckMetadata( Provider="aws", - CheckID="check2", + CheckID="iam_user_mfa_enabled_console_access", CheckTitle="Check 2", CheckType=["type2"], ServiceName="service2", @@ -158,44 +160,52 @@ class TestCompliance: } def test_update_checks_metadata(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata bulk_checks_metadata = self.get_custom_check_metadata() updated_metadata = update_checks_metadata_with_compliance( bulk_compliance_frameworks, bulk_checks_metadata ) - assert "check1" in updated_metadata - assert "check2" in updated_metadata + assert "accessanalyzer_enabled" in updated_metadata + assert "iam_user_mfa_enabled_console_access" in updated_metadata - check1_compliance = updated_metadata["check1"].Compliance[0] + accessanalyzer_enabled_compliance = updated_metadata[ + "accessanalyzer_enabled" + ].Compliance[0] - assert len(updated_metadata["check1"].Compliance) == 1 - assert check1_compliance.Framework == "Framework1" - assert check1_compliance.Provider == "aws" - assert check1_compliance.Version == "1.0" - assert check1_compliance.Description == "Framework 1 Description" - assert len(check1_compliance.Requirements) == 1 + assert len(updated_metadata["accessanalyzer_enabled"].Compliance) == 1 + assert accessanalyzer_enabled_compliance.Framework == "Framework1" + assert accessanalyzer_enabled_compliance.Provider == "aws" + assert accessanalyzer_enabled_compliance.Version == "1.0" + assert ( + accessanalyzer_enabled_compliance.Description == "Framework 1 Description" + ) + assert len(accessanalyzer_enabled_compliance.Requirements) == 1 - check1_requirement = check1_compliance.Requirements[0] - assert check1_requirement.Id == "1.1.1" - assert check1_requirement.Description == "description" - assert len(check1_requirement.Attributes) == 1 + accessanalyzer_enabled_requirement = ( + accessanalyzer_enabled_compliance.Requirements[0] + ) + assert accessanalyzer_enabled_requirement.Id == "1.1.1" + assert accessanalyzer_enabled_requirement.Description == "description" + assert len(accessanalyzer_enabled_requirement.Attributes) == 1 - check1_attribute = check1_requirement.Attributes[0] - assert check1_attribute.Section == "1. Identity" - assert check1_attribute.Profile == "Level 1" - assert check1_attribute.AssessmentStatus == "Manual" - assert check1_attribute.Description == "Description" - assert check1_attribute.RationaleStatement == "Rationale" - assert check1_attribute.ImpactStatement == "Impact" - assert check1_attribute.RemediationProcedure == "Remediation" - assert check1_attribute.AuditProcedure == "Audit" - assert check1_attribute.AdditionalInformation == "Additional" - assert check1_attribute.References == "References" + accessanalyzer_enabled_attribute = ( + accessanalyzer_enabled_requirement.Attributes[0] + ) + assert accessanalyzer_enabled_attribute.Section == "1. Identity" + assert accessanalyzer_enabled_attribute.Profile == "Level 1" + assert accessanalyzer_enabled_attribute.AssessmentStatus == "Manual" + assert accessanalyzer_enabled_attribute.Description == "Description" + assert accessanalyzer_enabled_attribute.RationaleStatement == "Rationale" + assert accessanalyzer_enabled_attribute.ImpactStatement == "Impact" + assert accessanalyzer_enabled_attribute.RemediationProcedure == "Remediation" + assert accessanalyzer_enabled_attribute.AuditProcedure == "Audit" + assert accessanalyzer_enabled_attribute.AdditionalInformation == "Additional" + assert accessanalyzer_enabled_attribute.References == "References" def test_list_no_provider(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata list_compliance = Compliance.list(bulk_compliance_frameworks) @@ -204,7 +214,7 @@ class TestCompliance: assert list_compliance[1] == "framework1_azure" def test_list_with_provider_aws(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata list_compliance = Compliance.list(bulk_compliance_frameworks, provider="aws") @@ -212,7 +222,7 @@ class TestCompliance: assert list_compliance[0] == "framework1_aws" def test_list_with_provider_azure(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata list_compliance = Compliance.list(bulk_compliance_frameworks, provider="azure") @@ -220,7 +230,7 @@ class TestCompliance: assert list_compliance[0] == "framework1_azure" def test_get_compliance_frameworks(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata compliance_framework = Compliance.get( bulk_compliance_frameworks, compliance_framework_name="framework1_aws" @@ -243,7 +253,7 @@ class TestCompliance: assert len(compliance_framework.Requirements) == 1 def test_get_non_existent_framework(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata compliance_framework = Compliance.get( bulk_compliance_frameworks, compliance_framework_name="non_existent" @@ -252,14 +262,14 @@ class TestCompliance: assert compliance_framework is None def test_list_compliance_requirements_no_compliance(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata list_requirements = Compliance.list_requirements(bulk_compliance_frameworks) assert len(list_requirements) == 0 def test_list_compliance_requirements_with_compliance(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata list_requirements = Compliance.list_requirements( bulk_compliance_frameworks, compliance_framework="framework1_aws" @@ -277,7 +287,7 @@ class TestCompliance: assert list_requirements[0] == "1.1.1" def test_get_compliance_requirement(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata compliance_requirement = Compliance.get_requirement( bulk_compliance_frameworks, @@ -310,7 +320,7 @@ class TestCompliance: assert len(compliance_requirement.Attributes) == 1 def test_get_compliance_requirement_not_found(self): - bulk_compliance_frameworks = self.get_custom_framework() + bulk_compliance_frameworks = custom_compliance_metadata compliance_requirement = Compliance.get_requirement( bulk_compliance_frameworks, diff --git a/tests/lib/check/models_test.py b/tests/lib/check/models_test.py index f373e0f80f..7cf470c7e6 100644 --- a/tests/lib/check/models_test.py +++ b/tests/lib/check/models_test.py @@ -1,6 +1,36 @@ from unittest import mock from prowler.lib.check.models import CheckMetadata +from tests.lib.check.compliance_check_test import custom_compliance_metadata + +mock_metadata = CheckMetadata( + Provider="aws", + CheckID="accessanalyzer_enabled", + CheckTitle="Check 1", + CheckType=["type1"], + ServiceName="service1", + SubServiceName="subservice1", + ResourceIdTemplate="template1", + Severity="high", + ResourceType="resource1", + Description="Description 1", + Risk="risk1", + RelatedUrl="url1", + Remediation={ + "Code": { + "CLI": "cli1", + "NativeIaC": "native1", + "Other": "other1", + "Terraform": "terraform1", + }, + "Recommendation": {"Text": "text1", "Url": "url1"}, + }, + Categories=["categoryone"], + DependsOn=["dependency1"], + RelatedTo=["related1"], + Notes="notes1", + Compliance=[], +) class TestCheckMetada: @@ -13,44 +43,215 @@ class TestCheckMetada: ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") ] - check_metadata = CheckMetadata( - Provider="aws", - CheckID="accessanalyzer_enabled", - CheckTitle="Check 1", - CheckType=["type1"], - ServiceName="service1", - SubServiceName="subservice1", - ResourceIdTemplate="template1", - Severity="high", - ResourceType="resource1", - Description="Description 1", - Risk="risk1", - RelatedUrl="url1", - Remediation={ - "Code": { - "CLI": "cli1", - "NativeIaC": "native1", - "Other": "other1", - "Terraform": "terraform1", - }, - "Recommendation": {"Text": "text1", "Url": "url1"}, - }, - Categories=["categoryone"], - DependsOn=["dependency1"], - RelatedTo=["related1"], - Notes="notes1", - Compliance=[], - ) - # Mock the return value of load_check_metadata - mock_load_metadata.return_value = check_metadata + mock_load_metadata.return_value = mock_metadata result = CheckMetadata.get_bulk(provider="aws") # Assertions assert "accessanalyzer_enabled" in result.keys() - assert result["accessanalyzer_enabled"] == check_metadata + assert result["accessanalyzer_enabled"] == mock_metadata mock_recover_checks.assert_called_once_with("aws") mock_load_metadata.assert_called_once_with( "/path/to/accessanalyzer_enabled/accessanalyzer_enabled.metadata.json" ) + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata) + + # Assertions + assert result == {"accessanalyzer_enabled"} + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_get(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata) + + # Assertions + assert result == {"accessanalyzer_enabled"} + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_severity(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata, severity="high") + + # Assertions + assert result == {"accessanalyzer_enabled"} + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_severity_not_values(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata, severity="low") + + # Assertions + assert result == set() + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_category(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list( + bulk_checks_metadata=bulk_metadata, category="categoryone" + ) + + # Assertions + assert result == {"accessanalyzer_enabled"} + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_category_not_valid(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list( + bulk_checks_metadata=bulk_metadata, category="categorytwo" + ) + + # Assertions + assert result == set() + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_service(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list( + bulk_checks_metadata=bulk_metadata, service="service1" + ) + + # Assertions + assert result == {"accessanalyzer_enabled"} + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_service_invalid(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list( + bulk_checks_metadata=bulk_metadata, service="service2" + ) + + # Assertions + assert result == set() + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_by_compliance(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of recover_checks_from_provider + mock_recover_checks.return_value = [ + ("accessanalyzer_enabled", "/path/to/accessanalyzer_enabled") + ] + + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + bulk_compliance_frameworks = custom_compliance_metadata + + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list( + bulk_checks_metadata=bulk_metadata, + bulk_compliance_frameworks=bulk_compliance_frameworks, + compliance_framework="framework1_aws", + ) + + # Assertions + assert result == {"accessanalyzer_enabled"} + + def test_list_by_compliance_empty(self): + bulk_compliance_frameworks = custom_compliance_metadata + result = CheckMetadata.list( + bulk_compliance_frameworks=bulk_compliance_frameworks, + compliance_framework="framework1_azure", + ) + + # Assertions + assert result == set() + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_list_only_check_metadata(self, mock_recover_checks, mock_load_metadata): + # Mock the return value of load_check_metadata + mock_load_metadata.return_value = mock_metadata + + bulk_metadata = CheckMetadata.get_bulk(provider="aws") + + result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata) + assert result == set() diff --git a/tests/lib/scan/scan_test.py b/tests/lib/scan/scan_test.py index ab278b3f41..715c4dbc77 100644 --- a/tests/lib/scan/scan_test.py +++ b/tests/lib/scan/scan_test.py @@ -121,6 +121,15 @@ def mock_load_check_metadata(): yield mock_load +@pytest.fixture +def mock_load_checks_to_execute(): + with mock.patch( + "prowler.lib.check.models.CheckMetadata.list", autospec=True + ) as mock_load: + mock_load.return_value = {"accessanalyzer_enabled"} + yield mock_load + + class TestScan: def test_init(mock_provider): checks_to_execute = { @@ -260,16 +269,16 @@ class TestScan: def test_init_with_no_checks( mock_provider, - mock_list_modules, mock_recover_checks_from_provider, mock_load_check_metadata, + mock_load_checks_to_execute, ): checks_to_execute = set() mock_provider.type = "aws" scan = Scan(mock_provider, checks=checks_to_execute) - mock_list_modules.assert_called_once_with("aws", None) mock_load_check_metadata.assert_called_once() + mock_load_checks_to_execute.assert_called_once() mock_recover_checks_from_provider.assert_called_once_with("aws") assert scan.provider == mock_provider