From 1a450aaa709388807c162a95fae62cf5e066d018 Mon Sep 17 00:00:00 2001 From: Pepe Fagoaga Date: Wed, 13 May 2026 08:04:13 +0200 Subject: [PATCH] feat(config): add SDK config's validator --- docs/developer-guide/configurable-checks.mdx | 67 ++++++ prowler/config/config.py | 11 +- prowler/config/schema/__init__.py | 0 prowler/config/schema/aws.py | 162 +++++++++++++ prowler/config/schema/azure.py | 34 +++ prowler/config/schema/base.py | 17 ++ prowler/config/schema/cloudflare.py | 10 + prowler/config/schema/gcp.py | 13 ++ prowler/config/schema/github.py | 9 + prowler/config/schema/kubernetes.py | 13 ++ prowler/config/schema/m365.py | 24 ++ prowler/config/schema/mongodbatlas.py | 9 + prowler/config/schema/registry.py | 28 +++ prowler/config/schema/validator.py | 61 +++++ prowler/config/schema/vercel.py | 15 ++ tests/config/schema/__init__.py | 0 tests/config/schema/aws_schema_test.py | 218 ++++++++++++++++++ .../config/schema/loader_integration_test.py | 123 ++++++++++ .../schema/other_providers_schema_test.py | 148 ++++++++++++ tests/config/schema/validator_test.py | 175 ++++++++++++++ 20 files changed, 1136 insertions(+), 1 deletion(-) create mode 100644 prowler/config/schema/__init__.py create mode 100644 prowler/config/schema/aws.py create mode 100644 prowler/config/schema/azure.py create mode 100644 prowler/config/schema/base.py create mode 100644 prowler/config/schema/cloudflare.py create mode 100644 prowler/config/schema/gcp.py create mode 100644 prowler/config/schema/github.py create mode 100644 prowler/config/schema/kubernetes.py create mode 100644 prowler/config/schema/m365.py create mode 100644 prowler/config/schema/mongodbatlas.py create mode 100644 prowler/config/schema/registry.py create mode 100644 prowler/config/schema/validator.py create mode 100644 prowler/config/schema/vercel.py create mode 100644 tests/config/schema/__init__.py create mode 100644 tests/config/schema/aws_schema_test.py create mode 100644 tests/config/schema/loader_integration_test.py create mode 100644 tests/config/schema/other_providers_schema_test.py create mode 100644 tests/config/schema/validator_test.py diff --git a/docs/developer-guide/configurable-checks.mdx b/docs/developer-guide/configurable-checks.mdx index 76b339601d..fb0c6ffcd0 100644 --- a/docs/developer-guide/configurable-checks.mdx +++ b/docs/developer-guide/configurable-checks.mdx @@ -40,9 +40,76 @@ When adding a new configurable check to Prowler, update the following files: # aws.awslambda_function_vpc_multi_az lambda_min_azs: 2 ``` +- **Provider Schema:** Add the typed field to the provider's Pydantic schema in `prowler/config/schema/.py`. This is required: the loader validates user configs against these schemas and the shipped `config.yaml` must round-trip with zero warnings. See [Adding a parameter to the provider schema](#adding-a-parameter-to-the-provider-schema) below. - **Test Fixtures:** If tests depend on this configuration, add the variable to `tests/config/fixtures/config.yaml`. - **Documentation:** Document the new variable in the list of configurable checks in `docs/tutorials/configuration_file.md`. For a complete list of checks that already support configuration, see the [Configuration File Tutorial](/user-guide/cli/tutorials/configuration_file). +## Adding a parameter to the provider schema + +Every provider has a typed Pydantic schema in `prowler/config/schema/`. When a config is loaded, `validate_provider_config` checks each user-supplied key against the schema, logs a warning, and drops any field that fails validation. The consumer's `.get(key, default)` then falls back to the built-in default. + +This catches typos in a value (for example, `0.2` typed as `20`, or `"medium"` for an enum that expects `"MEDIUM"`). It does NOT catch typos in a key name: `disalowed_regions` (one `l` missing) is treated as an unknown key and passes through untouched, because third-party check plugins legitimately rely on unknown keys being preserved. Reviewers should still check that any new key the YAML adds is named exactly the same as the field on the schema. + +### Where to add the field + +1. Open `prowler/config/schema/.py` (for example, `aws.py`). +2. Add a field on the provider's schema class. Always make it `Optional[...] = None` so the absence of the key is valid. +3. Apply the tightest type the value allows. Examples below. + +If you are introducing an entirely new provider rather than a new parameter, also add an entry mapping the provider name to its schema class in `prowler/config/schema/registry.py`. The loader uses that registry to find the schema for the provider it is loading. + +### Choosing the right type + +| Value kind | Field declaration | +|---|---| +| Boolean toggle | `Optional[bool] = None` | +| Strictly positive integer (days, counts) | `Optional[int] = Field(default=None, gt=0)` | +| Fraction in 0..1 (threshold) | `Optional[float] = Field(default=None, ge=0.0, le=1.0)` | +| Closed set of strings | `Optional[Literal["A", "B", "C"]] = None` | +| Free-form string | `Optional[str] = None` | +| List of strings or ints | `Optional[list[str]] = None` | + +Prefer `Literal[...]` over `str` whenever the value is one of a known set. Prefer `Field(gt=0)` over `int` whenever zero or negative would be nonsensical. The point of the schema is to catch real-world mistakes that previously passed silently. + +### Custom validators (only when needed) + +If the value has structural rules beyond type and range, add a `field_validator`. Examples already in `aws.py`: + +- `_validate_port_range` rejects ports outside `0..65535`. +- `_validate_account_ids` rejects anything that isn't a 12-digit AWS account ID. +- `_validate_trusted_ips` rejects entries that aren't a valid IP or CIDR. + +Raise `ValueError` from the validator. The framework converts the error into a warning and drops the offending key. + +### Example: adding a new parameter + +Say a new check needs `max_iam_role_session_hours`, a strictly positive integer that defaults to 12 in code. + +1. **Schema** (`prowler/config/schema/aws.py`): + ```python + # IAM + max_iam_role_session_hours: Optional[int] = Field(default=None, gt=0) + ``` +2. **Shipped config** (`prowler/config/config.yaml`): + ```yaml + # aws.iam_role_session_duration_within_limit + max_iam_role_session_hours: 12 + ``` +3. **Consumer** (the check): + ```python + max_hours = iam_client.audit_config.get("max_iam_role_session_hours", 12) + ``` +4. **Tests** in `tests/config/schema/aws_schema_test.py`: + - one test for a valid value that round-trips, + - one test for an invalid value (zero, negative, wrong type) that is dropped. + +### What the loader guarantees + +- **Unknown keys pass through.** Third-party check plugins can introduce arbitrary keys without schema edits; they will not be filtered. +- **Invalid values never crash the run.** They produce a single warning per field and the key is dropped. +- **Coerced values are normalized.** A YAML-quoted `"180"` for an `int` field arrives downstream as the integer `180`. +- **The shipped `config.yaml` must round-trip cleanly.** The integration test `test_shipped_default_config_loads_without_warnings` will fail if a key is added to the YAML without a matching schema field, so the two stay in sync. + This approach ensures that checks are easily configurable, making Prowler highly adaptable to different environments and requirements. diff --git a/prowler/config/config.py b/prowler/config/config.py index f318332e2f..92deb103df 100644 --- a/prowler/config/config.py +++ b/prowler/config/config.py @@ -221,6 +221,11 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict: Returns: dict: The configuration dictionary for the specified provider. """ + # Imported lazily to avoid an import cycle: schemas may eventually want to + # import from prowler.config.config (e.g. for shared constants). + from prowler.config.schema.registry import SCHEMAS + from prowler.config.schema.validator import validate_provider_config + try: with open(config_file_path, "r", encoding=encoding_format_utf_8) as f: config_file = yaml.safe_load(f) @@ -238,7 +243,11 @@ def load_and_validate_config_file(provider: str, config_file_path: str) -> dict: if provider in ["azure", "gcp", "kubernetes", "m365"]: config = {} - return config + return validate_provider_config( + provider=provider, + raw=config, + schema_cls=SCHEMAS.get(provider), + ) except FileNotFoundError as error: logger.error( diff --git a/prowler/config/schema/__init__.py b/prowler/config/schema/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/config/schema/aws.py b/prowler/config/schema/aws.py new file mode 100644 index 0000000000..d1eb74240f --- /dev/null +++ b/prowler/config/schema/aws.py @@ -0,0 +1,162 @@ +from ipaddress import ip_network +from typing import Annotated, Literal, Optional + +from pydantic import AfterValidator, Field + +from prowler.config.schema.base import ProviderConfigBase + + +class _DetectSecretsPlugin(ProviderConfigBase): + """One entry inside ``detect_secrets_plugins``. + + Only ``name`` is required by the upstream library. ``limit`` is used by + the entropy detectors. Any other plugin-specific kwarg is preserved by + the ``extra="allow"`` policy inherited from ProviderConfigBase. + """ + + name: str + limit: Optional[float] = None + + +def _validate_port_range(v: Optional[list[int]]) -> Optional[list[int]]: + if v is None: + return v + for port in v: + if not 0 <= port <= 65535: + raise ValueError(f"port {port} is outside the valid range 0..65535") + return v + + +def _validate_account_ids(v: Optional[list[str]]) -> Optional[list[str]]: + if v is None: + return v + for account_id in v: + if not (account_id.isdigit() and len(account_id) == 12): + raise ValueError( + f"trusted_account_ids entry {account_id!r} is not a 12-digit AWS account id" + ) + return v + + +def _validate_trusted_ips(v: Optional[list[str]]) -> Optional[list[str]]: + if v is None: + return v + for entry in v: + try: + ip_network(entry, strict=False) + except ValueError as exc: + raise ValueError( + f"trusted_ips entry {entry!r} is not a valid IP or CIDR ({exc})" + ) from exc + return v + + +class AWSProviderConfig(ProviderConfigBase): + # IAM + mute_non_default_regions: Optional[bool] = None + disallowed_regions: Optional[list[str]] = None + max_unused_access_keys_days: Optional[int] = Field(default=None, gt=0) + max_console_access_days: Optional[int] = Field(default=None, gt=0) + max_unused_sagemaker_access_days: Optional[int] = Field(default=None, gt=0) + + # EC2 + shodan_api_key: Optional[str] = None + max_security_group_rules: Optional[int] = Field(default=None, gt=0) + max_ec2_instance_age_in_days: Optional[int] = Field(default=None, gt=0) + ec2_allowed_interface_types: Optional[list[str]] = None + ec2_allowed_instance_owners: Optional[list[str]] = None + ec2_high_risk_ports: Annotated[ + Optional[list[int]], AfterValidator(_validate_port_range) + ] = None + + # ECS + fargate_linux_latest_version: Optional[str] = None + fargate_windows_latest_version: Optional[str] = None + + # Cross-account trust + trusted_account_ids: Annotated[ + Optional[list[str]], AfterValidator(_validate_account_ids) + ] = None + trusted_ips: Annotated[ + Optional[list[str]], AfterValidator(_validate_trusted_ips) + ] = None + + # CloudWatch / CloudFormation + log_group_retention_days: Optional[int] = Field(default=None, gt=0) + recommended_cdk_bootstrap_version: Optional[int] = Field(default=None, gt=0) + + # AppStream + max_idle_disconnect_timeout_in_seconds: Optional[int] = Field(default=None, gt=0) + max_disconnect_timeout_in_seconds: Optional[int] = Field(default=None, gt=0) + max_session_duration_seconds: Optional[int] = Field(default=None, gt=0) + + # Lambda + obsolete_lambda_runtimes: Optional[list[str]] = None + lambda_min_azs: Optional[int] = Field(default=None, gt=0) + + # Organizations + organizations_enabled_regions: Optional[list[str]] = None + organizations_trusted_delegated_administrators: Optional[list[str]] = None + organizations_trusted_ids: Optional[list[str]] = None + + # ECR + ecr_repository_vulnerability_minimum_severity: Optional[ + Literal["CRITICAL", "HIGH", "MEDIUM", "LOW"] + ] = None + + # Trusted Advisor + verify_premium_support_plans: Optional[bool] = None + + # CloudTrail threat detection + threat_detection_privilege_escalation_threshold: Optional[float] = Field( + default=None, ge=0.0, le=1.0 + ) + threat_detection_privilege_escalation_minutes: Optional[int] = Field( + default=None, gt=0 + ) + threat_detection_privilege_escalation_actions: Optional[list[str]] = None + + threat_detection_enumeration_threshold: Optional[float] = Field( + default=None, ge=0.0, le=1.0 + ) + threat_detection_enumeration_minutes: Optional[int] = Field(default=None, gt=0) + threat_detection_enumeration_actions: Optional[list[str]] = None + + threat_detection_llm_jacking_threshold: Optional[float] = Field( + default=None, ge=0.0, le=1.0 + ) + threat_detection_llm_jacking_minutes: Optional[int] = Field(default=None, gt=0) + threat_detection_llm_jacking_actions: Optional[list[str]] = None + + # RDS + check_rds_instance_replicas: Optional[bool] = None + + # ACM + days_to_expire_threshold: Optional[int] = Field(default=None, gt=0) + insecure_key_algorithms: Optional[list[str]] = None + + # EKS + eks_required_log_types: Optional[list[str]] = None + eks_cluster_oldest_version_supported: Optional[str] = None + + # CodeBuild + excluded_sensitive_environment_variables: Optional[list[str]] = None + codebuild_github_allowed_organizations: Optional[list[str]] = None + + # ELB / ELBv2 + elb_min_azs: Optional[int] = Field(default=None, gt=0) + elbv2_min_azs: Optional[int] = Field(default=None, gt=0) + + # ElastiCache + minimum_snapshot_retention_period: Optional[int] = Field(default=None, gt=0) + + # Secrets + secrets_ignore_patterns: Optional[list[str]] = None + max_days_secret_unused: Optional[int] = Field(default=None, gt=0) + max_days_secret_unrotated: Optional[int] = Field(default=None, gt=0) + + # Kinesis + min_kinesis_stream_retention_hours: Optional[int] = Field(default=None, gt=0) + + # detect-secrets plugins + detect_secrets_plugins: Optional[list[_DetectSecretsPlugin]] = None diff --git a/prowler/config/schema/azure.py b/prowler/config/schema/azure.py new file mode 100644 index 0000000000..2b04ccb2e9 --- /dev/null +++ b/prowler/config/schema/azure.py @@ -0,0 +1,34 @@ +from typing import Literal, Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class AzureProviderConfig(ProviderConfigBase): + # Network + shodan_api_key: Optional[str] = None + + # Defender + defender_attack_path_minimal_risk_level: Optional[ + Literal["Low", "Medium", "High", "Critical"] + ] = None + + # App Service + php_latest_version: Optional[str] = None + python_latest_version: Optional[str] = None + java_latest_version: Optional[str] = None + + # SQL + recommended_minimal_tls_versions: Optional[list[str]] = None + + # Virtual Machines + desired_vm_sku_sizes: Optional[list[str]] = None + vm_backup_min_daily_retention_days: Optional[int] = Field(default=None, gt=0) + + # API Management threat detection + apim_threat_detection_llm_jacking_threshold: Optional[float] = Field( + default=None, ge=0.0, le=1.0 + ) + apim_threat_detection_llm_jacking_minutes: Optional[int] = Field(default=None, gt=0) + apim_threat_detection_llm_jacking_actions: Optional[list[str]] = None diff --git a/prowler/config/schema/base.py b/prowler/config/schema/base.py new file mode 100644 index 0000000000..cc473a4545 --- /dev/null +++ b/prowler/config/schema/base.py @@ -0,0 +1,17 @@ +from pydantic import BaseModel, ConfigDict + + +class ProviderConfigBase(BaseModel): + """Base for every provider config schema. + + ``extra="allow"`` is REQUIRED for backwards compatibility: third-party + check plugins frequently introduce config keys we do not know about, + and pre-existing user configs may carry deprecated keys. Validation + must never reject these. + """ + + model_config = ConfigDict( + extra="allow", + str_strip_whitespace=True, + validate_assignment=False, + ) diff --git a/prowler/config/schema/cloudflare.py b/prowler/config/schema/cloudflare.py new file mode 100644 index 0000000000..0b612f047f --- /dev/null +++ b/prowler/config/schema/cloudflare.py @@ -0,0 +1,10 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class CloudflareProviderConfig(ProviderConfigBase): + # 0 disables retries; negative values would loop or assert in the client. + max_retries: Optional[int] = Field(default=None, ge=0) diff --git a/prowler/config/schema/gcp.py b/prowler/config/schema/gcp.py new file mode 100644 index 0000000000..22b3c29ce3 --- /dev/null +++ b/prowler/config/schema/gcp.py @@ -0,0 +1,13 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class GCPProviderConfig(ProviderConfigBase): + shodan_api_key: Optional[str] = None + mig_min_zones: Optional[int] = Field(default=None, gt=0) + max_snapshot_age_days: Optional[int] = Field(default=None, gt=0) + max_unused_account_days: Optional[int] = Field(default=None, gt=0) + storage_min_retention_days: Optional[int] = Field(default=None, gt=0) diff --git a/prowler/config/schema/github.py b/prowler/config/schema/github.py new file mode 100644 index 0000000000..ce1bf2dbb8 --- /dev/null +++ b/prowler/config/schema/github.py @@ -0,0 +1,9 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class GitHubProviderConfig(ProviderConfigBase): + inactive_not_archived_days_threshold: Optional[int] = Field(default=None, gt=0) diff --git a/prowler/config/schema/kubernetes.py b/prowler/config/schema/kubernetes.py new file mode 100644 index 0000000000..01612ac5bf --- /dev/null +++ b/prowler/config/schema/kubernetes.py @@ -0,0 +1,13 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class KubernetesProviderConfig(ProviderConfigBase): + audit_log_maxbackup: Optional[int] = Field(default=None, gt=0) + audit_log_maxsize: Optional[int] = Field(default=None, gt=0) + audit_log_maxage: Optional[int] = Field(default=None, gt=0) + apiserver_strong_ciphers: Optional[list[str]] = None + kubelet_strong_ciphers: Optional[list[str]] = None diff --git a/prowler/config/schema/m365.py b/prowler/config/schema/m365.py new file mode 100644 index 0000000000..01013ac2ae --- /dev/null +++ b/prowler/config/schema/m365.py @@ -0,0 +1,24 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class M365ProviderConfig(ProviderConfigBase): + # Entra + sign_in_frequency: Optional[int] = Field(default=None, gt=0) + + # Teams + allowed_cloud_storage_services: Optional[list[str]] = None + + # Exchange + recommended_mailtips_large_audience_threshold: Optional[int] = Field( + default=None, gt=0 + ) + + # Defender malware policy + default_recommended_extensions: Optional[list[str]] = None + + # Mailbox auditing + audit_log_age: Optional[int] = Field(default=None, gt=0) diff --git a/prowler/config/schema/mongodbatlas.py b/prowler/config/schema/mongodbatlas.py new file mode 100644 index 0000000000..d9a210184c --- /dev/null +++ b/prowler/config/schema/mongodbatlas.py @@ -0,0 +1,9 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class MongoDBAtlasProviderConfig(ProviderConfigBase): + max_service_account_secret_validity_hours: Optional[int] = Field(default=None, gt=0) diff --git a/prowler/config/schema/registry.py b/prowler/config/schema/registry.py new file mode 100644 index 0000000000..fa31f1e5eb --- /dev/null +++ b/prowler/config/schema/registry.py @@ -0,0 +1,28 @@ +"""Mapping of provider name to its Pydantic schema class. + +Kept in its own module so the validator stays free of provider-schema imports +and callers pay the import cost only when they actually need the registry. +""" + +from prowler.config.schema.aws import AWSProviderConfig +from prowler.config.schema.azure import AzureProviderConfig +from prowler.config.schema.base import ProviderConfigBase +from prowler.config.schema.cloudflare import CloudflareProviderConfig +from prowler.config.schema.gcp import GCPProviderConfig +from prowler.config.schema.github import GitHubProviderConfig +from prowler.config.schema.kubernetes import KubernetesProviderConfig +from prowler.config.schema.m365 import M365ProviderConfig +from prowler.config.schema.mongodbatlas import MongoDBAtlasProviderConfig +from prowler.config.schema.vercel import VercelProviderConfig + +SCHEMAS: dict[str, type[ProviderConfigBase]] = { + "aws": AWSProviderConfig, + "azure": AzureProviderConfig, + "gcp": GCPProviderConfig, + "kubernetes": KubernetesProviderConfig, + "m365": M365ProviderConfig, + "github": GitHubProviderConfig, + "mongodbatlas": MongoDBAtlasProviderConfig, + "cloudflare": CloudflareProviderConfig, + "vercel": VercelProviderConfig, +} diff --git a/prowler/config/schema/validator.py b/prowler/config/schema/validator.py new file mode 100644 index 0000000000..c2acb642e9 --- /dev/null +++ b/prowler/config/schema/validator.py @@ -0,0 +1,61 @@ +from typing import Any + +from pydantic import ValidationError + +from prowler.config.schema.base import ProviderConfigBase +from prowler.lib.logger import logger + + +def validate_provider_config( + provider: str, + raw: Any, + schema_cls: type[ProviderConfigBase] | None, +) -> dict: + """Validate a provider's config dict against its Pydantic schema. + + Behavior is intentionally lenient to preserve backwards compatibility: + + - If ``raw`` is not a dict, return an empty dict (mirrors prior loader). + - If no schema is registered for ``provider``, return ``raw`` untouched. + - On validation errors, log one WARNING per offending field, DROP those + keys from the result, and continue. Consumers fall back to their own + hard-coded defaults via ``audit_config.get(key, default)``. + - Coerced values (e.g. ``"180"`` -> ``180``) replace the user's input + so that downstream checks never receive a wrongly-typed value. + """ + if not isinstance(raw, dict): + return {} + + if schema_cls is None: + return raw + + try: + model = schema_cls.model_validate(raw) + return model.model_dump(exclude_unset=True) + except ValidationError as exc: + bad_keys: set[str] = set() + for err in exc.errors(): + loc = err.get("loc") or () + if not loc: + continue + key = loc[0] + if not isinstance(key, str): + continue + bad_keys.add(key) + logger.warning( + f"prowler.config[{provider}.{key}] = {raw.get(key)!r} is invalid " + f"({err.get('msg', 'validation error')}); the value will be ignored " + f"and the built-in default will be used." + ) + + cleaned = {k: v for k, v in raw.items() if k not in bad_keys} + try: + model = schema_cls.model_validate(cleaned) + return model.model_dump(exclude_unset=True) + except ValidationError as exc2: + logger.error( + f"prowler.config[{provider}] could not be revalidated after dropping " + f"invalid keys ({bad_keys}); passing through the cleaned dict as-is. " + f"Underlying errors: {exc2.errors()}" + ) + return cleaned diff --git a/prowler/config/schema/vercel.py b/prowler/config/schema/vercel.py new file mode 100644 index 0000000000..8d0f672fd3 --- /dev/null +++ b/prowler/config/schema/vercel.py @@ -0,0 +1,15 @@ +from typing import Optional + +from pydantic import Field + +from prowler.config.schema.base import ProviderConfigBase + + +class VercelProviderConfig(ProviderConfigBase): + stable_branches: Optional[list[str]] = None + days_to_expire_threshold: Optional[int] = Field(default=None, gt=0) + stale_token_threshold_days: Optional[int] = Field(default=None, gt=0) + stale_invitation_threshold_days: Optional[int] = Field(default=None, gt=0) + max_owner_percentage: Optional[int] = Field(default=None, ge=0, le=100) + max_owners: Optional[int] = Field(default=None, gt=0) + secret_suffixes: Optional[list[str]] = None diff --git a/tests/config/schema/__init__.py b/tests/config/schema/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/config/schema/aws_schema_test.py b/tests/config/schema/aws_schema_test.py new file mode 100644 index 0000000000..989ebde0e2 --- /dev/null +++ b/tests/config/schema/aws_schema_test.py @@ -0,0 +1,218 @@ +"""AWS-specific schema coverage — the biggest provider, with the richest +constraint surface (CIDRs, account IDs, port ranges, enums, thresholds).""" + +import pytest + +from prowler.config.schema.aws import AWSProviderConfig +from prowler.config.schema.validator import validate_provider_config + + +def _validate(raw): + return validate_provider_config("aws", raw, AWSProviderConfig) + + +class Test_AWS_Threat_Detection_Thresholds: + """All threat detection thresholds are documented as fractions in 0..1. + The biggest risk of mistyping them is silently disabling the check.""" + + @pytest.mark.parametrize( + "key", + [ + "threat_detection_privilege_escalation_threshold", + "threat_detection_enumeration_threshold", + "threat_detection_llm_jacking_threshold", + ], + ) + def test_valid_boundary_values(self, key): + assert _validate({key: 0.0}) == {key: 0.0} + assert _validate({key: 1.0}) == {key: 1.0} + assert _validate({key: 0.5}) == {key: 0.5} + + @pytest.mark.parametrize( + "key", + [ + "threat_detection_privilege_escalation_threshold", + "threat_detection_enumeration_threshold", + "threat_detection_llm_jacking_threshold", + ], + ) + def test_invalid_values_are_dropped(self, key): + # 20 instead of 0.2 — would never trigger + assert _validate({key: 20}) == {} + # negative + assert _validate({key: -0.1}) == {} + # string + assert _validate({key: "high"}) == {} + + +class Test_AWS_Trusted_Account_Ids: + def test_valid_twelve_digit_ids(self): + ids = ["123456789012", "098765432109"] + assert _validate({"trusted_account_ids": ids}) == {"trusted_account_ids": ids} + + def test_empty_list_is_valid(self): + assert _validate({"trusted_account_ids": []}) == {"trusted_account_ids": []} + + def test_short_id_is_dropped(self): + assert _validate({"trusted_account_ids": ["12345"]}) == {} + + def test_non_numeric_id_is_dropped(self): + assert _validate({"trusted_account_ids": ["1234abcd5678"]}) == {} + + def test_id_with_dashes_is_dropped(self): + # Some users format account IDs as "1234-5678-9012" + assert _validate({"trusted_account_ids": ["1234-5678-9012"]}) == {} + + +class Test_AWS_Trusted_Ips: + def test_single_ipv4_address(self): + assert _validate({"trusted_ips": ["1.2.3.4"]}) == {"trusted_ips": ["1.2.3.4"]} + + def test_ipv4_cidr(self): + assert _validate({"trusted_ips": ["10.0.0.0/8"]}) == { + "trusted_ips": ["10.0.0.0/8"] + } + + def test_ipv6_address(self): + assert _validate({"trusted_ips": ["2001:db8::1"]}) == { + "trusted_ips": ["2001:db8::1"] + } + + def test_ipv6_cidr(self): + assert _validate({"trusted_ips": ["2001:db8::/32"]}) == { + "trusted_ips": ["2001:db8::/32"] + } + + def test_mixed_list(self): + ips = ["1.2.3.4", "10.0.0.0/8", "2001:db8::1"] + assert _validate({"trusted_ips": ips}) == {"trusted_ips": ips} + + def test_garbage_entry_is_dropped(self): + assert _validate({"trusted_ips": ["definitely-not-an-ip"]}) == {} + + def test_cidr_with_host_bits_is_accepted(self): + # We use strict=False so "10.0.0.5/8" is accepted. This matches the + # behaviour of most security tools and avoids surprising users who + # paste real-world allowlists with non-canonical CIDR notation. + assert _validate({"trusted_ips": ["10.0.0.5/8"]}) == { + "trusted_ips": ["10.0.0.5/8"] + } + + +class Test_AWS_Ports: + def test_valid_ports_in_range(self): + ports = [25, 80, 443, 65535, 0] + assert _validate({"ec2_high_risk_ports": ports}) == { + "ec2_high_risk_ports": ports + } + + def test_out_of_range_port_is_dropped(self): + assert _validate({"ec2_high_risk_ports": [70000]}) == {} + + def test_negative_port_is_dropped(self): + assert _validate({"ec2_high_risk_ports": [-1]}) == {} + + +class Test_AWS_Enums: + @pytest.mark.parametrize("level", ["CRITICAL", "HIGH", "MEDIUM", "LOW"]) + def test_valid_severity_levels(self, level): + assert _validate({"ecr_repository_vulnerability_minimum_severity": level}) == { + "ecr_repository_vulnerability_minimum_severity": level + } + + @pytest.mark.parametrize("level", ["critical", "Medium", "ANY", "", "X"]) + def test_invalid_severity_levels_are_dropped(self, level): + assert _validate({"ecr_repository_vulnerability_minimum_severity": level}) == {} + + +class Test_AWS_Detect_Secrets_Plugins: + def test_plugin_without_limit(self): + out = _validate({"detect_secrets_plugins": [{"name": "AWSKeyDetector"}]}) + assert out == {"detect_secrets_plugins": [{"name": "AWSKeyDetector"}]} + + def test_plugin_with_limit(self): + out = _validate( + { + "detect_secrets_plugins": [ + {"name": "Base64HighEntropyString", "limit": 6.0} + ] + } + ) + assert out == { + "detect_secrets_plugins": [ + {"name": "Base64HighEntropyString", "limit": 6.0} + ] + } + + def test_plugin_missing_name_drops_whole_field(self): + # ``name`` is required by the upstream library. + out = _validate({"detect_secrets_plugins": [{"limit": 6.0}]}) + assert out == {} + + def test_extra_plugin_kwargs_pass_through(self): + # Plugins can have arbitrary extra params (extra="allow" on the + # nested model). They must round-trip. + out = _validate( + { + "detect_secrets_plugins": [ + {"name": "Custom", "my_param": "abc", "other": 42} + ] + } + ) + assert out == { + "detect_secrets_plugins": [ + {"name": "Custom", "my_param": "abc", "other": 42} + ] + } + + +class Test_AWS_Booleans: + @pytest.mark.parametrize( + "key", + [ + "mute_non_default_regions", + "verify_premium_support_plans", + "check_rds_instance_replicas", + ], + ) + def test_true_and_false_round_trip(self, key): + assert _validate({key: True}) == {key: True} + assert _validate({key: False}) == {key: False} + + def test_yaml_style_boolean_coercion(self): + # YAML can produce Python str "true"/"yes" if the user quoted it. + # Pydantic v2 will refuse string booleans by default. Verify it is + # dropped, not silently treated as True (which would be dangerous + # for verify_premium_support_plans). + out = _validate({"verify_premium_support_plans": "yes"}) + # Pydantic actually DOES coerce "yes"/"no"/"true"/"false" in lax mode. + # We accept either outcome but require it to be a real bool. + if "verify_premium_support_plans" in out: + assert isinstance(out["verify_premium_support_plans"], bool) + + +class Test_AWS_Full_Default_Config_Round_Trips: + """Loading the real shipped defaults through the schema must produce + exactly the same dict. This is the regression sentinel for backwards + compatibility.""" + + def test_full_default_config_round_trip(self): + # Subset that mirrors the shipped config.yaml semantics. + raw = { + "mute_non_default_regions": False, + "disallowed_regions": ["me-south-1", "me-central-1"], + "max_unused_access_keys_days": 45, + "max_ec2_instance_age_in_days": 180, + "trusted_account_ids": [], + "trusted_ips": [], + "ecr_repository_vulnerability_minimum_severity": "MEDIUM", + "threat_detection_privilege_escalation_threshold": 0.2, + "threat_detection_enumeration_threshold": 0.3, + "threat_detection_llm_jacking_threshold": 0.4, + "ec2_high_risk_ports": [25, 110, 8088], + "detect_secrets_plugins": [ + {"name": "AWSKeyDetector"}, + {"name": "Base64HighEntropyString", "limit": 6.0}, + ], + } + assert _validate(raw) == raw diff --git a/tests/config/schema/loader_integration_test.py b/tests/config/schema/loader_integration_test.py new file mode 100644 index 0000000000..c3337f4447 --- /dev/null +++ b/tests/config/schema/loader_integration_test.py @@ -0,0 +1,123 @@ +"""End-to-end tests that exercise the real ``load_and_validate_config_file`` +through a temp YAML file. Anything that breaks here would break the actual +``prowler aws -c …`` code path.""" + +import logging +import os +import pathlib + +import pytest + +from prowler.config.config import load_and_validate_config_file + + +@pytest.fixture +def write_config(tmp_path): + def _write(content: str) -> str: + path = tmp_path / "config.yaml" + path.write_text(content) + return str(path) + + return _write + + +class Test_Loader_With_Schema_Integration: + def test_shipped_default_config_loads_without_warnings(self, caplog): + """The default ``prowler/config/config.yaml`` must round-trip every + provider WITHOUT emitting any schema warnings. If this fails, + someone added a key to the YAML without updating the schema.""" + repo_root = pathlib.Path(os.path.dirname(os.path.realpath(__file__))).parents[2] + shipped = repo_root / "prowler" / "config" / "config.yaml" + with caplog.at_level(logging.WARNING, logger="prowler"): + for provider in [ + "aws", + "azure", + "gcp", + "kubernetes", + "m365", + "github", + "mongodbatlas", + "cloudflare", + "vercel", + ]: + cfg = load_and_validate_config_file(provider, str(shipped)) + # Provider always exists in the shipped file → non-empty. + assert cfg, f"{provider} returned an empty config" + + offending = [ + r.getMessage() + for r in caplog.records + if "prowler.config[" in r.getMessage() + ] + assert not offending, ( + "Shipped config.yaml triggered schema warnings — schema or YAML out of sync:\n" + + "\n".join(offending) + ) + + def test_user_config_with_bad_threshold_falls_back(self, write_config, caplog): + path = write_config( + "aws:\n" + " threat_detection_privilege_escalation_threshold: 5.0\n" + " lambda_min_azs: 2\n" + ) + with caplog.at_level(logging.WARNING, logger="prowler"): + cfg = load_and_validate_config_file("aws", path) + assert cfg == {"lambda_min_azs": 2} + assert any( + "threat_detection_privilege_escalation_threshold" in r.getMessage() + for r in caplog.records + ) + + def test_old_format_config_still_works(self, write_config): + # Old format = flat keys, no provider header. + path = write_config( + "max_ec2_instance_age_in_days: 90\n" + "ecr_repository_vulnerability_minimum_severity: HIGH\n" + ) + cfg = load_and_validate_config_file("aws", path) + assert cfg == { + "max_ec2_instance_age_in_days": 90, + "ecr_repository_vulnerability_minimum_severity": "HIGH", + } + + def test_unknown_keys_pass_through_via_loader(self, write_config): + path = write_config( + "aws:\n" " third_party_plugin_setting: hello\n" " lambda_min_azs: 2\n" + ) + cfg = load_and_validate_config_file("aws", path) + assert cfg == { + "third_party_plugin_setting": "hello", + "lambda_min_azs": 2, + } + + def test_quoted_numeric_is_coerced_via_loader(self, write_config): + # YAML quotes the number: ``"180"`` arrives as a Python str. + # The schema must coerce it to int so downstream comparisons work. + path = write_config('aws:\n max_ec2_instance_age_in_days: "180"\n') + cfg = load_and_validate_config_file("aws", path) + assert cfg == {"max_ec2_instance_age_in_days": 180} + assert isinstance(cfg["max_ec2_instance_age_in_days"], int) + + def test_invalid_yaml_shape_list_as_string_drops_key(self, write_config, caplog): + path = write_config( + "aws:\n" + " disallowed_regions: me-south-1\n" # forgot list dashes + " lambda_min_azs: 2\n" + ) + with caplog.at_level(logging.WARNING, logger="prowler"): + cfg = load_and_validate_config_file("aws", path) + assert cfg == {"lambda_min_azs": 2} + assert any("disallowed_regions" in r.getMessage() for r in caplog.records) + + def test_other_providers_unaffected_by_aws_block(self, write_config): + path = write_config( + "aws:\n max_ec2_instance_age_in_days: 90\n" "gcp:\n mig_min_zones: 5\n" + ) + assert load_and_validate_config_file("aws", path) == { + "max_ec2_instance_age_in_days": 90 + } + assert load_and_validate_config_file("gcp", path) == {"mig_min_zones": 5} + + def test_missing_provider_block_returns_empty(self, write_config): + path = write_config("aws:\n max_ec2_instance_age_in_days: 90\n") + assert load_and_validate_config_file("azure", path) == {} diff --git a/tests/config/schema/other_providers_schema_test.py b/tests/config/schema/other_providers_schema_test.py new file mode 100644 index 0000000000..163be36204 --- /dev/null +++ b/tests/config/schema/other_providers_schema_test.py @@ -0,0 +1,148 @@ +"""Smaller-provider schema coverage. One happy path + one invalid path +per field is enough to lock in the contract; the validator behaviour +itself is covered exhaustively in validator_test.py.""" + +import pytest + +from prowler.config.schema.registry import SCHEMAS +from prowler.config.schema.validator import validate_provider_config + + +def _validate(provider, raw): + return validate_provider_config(provider, raw, SCHEMAS[provider]) + + +class Test_Azure_Schema: + @pytest.mark.parametrize("level", ["Low", "Medium", "High", "Critical"]) + def test_defender_risk_level_valid_values(self, level): + assert _validate( + "azure", {"defender_attack_path_minimal_risk_level": level} + ) == {"defender_attack_path_minimal_risk_level": level} + + def test_defender_risk_level_lowercase_dropped(self): + # Case matters: the matching check uses Title-case comparison. + assert ( + _validate("azure", {"defender_attack_path_minimal_risk_level": "high"}) + == {} + ) + + def test_apim_threshold_in_range(self): + out = _validate("azure", {"apim_threat_detection_llm_jacking_threshold": 0.1}) + assert out == {"apim_threat_detection_llm_jacking_threshold": 0.1} + + def test_apim_threshold_out_of_range(self): + out = _validate("azure", {"apim_threat_detection_llm_jacking_threshold": 1.5}) + assert out == {} + + def test_vm_backup_retention_must_be_positive(self): + assert _validate("azure", {"vm_backup_min_daily_retention_days": 7}) == { + "vm_backup_min_daily_retention_days": 7 + } + assert _validate("azure", {"vm_backup_min_daily_retention_days": 0}) == {} + assert _validate("azure", {"vm_backup_min_daily_retention_days": -1}) == {} + + +class Test_GCP_Schema: + def test_valid_values_round_trip(self): + raw = { + "mig_min_zones": 2, + "max_snapshot_age_days": 90, + "max_unused_account_days": 180, + "storage_min_retention_days": 90, + } + assert _validate("gcp", raw) == raw + + def test_zero_zone_count_dropped(self): + assert _validate("gcp", {"mig_min_zones": 0}) == {} + + +class Test_Kubernetes_Schema: + def test_valid_values_round_trip(self): + raw = { + "audit_log_maxbackup": 10, + "audit_log_maxsize": 100, + "audit_log_maxage": 30, + } + assert _validate("kubernetes", raw) == raw + + def test_negative_audit_log_dropped(self): + assert _validate("kubernetes", {"audit_log_maxage": -1}) == {} + + +class Test_M365_Schema: + def test_valid_values_round_trip(self): + raw = { + "sign_in_frequency": 4, + "recommended_mailtips_large_audience_threshold": 25, + "audit_log_age": 90, + } + assert _validate("m365", raw) == raw + + def test_negative_audit_log_age_dropped(self): + assert _validate("m365", {"audit_log_age": -10}) == {} + + +class Test_GitHub_Schema: + def test_valid_threshold(self): + assert _validate("github", {"inactive_not_archived_days_threshold": 180}) == { + "inactive_not_archived_days_threshold": 180 + } + + def test_zero_threshold_dropped(self): + assert _validate("github", {"inactive_not_archived_days_threshold": 0}) == {} + + +class Test_MongoDBAtlas_Schema: + def test_valid(self): + assert _validate( + "mongodbatlas", {"max_service_account_secret_validity_hours": 8} + ) == {"max_service_account_secret_validity_hours": 8} + + def test_invalid_negative(self): + assert ( + _validate("mongodbatlas", {"max_service_account_secret_validity_hours": -1}) + == {} + ) + + +class Test_Cloudflare_Schema: + def test_zero_retries_allowed(self): + # 0 is explicitly documented as "disable retries" in config.yaml. + assert _validate("cloudflare", {"max_retries": 0}) == {"max_retries": 0} + + def test_positive_retries_allowed(self): + assert _validate("cloudflare", {"max_retries": 3}) == {"max_retries": 3} + + def test_negative_retries_dropped(self): + assert _validate("cloudflare", {"max_retries": -1}) == {} + + +class Test_Vercel_Schema: + def test_owner_percentage_in_range(self): + assert _validate("vercel", {"max_owner_percentage": 20}) == { + "max_owner_percentage": 20 + } + assert _validate("vercel", {"max_owner_percentage": 0}) == { + "max_owner_percentage": 0 + } + assert _validate("vercel", {"max_owner_percentage": 100}) == { + "max_owner_percentage": 100 + } + + def test_owner_percentage_over_100_dropped(self): + assert _validate("vercel", {"max_owner_percentage": 150}) == {} + + def test_owner_percentage_negative_dropped(self): + assert _validate("vercel", {"max_owner_percentage": -1}) == {} + + def test_full_default_config_round_trip(self): + raw = { + "stable_branches": ["main", "master"], + "days_to_expire_threshold": 7, + "stale_token_threshold_days": 90, + "stale_invitation_threshold_days": 30, + "max_owner_percentage": 20, + "max_owners": 3, + "secret_suffixes": ["_KEY", "_SECRET", "_TOKEN"], + } + assert _validate("vercel", raw) == raw diff --git a/tests/config/schema/validator_test.py b/tests/config/schema/validator_test.py new file mode 100644 index 0000000000..398a0d10e8 --- /dev/null +++ b/tests/config/schema/validator_test.py @@ -0,0 +1,175 @@ +"""Behavioural tests for ``validate_provider_config``. + +The validator is the gatekeeper for every provider schema: its job is to +keep backwards-compatible behaviour (no exceptions, drop only the bad +keys) while loudly logging type mistakes. +""" + +import logging + +import pytest + +from prowler.config.schema.aws import AWSProviderConfig +from prowler.config.schema.registry import SCHEMAS +from prowler.config.schema.validator import validate_provider_config + + +class Test_Validate_Provider_Config_Contract: + """Generic invariants that must hold for any schema.""" + + def test_returns_empty_dict_when_raw_is_not_a_dict(self): + assert validate_provider_config("aws", None, AWSProviderConfig) == {} + assert validate_provider_config("aws", "string", AWSProviderConfig) == {} + assert validate_provider_config("aws", 42, AWSProviderConfig) == {} + assert validate_provider_config("aws", [], AWSProviderConfig) == {} + + def test_returns_raw_unchanged_when_no_schema_registered(self): + raw = {"anything": "goes", "even": [1, 2, 3]} + assert validate_provider_config("mystery_provider", raw, None) == raw + + def test_unknown_keys_pass_through_for_plugin_compatibility(self): + # Third-party plugins inject arbitrary keys; the schema must NOT + # filter them. This is the contract that lets the plugin ecosystem + # keep working when we add validation. + raw = {"plugin_custom_key": "foo", "lambda_min_azs": 2} + assert validate_provider_config("aws", raw, AWSProviderConfig) == { + "plugin_custom_key": "foo", + "lambda_min_azs": 2, + } + + def test_empty_dict_returns_empty_dict(self): + assert validate_provider_config("aws", {}, AWSProviderConfig) == {} + + def test_known_valid_value_passes_through_unchanged(self): + raw = {"max_ec2_instance_age_in_days": 180} + assert validate_provider_config("aws", raw, AWSProviderConfig) == { + "max_ec2_instance_age_in_days": 180 + } + + +class Test_Validate_Provider_Config_Coercion: + """Pydantic v2 coerces common type-mistakes automatically. We want to + keep that behaviour so quoted numerics in user configs ``Just Work``.""" + + def test_string_numeric_is_coerced_to_int(self): + out = validate_provider_config( + "aws", {"max_ec2_instance_age_in_days": "180"}, AWSProviderConfig + ) + assert out == {"max_ec2_instance_age_in_days": 180} + assert isinstance(out["max_ec2_instance_age_in_days"], int) + + def test_string_numeric_is_coerced_to_float(self): + out = validate_provider_config( + "aws", + {"threat_detection_privilege_escalation_threshold": "0.4"}, + AWSProviderConfig, + ) + assert out == {"threat_detection_privilege_escalation_threshold": 0.4} + + +class Test_Validate_Provider_Config_Drops_Invalid_Keys: + """When a field fails validation, only that key is dropped from the + returned dict. The rest of the user's config is preserved so the + consumer's ``audit_config.get(key, default)`` falls back to its own + built-in default for the offending field and uses user values for + everything else.""" + + def test_out_of_range_threshold_is_dropped(self, caplog): + with caplog.at_level(logging.WARNING): + out = validate_provider_config( + "aws", + { + "threat_detection_privilege_escalation_threshold": 2.0, + "lambda_min_azs": 2, + }, + AWSProviderConfig, + ) + assert out == {"lambda_min_azs": 2} + assert any( + "threat_detection_privilege_escalation_threshold" in r.getMessage() + for r in caplog.records + ) + + def test_invalid_enum_is_dropped(self): + out = validate_provider_config( + "aws", + {"ecr_repository_vulnerability_minimum_severity": "medum"}, + AWSProviderConfig, + ) + assert out == {} + + def test_wrong_shape_list_as_string_is_dropped(self): + # Classic YAML mistake: ``disallowed_regions: me-south-1`` without dashes. + # Pydantic refuses to silently treat a str as a single-element list, + # which is exactly the safety guarantee we want. + out = validate_provider_config( + "aws", + {"disallowed_regions": "me-south-1", "lambda_min_azs": 2}, + AWSProviderConfig, + ) + assert out == {"lambda_min_azs": 2} + + def test_negative_positive_int_is_dropped(self): + out = validate_provider_config( + "aws", {"max_ec2_instance_age_in_days": -1}, AWSProviderConfig + ) + assert out == {} + + def test_zero_is_dropped_for_strictly_positive_field(self): + # max_ec2_instance_age_in_days is gt=0. Zero would silently cause every + # instance to FAIL the age check. + out = validate_provider_config( + "aws", {"max_ec2_instance_age_in_days": 0}, AWSProviderConfig + ) + assert out == {} + + def test_multiple_invalid_keys_yield_multiple_warnings(self, caplog): + with caplog.at_level(logging.WARNING): + out = validate_provider_config( + "aws", + { + "max_ec2_instance_age_in_days": "nope", + "ecr_repository_vulnerability_minimum_severity": "medum", + "valid_extra_key": "kept", + }, + AWSProviderConfig, + ) + assert out == {"valid_extra_key": "kept"} + messages = " ".join(r.getMessage() for r in caplog.records) + assert "max_ec2_instance_age_in_days" in messages + assert "ecr_repository_vulnerability_minimum_severity" in messages + + def test_warning_message_includes_provider_and_field(self, caplog): + with caplog.at_level(logging.WARNING): + validate_provider_config( + "aws", + {"threat_detection_privilege_escalation_threshold": 5.0}, + AWSProviderConfig, + ) + assert any( + "prowler.config[aws.threat_detection_privilege_escalation_threshold]" + in r.getMessage() + for r in caplog.records + ) + + +class Test_Schemas_Registry: + """Every provider mentioned in the YAML config must have a schema.""" + + @pytest.mark.parametrize( + "provider", + [ + "aws", + "azure", + "gcp", + "kubernetes", + "m365", + "github", + "mongodbatlas", + "cloudflare", + "vercel", + ], + ) + def test_schema_registered_for_provider(self, provider): + assert provider in SCHEMAS + assert SCHEMAS[provider] is not None