diff --git a/prowler/config/scan_config_schema.py b/prowler/config/scan_config_schema.py index d09b49f32a..431c4cec60 100644 --- a/prowler/config/scan_config_schema.py +++ b/prowler/config/scan_config_schema.py @@ -27,11 +27,14 @@ the UI can validate live with `ajv`. This module provides: """ import json +from functools import lru_cache from typing import Any from pydantic import ValidationError from prowler.config.schema.registry import SCHEMAS +from prowler.lib.check.check import list_services +from prowler.lib.check.models import CheckMetadata # Pydantic v2 prefixes messages emitted from a ``field_validator`` that # raises ``ValueError`` with this string. Strip it so the message that @@ -39,6 +42,18 @@ from prowler.config.schema.registry import SCHEMAS _PYDANTIC_VALUE_ERROR_PREFIX = "Value error, " +@lru_cache(maxsize=None) +def _get_provider_check_ids(provider: str) -> frozenset[str]: + """Return cached check identifiers for a provider.""" + return frozenset(CheckMetadata.get_bulk(provider)) + + +@lru_cache(maxsize=None) +def _get_provider_services(provider: str) -> frozenset[str]: + """Return cached service identifiers for a provider.""" + return frozenset(list_services(provider)) + + def _format_loc(loc: tuple) -> str: """Render a Pydantic error location as a dot-separated path. @@ -79,9 +94,9 @@ def validate_and_normalize_scan_config( sections and unknown keys inside registered sections are preserved untouched for forward compatibility with plugin-provided keys. - ``errors`` is a list of ``{"path": , "message": }`` - entries — one per Pydantic violation. When any error is present the - normalized dictionary is returned empty so the caller never - persists a partially validated configuration. + entries, one per schema or exclusion-catalog violation. When any error + is present the normalized dictionary is returned empty so the caller + never persists a partially validated configuration. The input payload is never mutated. """ @@ -156,6 +171,35 @@ def validate_and_normalize_scan_config( message = message[len(_PYDANTIC_VALUE_ERROR_PREFIX) :] errors.append({"path": path, "message": message}) continue + + if model.excluded_checks: + available_checks = _get_provider_check_ids(provider_key) + for index, check in enumerate(model.excluded_checks): + if check not in available_checks: + errors.append( + { + "path": f"{provider_key}.excluded_checks[{index}]", + "message": ( + f"Unknown check '{check}' for provider " + f"'{provider_key}'." + ), + } + ) + + if model.excluded_services: + available_services = _get_provider_services(provider_key) + for index, service in enumerate(model.excluded_services): + if service not in available_services: + errors.append( + { + "path": f"{provider_key}.excluded_services[{index}]", + "message": ( + f"Unknown service '{service}' for provider " + f"'{provider_key}'." + ), + } + ) + normalized[provider_key] = model.model_dump(mode="json", exclude_unset=True) if errors: diff --git a/tests/config/schema/scan_config_schema_test.py b/tests/config/schema/scan_config_schema_test.py index 0245276a9b..037fcd9501 100644 --- a/tests/config/schema/scan_config_schema_test.py +++ b/tests/config/schema/scan_config_schema_test.py @@ -8,15 +8,28 @@ the backend can persist verbatim in a Django ``JSONField``. """ import json +from unittest.mock import call, patch import pytest from prowler.config.scan_config_schema import ( + _get_provider_check_ids, + _get_provider_services, validate_and_normalize_scan_config, validate_scan_config, ) +@pytest.fixture(autouse=True) +def clear_provider_catalog_caches(): + """Keep provider catalog cache state isolated between tests.""" + _get_provider_check_ids.cache_clear() + _get_provider_services.cache_clear() + yield + _get_provider_check_ids.cache_clear() + _get_provider_services.cache_clear() + + class Test_Non_Dict_Root: @pytest.mark.parametrize("payload", [None, "string", 42, [], (1, 2)]) def test_non_mapping_root_is_rejected(self, payload): @@ -39,7 +52,7 @@ class Test_Success_Path: normalized, errors = validate_and_normalize_scan_config( { "aws": { - "excluded_checks": [" s3_bucket_public_access "], + "excluded_checks": [" s3_bucket_default_encryption "], "excluded_services": [" s3 "], } } @@ -47,7 +60,7 @@ class Test_Success_Path: assert errors == [] assert normalized == { "aws": { - "excluded_checks": ["s3_bucket_public_access"], + "excluded_checks": ["s3_bucket_default_encryption"], "excluded_services": ["s3"], } } @@ -62,6 +75,67 @@ class Test_Success_Path: assert errors == [] assert normalized == {"aws": {"plugin_option": "preserved", "another": 42}} + def test_plugin_catalog_identifiers_are_accepted_and_catalogs_are_cached(self): + payload = { + "aws": { + "excluded_checks": ["plugin_check"], + "excluded_services": ["plugin_service"], + } + } + with ( + patch( + "prowler.config.scan_config_schema.CheckMetadata.get_bulk", + return_value={"plugin_check": object()}, + ) as check_catalog, + patch( + "prowler.config.scan_config_schema.list_services", + return_value=["plugin_service"], + ) as service_catalog, + ): + first_result = validate_and_normalize_scan_config(payload) + second_result = validate_and_normalize_scan_config(payload) + + normalized, errors = first_result + assert errors == [] + assert normalized == { + "aws": { + "excluded_checks": ["plugin_check"], + "excluded_services": ["plugin_service"], + } + } + assert second_result == first_result + check_catalog.assert_called_once_with("aws") + service_catalog.assert_called_once_with("aws") + + def test_catalog_caches_are_keyed_by_provider(self): + with ( + patch( + "prowler.config.scan_config_schema.CheckMetadata.get_bulk", + side_effect=lambda provider: {f"{provider}_plugin_check": object()}, + ) as check_catalog, + patch( + "prowler.config.scan_config_schema.list_services", + side_effect=lambda provider: [f"{provider}_plugin_service"], + ) as service_catalog, + ): + payload = { + "aws": { + "excluded_checks": ["aws_plugin_check"], + "excluded_services": ["aws_plugin_service"], + }, + "azure": { + "excluded_checks": ["azure_plugin_check"], + "excluded_services": ["azure_plugin_service"], + }, + } + first_result = validate_and_normalize_scan_config(payload) + second_result = validate_and_normalize_scan_config(payload) + + assert first_result[1] == [] + assert second_result == first_result + assert check_catalog.call_args_list == [call("aws"), call("azure")] + assert service_catalog.call_args_list == [call("aws"), call("azure")] + def test_omitted_defaults_are_not_injected(self): normalized, errors = validate_and_normalize_scan_config( {"aws": {"max_ec2_instance_age_in_days": 90}} @@ -103,6 +177,89 @@ class Test_Success_Path: class Test_Error_Path: + def test_unknown_excluded_check_is_rejected(self): + normalized, errors = validate_and_normalize_scan_config( + {"aws": {"excluded_checks": ["aws_check_that_does_not_exist"]}} + ) + assert normalized == {} + assert errors == [ + { + "path": "aws.excluded_checks[0]", + "message": ( + "Unknown check 'aws_check_that_does_not_exist' for provider " + "'aws'." + ), + } + ] + + def test_unknown_excluded_service_is_rejected(self): + normalized, errors = validate_and_normalize_scan_config( + {"aws": {"excluded_services": ["not_a_real_aws_service"]}} + ) + assert normalized == {} + assert errors == [ + { + "path": "aws.excluded_services[0]", + "message": ( + "Unknown service 'not_a_real_aws_service' for provider 'aws'." + ), + } + ] + + def test_multiple_unknown_exclusions_return_deterministic_errors(self): + normalized, errors = validate_and_normalize_scan_config( + { + "aws": { + "excluded_checks": [ + "unknown_check_one", + "s3_bucket_default_encryption", + "unknown_check_two", + ], + "excluded_services": [ + "unknown_service_one", + "s3", + "unknown_service_two", + ], + } + } + ) + assert normalized == {} + assert errors == [ + { + "path": "aws.excluded_checks[0]", + "message": "Unknown check 'unknown_check_one' for provider 'aws'.", + }, + { + "path": "aws.excluded_checks[2]", + "message": "Unknown check 'unknown_check_two' for provider 'aws'.", + }, + { + "path": "aws.excluded_services[0]", + "message": ( + "Unknown service 'unknown_service_one' for provider 'aws'." + ), + }, + { + "path": "aws.excluded_services[2]", + "message": ( + "Unknown service 'unknown_service_two' for provider 'aws'." + ), + }, + ] + + def test_check_from_another_provider_is_rejected(self): + azure_check = "postgresql_flexible_server_allow_access_services_disabled" + normalized, errors = validate_and_normalize_scan_config( + {"aws": {"excluded_checks": [azure_check]}} + ) + assert normalized == {} + assert errors == [ + { + "path": "aws.excluded_checks[0]", + "message": f"Unknown check '{azure_check}' for provider 'aws'.", + } + ] + def test_invalid_input_returns_empty_normalized_and_errors(self): normalized, errors = validate_and_normalize_scan_config( {"aws": {"excluded_services": ["s3", " s3 "]}} @@ -215,6 +372,18 @@ class Test_Backward_Compatible_Wrapper: assert errors assert all(set(err) == {"path", "message"} for err in errors) + def test_unknown_exclusion_yields_the_semantic_error(self): + assert validate_scan_config( + {"aws": {"excluded_services": ["not_a_real_aws_service"]}} + ) == [ + { + "path": "aws.excluded_services[0]", + "message": ( + "Unknown service 'not_a_real_aws_service' for provider 'aws'." + ), + } + ] + def test_non_mapping_root_matches_new_contract(self): assert validate_scan_config(None) == [ {