diff --git a/prowler/__main__.py b/prowler/__main__.py index 31d555d0de..4f62fc521b 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -15,8 +15,6 @@ from prowler.config.config import ( ) from prowler.lib.banner import print_banner from prowler.lib.check.check import ( - bulk_load_checks_metadata, - bulk_load_compliance_frameworks, exclude_checks_to_run, exclude_services_to_run, execute_checks, @@ -36,10 +34,12 @@ from prowler.lib.check.check import ( ) from prowler.lib.check.checks_loader import load_checks_to_execute from prowler.lib.check.compliance import update_checks_metadata_with_compliance +from prowler.lib.check.compliance_models import Compliance from prowler.lib.check.custom_checks_metadata import ( parse_custom_checks_metadata_file, update_checks_metadata, ) +from prowler.lib.check.models import CheckMetadata from prowler.lib.cli.parser import ProwlerArgumentParser from prowler.lib.logger import logger, set_logging_config from prowler.lib.outputs.asff.asff import ASFF @@ -131,7 +131,7 @@ def prowler(): # Load checks metadata logger.debug("Loading checks metadata from .metadata.json files") - bulk_checks_metadata = bulk_load_checks_metadata(provider) + bulk_checks_metadata = CheckMetadata.get_bulk(provider) if args.list_categories: print_categories(list_categories(bulk_checks_metadata)) @@ -141,7 +141,7 @@ def prowler(): # Load compliance frameworks logger.debug("Loading compliance frameworks from .json files") - bulk_compliance_frameworks = bulk_load_compliance_frameworks(provider) + bulk_compliance_frameworks = Compliance.get_bulk(provider) # Complete checks metadata with the compliance framework specification bulk_checks_metadata = update_checks_metadata_with_compliance( bulk_compliance_frameworks, bulk_checks_metadata diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py index c45399045d..32a40349db 100644 --- a/prowler/lib/check/check.py +++ b/prowler/lib/check/check.py @@ -6,7 +6,6 @@ import re import shutil import sys import traceback -from pkgutil import walk_packages from types import ModuleType from typing import Any @@ -15,76 +14,15 @@ from colorama import Fore, Style import prowler from prowler.config.config import orange_color -from prowler.lib.check.compliance_models import load_compliance_framework from prowler.lib.check.custom_checks_metadata import update_check_metadata -from prowler.lib.check.models import Check, CheckMetadata, load_check_metadata +from prowler.lib.check.models import Check +from prowler.lib.check.utils import recover_checks_from_provider from prowler.lib.logger import logger from prowler.lib.outputs.outputs import report from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes from prowler.providers.common.models import Audit_Metadata -# Load all checks metadata -def bulk_load_checks_metadata(provider: str) -> dict[str, CheckMetadata]: - """ - Load the metadata of all checks for a given provider reading the check's metadata files. - Args: - provider (str): The name of the provider. - Returns: - dict[str, CheckMetadata]: A dictionary containing the metadata of all checks, with the CheckID as the key. - """ - - bulk_check_metadata = {} - checks = recover_checks_from_provider(provider) - # Build list of check's metadata files - for check_info in checks: - # Build check path name - check_name = check_info[0] - check_path = check_info[1] - # Ignore fixer files - if check_name.endswith("_fixer"): - continue - # Append metadata file extension - metadata_file = f"{check_path}/{check_name}.metadata.json" - # Load metadata - check_metadata = load_check_metadata(metadata_file) - bulk_check_metadata[check_metadata.CheckID] = check_metadata - - return bulk_check_metadata - - -# Bulk load all compliance frameworks specification -def bulk_load_compliance_frameworks(provider: str) -> dict: - """Bulk load all compliance frameworks specification into a dict""" - try: - bulk_compliance_frameworks = {} - available_compliance_framework_modules = list_compliance_modules() - for compliance_framework in available_compliance_framework_modules: - if provider in compliance_framework.name: - compliance_specification_dir_path = ( - f"{compliance_framework.module_finder.path}/{provider}" - ) - - # for compliance_framework in available_compliance_framework_modules: - for filename in os.listdir(compliance_specification_dir_path): - file_path = os.path.join( - compliance_specification_dir_path, filename - ) - # Check if it is a file and ti size is greater than 0 - if os.path.isfile(file_path) and os.stat(file_path).st_size > 0: - # Open Compliance file in JSON - # cis_v1.4_aws.json --> cis_v1.4_aws - compliance_framework_name = filename.split(".json")[0] - # Store the compliance info - bulk_compliance_frameworks[compliance_framework_name] = ( - load_compliance_framework(file_path) - ) - except Exception as e: - logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}") - - return bulk_compliance_frameworks - - # Exclude checks to run def exclude_checks_to_run(checks_to_execute: set, excluded_checks: list) -> set: for check in excluded_checks: @@ -381,65 +319,6 @@ def parse_checks_from_compliance_framework( return checks_to_execute -def recover_checks_from_provider( - provider: str, service: str = None, include_fixers: bool = False -) -> list[tuple]: - """ - Recover all checks from the selected provider and service - - Returns a list of tuples with the following format (check_name, check_path) - """ - try: - checks = [] - modules = list_modules(provider, service) - for module_name in modules: - # Format: "prowler.providers.{provider}.services.{service}.{check_name}.{check_name}" - check_module_name = module_name.name - # We need to exclude common shared libraries in services - if ( - check_module_name.count(".") == 6 - and "lib" not in check_module_name - and (not check_module_name.endswith("_fixer") or include_fixers) - ): - check_path = module_name.module_finder.path - # Check name is the last part of the check_module_name - check_name = check_module_name.split(".")[-1] - check_info = (check_name, check_path) - checks.append(check_info) - except ModuleNotFoundError: - logger.critical(f"Service {service} was not found for the {provider} provider.") - sys.exit(1) - except Exception as e: - logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}") - sys.exit(1) - else: - return checks - - -def list_compliance_modules(): - """ - list_compliance_modules returns the available compliance frameworks and returns their path - """ - # This module path requires the full path including "prowler." - module_path = "prowler.compliance" - return walk_packages( - importlib.import_module(module_path).__path__, - importlib.import_module(module_path).__name__ + ".", - ) - - -# List all available modules in the selected provider and service -def list_modules(provider: str, service: str): - # This module path requires the full path including "prowler." - module_path = f"prowler.providers.{provider}.services" - if service: - module_path += f".{service}" - return walk_packages( - importlib.import_module(module_path).__path__, - importlib.import_module(module_path).__name__ + ".", - ) - - # Import an input check using its path def import_check(check_path: str) -> ModuleType: lib = importlib.import_module(f"{check_path}") @@ -797,34 +676,3 @@ def update_audit_metadata( logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - - -def recover_checks_from_service(service_list: list, provider: str) -> set: - """ - Recover all checks from the selected provider and service - - Returns a set of checks from the given services - """ - try: - checks = set() - service_list = [ - "awslambda" if service == "lambda" else service for service in service_list - ] - for service in service_list: - service_checks = recover_checks_from_provider(provider, service) - if not service_checks: - logger.error(f"Service '{service}' does not have checks.") - - else: - for check in service_checks: - # Recover check name and module name from import path - # Format: "providers.{provider}.services.{service}.{check_name}.{check_name}" - check_name = check[0].split(".")[-1] - # If the service is present in the group list passed as parameters - # if service_name in group_list: checks_from_arn.add(check_name) - checks.add(check_name) - return checks - except Exception as error: - logger.error( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py index 2091e64561..b8496c2b2b 100644 --- a/prowler/lib/check/checks_loader.py +++ b/prowler/lib/check/checks_loader.py @@ -4,6 +4,8 @@ from prowler.config.config import valid_severities from prowler.lib.check.check import ( parse_checks_from_compliance_framework, parse_checks_from_file, +) +from prowler.lib.check.utils import ( recover_checks_from_provider, recover_checks_from_service, ) diff --git a/prowler/lib/check/compliance_models.py b/prowler/lib/check/compliance_models.py index ed7ba1341e..f53e49ab17 100644 --- a/prowler/lib/check/compliance_models.py +++ b/prowler/lib/check/compliance_models.py @@ -1,9 +1,11 @@ +import os import sys from enum import Enum from typing import Optional, Union from pydantic import BaseModel, ValidationError, root_validator +from prowler.lib.check.utils import list_compliance_modules from prowler.lib.logger import logger @@ -306,6 +308,36 @@ class Compliance(BaseModel): return requirement + @staticmethod + def get_bulk(provider: str) -> dict: + """Bulk load all compliance frameworks specification into a dict""" + try: + bulk_compliance_frameworks = {} + available_compliance_framework_modules = list_compliance_modules() + for compliance_framework in available_compliance_framework_modules: + if provider in compliance_framework.name: + compliance_specification_dir_path = ( + f"{compliance_framework.module_finder.path}/{provider}" + ) + # for compliance_framework in available_compliance_framework_modules: + for filename in os.listdir(compliance_specification_dir_path): + file_path = os.path.join( + compliance_specification_dir_path, filename + ) + # Check if it is a file and ti size is greater than 0 + if os.path.isfile(file_path) and os.stat(file_path).st_size > 0: + # Open Compliance file in JSON + # cis_v1.4_aws.json --> cis_v1.4_aws + compliance_framework_name = filename.split(".json")[0] + # Store the compliance info + bulk_compliance_frameworks[compliance_framework_name] = ( + load_compliance_framework(file_path) + ) + except Exception as e: + logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}") + + return bulk_compliance_frameworks + # Testing Pending def load_compliance_framework( diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index daf828e362..d95ae550ec 100644 --- a/prowler/lib/check/models.py +++ b/prowler/lib/check/models.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from pydantic import BaseModel, ValidationError, validator from prowler.config.config import valid_severities +from prowler.lib.check.utils import recover_checks_from_provider from prowler.lib.logger import logger @@ -129,6 +130,34 @@ class CheckMetadata(BaseModel): ) return severity + @staticmethod + def get_bulk(provider: str) -> dict[str, "CheckMetadata"]: + """ + Load the metadata of all checks for a given provider reading the check's metadata files. + Args: + provider (str): The name of the provider. + Returns: + dict[str, CheckMetadata]: A dictionary containing the metadata of all checks, with the CheckID as the key. + """ + + bulk_check_metadata = {} + checks = recover_checks_from_provider(provider) + # Build list of check's metadata files + for check_info in checks: + # Build check path name + check_name = check_info[0] + check_path = check_info[1] + # Ignore fixer files + if check_name.endswith("_fixer"): + continue + # Append metadata file extension + metadata_file = f"{check_path}/{check_name}.metadata.json" + # Load metadata + check_metadata = load_check_metadata(metadata_file) + bulk_check_metadata[check_metadata.CheckID] = check_metadata + + return bulk_check_metadata + class Check(ABC, CheckMetadata): """Prowler Check""" diff --git a/prowler/lib/check/utils.py b/prowler/lib/check/utils.py new file mode 100644 index 0000000000..c9e6d6de00 --- /dev/null +++ b/prowler/lib/check/utils.py @@ -0,0 +1,95 @@ +import importlib +import sys +from pkgutil import walk_packages + +from prowler.lib.logger import logger + + +def recover_checks_from_provider( + provider: str, service: str = None, include_fixers: bool = False +) -> list[tuple]: + """ + Recover all checks from the selected provider and service + + Returns a list of tuples with the following format (check_name, check_path) + """ + try: + checks = [] + modules = list_modules(provider, service) + for module_name in modules: + # Format: "prowler.providers.{provider}.services.{service}.{check_name}.{check_name}" + check_module_name = module_name.name + # We need to exclude common shared libraries in services + if ( + check_module_name.count(".") == 6 + and "lib" not in check_module_name + and (not check_module_name.endswith("_fixer") or include_fixers) + ): + check_path = module_name.module_finder.path + # Check name is the last part of the check_module_name + check_name = check_module_name.split(".")[-1] + check_info = (check_name, check_path) + checks.append(check_info) + except ModuleNotFoundError: + logger.critical(f"Service {service} was not found for the {provider} provider.") + sys.exit(1) + except Exception as e: + logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}") + sys.exit(1) + else: + return checks + + +# List all available modules in the selected provider and service +def list_modules(provider: str, service: str): + # This module path requires the full path including "prowler." + module_path = f"prowler.providers.{provider}.services" + if service: + module_path += f".{service}" + return walk_packages( + importlib.import_module(module_path).__path__, + importlib.import_module(module_path).__name__ + ".", + ) + + +def recover_checks_from_service(service_list: list, provider: str) -> set: + """ + Recover all checks from the selected provider and service + + Returns a set of checks from the given services + """ + try: + checks = set() + service_list = [ + "awslambda" if service == "lambda" else service for service in service_list + ] + for service in service_list: + service_checks = recover_checks_from_provider(provider, service) + if not service_checks: + logger.error(f"Service '{service}' does not have checks.") + + else: + for check in service_checks: + # Recover check name and module name from import path + # Format: "providers.{provider}.services.{service}.{check_name}.{check_name}" + check_name = check[0].split(".")[-1] + # If the service is present in the group list passed as parameters + # if service_name in group_list: checks_from_arn.add(check_name) + checks.add(check_name) + return checks + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + +def list_compliance_modules(): + """ + list_compliance_modules returns the available compliance frameworks and returns their path + """ + # This module path requires the full path including "prowler." + module_path = "prowler.compliance" + return walk_packages( + importlib.import_module(module_path).__path__, + importlib.import_module(module_path).__name__ + ".", + ) diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index cfb300cc81..b382ce9257 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -19,7 +19,7 @@ from prowler.config.config import ( load_and_validate_config_file, load_and_validate_fixer_config_file, ) -from prowler.lib.check.check import list_modules, recover_checks_from_service +from prowler.lib.check.utils import list_modules, recover_checks_from_service from prowler.lib.logger import logger from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes from prowler.providers.aws.config import ( diff --git a/tests/lib/check/check_test.py b/tests/lib/check/check_test.py index c14ee04905..16348cdca8 100644 --- a/tests/lib/check/check_test.py +++ b/tests/lib/check/check_test.py @@ -18,17 +18,19 @@ from prowler.lib.check.check import ( execute, list_categories, list_checks_json, - list_modules, list_services, parse_checks_from_file, parse_checks_from_folder, - recover_checks_from_provider, - recover_checks_from_service, remove_custom_checks_module, run_check, update_audit_metadata, ) from prowler.lib.check.models import load_check_metadata +from prowler.lib.check.utils import ( + list_modules, + recover_checks_from_provider, + recover_checks_from_service, +) from prowler.providers.aws.aws_provider import AwsProvider from prowler.providers.aws.services.accessanalyzer.accessanalyzer_service import ( Analyzer, @@ -575,7 +577,7 @@ class TestCheck: listed_categories = list_categories(test_bulk_checks_metadata) assert listed_categories == expected_categories - @patch("prowler.lib.check.check.list_modules", new=mock_list_modules) + @patch("prowler.lib.check.utils.list_modules", new=mock_list_modules) def test_recover_checks_from_provider(self): provider = "azure" service = "storage" @@ -636,7 +638,7 @@ class TestCheck: returned_checks = recover_checks_from_provider(provider, service) assert returned_checks == expected_checks - @patch("prowler.lib.check.check.walk_packages", new=mock_walk_packages) + @patch("prowler.lib.check.utils.walk_packages", new=mock_walk_packages) def test_list_modules(self): provider = "azure" service = "storage" @@ -644,7 +646,7 @@ class TestCheck: assert expected_modules == expected_packages @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider, ) def test_recover_checks_from_service(self): diff --git a/tests/lib/check/compliance_check_test.py b/tests/lib/check/compliance_check_test.py index 69a2ea76d7..476ad80ce4 100644 --- a/tests/lib/check/compliance_check_test.py +++ b/tests/lib/check/compliance_check_test.py @@ -1,3 +1,5 @@ +from unittest import mock + from prowler.lib.check.compliance import update_checks_metadata_with_compliance from prowler.lib.check.compliance_models import ( CIS_Requirement_Attribute, @@ -317,3 +319,39 @@ class TestCompliance: ) assert compliance_requirement is None + + @mock.patch("prowler.lib.check.compliance_models.load_compliance_framework") + @mock.patch("os.stat") + @mock.patch("os.path.isfile") + @mock.patch("os.listdir") + @mock.patch("prowler.lib.check.compliance_models.list_compliance_modules") + def test_get_bulk( + self, + mock_list_modules, + mock_listdir, + mock_isfile, + mock_stat, + mock_load_compliance, + ): + object = mock.Mock() + object.path = "/path/to/compliance" + object.name = "framework1_aws" + mock_list_modules.return_value = [object] + + mock_listdir.return_value = ["framework1_aws.json"] + + mock_isfile.return_value = True + + mock_stat.return_value.st_size = 100 + + mock_load_compliance.return_value = mock.Mock( + Framework="Framework1", Provider="aws" + ) + + from prowler.lib.check.compliance_models import Compliance + + result = Compliance.get_bulk(provider="aws") + + assert len(result) == 1 + assert "framework1_aws" in result.keys() + mock_list_modules.assert_called_once() diff --git a/tests/lib/check/models_test.py b/tests/lib/check/models_test.py new file mode 100644 index 0000000000..f373e0f80f --- /dev/null +++ b/tests/lib/check/models_test.py @@ -0,0 +1,56 @@ +from unittest import mock + +from prowler.lib.check.models import CheckMetadata + + +class TestCheckMetada: + + @mock.patch("prowler.lib.check.models.load_check_metadata") + @mock.patch("prowler.lib.check.models.recover_checks_from_provider") + def test_get_bulk(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") + ] + + 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 + + result = CheckMetadata.get_bulk(provider="aws") + + # Assertions + assert "accessanalyzer_enabled" in result.keys() + assert result["accessanalyzer_enabled"] == check_metadata + mock_recover_checks.assert_called_once_with("aws") + mock_load_metadata.assert_called_once_with( + "/path/to/accessanalyzer_enabled/accessanalyzer_enabled.metadata.json" + ) diff --git a/tests/providers/aws/aws_provider_test.py b/tests/providers/aws/aws_provider_test.py index 528a524ac0..d0bfcfc6a1 100644 --- a/tests/providers/aws/aws_provider_test.py +++ b/tests/providers/aws/aws_provider_test.py @@ -1423,7 +1423,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_elb_service, ) def test_get_checks_from_input_arn_elb(self): @@ -1444,7 +1444,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_efs_service, ) def test_get_checks_from_input_arn_efs(self): @@ -1465,7 +1465,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_lambda_service, ) def test_get_checks_from_input_arn_lambda(self): @@ -1485,7 +1485,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_iam_service, ) def test_get_checks_from_input_arn_iam(self): @@ -1507,7 +1507,7 @@ aws: @mock_aws @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_s3_service, ) def test_get_checks_from_input_arn_s3(self): @@ -1526,7 +1526,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_cloudwatch_service, ) def test_get_checks_from_input_arn_cloudwatch(self): @@ -1546,7 +1546,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_cognito_service, ) def test_get_checks_from_input_arn_cognito(self): @@ -1562,7 +1562,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_security_group(self): @@ -1578,7 +1578,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_acl(self): @@ -1594,7 +1594,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_rds_service, ) def test_get_checks_from_input_arn_rds_snapshots(self): @@ -1610,7 +1610,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_from_input_arn_ec2_ami(self): @@ -1738,7 +1738,7 @@ aws: @mock_aws @patch( - "prowler.lib.check.check.recover_checks_from_provider", + "prowler.lib.check.utils.recover_checks_from_provider", new=mock_recover_checks_from_aws_provider_ec2_service, ) def test_get_checks_to_execute_by_audit_resources(self):