feat(sdk): apply scan configuration exclusions (#12028)

prowler/changelog.d/sdk-scan-config-exclusions.added.md

@@ -0,0 +1 @@
+`excluded_checks` and `excluded_services` in scan configurations to narrow the execution scope

prowler/config/scan_config_schema.py

@@ -9,22 +9,35 @@ The Prowler App, however, needs to surface those errors to the user when
 they save a Scan Config from the UI, and to expose the schema as JSON so
 the UI can validate live with `ajv`. This module provides:
 
-- `validate_scan_config(payload)` — STRICT: returns a list of
-  `{path, message}` errors without silently dropping anything. The DRF
-  serializer (`api/.../v1/serializers.py:validate_scan_config_payload`)
-  turns each entry into a `ValidationError`.
+- `validate_and_normalize_scan_config(payload)` — STRICT: returns
+  ``(normalized, errors)``. When ``errors`` is non-empty the normalized
+  dictionary is empty so callers never persist a partially validated
+  configuration. On success the normalized payload is JSON-serializable
+  (`model_dump(mode="json", exclude_unset=True)`), so the API can store
+  it directly in a Django ``JSONField`` and consume it at scan time
+  without re-running schema validation.
+
+- `validate_scan_config(payload)` — thin backward-compatible wrapper that
+  returns only the validation errors, preserved for callers that don't
+  need the normalized payload.
 
 - `SCAN_CONFIG_SCHEMA` — aggregated JSON Schema derived from the Pydantic
   models via `model_json_schema()`. Served by the `/scan-configs/schema`
   endpoint and consumed by the UI editor for in-editor live validation.
 """
 
+import json
 from typing import Any
 
 from pydantic import ValidationError
 
 from prowler.config.schema.registry import SCHEMAS
 
+# Pydantic v2 prefixes messages emitted from a ``field_validator`` that
+# raises ``ValueError`` with this string. Strip it so the message that
+# reaches the UI is the one the validator actually wrote.
+_PYDANTIC_VALUE_ERROR_PREFIX = "Value error, "
+
 
 def _format_loc(loc: tuple) -> str:
     """Render a Pydantic error location as a dot-separated path.
@@ -50,48 +63,116 @@ def _format_loc(loc: tuple) -> str:
     return ".".join(parts) if parts else "<root>"
 
 
-def validate_scan_config(payload: Any) -> list[dict]:
-    """Validate a scan config payload against the registered provider schemas.
+def validate_and_normalize_scan_config(
+    payload: Any,
+) -> tuple[dict, list[dict[str, str]]]:
+    """Strict validation and normalization of a scan configuration payload.
 
-    Strict by design: every Pydantic violation surfaces as a `{path, message}`
-    entry so the caller can decide how to present it. Unknown provider
-    sections are accepted (consistent with `additionalProperties: True` at
-    the top level — the SDK simply has no opinion on them).
+    Returns ``(normalized, errors)``:
+
+    - ``normalized`` is a JSON-serializable dict that mirrors the layout of
+      ``prowler/config/config.yaml`` (keyed by provider type). Registered
+      provider sections are dumped from their Pydantic models with
+      ``mode="json"`` (so the API can persist the result in a Django
+      ``JSONField``) and ``exclude_unset=True`` (so omitted defaults are
+      not injected into pre-existing configurations). Unknown provider
+      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.
+
+    The input payload is never mutated.
     """
     if not isinstance(payload, dict):
-        return [
+        return {}, [
             {
                 "path": "<root>",
                 "message": "Scan config must be a mapping with provider sections.",
             }
         ]
 
-    errors: list[dict] = []
+    errors: list[dict[str, str]] = []
+    normalized: dict[str, Any] = {}
+
     for provider, section in payload.items():
-        schema_cls = SCHEMAS.get(provider)
+        # Reject non-string provider keys so distinct entries like ``123``
+        # and ``"123"`` don't collide after ``str()`` in the normalized dict.
+        # YAML always produces string keys at this level; anything else
+        # comes from a hand-built payload and is a caller bug.
+        if not isinstance(provider, str):
+            errors.append(
+                {
+                    "path": repr(provider),
+                    "message": "provider keys must be strings.",
+                }
+            )
+            continue
+
+        provider_key = provider
+        schema_cls = SCHEMAS.get(provider_key)
         if schema_cls is None:
-            # Unknown provider type: tolerated. The SDK will simply ignore it.
+            # Unknown provider type: tolerated, but only when its contents
+            # are already JSON-serializable. The API persists the returned
+            # payload in a Django ``JSONField`` and would blow up at write
+            # time if we let a ``set()`` or similar through here.
+            try:
+                json.dumps(section)
+            except (TypeError, ValueError) as exc:
+                errors.append(
+                    {
+                        "path": provider_key,
+                        "message": (
+                            "unknown provider section is not JSON-serializable: "
+                            f"{exc}"
+                        ),
+                    }
+                )
+                continue
+            normalized[provider_key] = section
             continue
         if not isinstance(section, dict):
             errors.append(
                 {
-                    "path": str(provider),
+                    "path": provider_key,
                     "message": "section must be a mapping.",
                 }
             )
             continue
         try:
-            schema_cls.model_validate(section)
+            model = schema_cls.model_validate(section)
         except ValidationError as exc:
             for err in exc.errors():
                 loc = err.get("loc") or ()
-                path = _format_loc((str(provider), *loc))
-                errors.append(
-                    {
-                        "path": path,
-                        "message": err.get("msg", "validation error"),
-                    }
-                )
+                path = _format_loc((provider_key, *loc))
+                message = err.get("msg", "validation error")
+                # Only strip on the specific error type that pydantic
+                # prefixes — a legitimate future message that happens to
+                # start with "Value error, " keeps its text intact.
+                if err.get("type") == "value_error" and message.startswith(
+                    _PYDANTIC_VALUE_ERROR_PREFIX
+                ):
+                    message = message[len(_PYDANTIC_VALUE_ERROR_PREFIX) :]
+                errors.append({"path": path, "message": message})
+            continue
+        normalized[provider_key] = model.model_dump(mode="json", exclude_unset=True)
+
+    if errors:
+        return {}, errors
+    return normalized, []
+
+
+def validate_scan_config(payload: Any) -> list[dict]:
+    """Backward-compatible wrapper returning only validation errors.
+
+    Preserved for callers that only need the strict-validation error list
+    (e.g. the DRF serializer that turns each entry into a
+    ``ValidationError``). New callers should prefer
+    :func:`validate_and_normalize_scan_config` to also receive the
+    normalized payload.
+    """
+    _, errors = validate_and_normalize_scan_config(payload)
     return errors
 
 

prowler/config/schema/base.py

@@ -1,4 +1,12 @@
-from pydantic import BaseModel, ConfigDict
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator
+
+# Item type for excluded_checks / excluded_services list entries. Item
+# whitespace is stripped via ``str_strip_whitespace`` on the base
+# ``model_config`` (no second stripping implementation added here), so
+# ``min_length=1`` catches "", "   ", and any all-whitespace input uniformly.
+NonEmptyScopeIdentifier = Annotated[str, StringConstraints(min_length=1)]
 
 
 class ProviderConfigBase(BaseModel):
@@ -15,3 +23,28 @@ class ProviderConfigBase(BaseModel):
         str_strip_whitespace=True,
         validate_assignment=False,
     )
+
+    excluded_checks: list[NonEmptyScopeIdentifier] = Field(
+        default_factory=list,
+        description="Check identifiers to exclude from the scan scope.",
+        json_schema_extra={"default": [], "uniqueItems": True},
+    )
+    excluded_services: list[NonEmptyScopeIdentifier] = Field(
+        default_factory=list,
+        description="Service identifiers to exclude from the scan scope.",
+        json_schema_extra={"default": [], "uniqueItems": True},
+    )
+
+    @field_validator("excluded_checks", "excluded_services")
+    @classmethod
+    def _reject_duplicates(cls, value: list[str]) -> list[str]:
+        seen: set[str] = set()
+        duplicates: set[str] = set()
+        for item in value:
+            if item in seen:
+                duplicates.add(item)
+            else:
+                seen.add(item)
+        if duplicates:
+            raise ValueError(f"duplicate values are not allowed: {sorted(duplicates)}")
+        return value

prowler/lib/scan/scan.py

@@ -178,25 +178,58 @@ class Scan:
                 )
             )
 
-        # Exclude checks
+        # Validate excluded checks against the FULL provider catalog — not
+        # just the selected scope — so a global config can exclude a valid
+        # check even when that check is not part of a particular scoped run.
+        excluded_check_set: set[str] = set()
         if excluded_checks:
-            for check in excluded_checks:
-                if check in self._checks_to_execute:
-                    self._checks_to_execute.remove(check)
-                else:
-                    raise ScanInvalidCheckError(
-                        f"Invalid check provided: {check}. Check does not exist in the provider."
-                    )
+            excluded_check_set = set(excluded_checks)
+            if len(excluded_check_set) != len(excluded_checks):
+                raise ScanInvalidCheckError(
+                    "Duplicate excluded checks are not allowed."
+                )
+            unknown_checks = excluded_check_set.difference(self._bulk_checks_metadata)
+            if unknown_checks:
+                raise ScanInvalidCheckError(
+                    f"Invalid excluded check(s) provided: {sorted(unknown_checks)}."
+                )
 
-        # Exclude services
+        # Validate excluded services against the provider service catalog.
+        # Only resolve the catalog when there is something to check to avoid
+        # walking the provider package tree unnecessarily.
+        excluded_service_set: set[str] = set()
         if excluded_services:
-            for check in self._checks_to_execute:
-                if get_service_name_from_check_name(check) in excluded_services:
-                    self._checks_to_execute.remove(check)
-                else:
-                    raise ScanInvalidServiceError(
-                        f"Invalid service provided: {check}. Service does not exist in the provider."
-                    )
+            excluded_service_set = set(excluded_services)
+            if len(excluded_service_set) != len(excluded_services):
+                raise ScanInvalidServiceError(
+                    "Duplicate excluded services are not allowed."
+                )
+            unknown_services = excluded_service_set.difference(
+                list_services(provider.type)
+            )
+            if unknown_services:
+                raise ScanInvalidServiceError(
+                    f"Invalid excluded service(s) provided: {sorted(unknown_services)}."
+                )
+
+        if excluded_check_set or excluded_service_set:
+            previous_scope = self._checks_to_execute
+            selected_checks = {
+                check
+                for check in previous_scope
+                if check not in excluded_check_set
+                and get_service_name_from_check_name(check) not in excluded_service_set
+            }
+            # Only complain when exclusions actually emptied a non-empty
+            # scope. If the scope was already empty (e.g. a severity or
+            # category filter matched nothing) the exclusions did not
+            # cause the emptiness and the misleading error would obscure
+            # the real reason.
+            if previous_scope and not selected_checks:
+                raise ScanInvalidCheckError(
+                    "The scan configuration excludes every selected check."
+                )
+            self._checks_to_execute = sorted(selected_checks)
 
         self._number_of_checks_to_execute = len(self._checks_to_execute)
 

tests/config/schema/exclusions_test.py

@@ -0,0 +1,113 @@
+"""Coverage for the ``excluded_checks`` / ``excluded_services`` fields
+added to :class:`prowler.config.schema.base.ProviderConfigBase`.
+
+Because the fields live on the base class, every registered provider
+schema exposes them and every provider must therefore share the same
+whitespace / uniqueness / non-empty guarantees. These tests lock in that
+contract at the base level and at the JSON-Schema level (which the UI
+editor consumes via ``ajv``).
+"""
+
+import pytest
+from pydantic import ValidationError
+
+from prowler.config.scan_config_schema import SCAN_CONFIG_SCHEMA
+from prowler.config.schema.aws import AWSProviderConfig
+from prowler.config.schema.registry import SCHEMAS
+from prowler.config.schema.validator import validate_provider_config
+
+EXCLUSION_FIELDS = ("excluded_checks", "excluded_services")
+
+
+class Test_JSON_Schema_Exposes_Exclusion_Fields:
+    @pytest.mark.parametrize("provider", sorted(SCHEMAS))
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_field_shape(self, provider, field):
+        field_schema = SCAN_CONFIG_SCHEMA["properties"][provider]["properties"][field]
+        assert field_schema["type"] == "array"
+        assert field_schema["items"] == {"type": "string", "minLength": 1}
+        assert field_schema["uniqueItems"] is True
+        assert field_schema["default"] == []
+
+
+class Test_Exclusion_Field_Validation:
+    def _model(self, **kwargs):
+        return AWSProviderConfig.model_validate(kwargs)
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_empty_string_is_rejected(self, field):
+        with pytest.raises(ValidationError):
+            self._model(**{field: [""]})
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_whitespace_only_string_is_rejected(self, field):
+        with pytest.raises(ValidationError):
+            self._model(**{field: ["   "]})
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_raw_duplicates_are_rejected(self, field):
+        with pytest.raises(ValidationError):
+            self._model(**{field: ["s3", "s3"]})
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_normalized_duplicates_are_rejected(self, field):
+        # After whitespace normalization ``" s3 "`` collapses to ``"s3"``
+        # and must be caught by the duplicate check.
+        with pytest.raises(ValidationError):
+            self._model(**{field: ["s3", " s3 "]})
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_whitespace_is_stripped(self, field):
+        model = self._model(**{field: [" identifier "]})
+        assert getattr(model, field) == ["identifier"]
+
+    @pytest.mark.parametrize("field", EXCLUSION_FIELDS)
+    def test_non_string_item_is_rejected(self, field):
+        with pytest.raises(ValidationError):
+            self._model(**{field: [123]})
+
+
+class Test_Exclusion_Defaults_Are_Not_Injected:
+    """The strict normalization path uses ``model_dump(exclude_unset=True)``
+    so pre-existing configs that never set an exclusion field must round-trip
+    without the default empty list being materialized."""
+
+    def test_absent_fields_are_not_injected_by_validator(self):
+        # The lenient SDK runtime path also uses ``exclude_unset=True`` under
+        # the hood; asserting on the validator output guards the promise
+        # against future refactors.
+        assert validate_provider_config("aws", {}, SCHEMAS["aws"]) == {}
+
+    def test_absent_fields_are_not_injected_when_other_keys_are_present(self):
+        assert validate_provider_config(
+            "aws",
+            {"max_ec2_instance_age_in_days": 180},
+            SCHEMAS["aws"],
+        ) == {"max_ec2_instance_age_in_days": 180}
+
+    def test_explicit_empty_list_round_trips(self):
+        # Explicitly setting ``excluded_checks: []`` is different from
+        # omitting it — the empty list is user-provided and must be
+        # preserved by the strict-normalization contract.
+        assert validate_provider_config(
+            "aws",
+            {"excluded_checks": []},
+            SCHEMAS["aws"],
+        ) == {"excluded_checks": []}
+
+
+class Test_Extra_Fields_Are_Preserved:
+    """``extra="allow"`` must keep plugin-provided keys around so the
+    ecosystem contract in ``validator_test.py`` still holds after adding
+    the exclusion fields."""
+
+    def test_unknown_keys_are_preserved_alongside_exclusions(self):
+        out = validate_provider_config(
+            "aws",
+            {"excluded_checks": ["s3_bucket_public_access"], "plugin_option": "kept"},
+            SCHEMAS["aws"],
+        )
+        assert out == {
+            "excluded_checks": ["s3_bucket_public_access"],
+            "plugin_option": "kept",
+        }

tests/config/schema/scan_config_schema_test.py

@@ -0,0 +1,224 @@
+"""Coverage for the strict scan-config validation and normalization
+contract exposed to the Prowler App backend.
+
+Split from :mod:`tests.config.schema.validator_test` because the strict
+API (``validate_and_normalize_scan_config``) has different guarantees:
+it never silently drops keys, and it returns a JSON-serializable payload
+the backend can persist verbatim in a Django ``JSONField``.
+"""
+
+import json
+
+import pytest
+
+from prowler.config.scan_config_schema import (
+    validate_and_normalize_scan_config,
+    validate_scan_config,
+)
+
+
+class Test_Non_Dict_Root:
+    @pytest.mark.parametrize("payload", [None, "string", 42, [], (1, 2)])
+    def test_non_mapping_root_is_rejected(self, payload):
+        normalized, errors = validate_and_normalize_scan_config(payload)
+        assert normalized == {}
+        assert len(errors) == 1
+        assert errors[0]["path"] == "<root>"
+
+
+class Test_Registered_Provider_Section_Must_Be_Mapping:
+    @pytest.mark.parametrize("section", ["a-string", 42, ["s3"], None])
+    def test_non_mapping_section_reports_provider_path(self, section):
+        normalized, errors = validate_and_normalize_scan_config({"aws": section})
+        assert normalized == {}
+        assert errors == [{"path": "aws", "message": "section must be a mapping."}]
+
+
+class Test_Success_Path:
+    def test_whitespace_is_normalized_in_exclusions(self):
+        normalized, errors = validate_and_normalize_scan_config(
+            {
+                "aws": {
+                    "excluded_checks": [" s3_bucket_public_access "],
+                    "excluded_services": [" s3 "],
+                }
+            }
+        )
+        assert errors == []
+        assert normalized == {
+            "aws": {
+                "excluded_checks": ["s3_bucket_public_access"],
+                "excluded_services": ["s3"],
+            }
+        }
+
+    def test_plugin_options_are_preserved(self):
+        # Third-party plugins inject arbitrary keys inside a provider
+        # section; ``extra="allow"`` on the schema keeps them alive
+        # through the dump/normalize round-trip.
+        normalized, errors = validate_and_normalize_scan_config(
+            {"aws": {"plugin_option": "preserved", "another": 42}}
+        )
+        assert errors == []
+        assert normalized == {"aws": {"plugin_option": "preserved", "another": 42}}
+
+    def test_omitted_defaults_are_not_injected(self):
+        normalized, errors = validate_and_normalize_scan_config(
+            {"aws": {"max_ec2_instance_age_in_days": 90}}
+        )
+        assert errors == []
+        assert normalized == {"aws": {"max_ec2_instance_age_in_days": 90}}
+        assert "excluded_checks" not in normalized["aws"]
+        assert "excluded_services" not in normalized["aws"]
+
+    def test_unknown_provider_sections_are_preserved_verbatim(self):
+        payload = {"future_provider": {"custom_option": True, "nested": {"k": 1}}}
+        normalized, errors = validate_and_normalize_scan_config(payload)
+        assert errors == []
+        assert normalized == payload
+
+    def test_normalized_payload_is_json_serializable(self):
+        normalized, _ = validate_and_normalize_scan_config(
+            {
+                "aws": {
+                    "excluded_checks": ["s3_bucket_public_access"],
+                    "excluded_services": ["s3"],
+                }
+            }
+        )
+        # If ``model_dump(mode="json", ...)`` is ever dropped this
+        # ``json.dumps`` call is what will notice.
+        json.dumps(normalized)
+
+    def test_input_payload_is_not_mutated(self):
+        payload = {
+            "aws": {
+                "excluded_checks": [" s3_bucket_public_access "],
+                "excluded_services": [" s3 "],
+            }
+        }
+        snapshot = json.loads(json.dumps(payload))
+        validate_and_normalize_scan_config(payload)
+        assert payload == snapshot
+
+
+class Test_Error_Path:
+    def test_invalid_input_returns_empty_normalized_and_errors(self):
+        normalized, errors = validate_and_normalize_scan_config(
+            {"aws": {"excluded_services": ["s3", " s3 "]}}
+        )
+        assert normalized == {}
+        assert errors
+        assert any(err["path"].startswith("aws.excluded_services") for err in errors)
+
+    def test_partial_error_zeros_the_normalized_payload(self):
+        # One valid provider + one invalid provider must not leak the
+        # valid section into a partially normalized result.
+        normalized, errors = validate_and_normalize_scan_config(
+            {
+                "aws": {"excluded_services": ["s3", "s3"]},
+                "azure": {"vm_backup_min_daily_retention_days": 7},
+            }
+        )
+        assert normalized == {}
+        assert errors
+        assert any(err["path"].startswith("aws.") for err in errors)
+
+    def test_value_error_prefix_is_stripped_from_user_facing_messages(self):
+        # Pydantic prefixes messages emitted from ``field_validator``
+        # ValueError with ``"Value error, "``. If this test starts to fail
+        # because the prefix reappears, either pydantic changed the format
+        # or the strip in ``validate_and_normalize_scan_config`` was
+        # dropped — either way the UI would render the noisy prefix, so
+        # we lock the cleaned message in explicitly.
+        _, errors = validate_and_normalize_scan_config(
+            {"aws": {"excluded_services": ["s3", "s3"]}}
+        )
+        assert errors
+        message = errors[0]["message"]
+        assert not message.startswith("Value error, ")
+        assert "duplicate values are not allowed" in message
+
+    def test_all_errors_are_reported_not_only_the_first(self):
+        normalized, errors = validate_and_normalize_scan_config(
+            {
+                "aws": {
+                    "excluded_checks": ["", ""],
+                    "excluded_services": ["", ""],
+                }
+            }
+        )
+        assert normalized == {}
+        # ``excluded_checks`` yields per-item empty-string errors AND a
+        # duplicate error; ``excluded_services`` yields the same set.
+        paths = {err["path"] for err in errors}
+        assert any(p.startswith("aws.excluded_checks") for p in paths)
+        assert any(p.startswith("aws.excluded_services") for p in paths)
+
+
+class Test_Non_String_Provider_Keys:
+    """The normalized payload is later persisted in a Django JSONField
+    keyed by provider. Two entries whose ``str()`` collide (e.g. ``123``
+    and ``"123"``) would silently overwrite each other, so non-string
+    keys are rejected up front instead of silently coerced."""
+
+    def test_non_string_key_is_rejected(self):
+        normalized, errors = validate_and_normalize_scan_config({123: {}})
+        assert normalized == {}
+        assert errors == [{"path": "123", "message": "provider keys must be strings."}]
+
+    def test_string_and_int_collision_does_not_silently_overwrite(self):
+        # If only ``str()`` coercion happened both keys would collapse to
+        # ``"aws"`` in the output — this test guards against that regression.
+        normalized, errors = validate_and_normalize_scan_config(
+            {"aws": {}, 123: {"a": 1}}
+        )
+        assert normalized == {}
+        assert any(err["path"] == "123" for err in errors)
+
+
+class Test_Unknown_Sections_Must_Be_JSON_Serializable:
+    """``normalized`` is persisted by the API in a Django JSONField, so
+    unknown provider sections must fail fast here instead of blowing up
+    at persist time. Registered sections cannot hit this path — they go
+    through ``model_dump(mode="json", ...)`` which already coerces."""
+
+    def test_set_inside_unknown_section_is_rejected(self):
+        # ``set`` is a common trap: ``yaml.safe_load`` never produces it,
+        # but a hand-built dict might.
+        normalized, errors = validate_and_normalize_scan_config(
+            {"future_provider": {"values": {1, 2, 3}}}
+        )
+        assert normalized == {}
+        assert errors
+        assert errors[0]["path"] == "future_provider"
+        assert "JSON-serializable" in errors[0]["message"]
+
+    def test_json_safe_unknown_section_is_still_preserved(self):
+        payload = {"future_provider": {"nested": {"k": [1, 2, 3]}}}
+        normalized, errors = validate_and_normalize_scan_config(payload)
+        assert errors == []
+        assert normalized == payload
+
+
+class Test_Backward_Compatible_Wrapper:
+    def test_valid_payload_yields_no_errors(self):
+        assert (
+            validate_scan_config(
+                {"aws": {"excluded_checks": ["s3_bucket_public_access"]}}
+            )
+            == []
+        )
+
+    def test_invalid_payload_yields_only_the_errors(self):
+        errors = validate_scan_config({"aws": {"excluded_checks": ["", ""]}})
+        assert errors
+        assert all(set(err) == {"path", "message"} for err in errors)
+
+    def test_non_mapping_root_matches_new_contract(self):
+        assert validate_scan_config(None) == [
+            {
+                "path": "<root>",
+                "message": "Scan config must be a mapping with provider sections.",
+            }
+        ]

tests/lib/scan/scan_exclusions_test.py

@@ -0,0 +1,178 @@
+"""Coverage for ``Scan`` constructor exclusion semantics.
+
+The Scan class is the single execution entry point used by both the CLI
+and the API worker. Its exclusion validation must:
+
+- Reject duplicates and unknown identifiers with actionable errors.
+- Validate excluded checks against the FULL provider catalog so a global
+  configuration can exclude a valid check that is not part of a scoped
+  run (see the SDK acceptance criteria for scan-configuration exclusions).
+- Refuse a configuration that would leave nothing to execute.
+- Produce a deterministic, sorted final scope.
+
+The catalog dependencies (``CheckMetadata.get_bulk``, ``Compliance.get_bulk``,
+``list_services``, ``load_checks_to_execute``) are patched so tests stay
+focused on the exclusion logic and avoid walking the provider package tree.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from prowler.lib.scan.exceptions.exceptions import (
+    ScanInvalidCheckError,
+    ScanInvalidServiceError,
+)
+from prowler.lib.scan.scan import Scan
+from tests.providers.aws.utils import set_mocked_aws_provider
+
+# The provider catalog for these tests: three checks spread across two
+# services (``accessanalyzer`` and ``s3``). Keeps assertions readable.
+PROVIDER_CATALOG = {
+    "accessanalyzer_enabled",
+    "s3_bucket_encryption_enabled",
+    "s3_bucket_public_access",
+}
+PROVIDER_SERVICES = ["accessanalyzer", "s3"]
+
+
+@pytest.fixture
+def scan_provider():
+    provider = set_mocked_aws_provider()
+    metadata = MagicMock()
+    metadata.Categories = []
+    bulk = {check: metadata for check in PROVIDER_CATALOG}
+
+    with (
+        patch(
+            "prowler.lib.scan.scan.CheckMetadata.get_bulk",
+            return_value=bulk,
+        ),
+        patch("prowler.lib.scan.scan.Compliance.get_bulk", return_value={}),
+        patch(
+            "prowler.lib.scan.scan.update_checks_metadata_with_compliance",
+            side_effect=lambda _compliance, checks: checks,
+        ),
+        patch(
+            "prowler.lib.scan.scan.load_checks_to_execute",
+            side_effect=lambda **kwargs: set(kwargs["check_list"] or PROVIDER_CATALOG),
+        ),
+        patch(
+            "prowler.lib.scan.scan.list_services",
+            return_value=PROVIDER_SERVICES,
+        ),
+    ):
+        yield provider
+
+
+class Test_Exclusion_No_Ops:
+    def test_none_lists_are_no_ops(self, scan_provider):
+        scan = Scan(scan_provider, excluded_checks=None, excluded_services=None)
+        assert scan.checks_to_execute == sorted(PROVIDER_CATALOG)
+
+    def test_empty_lists_are_no_ops(self, scan_provider):
+        scan = Scan(scan_provider, excluded_checks=[], excluded_services=[])
+        assert scan.checks_to_execute == sorted(PROVIDER_CATALOG)
+
+
+class Test_Excluded_Checks:
+    def test_valid_check_is_removed_from_the_scope(self, scan_provider):
+        scan = Scan(
+            scan_provider,
+            excluded_checks=["s3_bucket_public_access"],
+        )
+        assert scan.checks_to_execute == sorted(
+            PROVIDER_CATALOG - {"s3_bucket_public_access"}
+        )
+
+    def test_excluded_check_may_be_outside_the_selected_scope(self, scan_provider):
+        # ``s3_bucket_public_access`` is not in the explicitly selected
+        # ``checks`` list but is still a valid provider check, so the
+        # global exclusion must be accepted and be a no-op for this run.
+        scan = Scan(
+            scan_provider,
+            checks=["accessanalyzer_enabled"],
+            excluded_checks=["s3_bucket_public_access"],
+        )
+        assert scan.checks_to_execute == ["accessanalyzer_enabled"]
+
+    def test_unknown_check_is_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidCheckError):
+            Scan(scan_provider, excluded_checks=["not_a_real_check"])
+
+    def test_duplicate_checks_are_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidCheckError):
+            Scan(
+                scan_provider,
+                excluded_checks=[
+                    "s3_bucket_public_access",
+                    "s3_bucket_public_access",
+                ],
+            )
+
+
+class Test_Excluded_Services:
+    def test_service_exclusion_removes_every_check_in_the_service(self, scan_provider):
+        scan = Scan(scan_provider, excluded_services=["s3"])
+        assert scan.checks_to_execute == ["accessanalyzer_enabled"]
+
+    def test_unknown_service_is_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidServiceError):
+            Scan(scan_provider, excluded_services=["not_a_real_service"])
+
+    def test_duplicate_services_are_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidServiceError):
+            Scan(scan_provider, excluded_services=["s3", "s3"])
+
+
+class Test_Combined_Exclusions:
+    def test_selected_checks_plus_excluded_checks_and_services(self, scan_provider):
+        scan = Scan(
+            scan_provider,
+            checks=["accessanalyzer_enabled", "s3_bucket_encryption_enabled"],
+            excluded_checks=["s3_bucket_public_access"],
+            excluded_services=["s3"],
+        )
+        # The explicit ``checks`` selection is narrowed by both the
+        # excluded_checks (drops nothing extra here) and excluded_services
+        # (drops every s3 check), leaving accessanalyzer alone.
+        assert scan.checks_to_execute == ["accessanalyzer_enabled"]
+
+    def test_result_is_sorted_and_deterministic(self, scan_provider):
+        scan = Scan(
+            scan_provider,
+            excluded_checks=["s3_bucket_public_access"],
+        )
+        assert scan.checks_to_execute == sorted(scan.checks_to_execute)
+
+
+class Test_Empty_Final_Scope_Is_Rejected:
+    def test_excluding_every_service_is_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidCheckError):
+            Scan(scan_provider, excluded_services=PROVIDER_SERVICES)
+
+    def test_excluding_every_check_is_rejected(self, scan_provider):
+        with pytest.raises(ScanInvalidCheckError):
+            Scan(scan_provider, excluded_checks=sorted(PROVIDER_CATALOG))
+
+
+class Test_Already_Empty_Scope_Does_Not_Blame_Exclusions:
+    """When a positive filter (severity, categories, checks that resolve
+    to nothing) leaves the scope empty *before* exclusions run, the
+    exclusion pass must not falsely claim to be the cause. Otherwise the
+    real reason (empty selection) is masked by a misleading error."""
+
+    def test_empty_initial_scope_with_valid_exclusions_does_not_raise(
+        self, scan_provider
+    ):
+        # Force ``load_checks_to_execute`` to return an empty scope while
+        # keeping the exclusion inputs valid against the provider catalog.
+        with patch(
+            "prowler.lib.scan.scan.load_checks_to_execute",
+            return_value=set(),
+        ):
+            scan = Scan(
+                scan_provider,
+                excluded_checks=["s3_bucket_public_access"],
+            )
+        assert scan.checks_to_execute == []