refactor(check_metadata): move bulk_load_checks_metadata inside class (#4934)

This commit is contained in:
Pedro Martín
2024-09-06 09:50:14 +02:00
committed by GitHub
parent f169599a56
commit 4eed5c7a99
11 changed files with 279 additions and 177 deletions
+4 -4
View File
@@ -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
+2 -154
View File
@@ -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}"
)
+2
View File
@@ -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,
)
+32
View File
@@ -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(
+29
View File
@@ -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"""
+95
View File
@@ -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__ + ".",
)
+1 -1
View File
@@ -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 (
+8 -6
View File
@@ -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):
+38
View File
@@ -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()
+56
View File
@@ -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"
)
+12 -12
View File
@@ -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):