diff --git a/prowler/.env.sample b/prowler/.env.sample index 30a08961..15b351a1 100644 --- a/prowler/.env.sample +++ b/prowler/.env.sample @@ -8,3 +8,6 @@ OPENAEV_TENANT_ID=ChangeMe INJECTOR_ID=ChangeMe INJECTOR_NAME=Prowler INJECTOR_LOG_LEVEL=error + +# Prowler runtime (must be a nonblank absolute path) +PROWLER_EXECUTABLE_PATH=/usr/local/bin/prowler diff --git a/prowler/README.md b/prowler/README.md index 67a36038..64416517 100644 --- a/prowler/README.md +++ b/prowler/README.md @@ -1,8 +1,8 @@ # OpenAEV Prowler Injector -The Prowler injector foundation registers Prowler with OpenAEV. Assessment -providers, credentials, execution, mapping, routes, and contracts are outside -CHK.001 and are intentionally not configured here. +The Prowler injector foundation registers Prowler with OpenAEV. Provider and +account inputs, credentials, execution, mapping, routes, and contracts are not +startup configuration and remain outside this chunk. ## Configuration @@ -14,10 +14,15 @@ CHK.001 and are intentionally not configured here. | `INJECTOR_ID` | `injector.id` | Unique injector identifier | | `INJECTOR_NAME` | `injector.name` | Injector display name | | `INJECTOR_LOG_LEVEL` | `injector.log_level` | Runtime log level | +| `PROWLER_EXECUTABLE_PATH` | `prowler.executable_path` | Absolute path to the Prowler executable (default: `/usr/local/bin/prowler`) | Copy `config.yml.sample` to the ignored `config.yml` for local use, or supply the equivalent environment variables. Never commit real tokens. +`prowler.executable_path` must be nonblank and absolute. Startup does not +require the file to exist; executable resolution happens immediately before a +future assessment execution. + ## Run ```shell @@ -26,3 +31,18 @@ python -m prowler The CHK.001 foundation starts with zero assessment contracts. Contract catalog registration is deferred to CHK.006. + +## Provider input boundary + +Provider selection, account or target values, and credentials are not injector +startup configuration. They will be supplied per OpenAEV form contract so that +credential changes do not require redeploying the injector. CHK.002 provides +only reusable, strict provider input models; it does not register forms, routes, +or contracts. + +`aws_endpoint_url` is an optional per-assessment provider input for AWS, like +its credentials. When supplied, it must be an absolute HTTP or HTTPS URL with a +host and must not contain user information, a query, a fragment, or whitespace. +Paths and valid ports are allowed, including endpoints on localhost, private +networks, and container services. The accepted value remains an ordinary string, +and validation does not check network reachability. diff --git a/prowler/config.yml.sample b/prowler/config.yml.sample index 91497583..fc6f4274 100644 --- a/prowler/config.yml.sample +++ b/prowler/config.yml.sample @@ -8,3 +8,7 @@ injector: id: 'ChangeMe' name: 'Prowler' log_level: 'error' + +prowler: + # Must be a nonblank absolute path. Existence is checked only before execution. + executable_path: '/usr/local/bin/prowler' diff --git a/prowler/prowler/models/configs/config_loader.py b/prowler/prowler/models/configs/config_loader.py index 81ef30e2..7c4293bc 100644 --- a/prowler/prowler/models/configs/config_loader.py +++ b/prowler/prowler/models/configs/config_loader.py @@ -1,6 +1,8 @@ """Configuration foundation for the Prowler injector.""" -from pydantic import Field +from pathlib import Path + +from pydantic import BaseModel, Field, field_validator from pyoaev.configuration import ( ConfigLoaderCollector, ConfigLoaderOAEV, @@ -18,11 +20,31 @@ class InjectorConfig(ConfigLoaderCollector): ) +class ProwlerConfig(BaseModel): + """Prowler command runtime settings.""" + + executable_path: Path = Field( + default=Path("/usr/local/bin/prowler"), + description="Absolute path to the Prowler executable.", + ) + + @field_validator("executable_path", mode="before") + @classmethod + def validate_executable_path(cls, value: object) -> object: + """Reject blank and non-absolute executable paths.""" + if isinstance(value, str) and not value.strip(): + raise ValueError("executable path must not be blank") + if isinstance(value, (str, Path)) and not Path(value).is_absolute(): + raise ValueError("executable path must be absolute") + return value + + class ConfigLoader(SettingsLoader): - """Load only the standard OpenAEV and injector settings.""" + """Load standard settings and the Prowler runtime section.""" openaev: ConfigLoaderOAEV = Field(default_factory=ConfigLoaderOAEV) injector: InjectorConfig = Field(default_factory=InjectorConfig) + prowler: ProwlerConfig = Field(default_factory=ProwlerConfig) def to_daemon_config(self) -> Configuration: """Translate settings into the OpenAEV daemon configuration.""" diff --git a/prowler/prowler/models/provider_inputs.py b/prowler/prowler/models/provider_inputs.py new file mode 100644 index 00000000..ebc47550 --- /dev/null +++ b/prowler/prowler/models/provider_inputs.py @@ -0,0 +1,134 @@ +"""Strict, secret-safe provider inputs for future OpenAEV form contracts.""" + +from typing import Annotated, Any, Literal, NoReturn +from urllib.parse import urlsplit + +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + SecretStr, + TypeAdapter, + field_validator, +) + + +def _reject_blank(value: object) -> object: + """Reject blank strings before they can enter a provider field.""" + if isinstance(value, str) and not value.strip(): + raise ValueError("value must not be blank") + return value + + +NonBlankStr = Annotated[str, BeforeValidator(_reject_blank)] +NonBlankSecretStr = Annotated[SecretStr, BeforeValidator(_reject_blank)] + + +class ImmutableProviderInput(BaseModel): + """Provider boundary that rejects assignment without retaining its value.""" + + model_config = ConfigDict( + extra="forbid", strict=True, hide_input_in_errors=True, frozen=True + ) + + def __setattr__(self, name: str, value: Any) -> NoReturn: + """Reject all post-construction assignment with a value-free error.""" + raise TypeError("Provider inputs are immutable") + + +class AwsProviderInput(ImmutableProviderInput): + """AWS provider form input.""" + + model_config = ConfigDict( + extra="forbid", strict=True, hide_input_in_errors=True, frozen=True + ) + + provider: Literal["aws"] + aws_access_key_id: NonBlankStr + aws_secret_access_key: NonBlankSecretStr + aws_account_id: NonBlankStr + aws_region: NonBlankStr + aws_session_token: NonBlankSecretStr | None = None + aws_endpoint_url: str | None = None + + @field_validator("aws_endpoint_url", mode="before") + @classmethod + def validate_aws_endpoint_url(cls, value: object) -> object: + """Accept only absolute HTTP(S) endpoints without unsafe URL extras.""" + if value is None: + return None + if not isinstance(value, str): + raise ValueError("AWS endpoint URL must be a string") + if not value.strip(): + raise ValueError("AWS endpoint URL must not be blank") + if any(character.isspace() for character in value): + raise ValueError("AWS endpoint URL must not contain whitespace") + if "?" in value or "#" in value: + raise ValueError("AWS endpoint URL must not include query or fragment") + try: + endpoint = urlsplit(value) + if endpoint.netloc.rsplit("@", maxsplit=1)[-1].endswith(":"): + raise ValueError("AWS endpoint URL port must not be empty") + port = endpoint.port + except ValueError as error: + raise ValueError("AWS endpoint URL must be valid") from error + if endpoint.scheme not in {"http", "https"}: + raise ValueError("AWS endpoint URL must use HTTP or HTTPS") + if not endpoint.netloc or endpoint.hostname is None: + raise ValueError("AWS endpoint URL must include a host") + if endpoint.username is not None or endpoint.password is not None: + raise ValueError("AWS endpoint URL must not include user information") + if port is not None and not 1 <= port <= 65535: + raise ValueError("AWS endpoint URL port must be between 1 and 65535") + return value + + +class AzureProviderInput(ImmutableProviderInput): + """Azure provider form input.""" + + model_config = ConfigDict( + extra="forbid", strict=True, hide_input_in_errors=True, frozen=True + ) + + provider: Literal["azure"] + azure_tenant_id: NonBlankStr + azure_client_id: NonBlankStr + azure_client_secret: NonBlankSecretStr + azure_subscription_id: NonBlankStr + azure_provider: NonBlankStr + + +class GcpProviderInput(ImmutableProviderInput): + """GCP provider form input.""" + + model_config = ConfigDict( + extra="forbid", strict=True, hide_input_in_errors=True, frozen=True + ) + + provider: Literal["gcp"] + gcp_service_account_json: NonBlankSecretStr + gcp_project_id: NonBlankStr + + +class KubernetesProviderInput(ImmutableProviderInput): + """Kubernetes provider form input.""" + + model_config = ConfigDict( + extra="forbid", strict=True, hide_input_in_errors=True, frozen=True + ) + + provider: Literal["kubernetes"] + kubernetes_kubeconfig: NonBlankSecretStr + kubernetes_context: NonBlankStr + + +ProviderInput = Annotated[ + AwsProviderInput | AzureProviderInput | GcpProviderInput | KubernetesProviderInput, + Field(discriminator="provider"), +] + +PROVIDER_INPUT_ADAPTER: TypeAdapter[ProviderInput] = TypeAdapter( + ProviderInput, + config=ConfigDict(hide_input_in_errors=True), +) diff --git a/prowler/tests/behaviour/chk001_catalog_scaffold/chk001_catalog_scaffold.feature b/prowler/tests/behaviour/chk001_catalog_scaffold/chk001_catalog_scaffold.feature index 0e00c017..1ab4579a 100644 --- a/prowler/tests/behaviour/chk001_catalog_scaffold/chk001_catalog_scaffold.feature +++ b/prowler/tests/behaviour/chk001_catalog_scaffold/chk001_catalog_scaffold.feature @@ -9,7 +9,7 @@ Feature: Prowler catalog registration and project scaffold Scenario: Foundation configuration excludes future provider settings Given CHK.001 owns only the injector configuration foundation When the available configuration is inspected - Then only the standard OpenAEV and injector settings are present + Then only the standard OpenAEV and injector settings plus the Prowler runtime section are present Scenario: Foundation startup registers no assessment contracts Given assessment contracts are deferred to CHK.006 diff --git a/prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py b/prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py index 22bd633e..8ec1daf9 100644 --- a/prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py +++ b/prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py @@ -21,6 +21,7 @@ "INJECTOR_ID", "INJECTOR_NAME", "INJECTOR_LOG_LEVEL", + "PROWLER_EXECUTABLE_PATH", } VALIDATION_CANARY = "PYDANTIC_VALIDATION_CANARY" UNEXPECTED_CANARY = "UNEXPECTED_EXCEPTION_CANARY" @@ -61,7 +62,7 @@ def _when_sample_environment_is_loaded(project_root: Path) -> set[str]: } -def _then_only_standard_settings_are_available(settings: set[str]) -> None: +def _then_only_standard_and_prowler_settings_are_available(settings: set[str]) -> None: assert settings == STANDARD_ENV_SETTINGS @@ -134,7 +135,7 @@ def test_foundation_configuration_excludes_future_provider_settings() -> None: """The foundation exposes only standard injector settings.""" project_root = _given_the_prowler_project() settings = _when_sample_environment_is_loaded(project_root) - _then_only_standard_settings_are_available(settings) + _then_only_standard_and_prowler_settings_are_available(settings) def test_foundation_startup_registers_no_assessment_contracts( diff --git a/prowler/tests/behaviour/chk002_multi_provider_configuration/__init__.py b/prowler/tests/behaviour/chk002_multi_provider_configuration/__init__.py new file mode 100644 index 00000000..00d931d9 --- /dev/null +++ b/prowler/tests/behaviour/chk002_multi_provider_configuration/__init__.py @@ -0,0 +1 @@ +"""CHK.002 multi-provider configuration behaviour tests.""" diff --git a/prowler/tests/behaviour/chk002_multi_provider_configuration/chk002_multi_provider_configuration.feature b/prowler/tests/behaviour/chk002_multi_provider_configuration/chk002_multi_provider_configuration.feature new file mode 100644 index 00000000..84df9b27 --- /dev/null +++ b/prowler/tests/behaviour/chk002_multi_provider_configuration/chk002_multi_provider_configuration.feature @@ -0,0 +1,161 @@ +Feature: Prowler multi-provider form input + An OpenAEV form submission selects exactly one supported Prowler provider and + keeps provider credentials out of ordinary serialized output. + + Scenario Outline: Select one supported provider input + Given a complete "" provider form input + When the provider input is accepted + Then exactly the "" provider is selected + + Examples: + | provider | + | aws | + | azure | + | gcp | + | kubernetes | + + Scenario Outline: Protect credential secrets + Given a complete "" provider form input + When the provider input is accepted + Then its credential secrets are protected from ordinary output + + Examples: + | provider | + | aws | + | azure | + | gcp | + | kubernetes | + + Scenario Outline: Keep accepted provider input immutable + Given a complete "" provider form input + When the provider input is accepted + Then neither ordinary fields nor credential secrets can be replaced + + Examples: + | provider | + | aws | + | azure | + | gcp | + | kubernetes | + + Scenario Outline: Safely snapshot accepted provider input + Given a complete "" provider form input + When the provider input is deeply copied + Then the snapshot preserves protected credential values without exposing them + + Examples: + | provider | + | aws | + | azure | + | gcp | + | kubernetes | + + Scenario: Accept an optional AWS session token safely + Given a complete AWS provider form input with a session token + When the provider input is accepted + Then the session token is protected from ordinary output + + Scenario: Reject a missing provider + Given a form input without a provider + When the provider input is submitted + Then the provider input is rejected + + Scenario: Reject an unknown provider + Given a form input with an unknown provider + When the provider input is submitted + Then the provider input is rejected without exposing submitted credentials + + Scenario: Reject fields outside the selected provider + Given an AWS form input containing an Azure credential field + When the provider input is submitted + Then the provider input is rejected without exposing the rejected credential + + Scenario: Startup configuration remains provider-free + Given the six standard injector startup settings + When startup configuration is loaded + Then no provider input is present in startup configuration + + Scenario: Keep AWS endpoint overrides out of startup configuration + Given an AWS endpoint override appears in environment or YAML startup input + When startup configuration is loaded + Then the AWS endpoint override is not exposed by startup configuration + + Scenario: Use the recommended Prowler executable by default + Given no Prowler executable path is configured + When startup configuration is loaded + Then the Prowler executable path is "/usr/local/bin/prowler" + + Scenario: Configure an absolute Prowler executable path + Given an absolute Prowler executable path is configured + When startup configuration is loaded + Then that Prowler executable path is available as ordinary runtime configuration + + Scenario Outline: Reject an invalid Prowler executable path + Given the Prowler executable path is "" + When startup configuration is loaded + Then the startup configuration is rejected + + Examples: + | path | + | | + | | + | bin/prowler | + + Scenario: Use the default AWS service endpoint for an assessment + Given an AWS provider input without an endpoint override + When the provider input is accepted + Then the AWS endpoint override is absent + + Scenario: Supply a trusted AWS endpoint override per assessment + Given an AWS provider input with an absolute HTTP or HTTPS endpoint URL + When the provider input is accepted + Then that endpoint is available as an ordinary string without a reachability check + + Scenario Outline: Reject an unsafe AWS endpoint override + Given an AWS provider input whose endpoint override is "" + When the provider input is submitted + Then the provider input is rejected + + Examples: + | endpoint | + | | + | /relative | + | https:///missing-host | + | ftp://localhost:4566 | + | https://user:password@aws.example.com | + | https://aws.example.com?region=local | + | https://aws.example.com#credentials | + + # ---- Constraints identified ---- + + Scenario Outline: Provider selection is a strict discriminator + Given a form input with provider value "" + When the provider input is submitted + Then the provider input is rejected + + Examples: + | provider | + | AWS | + | aws | + + Scenario Outline: Required provider values cannot be blank + Given a "" form input whose required value "" is blank + When the provider input is submitted + Then the provider input is rejected without exposing the submitted value + + Examples: + | provider | field | + | aws | aws_secret_access_key | + | azure | azure_client_secret | + | gcp | gcp_service_account_json | + | kubernetes | kubernetes_kubeconfig | + + Scenario: Reject a non-string AWS endpoint override + Given an AWS provider input whose endpoint override is a non-string value + When the provider input is submitted + Then the provider input is rejected before type coercion + + Scenario: Reject an AWS endpoint override with an empty explicit port + Given an AWS provider input whose endpoint override has a port delimiter without a port + When the provider input is submitted + Then the provider input is rejected diff --git a/prowler/tests/behaviour/chk002_multi_provider_configuration/conftest.py b/prowler/tests/behaviour/chk002_multi_provider_configuration/conftest.py new file mode 100644 index 00000000..0b4c097a --- /dev/null +++ b/prowler/tests/behaviour/chk002_multi_provider_configuration/conftest.py @@ -0,0 +1,20 @@ +"""Fixtures local to CHK.002 multi-provider configuration behaviour.""" + +import pytest + + +@pytest.fixture +def standard_injector_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """Configure the standard injector environment owned by CHK.001.""" + monkeypatch.setenv("OPENAEV_URL", "http://localhost:8080") + monkeypatch.setenv("OPENAEV_TOKEN", "test-only-openaev-token") + monkeypatch.setenv("OPENAEV_TENANT_ID", "00000000-0000-4000-8000-000000000002") + monkeypatch.setenv("INJECTOR_ID", "test-prowler-injector") + monkeypatch.setenv("INJECTOR_NAME", "Prowler") + monkeypatch.setenv("INJECTOR_LOG_LEVEL", "debug") + + +@pytest.fixture +def clean_prowler_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure optional Prowler runtime settings are not overridden externally.""" + monkeypatch.delenv("PROWLER_EXECUTABLE_PATH", raising=False) diff --git a/prowler/tests/behaviour/chk002_multi_provider_configuration/test_chk002_multi_provider_configuration_bdd.py b/prowler/tests/behaviour/chk002_multi_provider_configuration/test_chk002_multi_provider_configuration_bdd.py new file mode 100644 index 00000000..298bbc3d --- /dev/null +++ b/prowler/tests/behaviour/chk002_multi_provider_configuration/test_chk002_multi_provider_configuration_bdd.py @@ -0,0 +1,449 @@ +"""Behaviour tests for CHK.002 provider form input models.""" + +import importlib +import json +from pathlib import Path +from typing import Any + +import pytest +from pydantic import HttpUrl, SecretStr, TypeAdapter, ValidationError + +from prowler.models.configs.config_loader import ConfigLoader, ProwlerConfig + +PROVIDER_PAYLOADS: dict[str, dict[str, str]] = { + "aws": { + "provider": "aws", + "aws_access_key_id": "EXAMPLEACCESSKEY", + "aws_secret_access_key": "example-aws-secret", + "aws_session_token": "example-aws-session-token", + "aws_account_id": "123456789012", + "aws_region": "eu-west-1", + }, + "azure": { + "provider": "azure", + "azure_tenant_id": "example-tenant", + "azure_client_id": "example-client", + "azure_client_secret": "example-azure-secret", + "azure_subscription_id": "example-subscription", + "azure_provider": "Microsoft.Compute", + }, + "gcp": { + "provider": "gcp", + "gcp_service_account_json": ( + '{"type":"service_account","private_key":"example-gcp-secret"}' + ), + "gcp_project_id": "example-project", + }, + "kubernetes": { + "provider": "kubernetes", + "kubernetes_kubeconfig": ( + "apiVersion: v1\nusers: []\n# example-kubernetes-secret" + ), + "kubernetes_context": "example-context", + }, +} + +SECRET_FIELDS = { + "aws": ("aws_secret_access_key", "aws_session_token"), + "azure": ("azure_client_secret",), + "gcp": ("gcp_service_account_json",), + "kubernetes": ("kubernetes_kubeconfig",), +} + +ORDINARY_FIELDS = { + "aws": "aws_region", + "azure": "azure_tenant_id", + "gcp": "gcp_project_id", + "kubernetes": "kubernetes_context", +} + + +def _provider_input_adapter() -> TypeAdapter[Any]: + try: + module = importlib.import_module("prowler.models.provider_inputs") + except ModuleNotFoundError: + pytest.fail("reusable provider input models are absent") + return module.PROVIDER_INPUT_ADAPTER # type: ignore[no-any-return] + + +def _when_submitted(payload: dict[str, str]) -> Any | ValidationError: + try: + return _provider_input_adapter().validate_python(payload) + except ValidationError as error: + return error + + +def _ordinary_outputs(provider_input: Any) -> tuple[str, str, str]: + json_dump = provider_input.model_dump(mode="json") + return repr(provider_input), str(provider_input), json.dumps(json_dump) + + +@pytest.mark.parametrize("provider", ["aws", "azure", "gcp", "kubernetes"]) +def test_select_exactly_one_supported_provider(provider: str) -> None: + """Accept each supported discriminator as exactly one provider model.""" + result = _when_submitted(PROVIDER_PAYLOADS[provider]) + + assert not isinstance(result, ValidationError) + assert result.provider == provider + assert type(result).__name__.lower().startswith(provider) + + +@pytest.mark.parametrize("provider", ["aws", "azure", "gcp", "kubernetes"]) +def test_protect_credentials_from_ordinary_output(provider: str) -> None: + """Redact provider credentials from repr, str, and JSON-mode dumps.""" + payload = PROVIDER_PAYLOADS[provider] + result = _when_submitted(payload) + + assert not isinstance(result, ValidationError) + outputs = _ordinary_outputs(result) + for field in SECRET_FIELDS[provider]: + assert all(payload[field] not in output for output in outputs) + assert all("**********" in output for output in outputs) + + +@pytest.mark.parametrize("provider", ["aws", "azure", "gcp", "kubernetes"]) +def test_reject_mutation_of_provider_fields_without_leaking_secrets( + provider: str, +) -> None: + """Reject raw assignment before attempted credentials can enter an error.""" + payload = PROVIDER_PAYLOADS[provider] + result = _when_submitted(payload) + + assert not isinstance(result, ValidationError) + for field in (ORDINARY_FIELDS[provider], *SECRET_FIELDS[provider]): + replacement_value = f"replacement-{provider}-secret" + original_value = getattr(result, field) + with pytest.raises( + TypeError, match="^Provider inputs are immutable$" + ) as raised: + setattr(result, field, replacement_value) + assert replacement_value not in str(raised.value) + assert replacement_value not in repr(raised.value) + assert all( + payload[secret] not in str(raised.value) + for secret in SECRET_FIELDS[provider] + ) + assert all( + payload[secret] not in repr(raised.value) + for secret in SECRET_FIELDS[provider] + ) + assert raised.value.__cause__ is None + assert raised.value.__context__ is None + assert getattr(result, field) == original_value + + +@pytest.mark.parametrize("provider", ["aws", "azure", "gcp", "kubernetes"]) +def test_deep_copy_preserves_secret_values_and_redaction(provider: str) -> None: + """Deep-copy provider input without losing or exposing protected values.""" + payload = PROVIDER_PAYLOADS[provider] + result = _when_submitted(payload) + + assert not isinstance(result, ValidationError) + snapshot = result.model_copy(deep=True) + assert snapshot is not result + for field in SECRET_FIELDS[provider]: + source_secret = getattr(result, field) + copied_secret = getattr(snapshot, field) + assert isinstance(source_secret, SecretStr) + assert isinstance(copied_secret, SecretStr) + assert copied_secret is not source_secret + assert copied_secret.get_secret_value() == payload[field] + assert payload[field] not in repr(snapshot) + assert payload[field] not in str(snapshot) + assert payload[field] not in json.dumps(snapshot.model_dump(mode="json")) + for provider_input in (result, snapshot): + attempted_value = f"deep-copy-replacement-{field}" + with pytest.raises( + TypeError, match="^Provider inputs are immutable$" + ) as raised: + setattr(provider_input, field, attempted_value) + assert attempted_value not in str(raised.value) + assert attempted_value not in repr(raised.value) + assert getattr(provider_input, field).get_secret_value() == payload[field] + + +def test_protect_optional_aws_session_token() -> None: + """Accept and redact the optional nonblank AWS session token.""" + submitted_value = "example-aws-session-token" + result = _when_submitted( + {**PROVIDER_PAYLOADS["aws"], "aws_session_token": submitted_value} + ) + + assert not isinstance(result, ValidationError) + assert all(submitted_value not in output for output in _ordinary_outputs(result)) + + +def test_reject_missing_provider_selection() -> None: + """Reject form input without its provider discriminator.""" + result = _when_submitted({}) + + assert isinstance(result, ValidationError) + + +def test_reject_unknown_provider_without_leaking_input() -> None: + """Reject an unknown discriminator without echoing submitted values.""" + submitted_value = "example-unknown-secret" + result = _when_submitted({"provider": "oracle", "credential": submitted_value}) + + assert isinstance(result, ValidationError) + assert submitted_value not in str(result) + + +def test_reject_cross_provider_field_without_leaking_it() -> None: + """Forbid extra cross-provider fields without echoing their values.""" + submitted_value = "example-cross-provider-secret" + payload = {**PROVIDER_PAYLOADS["aws"], "azure_client_secret": submitted_value} + + result = _when_submitted(payload) + + assert isinstance(result, ValidationError) + assert "azure_client_secret" in str(result) + assert submitted_value not in str(result) + + +@pytest.mark.parametrize("provider", ["AWS", " aws "]) +def test_provider_discriminator_is_strict(provider: str) -> None: + """Reject discriminator values whose case or whitespace is altered.""" + payload = {**PROVIDER_PAYLOADS["aws"], "provider": provider} + + result = _when_submitted(payload) + + assert isinstance(result, ValidationError) + + +@pytest.mark.parametrize( + ("provider", "field"), + [ + ("aws", "aws_secret_access_key"), + ("azure", "azure_client_secret"), + ("gcp", "gcp_service_account_json"), + ("kubernetes", "kubernetes_kubeconfig"), + ], +) +def test_reject_blank_required_values_without_leaking_them( + provider: str, + field: str, +) -> None: + """Reject blank required values without including input in errors.""" + payload = {**PROVIDER_PAYLOADS[provider], field: " \t "} + + result = _when_submitted(payload) + + assert isinstance(result, ValidationError) + assert field in str(result) + assert "input_value" not in str(result) + + +def test_startup_configuration_remains_provider_free( + standard_injector_environment: None, +) -> None: + """Keep provider form input outside the six-setting startup boundary.""" + config = ConfigLoader() + + assert set(type(config).model_fields) == {"openaev", "injector", "prowler"} + assert not hasattr(config, "provider") + assert not hasattr(config, "provider_config") + assert not hasattr(config, "selected_provider_config") + + +def test_recommended_prowler_executable_path_is_default( + standard_injector_environment: None, + clean_prowler_environment: None, +) -> None: + """Use the production Prowler executable location by default.""" + config = ConfigLoader() + + assert config.prowler.executable_path == Path("/usr/local/bin/prowler") + + +def test_absolute_prowler_executable_path_can_be_configured( + standard_injector_environment: None, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Load a non-secret executable path from its environment setting.""" + configured_path = "/opt/prowler/bin/prowler" + monkeypatch.setenv("PROWLER_EXECUTABLE_PATH", configured_path) + + config = ConfigLoader() + + assert config.prowler.executable_path == Path(configured_path) + assert configured_path in config.model_dump_json() + + +def test_absolute_prowler_executable_path_can_be_loaded_from_yaml( + standard_injector_environment: None, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Load the executable path from the Prowler YAML runtime section.""" + configured_path = "/srv/prowler/bin/prowler" + (tmp_path / "config.yml").write_text( + f"prowler:\n executable_path: '{configured_path}'\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + monkeypatch.setitem(ConfigLoader.model_config, "yaml_file", None) + + config = ConfigLoader() + + assert config.prowler.executable_path == Path(configured_path) + + +@pytest.mark.parametrize("executable_path", ["", " ", "bin/prowler"]) +def test_reject_invalid_prowler_executable_path( + standard_injector_environment: None, + monkeypatch: pytest.MonkeyPatch, + executable_path: str, +) -> None: + """Reject blank and relative executable paths at startup.""" + monkeypatch.setenv("PROWLER_EXECUTABLE_PATH", executable_path) + + with pytest.raises(ValidationError): + ConfigLoader() + + +def test_aws_endpoint_url_defaults_to_none() -> None: + """Use the AWS SDK service endpoint when an assessment supplies no override.""" + result = _when_submitted(PROVIDER_PAYLOADS["aws"]) + + assert not isinstance(result, ValidationError) + assert result.aws_endpoint_url is None + + +@pytest.mark.parametrize( + "configured_url", + [ + "https://s3.us-east-1.amazonaws.com", + "http://localhost:4566", + "http://localstack:4566", + "http://10.0.0.25:4566", + "https://aws.example.com:1/service/path", + "https://aws.example.com:65535/service/path", + ], +) +def test_trusted_aws_endpoint_url_can_be_supplied_per_assessment( + configured_url: str, +) -> None: + """Accept assessment-trusted HTTP endpoints without contacting their hosts.""" + result = _when_submitted( + {**PROVIDER_PAYLOADS["aws"], "aws_endpoint_url": configured_url} + ) + + assert not isinstance(result, ValidationError) + assert result.aws_endpoint_url == configured_url + assert type(result.aws_endpoint_url) is str + + +def test_aws_endpoint_url_is_not_loaded_from_startup_environment( + standard_injector_environment: None, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep the per-assessment endpoint outside environment startup settings.""" + monkeypatch.setenv("PROWLER_AWS_ENDPOINT_URL", "http://localhost:4566") + + config = ConfigLoader() + + assert "aws_endpoint_url" not in ProwlerConfig.model_fields + assert not hasattr(config.prowler, "aws_endpoint_url") + + +def test_aws_endpoint_url_is_not_loaded_from_startup_yaml( + standard_injector_environment: None, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Keep the per-assessment endpoint outside YAML startup settings.""" + (tmp_path / "config.yml").write_text( + "prowler:\n aws_endpoint_url: 'http://localhost:4566'\n", + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + monkeypatch.setitem(ConfigLoader.model_config, "yaml_file", None) + + config = ConfigLoader() + + assert "aws_endpoint_url" not in ProwlerConfig.model_fields + assert not hasattr(config.prowler, "aws_endpoint_url") + + +@pytest.mark.parametrize( + "configured_url", + [ + b"https://aws.example.com/unchecked", + HttpUrl("https://aws.example.com/service"), + 4566, + ], +) +def test_reject_non_string_aws_endpoint_url(configured_url: object) -> None: + """Reject endpoint values that could bypass checks through later coercion.""" + payload: dict[str, object] = { + **PROVIDER_PAYLOADS["aws"], + "aws_endpoint_url": configured_url, + } + + with pytest.raises(ValidationError): + _provider_input_adapter().validate_python(payload) + + +@pytest.mark.parametrize( + "configured_url", + [ + "", + " ", + "/relative", + "//localhost:4566", + "https:///missing-host", + "ftp://localhost:4566", + "https://user:password@aws.example.com", + "https://aws.example.com?region=local", + "https://aws.example.com#credentials", + "https://aws.example.com /service", + "https://aws.example.com:\t4566", + "https://aws.example.com:", + "https://aws.example.com:abc", + "https://aws.example.com:0", + "https://aws.example.com:65536", + ], +) +def test_reject_invalid_trusted_aws_endpoint_url( + configured_url: str, +) -> None: + """Reject endpoint overrides that cross the provider-input boundary.""" + payload = {**PROVIDER_PAYLOADS["aws"], "aws_endpoint_url": configured_url} + + with pytest.raises(ValidationError): + _provider_input_adapter().validate_python(payload) + + +def test_aws_endpoint_url_samples_and_documentation_are_consistent() -> None: + """Document the endpoint as provider input, never as a startup setting.""" + project_root = Path(__file__).parents[3] + + assert "PROWLER_AWS_ENDPOINT_URL" not in (project_root / ".env.sample").read_text( + encoding="utf-8" + ) + assert "aws_endpoint_url" not in (project_root / "config.yml.sample").read_text( + encoding="utf-8" + ) + readme = (project_root / "README.md").read_text(encoding="utf-8") + assert "`PROWLER_AWS_ENDPOINT_URL`" not in readme + assert "`prowler.aws_endpoint_url`" not in readme + assert "`aws_endpoint_url`" in readme + assert "per-assessment provider input" in readme + + +def test_runtime_path_samples_and_documentation_are_consistent() -> None: + """Expose the same recommended runtime path in all operator guidance.""" + project_root = Path(__file__).parents[3] + expected_path = "/usr/local/bin/prowler" + + assert f"PROWLER_EXECUTABLE_PATH={expected_path}" in ( + project_root / ".env.sample" + ).read_text(encoding="utf-8") + assert f"executable_path: '{expected_path}'" in ( + project_root / "config.yml.sample" + ).read_text(encoding="utf-8") + readme = (project_root / "README.md").read_text(encoding="utf-8") + assert "`PROWLER_EXECUTABLE_PATH`" in readme + assert "`prowler.executable_path`" in readme + assert f"`{expected_path}`" in readme