ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(scan-class): add new scan class (#4564)

Browse Source 
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
Pedro Martín
2024-08-05 08:21:13 +02:00
committed by GitHub
parent 112f48ac08
commit ffd9b2a2f6
5 changed files with 450 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/check.py
+14 -13
View File
@@ -593,12 +593,17 @@ def execute_checks(
service,
check_name,
global_provider,
services_executed,
checks_executed,
custom_checks_metadata,
)
all_findings.extend(check_findings)
# Update Audit Status
services_executed.add(service)
checks_executed.add(check_name)
global_provider.audit_metadata = update_audit_metadata(
global_provider.audit_metadata, services_executed, checks_executed
)
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
logger.error(
@@ -651,11 +656,16 @@ def execute_checks(
service,
check_name,
global_provider,
services_executed,
checks_executed,
custom_checks_metadata,
)
all_findings.extend(check_findings)
services_executed.add(service)
checks_executed.add(check_name)
global_provider.audit_metadata = update_audit_metadata(
global_provider.audit_metadata,
services_executed,
checks_executed,
)
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
@@ -677,8 +687,6 @@ def execute(
service: str,
check_name: str,
global_provider: Any,
services_executed: set,
checks_executed: set,
custom_checks_metadata: Any,
):
try:
@@ -714,13 +722,6 @@ def execute(
if finding.status in global_provider.output_options.status
]
# Update Audit Status
services_executed.add(service)
checks_executed.add(check_name)
global_provider.audit_metadata = update_audit_metadata(
global_provider.audit_metadata, services_executed, checks_executed
)
# Mutelist findings
if hasattr(global_provider, "mutelist") and global_provider.mutelist.mutelist:
# TODO: make this prettier
prowler/lib/scan/scan.py
+191 -49
View File
@@ -1,60 +1,202 @@
from typing import Any
from typing import Generator
from prowler.lib.check.check import execute
from prowler.lib.check.models import Check_Report
from prowler.lib.check.check import execute, update_audit_metadata
from prowler.lib.logger import logger
from prowler.lib.outputs.finding import Finding
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.provider import Provider
def scan(
checks_to_execute: list,
global_provider: Any,
custom_checks_metadata: Any,
) -> list[Check_Report]:
try:
# List to store all the check's findings
all_findings = []
# Services and checks executed for the Audit Status
services_executed = set()
checks_executed = set()
class Scan:
_provider: Provider
# Refactor(Core): This should replace the Audit_Metadata
_number_of_checks_to_execute: int = 0
_number_of_checks_completed: int = 0
# TODO the str should be a set of Check objects
_checks_to_execute: list[str]
_service_checks_to_execute: dict[str, set[str]]
_service_checks_completed: dict[str, set[str]]
_progress: float = 0.0
_findings: list = []
# Initialize the Audit Metadata
# TODO: this should be done in the provider class
# Refactor(Core): Audit manager?
global_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
expected_checks=checks_to_execute,
completed_checks=0,
audit_progress=0,
def __init__(self, provider: Provider, checks_to_execute: list[str]):
"""
Scan is the class that executes the checks and yields the progress and the findings.
Params:
provider: Provider -> The provider to scan
checks_to_execute: list[str] -> The checks to execute
"""
self._provider = provider
# Remove duplicated checks and sort them
self._checks_to_execute = sorted(list(set(checks_to_execute)))
self._number_of_checks_to_execute = len(checks_to_execute)
service_checks_to_execute = get_service_checks_to_execute(checks_to_execute)
service_checks_completed = dict()
self._service_checks_to_execute = service_checks_to_execute
self._service_checks_completed = service_checks_completed
@property
def checks_to_execute(self) -> set[str]:
return self._checks_to_execute
@property
def service_checks_to_execute(self) -> dict[str, set[str]]:
return self._service_checks_to_execute
@property
def service_checks_completed(self) -> dict[str, set[str]]:
return self._service_checks_completed
@property
def provider(self) -> Provider:
return self._provider
@property
def progress(self) -> float:
return (
self._number_of_checks_completed / self._number_of_checks_to_execute * 100
)
for check_name in checks_to_execute:
try:
# Recover service from check name
service = check_name.split("_")[0]
@property
def findings(self) -> list:
return self._findings
check_findings = execute(
service,
check_name,
global_provider,
services_executed,
checks_executed,
custom_checks_metadata,
)
all_findings.extend(check_findings)
def scan(
self,
custom_checks_metadata: dict = {},
) -> Generator[tuple[float, list[Finding]], None, None]:
"""
Executes the scan by iterating over the checks to execute and executing each check.
Yields the progress and findings for each check.
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
logger.error(
f"Check '{check_name}' was not found for the {global_provider.type.upper()} provider"
)
except Exception as error:
logger.error(
f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
Args:
custom_checks_metadata (dict): Custom metadata for the checks (default: {}).
return all_findings
Yields:
Tuple[float, list[Finding]]: A tuple containing the progress and findings for each check.
Raises:
ModuleNotFoundError: If the check does not exist in the provider or is from another provider.
Exception: If any other error occurs during the execution of a check.
"""
try:
checks_to_execute = self.checks_to_execute
# Initialize the Audit Metadata
# TODO: this should be done in the provider class
# Refactor(Core): Audit manager?
self._provider.audit_metadata = Audit_Metadata(
services_scanned=0,
expected_checks=checks_to_execute,
completed_checks=0,
audit_progress=0,
)
for check_name in checks_to_execute:
try:
# Recover service from check name
service = get_service_name_from_check_name(check_name)
# Execute the check
check_findings = execute(
service,
check_name,
self._provider,
custom_checks_metadata,
)
# Store findings
self._findings.extend(check_findings)
# Remove the executed check
self._service_checks_to_execute[service].remove(check_name)
if len(self._service_checks_to_execute[service]) == 0:
self._service_checks_to_execute.pop(service, None)
# Add the completed check
if service not in self._service_checks_completed:
self._service_checks_completed[service] = set()
self._service_checks_completed[service].add(check_name)
self._number_of_checks_completed += 1
# This should be done just once all the service's checks are completed
# This metadata needs to get to the services not within the provider
# since it is present in the Scan class
self._provider.audit_metadata = update_audit_metadata(
self._provider.audit_metadata,
self.get_completed_services(),
self.get_completed_checks(),
)
findings = [
Finding.generate_output(self._provider, finding)
for finding in check_findings
]
yield self.progress, findings
# If check does not exists in the provider or is from another provider
except ModuleNotFoundError:
logger.error(
f"Check '{check_name}' was not found for the {self._provider.type.upper()} provider"
)
except Exception as error:
logger.error(
f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def get_completed_services(self) -> set[str]:
"""
get_completed_services returns the services that have been completed.
Example:
get_completed_services() -> {"ec2", "s3"}
"""
return self._service_checks_completed.keys()
def get_completed_checks(self) -> set[str]:
"""
get_completed_checks returns the checks that have been completed.
Example:
get_completed_checks() -> {"ec2_instance_public_ip", "s3_bucket_public"}
"""
completed_checks = set()
for checks in self._service_checks_completed.values():
completed_checks.update(checks)
return completed_checks
def get_service_name_from_check_name(check_name: str) -> str:
"""
get_service_name_from_check_name returns the service name for a given check name.
Example:
get_service_name_from_check_name("ec2_instance_public") -> "ec2"
"""
return check_name.split("_")[0]
def get_service_checks_to_execute(checks_to_execute: set[str]) -> dict[str, set[str]]:
"""
get_service_checks_to_execute returns a dictionary with the services and the checks to execute.
Example:
get_service_checks_to_execute({"accessanalyzer_enabled", "ec2_instance_public_ip"})
-> {"accessanalyzer": {"accessanalyzer_enabled"}, "ec2": {"ec2_instance_public_ip"}}
"""
service_checks_to_execute = dict()
for check in checks_to_execute:
# check -> accessanalyzer_enabled
# service -> accessanalyzer
service = get_service_name_from_check_name(check)
if service not in service_checks_to_execute:
service_checks_to_execute[service] = set()
service_checks_to_execute[service].add(check)
return service_checks_to_execute
prowler/providers/aws/lib/service/service.py
+1
View File
@@ -40,6 +40,7 @@ class AWSService:
self.audited_account_arn = provider.identity.account_arn
self.audited_partition = provider.identity.partition
self.audit_resources = provider.audit_resources
# TODO: remove this
self.audited_checks = provider.audit_metadata.expected_checks
self.audit_config = provider.audit_config
self.fixer_config = provider.fixer_config
tests/lib/check/check_test.py
-4
View File
@@ -825,8 +825,6 @@ class TestCheck:
global_provider=set_mocked_aws_provider(
expected_checks=["accessanalyzer_enabled"]
),
services_executed={"accessanalyzer"},
checks_executed={"accessanalyzer_enabled"},
custom_checks_metadata=None,
)
assert len(findings) == 1
@@ -855,8 +853,6 @@ class TestCheck:
global_provider=set_mocked_aws_provider(
status=status, expected_checks=["accessanalyzer_enabled"]
),
services_executed={"accessanalyzer"},
checks_executed={"accessanalyzer_enabled"},
custom_checks_metadata=None,
)
assert len(findings) == 0
tests/lib/scan/scan_test.py
+244
View File
@@ -0,0 +1,244 @@
from unittest import mock
import pytest
from prowler.lib.scan.scan import Scan, get_service_checks_to_execute
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
from tests.providers.aws.utils import set_mocked_aws_provider
finding = generate_finding_output(
status="PASS",
status_extended="status-extended",
resource_uid="resource-123",
resource_name="Example Resource",
resource_details="Detailed information about the resource",
resource_tags="tag1,tag2",
partition="aws",
description="Description of the finding",
risk="High",
related_url="http://example.com",
remediation_recommendation_text="Recommendation text",
remediation_recommendation_url="http://example.com/remediation",
remediation_code_nativeiac="native-iac-code",
remediation_code_terraform="terraform-code",
remediation_code_other="other-code",
remediation_code_cli="cli-code",
compliance={"compliance_key": "compliance_value"},
categories="category1,category2",
depends_on="dependency",
related_to="related finding",
notes="Notes about the finding",
)
@pytest.fixture
def mock_provider():
return set_mocked_aws_provider()
@pytest.fixture
def mock_execute():
with mock.patch("prowler.lib.scan.scan.execute", autospec=True) as mock_exec:
findings = [finding]
mock_exec.side_effect = lambda *args, **kwargs: findings
yield mock_exec
@pytest.fixture
def mock_logger():
with mock.patch("prowler.lib.logger.logger", autospec=True) as mock_log:
yield mock_log
@pytest.fixture
def mock_global_provider(mock_provider):
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=mock_provider,
):
yield mock_provider
@pytest.fixture
def mock_generate_output():
with mock.patch(
"prowler.lib.outputs.finding.Finding.generate_output", autospec=True
) as mock_gen_output:
mock_gen_output.side_effect = lambda provider, finding: finding
yield mock_gen_output
class TestScan:
def test_init(mock_provider):
checks_to_execute = {
"workspaces_vpc_2private_1public_subnets_nat",
"workspaces_vpc_2private_1public_subnets_nat",
"accessanalyzer_enabled",
"accessanalyzer_enabled_without_findings",
"account_maintain_current_contact_details",
"account_maintain_different_contact_details_to_security_billing_and_operations",
"account_security_contact_information_is_registered",
"account_security_questions_are_registered_in_the_aws_account",
"acm_certificates_expiration_check",
"acm_certificates_transparency_logs_enabled",
"apigateway_restapi_authorizers_enabled",
"apigateway_restapi_client_certificate_enabled",
"apigateway_restapi_logging_enabled",
"apigateway_restapi_public",
"awslambda_function_not_publicly_accessible",
"awslambda_function_url_cors_policy",
"awslambda_function_url_public",
"awslambda_function_using_supported_runtimes",
"backup_plans_exist",
"backup_reportplans_exist",
"backup_vaults_encrypted",
"backup_vaults_exist",
"cloudformation_stack_outputs_find_secrets",
"cloudformation_stacks_termination_protection_enabled",
"cloudwatch_cross_account_sharing_disabled",
"cloudwatch_log_group_kms_encryption_enabled",
"cloudwatch_log_group_no_secrets_in_logs",
"cloudwatch_log_group_retention_policy_specific_days_enabled",
"cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled",
"cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_changes_enabled",
"cloudwatch_log_metric_filter_authentication_failures",
"cloudwatch_log_metric_filter_aws_organizations_changes",
"cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk",
"cloudwatch_log_metric_filter_for_s3_bucket_policy_changes",
"cloudwatch_log_metric_filter_policy_changes",
"cloudwatch_log_metric_filter_root_usage",
"cloudwatch_log_metric_filter_security_group_changes",
"cloudwatch_log_metric_filter_sign_in_without_mfa",
"cloudwatch_log_metric_filter_unauthorized_api_calls",
"codeartifact_packages_external_public_publishing_disabled",
"codebuild_project_older_90_days",
"codebuild_project_user_controlled_buildspec",
"cognito_identity_pool_guest_access_disabled",
"cognito_user_pool_advanced_security_enabled",
"cognito_user_pool_blocks_compromised_credentials_sign_in_attempts",
"cognito_user_pool_blocks_potential_malicious_sign_in_attempts",
"cognito_user_pool_client_prevent_user_existence_errors",
"cognito_user_pool_client_token_revocation_enabled",
"cognito_user_pool_deletion_protection_enabled",
"cognito_user_pool_mfa_enabled",
"cognito_user_pool_password_policy_lowercase",
"cognito_user_pool_password_policy_minimum_length_14",
"cognito_user_pool_password_policy_number",
"cognito_user_pool_password_policy_symbol",
"cognito_user_pool_password_policy_uppercase",
"cognito_user_pool_self_registration_disabled",
"cognito_user_pool_temporary_password_expiration",
"cognito_user_pool_waf_acl_attached",
"config_recorder_all_regions_enabled",
}
scan = Scan(mock_provider, checks_to_execute)
assert scan.provider == mock_provider
# Check that the checks to execute are sorted and without duplicates
assert scan.checks_to_execute == [
"accessanalyzer_enabled",
"accessanalyzer_enabled_without_findings",
"account_maintain_current_contact_details",
"account_maintain_different_contact_details_to_security_billing_and_operations",
"account_security_contact_information_is_registered",
"account_security_questions_are_registered_in_the_aws_account",
"acm_certificates_expiration_check",
"acm_certificates_transparency_logs_enabled",
"apigateway_restapi_authorizers_enabled",
"apigateway_restapi_client_certificate_enabled",
"apigateway_restapi_logging_enabled",
"apigateway_restapi_public",
"awslambda_function_not_publicly_accessible",
"awslambda_function_url_cors_policy",
"awslambda_function_url_public",
"awslambda_function_using_supported_runtimes",
"backup_plans_exist",
"backup_reportplans_exist",
"backup_vaults_encrypted",
"backup_vaults_exist",
"cloudformation_stack_outputs_find_secrets",
"cloudformation_stacks_termination_protection_enabled",
"cloudwatch_cross_account_sharing_disabled",
"cloudwatch_log_group_kms_encryption_enabled",
"cloudwatch_log_group_no_secrets_in_logs",
"cloudwatch_log_group_retention_policy_specific_days_enabled",
"cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled",
"cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_changes_enabled",
"cloudwatch_log_metric_filter_authentication_failures",
"cloudwatch_log_metric_filter_aws_organizations_changes",
"cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk",
"cloudwatch_log_metric_filter_for_s3_bucket_policy_changes",
"cloudwatch_log_metric_filter_policy_changes",
"cloudwatch_log_metric_filter_root_usage",
"cloudwatch_log_metric_filter_security_group_changes",
"cloudwatch_log_metric_filter_sign_in_without_mfa",
"cloudwatch_log_metric_filter_unauthorized_api_calls",
"codeartifact_packages_external_public_publishing_disabled",
"codebuild_project_older_90_days",
"codebuild_project_user_controlled_buildspec",
"cognito_identity_pool_guest_access_disabled",
"cognito_user_pool_advanced_security_enabled",
"cognito_user_pool_blocks_compromised_credentials_sign_in_attempts",
"cognito_user_pool_blocks_potential_malicious_sign_in_attempts",
"cognito_user_pool_client_prevent_user_existence_errors",
"cognito_user_pool_client_token_revocation_enabled",
"cognito_user_pool_deletion_protection_enabled",
"cognito_user_pool_mfa_enabled",
"cognito_user_pool_password_policy_lowercase",
"cognito_user_pool_password_policy_minimum_length_14",
"cognito_user_pool_password_policy_number",
"cognito_user_pool_password_policy_symbol",
"cognito_user_pool_password_policy_uppercase",
"cognito_user_pool_self_registration_disabled",
"cognito_user_pool_temporary_password_expiration",
"cognito_user_pool_waf_acl_attached",
"config_recorder_all_regions_enabled",
"workspaces_vpc_2private_1public_subnets_nat",
]
assert scan.service_checks_to_execute == get_service_checks_to_execute(
checks_to_execute
)
assert scan.service_checks_completed == {}
assert scan.progress == 0
assert scan.get_completed_services() == set()
assert scan.get_completed_checks() == set()
def test_scan(
mock_global_provider, mock_execute, mock_logger, mock_generate_output
):
checks_to_execute = {"accessanalyzer_enabled", "ec2_instance_public"}
custom_checks_metadata = {}
# Create a Scan object
scan = Scan(mock_global_provider, checks_to_execute)
# Execute the scan
results = list(scan.scan(custom_checks_metadata))
# Verify that generate_output was called with the correct findings
assert mock_generate_output.call_count == 2 * len(mock_execute.side_effect())
# Verify that execute was called twice
assert mock_execute.call_count == 2
assert len(results) == 2
assert results[0][1] == mock_execute.side_effect()
assert results[1][1] == mock_execute.side_effect()
# Check the audit progress for the last result
assert results[1][0] == 100.0
# Verify that the progress is 100.0
assert scan.progress == 100.0 # 100% progress is 100
assert scan._number_of_checks_completed == 2
assert scan.service_checks_to_execute == {}
assert scan.service_checks_completed == {
"ec2": {"ec2_instance_public"},
"accessanalyzer": {"accessanalyzer_enabled"},
}
# Verify that the findings are correct
assert scan.findings == mock_execute.side_effect() + mock_execute.side_effect()
# Verify that no error was logged
mock_logger.error.assert_not_called()