mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-27 22:04:20 +00:00
feat(sdk): scan configuration schema and validation for Prowler Cloud
This commit is contained in:
13 files changed
+1149
-124
No files matched your search
@@ -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 "<root>"
|
||||
|
||||
|
||||
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": "<root>",
|
||||
"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()
|
||||
+352
-65
@@ -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
|
||||
@@ -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
|
||||
@@ -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)."
|
||||
),
|
||||
)
|
||||
@@ -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.",
|
||||
)
|
||||
@@ -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)."
|
||||
),
|
||||
)
|
||||
@@ -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.",
|
||||
)
|
||||
@@ -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)."
|
||||
),
|
||||
)
|
||||
@@ -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)."
|
||||
),
|
||||
)
|
||||
@@ -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.",
|
||||
)
|
||||
@@ -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]}) == {}
|
||||
|
||||
|
||||
@@ -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"] == "<root>"
|
||||
|
||||
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")
|
||||
@@ -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):
|
||||
|
||||
Reference in new issue
Block a user