diff --git a/prowler/changelog.d/sdk-scan-config-exclusions.added.md b/prowler/changelog.d/sdk-scan-config-exclusions.added.md new file mode 100644 index 0000000000..6955af0e70 --- /dev/null +++ b/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 diff --git a/prowler/config/scan_config_schema.py b/prowler/config/scan_config_schema.py index ac00250c78..d09b49f32a 100644 --- a/prowler/config/scan_config_schema.py +++ b/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 "" -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": , "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. + + The input payload is never mutated. """ if not isinstance(payload, dict): - return [ + return {}, [ { "path": "", "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 diff --git a/prowler/config/schema/base.py b/prowler/config/schema/base.py index cc473a4545..fc5a76af43 100644 --- a/prowler/config/schema/base.py +++ b/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 diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py index 4bef660d33..b87cfcdf1d 100644 --- a/prowler/lib/scan/scan.py +++ b/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) diff --git a/tests/config/schema/exclusions_test.py b/tests/config/schema/exclusions_test.py new file mode 100644 index 0000000000..a13ed2f786 --- /dev/null +++ b/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", + } diff --git a/tests/config/schema/scan_config_schema_test.py b/tests/config/schema/scan_config_schema_test.py new file mode 100644 index 0000000000..0245276a9b --- /dev/null +++ b/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"] == "" + + +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": "", + "message": "Scan config must be a mapping with provider sections.", + } + ] diff --git a/tests/lib/scan/scan_exclusions_test.py b/tests/lib/scan/scan_exclusions_test.py new file mode 100644 index 0000000000..2c8f260c20 --- /dev/null +++ b/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 == []