fix(sdk): validate scan configuration exclusions (#12033)

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": <dotted-path>, "message": <str>}``
-      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:

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) == [
             {