From ed17dc4b09d17a8d6a9b8c559b31f494e0956aa1 Mon Sep 17 00:00:00 2001 From: pedrooot Date: Tue, 9 Jun 2026 16:34:37 +0200 Subject: [PATCH] feat(sdk): scan configuration schema and validation for Prowler Cloud --- prowler/config/scan_config_schema.py | 106 +++++ prowler/config/schema/aws.py | 417 +++++++++++++++--- prowler/config/schema/azure.py | 93 +++- prowler/config/schema/cloudflare.py | 12 +- prowler/config/schema/gcp.py | 42 +- prowler/config/schema/github.py | 13 +- prowler/config/schema/kubernetes.py | 42 +- prowler/config/schema/m365.py | 56 ++- prowler/config/schema/mongodbatlas.py | 12 +- prowler/config/schema/vercel.py | 60 ++- tests/config/schema/aws_schema_test.py | 6 +- tests/config/schema/bounds_test.py | 398 +++++++++++++++++ .../schema/other_providers_schema_test.py | 16 +- 13 files changed, 1149 insertions(+), 124 deletions(-) create mode 100644 prowler/config/scan_config_schema.py create mode 100644 tests/config/schema/bounds_test.py diff --git a/prowler/config/scan_config_schema.py b/prowler/config/scan_config_schema.py new file mode 100644 index 0000000000..d4dd694e04 --- /dev/null +++ b/prowler/config/scan_config_schema.py @@ -0,0 +1,106 @@ +"""Bridge between the Pydantic-based provider schemas in +`prowler.config.schema` and the Prowler App backend (Django) + UI. + +The SDK runtime is intentionally LENIENT: invalid keys are dropped with a +warning and downstream checks fall back to their defaults +(`prowler.config.schema.validator.validate_provider_config`). + +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`. + +- `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. +""" + +from typing import Any + +from pydantic import ValidationError + +from prowler.config.schema.registry import SCHEMAS + + +def _format_loc(loc: tuple) -> str: + """Render a Pydantic error location as `key[idx].nested`.""" + parts: list[str] = [] + for piece in loc: + if isinstance(piece, int): + if parts: + parts[-1] = f"{parts[-1]}[{piece}]" + else: + parts.append(f"[{piece}]") + else: + parts.append(str(piece)) + return ".".join(parts) if parts else "" + + +def validate_scan_config(payload: Any) -> list[dict]: + """Validate a scan config payload against the registered provider schemas. + + 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). + """ + if not isinstance(payload, dict): + return [ + { + "path": "", + "message": "Scan config must be a mapping with provider sections.", + } + ] + + errors: list[dict] = [] + for provider, section in payload.items(): + schema_cls = SCHEMAS.get(provider) + if schema_cls is None: + # Unknown provider type: tolerated. The SDK will simply ignore it. + continue + if not isinstance(section, dict): + errors.append( + { + "path": str(provider), + "message": "section must be a mapping.", + } + ) + continue + try: + 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"), + } + ) + return errors + + +def _build_aggregated_schema() -> dict: + """Compose one JSON Schema per provider into a single top-level schema. + + The output mirrors the layout of `prowler/config/config.yaml` (a mapping + keyed by provider type) and is what the UI consumes via `ajv`. + """ + properties: dict[str, dict] = {} + for provider, schema_cls in SCHEMAS.items(): + properties[provider] = schema_cls.model_json_schema() + return { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "Prowler Scan Config", + "type": "object", + "additionalProperties": True, + "properties": properties, + } + + +SCAN_CONFIG_SCHEMA: dict = _build_aggregated_schema() diff --git a/prowler/config/schema/aws.py b/prowler/config/schema/aws.py index d1eb74240f..7f61e5ad11 100644 --- a/prowler/config/schema/aws.py +++ b/prowler/config/schema/aws.py @@ -1,3 +1,17 @@ +"""AWS provider config schema. + +Bounds on every field are intentionally conservative: they are not the +absolute service maxima but the values that produce a useful security +check. A user is free to keep the built-in default by omitting the key — +out-of-range values are dropped with a warning at SDK runtime, and +rejected at the Prowler App backend. + +Whenever an upper bound is uncertain, the cap is set to a value that +still keeps the check meaningful (e.g. a 10-year window for date-based +thresholds) and avoids ints that obviously break downstream maths +(`min_kinesis_stream_retention_hours = 99999`). +""" + from ipaddress import ip_network from typing import Annotated, Literal, Optional @@ -5,25 +19,68 @@ from pydantic import AfterValidator, Field from prowler.config.schema.base import ProviderConfigBase +# ---- Reusable constants ----------------------------------------------------- -class _DetectSecretsPlugin(ProviderConfigBase): - """One entry inside ``detect_secrets_plugins``. +# CloudWatch Logs only accepts these retention values (in days). Anything else +# is silently coerced to the next valid value by the API — we reject upfront. +_CLOUDWATCH_RETENTION_DAYS = ( + 1, + 3, + 5, + 7, + 14, + 30, + 60, + 90, + 120, + 150, + 180, + 365, + 400, + 545, + 731, + 1827, + 2192, + 2557, + 2922, + 3288, + 3653, +) - 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. - """ +_VALID_CW_RETENTION_LITERAL = Literal[ + 1, + 3, + 5, + 7, + 14, + 30, + 60, + 90, + 120, + 150, + 180, + 365, + 400, + 545, + 731, + 1827, + 2192, + 2557, + 2922, + 3288, + 3653, +] - name: str - limit: Optional[float] = None + +# ---- Custom validators ------------------------------------------------------ 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") + if not 1 <= port <= 65535: + raise ValueError(f"port {port} is outside the valid range 1..65535") return v @@ -51,112 +108,342 @@ def _validate_trusted_ips(v: Optional[list[str]]) -> Optional[list[str]]: return v +def _validate_semver(v: Optional[str]) -> Optional[str]: + """Accept "1.4.0" style strings (used by Fargate platform versions).""" + if v is None: + return v + parts = v.split(".") + if len(parts) != 3 or not all(p.isdigit() for p in parts): + raise ValueError(f"{v!r} is not a valid semantic version (expected X.Y.Z)") + return v + + +def _validate_eks_minor(v: Optional[str]) -> Optional[str]: + """Accept "1.28" style strings (EKS minor versions).""" + if v is None: + return v + parts = v.split(".") + if len(parts) != 2 or not all(p.isdigit() for p in parts): + raise ValueError(f"{v!r} is not a valid EKS version (expected X.Y)") + return v + + +# ---- Nested models ---------------------------------------------------------- + + +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] = Field( + default=None, + ge=0.0, + le=10.0, + description=( + "Entropy threshold for detect-secrets entropy plugins. Range: 0..10 " + "(Shannon entropy is bounded by log2(256)=8; >10 is meaningless)." + ), + ) + + +# ---- Main schema ------------------------------------------------------------ + + class AWSProviderConfig(ProviderConfigBase): - # IAM + # --- 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) + max_unused_access_keys_days: Optional[int] = Field( + default=None, + ge=30, + le=180, + description=( + "Days an IAM user access key can stay unused before being flagged. " + "Range: 30..180 days (CIS AWS 1.13 recommends 45; NIST IA-5 ≤90)." + ), + ) + max_console_access_days: Optional[int] = Field( + default=None, + ge=30, + le=180, + description=( + "Days an IAM console password can stay unused before being flagged. " + "Range: 30..180 days (CIS AWS 1.12 recommends 45)." + ), + ) + max_unused_sagemaker_access_days: Optional[int] = Field( + default=None, + ge=7, + le=180, + description=( + "Days a SageMaker user access key can stay unused. Range: 7..180 " + "(SageMaker tokens are usually high-privilege over S3/KMS)." + ), + ) - # 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 --------------------------------------------------------------- + shodan_api_key: Optional[str] = Field( + default=None, + max_length=512, + description="API key for Shodan lookups on EC2 public IPs.", + ) + max_security_group_rules: Optional[int] = Field( + default=None, + ge=1, + le=1000, + description="Max ingress+egress rules per security group. AWS hard limit is 1000.", + ) + max_ec2_instance_age_in_days: Optional[int] = Field( + default=None, + ge=1, + le=1095, + description=( + "Days an EC2 instance can run before being flagged as old. " + "Range: 1..1095 (3 years; instances should be refreshed for patching " + "per NIST CM-3 — anything older is a security smell)." + ), + ) 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 + ] = Field( + default=None, + description="TCP/UDP ports considered high-risk when reachable from the Internet (1..65535; port 0 is reserved).", + ) - # ECS - fargate_linux_latest_version: Optional[str] = None - fargate_windows_latest_version: Optional[str] = None + # --- ECS --------------------------------------------------------------- + fargate_linux_latest_version: Annotated[ + Optional[str], AfterValidator(_validate_semver) + ] = Field(default=None, description="Fargate Linux platform version (X.Y.Z).") + fargate_windows_latest_version: Annotated[ + Optional[str], AfterValidator(_validate_semver) + ] = Field(default=None, description="Fargate Windows platform version (X.Y.Z).") - # Cross-account trust + # --- Cross-account trust ---------------------------------------------- trusted_account_ids: Annotated[ Optional[list[str]], AfterValidator(_validate_account_ids) - ] = None + ] = Field( + default=None, + description="Additional 12-digit AWS account IDs trusted by cross-account checks.", + ) trusted_ips: Annotated[ Optional[list[str]], AfterValidator(_validate_trusted_ips) - ] = None + ] = Field( + default=None, + description="IPv4/IPv6 addresses or CIDR ranges that are NOT considered public.", + ) - # CloudWatch / CloudFormation - log_group_retention_days: Optional[int] = Field(default=None, gt=0) - recommended_cdk_bootstrap_version: Optional[int] = Field(default=None, gt=0) + # --- CloudWatch / CloudFormation -------------------------------------- + log_group_retention_days: Optional[_VALID_CW_RETENTION_LITERAL] = Field( + default=None, + description=( + "Required CloudWatch Logs retention in days. Must match one of the " + f"values accepted by the AWS API: {list(_CLOUDWATCH_RETENTION_DAYS)}." + ), + ) + recommended_cdk_bootstrap_version: Optional[int] = Field( + default=None, + ge=1, + le=100, + description="Min CDK bootstrap version expected on the account.", + ) - # 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) + # --- AppStream -------------------------------------------------------- + max_idle_disconnect_timeout_in_seconds: Optional[int] = Field( + default=None, + ge=60, + le=1800, + description=( + "AppStream idle disconnect timeout (seconds). Range: 60..1800 " + "(NIST AC-12: sensitive sessions ≤15 min — cap at 30 min)." + ), + ) + max_disconnect_timeout_in_seconds: Optional[int] = Field( + default=None, + ge=60, + le=3600, + description="AppStream disconnect timeout (seconds). Range: 60..3600.", + ) + max_session_duration_seconds: Optional[int] = Field( + default=None, + ge=600, + le=86400, + description=( + "AppStream max session duration (seconds). Range: 600..86400 " + "(10 min .. 24 h — AWS AppStream hard limit per session)." + ), + ) - # Lambda + # --- Lambda ----------------------------------------------------------- obsolete_lambda_runtimes: Optional[list[str]] = None - lambda_min_azs: Optional[int] = Field(default=None, gt=0) + lambda_min_azs: Optional[int] = Field( + default=None, + ge=1, + le=6, + description="Min number of AZs a VPC-bound Lambda must span. Range: 1..6.", + ) - # Organizations + # --- Organizations ---------------------------------------------------- organizations_enabled_regions: Optional[list[str]] = None - organizations_trusted_delegated_administrators: Optional[list[str]] = None + organizations_trusted_delegated_administrators: Annotated[ + Optional[list[str]], AfterValidator(_validate_account_ids) + ] = None organizations_trusted_ids: Optional[list[str]] = None - # ECR + # --- ECR -------------------------------------------------------------- ecr_repository_vulnerability_minimum_severity: Optional[ - Literal["CRITICAL", "HIGH", "MEDIUM", "LOW"] - ] = None + Literal["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"] + ] = Field( + default=None, + description="Highest severity tolerated for ECR images.", + ) - # Trusted Advisor + # --- Trusted Advisor -------------------------------------------------- verify_premium_support_plans: Optional[bool] = None - # CloudTrail threat detection + # --- CloudTrail threat detection: privilege escalation ---------------- threat_detection_privilege_escalation_threshold: Optional[float] = Field( - default=None, ge=0.0, le=1.0 + default=None, + ge=0.0, + le=1.0, + description="Fraction of suspicious actions that triggers the priv-esc detection.", ) threat_detection_privilege_escalation_minutes: Optional[int] = Field( - default=None, gt=0 + default=None, + ge=5, + le=43200, + description=( + "Lookback window (minutes) for priv-esc detection. Range: 5..43200 " + "(under 5 min the signal is dominated by false positives)." + ), ) threat_detection_privilege_escalation_actions: Optional[list[str]] = None + # --- CloudTrail threat detection: enumeration ------------------------- threat_detection_enumeration_threshold: Optional[float] = Field( - default=None, ge=0.0, le=1.0 + default=None, + ge=0.0, + le=1.0, + description="Fraction of suspicious actions that triggers the enumeration detection.", + ) + threat_detection_enumeration_minutes: Optional[int] = Field( + default=None, + ge=5, + le=43200, + description="Lookback window (minutes) for enumeration detection. Range: 5..43200.", ) - threat_detection_enumeration_minutes: Optional[int] = Field(default=None, gt=0) threat_detection_enumeration_actions: Optional[list[str]] = None + # --- CloudTrail threat detection: LLM jacking ------------------------- threat_detection_llm_jacking_threshold: Optional[float] = Field( - default=None, ge=0.0, le=1.0 + default=None, + ge=0.0, + le=1.0, + description="Fraction of suspicious actions that triggers the LLM-jacking detection.", + ) + threat_detection_llm_jacking_minutes: Optional[int] = Field( + default=None, + ge=5, + le=43200, + description="Lookback window (minutes) for LLM-jacking detection. Range: 5..43200.", ) - threat_detection_llm_jacking_minutes: Optional[int] = Field(default=None, gt=0) threat_detection_llm_jacking_actions: Optional[list[str]] = None - # RDS + # --- RDS -------------------------------------------------------------- check_rds_instance_replicas: Optional[bool] = None - # ACM - days_to_expire_threshold: Optional[int] = Field(default=None, gt=0) + # --- ACM -------------------------------------------------------------- + days_to_expire_threshold: Optional[int] = Field( + default=None, + ge=7, + le=365, + description=( + "Days before certificate expiration to flag. Range: 7..365 " + "(PCI-DSS 4.2.1.1: alert ≥30 days before expiry; <7 days is too " + "tight to actually act on)." + ), + ) insecure_key_algorithms: Optional[list[str]] = None - # EKS - eks_required_log_types: Optional[list[str]] = None - eks_cluster_oldest_version_supported: Optional[str] = None + # --- EKS -------------------------------------------------------------- + eks_required_log_types: Optional[ + list[ + Literal[ + "api", + "audit", + "authenticator", + "controllerManager", + "scheduler", + ] + ] + ] = Field( + default=None, + description="EKS control plane log types that must be enabled.", + ) + eks_cluster_oldest_version_supported: Annotated[ + Optional[str], AfterValidator(_validate_eks_minor) + ] = Field( + default=None, + description='Minimum supported EKS minor version, expected as "X.Y".', + ) - # CodeBuild + # --- 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) + # --- ELB / ELBv2 ------------------------------------------------------ + elb_min_azs: Optional[int] = Field( + default=None, + ge=1, + le=6, + description="Min AZs a Classic ELB must span. Range: 1..6.", + ) + elbv2_min_azs: Optional[int] = Field( + default=None, + ge=1, + le=6, + description="Min AZs an Application/Network LB must span. Range: 1..6.", + ) - # ElastiCache - minimum_snapshot_retention_period: Optional[int] = Field(default=None, gt=0) + # --- ElastiCache ----------------------------------------------------- + minimum_snapshot_retention_period: Optional[int] = Field( + default=None, + ge=1, + le=35, + description="Days an ElastiCache backup must be retained. Range: 1..35 (service hard limit).", + ) - # Secrets + # --- 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) + max_days_secret_unused: Optional[int] = Field( + default=None, + ge=7, + le=365, + description="Days a Secrets Manager secret can stay unused. Range: 7..365.", + ) + max_days_secret_unrotated: Optional[int] = Field( + default=None, + ge=1, + le=180, + description=( + "Days a Secrets Manager secret can go without rotation. Range: 1..180 " + "(NIST IA-5: rotate quarterly; CIS recommends ≤90)." + ), + ) - # Kinesis - min_kinesis_stream_retention_hours: Optional[int] = Field(default=None, gt=0) + # --- Kinesis --------------------------------------------------------- + min_kinesis_stream_retention_hours: Optional[int] = Field( + default=None, + ge=24, + le=8760, + description="Hours of Kinesis stream retention. Range: 24..8760 (1 day .. 1 year).", + ) - # detect-secrets plugins + # --- detect-secrets plugin list ------------------------------------- detect_secrets_plugins: Optional[list[_DetectSecretsPlugin]] = None diff --git a/prowler/config/schema/azure.py b/prowler/config/schema/azure.py index 2b04ccb2e9..26953d8029 100644 --- a/prowler/config/schema/azure.py +++ b/prowler/config/schema/azure.py @@ -1,34 +1,91 @@ -from typing import Literal, Optional +"""Azure provider config schema with safety bounds. -from pydantic import Field +Bounds aim for values that produce a meaningful security check; out-of-range +values are dropped (SDK runtime) or rejected (Prowler App backend). +""" + +from typing import Annotated, Literal, Optional + +from pydantic import AfterValidator, Field from prowler.config.schema.base import ProviderConfigBase -class AzureProviderConfig(ProviderConfigBase): - # Network - shodan_api_key: Optional[str] = None +def _validate_dotted_version(v: Optional[str]) -> Optional[str]: + """Accept ``"8.2"``, ``"3.12"``, ``"17"`` style version strings. - # Defender + Used by App Service language version fields where the upstream APIs + accept either ``MAJOR`` or ``MAJOR.MINOR`` notation. + """ + if v is None: + return v + parts = v.split(".") + if not (1 <= len(parts) <= 2) or not all(p.isdigit() for p in parts): + raise ValueError(f"{v!r} is not a valid version (expected 'X' or 'X.Y')") + return v + + +class AzureProviderConfig(ProviderConfigBase): + # --- Network --------------------------------------------------------- + shodan_api_key: Optional[str] = Field( + default=None, + max_length=512, + description="API key for Shodan lookups on Azure public IPs.", + ) + + # --- Defender -------------------------------------------------------- defender_attack_path_minimal_risk_level: Optional[ Literal["Low", "Medium", "High", "Critical"] - ] = None + ] = Field( + default=None, + description="Minimum attack-path risk level worth a notification.", + ) - # App Service - php_latest_version: Optional[str] = None - python_latest_version: Optional[str] = None - java_latest_version: Optional[str] = None + # --- App Service ---------------------------------------------------- + php_latest_version: Annotated[ + Optional[str], AfterValidator(_validate_dotted_version) + ] = Field(default=None, description='PHP minimum acceptable version, e.g. "8.2".') + python_latest_version: Annotated[ + Optional[str], AfterValidator(_validate_dotted_version) + ] = Field( + default=None, description='Python minimum acceptable version, e.g. "3.12".' + ) + java_latest_version: Annotated[ + Optional[str], AfterValidator(_validate_dotted_version) + ] = Field(default=None, description='Java minimum acceptable version, e.g. "17".') - # SQL - recommended_minimal_tls_versions: Optional[list[str]] = None + # --- SQL ------------------------------------------------------------ + recommended_minimal_tls_versions: Optional[list[Literal["1.2", "1.3"]]] = Field( + default=None, + description="TLS versions accepted on Azure SQL Server.", + ) - # Virtual Machines + # --- Virtual Machines ----------------------------------------------- desired_vm_sku_sizes: Optional[list[str]] = None - vm_backup_min_daily_retention_days: Optional[int] = Field(default=None, gt=0) + vm_backup_min_daily_retention_days: Optional[int] = Field( + default=None, + ge=7, + le=9999, + description=( + "Min daily backup retention days. Range: 7..9999 " + "(Azure Backup hard limit; <7 days defeats DR/ransomware recovery)." + ), + ) - # API Management threat detection + # --- API Management threat detection (LLM jacking) ----------------- apim_threat_detection_llm_jacking_threshold: Optional[float] = Field( - default=None, ge=0.0, le=1.0 + default=None, + ge=0.0, + le=1.0, + description="Fraction of suspicious actions that triggers the detection.", + ) + apim_threat_detection_llm_jacking_minutes: Optional[int] = Field( + default=None, + ge=5, + le=43200, + description=( + "Lookback window (minutes) for LLM-jacking detection. Range: 5..43200 " + "(under 5 min the signal is dominated by false positives)." + ), ) - 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/cloudflare.py b/prowler/config/schema/cloudflare.py index 0b612f047f..914aa754aa 100644 --- a/prowler/config/schema/cloudflare.py +++ b/prowler/config/schema/cloudflare.py @@ -1,3 +1,5 @@ +"""Cloudflare provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,5 +8,11 @@ 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) + max_retries: Optional[int] = Field( + default=None, + ge=0, + le=10, + description=( + "Max retries for Cloudflare API requests. Range: 0..10 (0 disables retries)." + ), + ) diff --git a/prowler/config/schema/gcp.py b/prowler/config/schema/gcp.py index 22b3c29ce3..59e1a162fa 100644 --- a/prowler/config/schema/gcp.py +++ b/prowler/config/schema/gcp.py @@ -1,3 +1,5 @@ +"""GCP provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,8 +8,38 @@ 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) + shodan_api_key: Optional[str] = Field( + default=None, + max_length=512, + description="API key for Shodan lookups on GCP public IPs.", + ) + mig_min_zones: Optional[int] = Field( + default=None, + ge=1, + le=5, + description="Min zones a Managed Instance Group must span. Range: 1..5.", + ) + max_snapshot_age_days: Optional[int] = Field( + default=None, + ge=1, + le=1095, + description=( + "Days a disk snapshot can age before being flagged. Range: 1..1095 " + "(3 years; older snapshots typically miss data-class compliance)." + ), + ) + max_unused_account_days: Optional[int] = Field( + default=None, + ge=30, + le=365, + description=( + "Days a service account or user-managed key can stay unused. " + "Range: 30..365." + ), + ) + storage_min_retention_days: Optional[int] = Field( + default=None, + ge=1, + le=3650, + description="Min retention period on Cloud Storage buckets. Range: 1..3650.", + ) diff --git a/prowler/config/schema/github.py b/prowler/config/schema/github.py index ce1bf2dbb8..30b5a13b85 100644 --- a/prowler/config/schema/github.py +++ b/prowler/config/schema/github.py @@ -1,3 +1,5 @@ +"""GitHub provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,4 +8,13 @@ from prowler.config.schema.base import ProviderConfigBase class GitHubProviderConfig(ProviderConfigBase): - inactive_not_archived_days_threshold: Optional[int] = Field(default=None, gt=0) + inactive_not_archived_days_threshold: Optional[int] = Field( + default=None, + ge=30, + le=3650, + description=( + "Days a repository can stay inactive without being archived before " + "being flagged. Range: 30..3650 (CIS GitHub recommends 180; " + "<30 days produces false positives on seasonal projects)." + ), + ) diff --git a/prowler/config/schema/kubernetes.py b/prowler/config/schema/kubernetes.py index 01612ac5bf..3235971500 100644 --- a/prowler/config/schema/kubernetes.py +++ b/prowler/config/schema/kubernetes.py @@ -1,3 +1,5 @@ +"""Kubernetes provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,8 +8,38 @@ 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 + audit_log_maxbackup: Optional[int] = Field( + default=None, + ge=2, + le=1000, + description=( + "API server audit log file rotations to keep. Range: 2..1000 " + "(CIS Kubernetes 1.2.18 recommends ≥10)." + ), + ) + audit_log_maxsize: Optional[int] = Field( + default=None, + ge=10, + le=10000, + description=( + "Max MB per audit log file before rotation. Range: 10..10000 MB " + "(CIS Kubernetes 1.2.19 recommends ≥100 MB)." + ), + ) + audit_log_maxage: Optional[int] = Field( + default=None, + ge=7, + le=3650, + description=( + "Days an audit log file is retained. Range: 7..3650 " + "(CIS Kubernetes 1.2.17 recommends ≥30 days)." + ), + ) + apiserver_strong_ciphers: Optional[list[str]] = Field( + default=None, + description="Whitelist of strong TLS cipher suites required on the API server.", + ) + kubelet_strong_ciphers: Optional[list[str]] = Field( + default=None, + description="Whitelist of strong TLS cipher suites required on kubelet.", + ) diff --git a/prowler/config/schema/m365.py b/prowler/config/schema/m365.py index 01013ac2ae..ff1ff98285 100644 --- a/prowler/config/schema/m365.py +++ b/prowler/config/schema/m365.py @@ -1,3 +1,5 @@ +"""M365 provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,19 +8,47 @@ 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 + # --- Entra (sign-in policy) ---------------------------------------- + sign_in_frequency: Optional[int] = Field( + default=None, + ge=1, + le=168, + description=( + "Hours between forced sign-ins for admin users. Range: 1..168 (1 h .. 7 days). " + "Microsoft Conditional Access baseline for admin roles is ≤24 h." + ), ) - # Defender malware policy - default_recommended_extensions: Optional[list[str]] = None + # --- Teams --------------------------------------------------------- + allowed_cloud_storage_services: Optional[list[str]] = Field( + default=None, + description="External cloud storage services allowed in Teams.", + ) - # Mailbox auditing - audit_log_age: Optional[int] = Field(default=None, gt=0) + # --- Exchange ------------------------------------------------------ + recommended_mailtips_large_audience_threshold: Optional[int] = Field( + default=None, + ge=5, + le=10000, + description=( + "Recipient count that should trigger a 'large audience' MailTip. " + "Range: 5..10000 (Microsoft default 25)." + ), + ) + + # --- Defender malware policy -------------------------------------- + default_recommended_extensions: Optional[list[str]] = Field( + default=None, + description="File extensions blocked by the malware policy.", + ) + + # --- Mailbox auditing --------------------------------------------- + audit_log_age: Optional[int] = Field( + default=None, + ge=30, + le=3650, + description=( + "Days mailbox audit logs must be retained. Range: 30..3650 " + "(M365 E3 default is 90 days; SEC/FINRA require ≥7 years)." + ), + ) diff --git a/prowler/config/schema/mongodbatlas.py b/prowler/config/schema/mongodbatlas.py index d9a210184c..ab8b9a79d6 100644 --- a/prowler/config/schema/mongodbatlas.py +++ b/prowler/config/schema/mongodbatlas.py @@ -1,3 +1,5 @@ +"""MongoDB Atlas provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,4 +8,12 @@ from prowler.config.schema.base import ProviderConfigBase class MongoDBAtlasProviderConfig(ProviderConfigBase): - max_service_account_secret_validity_hours: Optional[int] = Field(default=None, gt=0) + max_service_account_secret_validity_hours: Optional[int] = Field( + default=None, + ge=1, + le=720, + description=( + "Max hours a service account secret can stay valid. " + "Range: 1..720 (1 h .. 30 days)." + ), + ) diff --git a/prowler/config/schema/vercel.py b/prowler/config/schema/vercel.py index 8d0f672fd3..69e5511a83 100644 --- a/prowler/config/schema/vercel.py +++ b/prowler/config/schema/vercel.py @@ -1,3 +1,5 @@ +"""Vercel provider config schema with safety bounds.""" + from typing import Optional from pydantic import Field @@ -6,10 +8,54 @@ 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 + stable_branches: Optional[list[str]] = Field( + default=None, + description="Branches considered stable for production deployments.", + ) + days_to_expire_threshold: Optional[int] = Field( + default=None, + ge=7, + le=365, + description=( + "Days before token/certificate expiration to flag. Range: 7..365 " + "(PCI-DSS 4.2.1.1: alert ≥30 days before expiry)." + ), + ) + stale_token_threshold_days: Optional[int] = Field( + default=None, + ge=30, + le=3650, + description=( + "Days of inactivity before a token is considered stale. Range: 30..3650 " + "(NIST AC-2(3) typical window 30..90 days)." + ), + ) + stale_invitation_threshold_days: Optional[int] = Field( + default=None, + ge=7, + le=365, + description=( + "Days a pending invitation can stay open. Range: 7..365 " + "(OWASP ASVS 2.7.1 recommends short-lived invitations)." + ), + ) + max_owner_percentage: Optional[int] = Field( + default=None, + ge=1, + le=50, + description=( + "Max percentage of team members that can have the OWNER role. " + "Range: 1..50 (PoLP — having >50% of a team as OWNER defeats RBAC; " + "industry guidance recommends ≤25%)." + ), + ) + max_owners: Optional[int] = Field( + default=None, + ge=1, + le=1000, + description="Absolute max owners (overrides percentage for large teams). Range: 1..1000.", + ) + secret_suffixes: Optional[list[str]] = Field( + default=None, + description="Suffixes that mark a project env var as secret-like.", + ) diff --git a/tests/config/schema/aws_schema_test.py b/tests/config/schema/aws_schema_test.py index 989ebde0e2..9bc7e02d49 100644 --- a/tests/config/schema/aws_schema_test.py +++ b/tests/config/schema/aws_schema_test.py @@ -101,11 +101,15 @@ class Test_AWS_Trusted_Ips: class Test_AWS_Ports: def test_valid_ports_in_range(self): - ports = [25, 80, 443, 65535, 0] + ports = [25, 80, 443, 65535, 1] assert _validate({"ec2_high_risk_ports": ports}) == { "ec2_high_risk_ports": ports } + def test_port_zero_is_dropped(self): + # Port 0 is reserved and not a valid security signal. + assert _validate({"ec2_high_risk_ports": [0]}) == {} + def test_out_of_range_port_is_dropped(self): assert _validate({"ec2_high_risk_ports": [70000]}) == {} diff --git a/tests/config/schema/bounds_test.py b/tests/config/schema/bounds_test.py new file mode 100644 index 0000000000..4235a81582 --- /dev/null +++ b/tests/config/schema/bounds_test.py @@ -0,0 +1,398 @@ +"""Boundary tests for the safety bounds added on top of the upstream schemas. + +Each parametrised case checks (a) the min and max values are accepted and +(b) one step outside the range is rejected. Custom validators (semver, +EKS minor, dotted version, port range, account IDs, IPs) get focused +positive/negative tests. + +Tests use the public adapter ``prowler.config.scan_config_schema``: a +schema violation surfaces as a list of ``{"path", "message"}`` entries. +This keeps the contract the Prowler App backend depends on under test. +""" + +import pytest + +from prowler.config.scan_config_schema import validate_scan_config + + +def _has_error_for(errors, path_substr: str) -> bool: + return any(path_substr in e["path"] for e in errors) + + +# Each tuple: (provider, key, min_allowed, max_allowed) +INT_BOUND_CASES = [ + # AWS + ("aws", "max_unused_access_keys_days", 30, 180), + ("aws", "max_console_access_days", 30, 180), + ("aws", "max_unused_sagemaker_access_days", 7, 180), + ("aws", "max_security_group_rules", 1, 1000), + ("aws", "max_ec2_instance_age_in_days", 1, 1095), + ("aws", "recommended_cdk_bootstrap_version", 1, 100), + ("aws", "max_idle_disconnect_timeout_in_seconds", 60, 1800), + ("aws", "max_disconnect_timeout_in_seconds", 60, 3600), + ("aws", "max_session_duration_seconds", 600, 86400), + ("aws", "lambda_min_azs", 1, 6), + ("aws", "threat_detection_privilege_escalation_minutes", 5, 43200), + ("aws", "threat_detection_enumeration_minutes", 5, 43200), + ("aws", "threat_detection_llm_jacking_minutes", 5, 43200), + ("aws", "days_to_expire_threshold", 7, 365), + ("aws", "elb_min_azs", 1, 6), + ("aws", "elbv2_min_azs", 1, 6), + ("aws", "minimum_snapshot_retention_period", 1, 35), + ("aws", "max_days_secret_unused", 7, 365), + ("aws", "max_days_secret_unrotated", 1, 180), + ("aws", "min_kinesis_stream_retention_hours", 24, 8760), + # Azure + ("azure", "vm_backup_min_daily_retention_days", 7, 9999), + ("azure", "apim_threat_detection_llm_jacking_minutes", 5, 43200), + # GCP + ("gcp", "mig_min_zones", 1, 5), + ("gcp", "max_snapshot_age_days", 1, 1095), + ("gcp", "max_unused_account_days", 30, 365), + ("gcp", "storage_min_retention_days", 1, 3650), + # Kubernetes + ("kubernetes", "audit_log_maxbackup", 2, 1000), + ("kubernetes", "audit_log_maxsize", 10, 10000), + ("kubernetes", "audit_log_maxage", 7, 3650), + # M365 + ("m365", "sign_in_frequency", 1, 168), + ("m365", "recommended_mailtips_large_audience_threshold", 5, 10000), + ("m365", "audit_log_age", 30, 3650), + # GitHub + ("github", "inactive_not_archived_days_threshold", 30, 3650), + # MongoDB Atlas + ("mongodbatlas", "max_service_account_secret_validity_hours", 1, 720), + # Cloudflare + ("cloudflare", "max_retries", 0, 10), + # Vercel + ("vercel", "days_to_expire_threshold", 7, 365), + ("vercel", "stale_token_threshold_days", 30, 3650), + ("vercel", "stale_invitation_threshold_days", 7, 365), + ("vercel", "max_owner_percentage", 1, 50), + ("vercel", "max_owners", 1, 1000), +] + + +FLOAT_THRESHOLD_FIELDS = [ + ("aws", "threat_detection_privilege_escalation_threshold"), + ("aws", "threat_detection_enumeration_threshold"), + ("aws", "threat_detection_llm_jacking_threshold"), + ("azure", "apim_threat_detection_llm_jacking_threshold"), +] + + +class TestIntegerBounds: + """Each int field accepts both ends of its range and rejects ±1 outside.""" + + @pytest.mark.parametrize("provider, key, lo, hi", INT_BOUND_CASES) + def test_min_accepted(self, provider, key, lo, hi): + assert validate_scan_config({provider: {key: lo}}) == [] + + @pytest.mark.parametrize("provider, key, lo, hi", INT_BOUND_CASES) + def test_max_accepted(self, provider, key, lo, hi): + assert validate_scan_config({provider: {key: hi}}) == [] + + @pytest.mark.parametrize("provider, key, lo, hi", INT_BOUND_CASES) + def test_below_min_rejected(self, provider, key, lo, hi): + errors = validate_scan_config({provider: {key: lo - 1}}) + assert _has_error_for(errors, f"{provider}.{key}"), errors + + @pytest.mark.parametrize("provider, key, lo, hi", INT_BOUND_CASES) + def test_above_max_rejected(self, provider, key, lo, hi): + errors = validate_scan_config({provider: {key: hi + 1}}) + assert _has_error_for(errors, f"{provider}.{key}"), errors + + +class TestFloatThresholds: + """Threshold floats must stay within 0..1 inclusive.""" + + @pytest.mark.parametrize("provider, key", FLOAT_THRESHOLD_FIELDS) + def test_zero_and_one_accepted(self, provider, key): + assert validate_scan_config({provider: {key: 0.0}}) == [] + assert validate_scan_config({provider: {key: 1.0}}) == [] + assert validate_scan_config({provider: {key: 0.5}}) == [] + + @pytest.mark.parametrize("provider, key", FLOAT_THRESHOLD_FIELDS) + def test_negative_rejected(self, provider, key): + errors = validate_scan_config({provider: {key: -0.01}}) + assert _has_error_for(errors, f"{provider}.{key}") + + @pytest.mark.parametrize("provider, key", FLOAT_THRESHOLD_FIELDS) + def test_above_one_rejected(self, provider, key): + errors = validate_scan_config({provider: {key: 1.01}}) + assert _has_error_for(errors, f"{provider}.{key}") + + +class TestCloudWatchRetention: + """`log_group_retention_days` only accepts the AWS-approved enum values.""" + + @pytest.mark.parametrize("value", [1, 7, 30, 365, 731, 3653]) + def test_valid_values_accepted(self, value): + assert validate_scan_config({"aws": {"log_group_retention_days": value}}) == [] + + @pytest.mark.parametrize("value", [0, 2, 42, 500, 999, 4000]) + def test_invalid_values_rejected(self, value): + errors = validate_scan_config({"aws": {"log_group_retention_days": value}}) + assert _has_error_for(errors, "aws.log_group_retention_days") + + +class TestSemverValidator: + """AWS Fargate platform versions: X.Y.Z.""" + + @pytest.mark.parametrize("value", ["1.4.0", "1.0.0", "0.0.1", "10.20.30"]) + def test_accepts_semver(self, value): + assert ( + validate_scan_config({"aws": {"fargate_linux_latest_version": value}}) == [] + ) + + @pytest.mark.parametrize("value", ["1.4", "1", "v1.4.0", "1.4.0-beta", "a.b.c", ""]) + def test_rejects_non_semver(self, value): + errors = validate_scan_config({"aws": {"fargate_linux_latest_version": value}}) + assert _has_error_for(errors, "aws.fargate_linux_latest_version") + + +class TestEksVersionValidator: + """`eks_cluster_oldest_version_supported` expects MAJOR.MINOR.""" + + @pytest.mark.parametrize("value", ["1.28", "1.29", "1.30", "2.0"]) + def test_accepts_minor(self, value): + assert ( + validate_scan_config( + {"aws": {"eks_cluster_oldest_version_supported": value}} + ) + == [] + ) + + @pytest.mark.parametrize("value", ["1.28.0", "v1.28", "1", "1.x", ""]) + def test_rejects_invalid(self, value): + errors = validate_scan_config( + {"aws": {"eks_cluster_oldest_version_supported": value}} + ) + assert _has_error_for(errors, "aws.eks_cluster_oldest_version_supported") + + +class TestEksLogTypesEnum: + """Only the documented log types are accepted.""" + + def test_full_enum_accepted(self): + assert ( + validate_scan_config( + { + "aws": { + "eks_required_log_types": [ + "api", + "audit", + "authenticator", + "controllerManager", + "scheduler", + ] + } + } + ) + == [] + ) + + def test_unknown_type_rejected(self): + errors = validate_scan_config( + {"aws": {"eks_required_log_types": ["api", "telemetry"]}} + ) + assert _has_error_for(errors, "aws.eks_required_log_types") + + +class TestAzureDottedVersion: + """App Service versions accept 'X' and 'X.Y' but not 'X.Y.Z' or junk.""" + + @pytest.mark.parametrize("value", ["8.2", "3.12", "17"]) + def test_accepts(self, value): + assert validate_scan_config({"azure": {"php_latest_version": value}}) == [] + assert validate_scan_config({"azure": {"python_latest_version": value}}) == [] + assert validate_scan_config({"azure": {"java_latest_version": value}}) == [] + + @pytest.mark.parametrize("value", ["8.2.0", "v8", "8.x", ""]) + def test_rejects(self, value): + errors = validate_scan_config({"azure": {"php_latest_version": value}}) + assert _has_error_for(errors, "azure.php_latest_version") + + +class TestAzureTlsLiteralEnum: + """Only TLS 1.2 and 1.3 are tolerated by the recommended list.""" + + def test_accepted_versions(self): + assert ( + validate_scan_config( + {"azure": {"recommended_minimal_tls_versions": ["1.2", "1.3"]}} + ) + == [] + ) + + @pytest.mark.parametrize("value", ["1.0", "1.1", "2.0", ""]) + def test_unknown_version_rejected(self, value): + errors = validate_scan_config( + {"azure": {"recommended_minimal_tls_versions": [value]}} + ) + assert _has_error_for(errors, "azure.recommended_minimal_tls_versions") + + +class TestAzureRiskLevelLiteral: + """Defender attack-path risk level is a closed enum.""" + + @pytest.mark.parametrize("value", ["Low", "Medium", "High", "Critical"]) + def test_accepted(self, value): + assert ( + validate_scan_config( + {"azure": {"defender_attack_path_minimal_risk_level": value}} + ) + == [] + ) + + @pytest.mark.parametrize("value", ["low", "CRITICAL", "Severe", ""]) + def test_rejected(self, value): + errors = validate_scan_config( + {"azure": {"defender_attack_path_minimal_risk_level": value}} + ) + assert _has_error_for(errors, "azure.defender_attack_path_minimal_risk_level") + + +class TestECRSeverityLiteral: + """ECR severity is a closed enum (with INFORMATIONAL allowed).""" + + @pytest.mark.parametrize( + "value", + ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFORMATIONAL"], + ) + def test_accepted(self, value): + assert ( + validate_scan_config( + {"aws": {"ecr_repository_vulnerability_minimum_severity": value}} + ) + == [] + ) + + @pytest.mark.parametrize("value", ["URGENT", "low", "Crit", ""]) + def test_rejected(self, value): + errors = validate_scan_config( + {"aws": {"ecr_repository_vulnerability_minimum_severity": value}} + ) + assert _has_error_for( + errors, "aws.ecr_repository_vulnerability_minimum_severity" + ) + + +class TestPortRangeValidator: + """Each entry of `ec2_high_risk_ports` must be 1..65535 (0 is reserved).""" + + def test_valid_ports(self): + assert ( + validate_scan_config({"aws": {"ec2_high_risk_ports": [1, 22, 8080, 65535]}}) + == [] + ) + + @pytest.mark.parametrize("value", [-1, 0, 65536, 99999]) + def test_invalid_port_rejected(self, value): + errors = validate_scan_config({"aws": {"ec2_high_risk_ports": [80, value]}}) + assert _has_error_for(errors, "aws.ec2_high_risk_ports") + + +class TestAccountIdsValidator: + """AWS account IDs are 12-digit strings.""" + + def test_valid(self): + assert ( + validate_scan_config( + {"aws": {"trusted_account_ids": ["123456789012", "098765432109"]}} + ) + == [] + ) + + @pytest.mark.parametrize( + "value", ["12345", "12345678901", "1234567890123", "12345678901a"] + ) + def test_invalid_rejected(self, value): + errors = validate_scan_config({"aws": {"trusted_account_ids": [value]}}) + assert _has_error_for(errors, "aws.trusted_account_ids") + + +class TestTrustedIpsValidator: + """Trusted IPs accept IPv4, IPv6, and CIDR; reject junk.""" + + @pytest.mark.parametrize( + "value", + ["1.2.3.4", "10.0.0.0/8", "2001:db8::1", "2001:db8::/32"], + ) + def test_valid(self, value): + assert validate_scan_config({"aws": {"trusted_ips": [value]}}) == [] + + @pytest.mark.parametrize( + "value", ["not.an.ip", "1.2.3.300", "10.0.0.0/40", "::ffff:::"] + ) + def test_invalid_rejected(self, value): + errors = validate_scan_config({"aws": {"trusted_ips": [value]}}) + assert _has_error_for(errors, "aws.trusted_ips") + + +class TestDetectSecretsEntropyBound: + """`detect_secrets_plugins[].limit` is Shannon entropy: 0..10.""" + + @pytest.mark.parametrize("value", [0.0, 3.5, 4.5, 8.0, 10.0]) + def test_valid(self, value): + assert ( + validate_scan_config( + { + "aws": { + "detect_secrets_plugins": [ + {"name": "Base64HighEntropyString", "limit": value} + ] + } + } + ) + == [] + ) + + @pytest.mark.parametrize("value", [-0.1, 10.01, 50]) + def test_invalid(self, value): + errors = validate_scan_config( + { + "aws": { + "detect_secrets_plugins": [ + {"name": "Base64HighEntropyString", "limit": value} + ] + } + } + ) + assert _has_error_for(errors, "aws.detect_secrets_plugins") + + +class TestAdapterRobustness: + """Top-level adapter behaviour the Prowler App backend depends on.""" + + def test_non_dict_payload(self): + errors = validate_scan_config([1, 2, 3]) + assert len(errors) == 1 + assert errors[0]["path"] == "" + + def test_unknown_provider_section_tolerated(self): + # additionalProperties: True at the root level by design. + assert validate_scan_config({"newprovider": {"foo": "bar"}}) == [] + + def test_unknown_key_tolerated_by_pydantic_extra_allow(self): + # ProviderConfigBase has extra="allow" for forward compatibility. + assert validate_scan_config({"aws": {"completely_new_knob": 1}}) == [] + + def test_provider_section_must_be_mapping(self): + errors = validate_scan_config({"aws": "not a mapping"}) + assert _has_error_for(errors, "aws") + + def test_multiple_errors_surfaced(self): + errors = validate_scan_config( + { + "aws": { + "max_unused_access_keys_days": 5, # below min 30 + "max_security_group_rules": 99999, # above max 1000 + "ec2_high_risk_ports": [80, 70000], # port out of range + } + } + ) + # All three should surface independently. + assert _has_error_for(errors, "aws.max_unused_access_keys_days") + assert _has_error_for(errors, "aws.max_security_group_rules") + assert _has_error_for(errors, "aws.ec2_high_risk_ports") diff --git a/tests/config/schema/other_providers_schema_test.py b/tests/config/schema/other_providers_schema_test.py index 163be36204..f4dc5184ab 100644 --- a/tests/config/schema/other_providers_schema_test.py +++ b/tests/config/schema/other_providers_schema_test.py @@ -122,17 +122,21 @@ class Test_Vercel_Schema: 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": 1}) == { + "max_owner_percentage": 1 } - assert _validate("vercel", {"max_owner_percentage": 100}) == { - "max_owner_percentage": 100 + assert _validate("vercel", {"max_owner_percentage": 50}) == { + "max_owner_percentage": 50 } - def test_owner_percentage_over_100_dropped(self): + def test_owner_percentage_over_max_dropped(self): + # Tightened to 1..50 — anything above (incl. previous 100) is dropped. + assert _validate("vercel", {"max_owner_percentage": 51}) == {} assert _validate("vercel", {"max_owner_percentage": 150}) == {} - def test_owner_percentage_negative_dropped(self): + def test_owner_percentage_zero_or_negative_dropped(self): + # 0 is no longer a valid configuration (defeats PoLP signal). + assert _validate("vercel", {"max_owner_percentage": 0}) == {} assert _validate("vercel", {"max_owner_percentage": -1}) == {} def test_full_default_config_round_trip(self):