ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 21:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(metadata): add validator for ResourceType (#8036)

Browse Source 
Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
This commit is contained in:
Prowler Bot
2025-06-17 10:41:14 +02:00
committed by GitHub
parent 0df95e9088
commit 52b68451d1
6 changed files with 79 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/CHANGELOG.md
+1
View File
@@ -41,6 +41,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Github provider to `usage` section of `prowler -h`: [(#7906)](https://github.com/prowler-cloud/prowler/pull/7906)
- `network_flow_log_more_than_90_days` check to pass when retention policy is 0 days [(#7975)](https://github.com/prowler-cloud/prowler/pull/7975)
- Update SDK Azure call for ftps_state in the App Service [(#7923)](https://github.com/prowler-cloud/prowler/pull/7923)
- Validate ResourceType in CheckMetadata [(#8035)](https://github.com/prowler-cloud/prowler/pull/8035)
- Missing ResourceType values in check's metadata [(#8028)](https://github.com/prowler-cloud/prowler/pull/8028)
---
prowler/lib/check/models.py
+8 -2
View File
@@ -96,6 +96,7 @@ class CheckMetadata(BaseModel):
severity_to_lower(severity): Validator function to convert the severity to lowercase.
valid_severity(severity): Validator function to validate the severity of the check.
valid_cli_command(remediation): Validator function to validate the CLI command is not an URL.
valid_resource_type(resource_type): Validator function to validate the resource type is not empty.
"""
Provider: str
@@ -141,6 +142,12 @@ class CheckMetadata(BaseModel):
raise ValueError("CLI command cannot be an URL")
return remediation
@validator("ResourceType", pre=True, always=True)
def valid_resource_type(resource_type):
if not resource_type or not isinstance(resource_type, str):
raise ValueError("ResourceType must be a non-empty string")
return resource_type
@staticmethod
def get_bulk(provider: str) -> dict[str, "CheckMetadata"]:
"""
@@ -642,7 +649,6 @@ def load_check_metadata(metadata_file: str) -> CheckMetadata:
check_metadata = CheckMetadata.parse_file(metadata_file)
except ValidationError as error:
logger.critical(f"Metadata from {metadata_file} is not valid: {error}")
# TODO: remove this exit and raise an exception
sys.exit(1)
raise error
else:
return check_metadata
tests/lib/check/check_loader_test.py
+4
View File
@@ -180,6 +180,9 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_compliance_frameworks(
self,
):
bulk_checks_metatada = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
bulk_compliance_frameworks = {
"soc2_aws": Compliance(
Framework="SOC2",
@@ -199,6 +202,7 @@ class TestCheckLoader:
compliance_frameworks = ["soc2_aws"]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_compliance_frameworks=bulk_compliance_frameworks,
compliance_frameworks=compliance_frameworks,
provider=self.provider,
tests/lib/check/models_test.py
+3 -2
View File
@@ -304,13 +304,14 @@ class TestCheckMetada:
# Assertions
assert result == {"accessanalyzer_enabled"}
def test_list_by_compliance_empty(self):
@mock.patch("prowler.lib.check.models.CheckMetadata.get_bulk")
def test_list_by_compliance_empty(self, mock_get_bulk):
mock_get_bulk.return_value = {}
bulk_compliance_frameworks = custom_compliance_metadata
result = CheckMetadata.list(
bulk_compliance_frameworks=bulk_compliance_frameworks,
compliance_framework="framework1_azure",
)
# Assertions
assert result == set()
tests/lib/outputs/finding_test.py
+6 -6
View File
@@ -28,7 +28,7 @@ def mock_check_metadata(provider):
SubServiceName="",
ResourceIdTemplate="",
Severity="high",
ResourceType="",
ResourceType="mock_resource_type",
Description="",
Risk="",
RelatedUrl="",
@@ -209,7 +209,7 @@ class TestFinding:
assert finding_output.metadata.SubServiceName == ""
assert finding_output.metadata.ResourceIdTemplate == ""
assert finding_output.metadata.Severity == Severity.high
assert finding_output.metadata.ResourceType == ""
assert finding_output.metadata.ResourceType == "mock_resource_type"
assert finding_output.metadata.Description == ""
assert finding_output.metadata.Risk == ""
assert finding_output.metadata.RelatedUrl == ""
@@ -230,7 +230,7 @@ class TestFinding:
assert finding_output.check_id == "mock_check_id"
assert finding_output.severity == Severity.high.value
assert finding_output.status == Status.PASS.value
assert finding_output.resource_type == ""
assert finding_output.resource_type == "mock_resource_type"
assert finding_output.service_name == "mock_service_name"
assert finding_output.raw == {}
@@ -310,7 +310,7 @@ class TestFinding:
assert finding_output.metadata.SubServiceName == ""
assert finding_output.metadata.ResourceIdTemplate == ""
assert finding_output.metadata.Severity == Severity.high
assert finding_output.metadata.ResourceType == ""
assert finding_output.metadata.ResourceType == "mock_resource_type"
assert finding_output.metadata.Description == ""
assert finding_output.metadata.Risk == ""
assert finding_output.metadata.RelatedUrl == ""
@@ -405,7 +405,7 @@ class TestFinding:
assert finding_output.metadata.SubServiceName == ""
assert finding_output.metadata.ResourceIdTemplate == ""
assert finding_output.metadata.Severity == Severity.high
assert finding_output.metadata.ResourceType == ""
assert finding_output.metadata.ResourceType == "mock_resource_type"
assert finding_output.metadata.Description == ""
assert finding_output.metadata.Risk == ""
assert finding_output.metadata.RelatedUrl == ""
@@ -490,7 +490,7 @@ class TestFinding:
assert finding_output.metadata.SubServiceName == ""
assert finding_output.metadata.ResourceIdTemplate == ""
assert finding_output.metadata.Severity == Severity.high
assert finding_output.metadata.ResourceType == ""
assert finding_output.metadata.ResourceType == "mock_resource_type"
assert finding_output.metadata.Description == ""
assert finding_output.metadata.Risk == ""
assert finding_output.metadata.RelatedUrl == ""
tests/lib/scan/scan_test.py
+57 -88
View File
@@ -117,6 +117,7 @@ def mock_load_check_metadata():
) as mock_load:
mock_metadata = MagicMock()
mock_metadata.CheckID = "accessanalyzer_enabled"
mock_metadata.ResourceType = "AWS::IAM::AccessAnalyzer"
mock_load.return_value = mock_metadata
yield mock_load
@@ -130,8 +131,22 @@ def mock_load_checks_to_execute():
yield mock_load
@pytest.fixture
def mock_check_metadata_get_bulk():
with mock.patch(
"prowler.lib.check.models.CheckMetadata.get_bulk", autospec=True
) as mock_get_bulk:
mock_metadata = MagicMock()
mock_metadata.CheckID = "accessanalyzer_enabled"
mock_metadata.ResourceType = "AWS::IAM::AccessAnalyzer"
mock_get_bulk.return_value = {"accessanalyzer_enabled": mock_metadata}
yield mock_get_bulk
class TestScan:
def test_init(mock_provider):
def test_init(
mock_provider,
):
checks_to_execute = {
"workspaces_vpc_2private_1public_subnets_nat",
"workspaces_vpc_2private_1public_subnets_nat",
@@ -194,102 +209,56 @@ class TestScan:
"config_recorder_all_regions_enabled",
}
mock_provider.type = "aws"
scan = Scan(mock_provider, checks=checks_to_execute)
# Patch get_bulk to return all these checks
with mock.patch(
"prowler.lib.check.models.CheckMetadata.get_bulk"
) as mock_get_bulk:
mock_metadata = MagicMock()
mock_metadata.ResourceType = "AWS::IAM::AccessAnalyzer"
mock_metadata.Categories = []
mock_get_bulk.return_value = {
check: mock_metadata for check in checks_to_execute
}
scan = Scan(mock_provider, checks=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.duration == 0
assert scan.get_completed_services() == set()
assert scan.get_completed_checks() == set()
assert scan.provider == mock_provider
# Check that the checks to execute are sorted and without duplicates
assert scan.checks_to_execute == sorted(list(checks_to_execute))
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.duration == 0
assert scan.get_completed_services() == set()
assert scan.get_completed_checks() == set()
def test_init_with_no_checks(
mock_provider,
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_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
assert scan.checks_to_execute == ["accessanalyzer_enabled"]
assert scan.service_checks_to_execute == get_service_checks_to_execute(
["accessanalyzer_enabled"]
)
assert scan.service_checks_completed == {}
assert scan.progress == 0
assert scan.get_completed_services() == set()
assert scan.get_completed_checks() == set()
# Patch get_bulk to return only accessanalyzer_enabled
with mock.patch(
"prowler.lib.check.models.CheckMetadata.get_bulk"
) as mock_get_bulk:
mock_metadata = MagicMock()
mock_metadata.ResourceType = "AWS::IAM::AccessAnalyzer"
mock_metadata.Categories = []
mock_get_bulk.return_value = {"accessanalyzer_enabled": mock_metadata}
scan = Scan(mock_provider, checks=checks_to_execute)
# Remove assertion for mock_load_check_metadata
assert scan.provider == mock_provider
assert scan.checks_to_execute == ["accessanalyzer_enabled"]
assert scan.service_checks_to_execute == get_service_checks_to_execute(
["accessanalyzer_enabled"]
)
assert scan.service_checks_completed == {}
assert scan.progress == 0
assert scan.get_completed_services() == set()
assert scan.get_completed_checks() == set()
@patch("importlib.import_module")
def test_scan(