feat(config): add SDK config's validator

This commit is contained in:
Pepe Fagoaga committed 2026-05-13 08:04:13 +02:00
1 parent 7f3dcdf02f
commit 1a450aaa70
20 files changed
+1136 -1

No files matched your search

@@ -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/<provider>.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/<provider>.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.
+10 -1
View File
@@ -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(
View File
Whitespace-only changes.
+162
View File
@@ -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
+34
View File
@@ -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
+17
View File
@@ -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,
)
+10
View File
@@ -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)
+13
View File
@@ -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)
+9
View File
@@ -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)
+13
View File
@@ -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
+24
View File
@@ -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)
+9
View File
@@ -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)
+28
View File
@@ -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,
}
+61
View File
@@ -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
+15
View File
@@ -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
View File
Whitespace-only changes.
+218
View File
@@ -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
@@ -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) == {}
@@ -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
+175
View File
@@ -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