From c49752ba6a73c0754c98f7deab109b4fa120b585 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Fri, 28 Aug 2026 13:25:37 +0200 Subject: [PATCH 1/4] docs(prowler): define CHK.015 compliance behaviour (#422) --- .../chk015_compliance_nis2_iso27001.feature | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 prowler/tests/behaviour/chk015_compliance_nis2_iso27001/chk015_compliance_nis2_iso27001.feature diff --git a/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/chk015_compliance_nis2_iso27001.feature b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/chk015_compliance_nis2_iso27001.feature new file mode 100644 index 00000000..7b1ddce0 --- /dev/null +++ b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/chk015_compliance_nis2_iso27001.feature @@ -0,0 +1,77 @@ +Feature: CHK.015 NIS2 and ISO 27001 assessments + OpenAEV operators can run the provider-supported Prowler 5.36 NIS2 and ISO + 27001 frameworks through fixed routes without exposing framework routing as + form input or changing existing check and service selection channels. + + Scenario Outline: A canonical route selects one exact Prowler compliance framework + Given the executable Prowler registry contains the canonical route "" + When a valid provider assessment is executed through "" + Then the Prowler client is called exactly once with compliance "" + And no check filter or service selector is supplied + + Examples: + | route | compliance | + | nis2/aws | nis2_aws | + | nis2/azure | nis2_azure | + | nis2/gcp | nis2_gcp | + | iso27001/aws | iso27001_2022_aws | + | iso27001/azure | iso27001_2022_azure | + | iso27001/gcp | iso27001_2022_gcp | + | iso27001/kubernetes | iso27001_2022_kubernetes | + + Scenario: The twenty-two executable contracts have stable canonical identities + Given the executable Prowler registry + Then it contains the existing fifteen routes followed by the seven CHK.015 routes + And every CHK.015 route has compliance-identifying labels and no selector field + + Scenario: Unsupported compliance combinations stop before the client + Given a CHK.015 contract is paired with unsupported route metadata + When the invalid contract execution is attempted + Then the request is rejected before the Prowler client is called + + Scenario: Findings preserve the CHK.005 model and compliance values + Given ordered duplicate provider and non-provider OCSF records for a CHK.015 route + When the compliance assessment completes + Then provider findings retain source order, duplicates, and compliance tags + And the same findings feed shared Text and Vulnerability outputs and dynamic Rich trace + + Scenario Outline: Runtime dispatch emits exact compliance argv without real execution + Given a fake CLI engine and valid provider runtime message for "" + When the injector processes the runtime message + Then exactly one fake CLI request contains "--compliance" followed by "" + And provider credentials and temporary leases follow their existing lifecycle + And credentials, stderr, and temporary-path canaries are absent from the callback + + Examples: + | route | compliance | + | nis2/aws | nis2_aws | + | nis2/azure | nis2_azure | + | nis2/gcp | nis2_gcp | + | iso27001/aws | iso27001_2022_aws | + | iso27001/azure | iso27001_2022_azure | + | iso27001/gcp | iso27001_2022_gcp | + | iso27001/kubernetes | iso27001_2022_kubernetes | + +# ---- Constraints identified ---- + Scenario: Existing check and service channels remain dedicated + Given fake Prowler requests using an existing check filter or service selector + When those requests are rendered without compliance selection + Then check filters still use "-c" and services still use "--services" + And neither request emits "--compliance" + + Scenario: A compliance selector cannot target another provider + Given a fake provider request with another provider's NIS2 or ISO 27001 compliance + When the client request is validated + Then it is rejected before credential materialization or fake CLI execution + + Scenario: Framework routing constants are internal + Given a serialized CHK.015 OpenAEV contract + When its fields are inspected + Then only the existing provider fields are public + And no compliance or framework selector field is present + + Scenario: Existing finding semantics are not replaced by reports + Given duplicate OCSF findings from one selected framework + When the assessment output is built + Then duplicate findings and compliance tags remain unchanged + And no framework report, aggregation, or deduplication output is invented From 040c46dd1434e57308a32fd5f6c772da058ca2a3 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Fri, 28 Aug 2026 13:33:56 +0200 Subject: [PATCH 2/4] test(prowler): prove CHK.015 compliance RED (#422) --- .../__init__.py | 1 + .../conftest.py | 136 ++++ ...est_chk015_compliance_nis2_iso27001_bdd.py | 740 ++++++++++++++++++ .../test_outputs_registry_runtime.py | 11 +- 4 files changed, 886 insertions(+), 2 deletions(-) create mode 100644 prowler/tests/behaviour/chk015_compliance_nis2_iso27001/__init__.py create mode 100644 prowler/tests/behaviour/chk015_compliance_nis2_iso27001/conftest.py create mode 100644 prowler/tests/behaviour/chk015_compliance_nis2_iso27001/test_chk015_compliance_nis2_iso27001_bdd.py diff --git a/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/__init__.py b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/__init__.py new file mode 100644 index 00000000..b2306831 --- /dev/null +++ b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/__init__.py @@ -0,0 +1 @@ +"""CHK.015 NIS2 and ISO 27001 executable behavior.""" diff --git a/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/conftest.py b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/conftest.py new file mode 100644 index 00000000..2c3e1c09 --- /dev/null +++ b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/conftest.py @@ -0,0 +1,136 @@ +"""Local deterministic CHK.015 fixtures.""" + +from dataclasses import dataclass, field +from typing import Any + +import pytest + + +@dataclass(frozen=True) +class RecordedLog: + """One AppLogger-compatible call captured without formatting side effects.""" + + level: str + message: str + metadata: dict[str, object] | None = None + exc_info: bool | None = None + + +class _RecordingLocalLogger: + """Capture the direct standard-library ERROR path used by the injector.""" + + def __init__(self, events: list[RecordedLog]) -> None: + self._events = events + + def error( + self, + message: str, + *, + exc_info: bool, + extra: dict[str, object], + ) -> None: + """Record safe ERROR metadata in the same shape accepted by AppLogger.""" + attributes = extra.get("attributes") + metadata = dict(attributes) if isinstance(attributes, dict) else None + self._events.append(RecordedLog("error", message, metadata, exc_info)) + + +@dataclass +class RecordingLogger: + """Record the lifecycle logger surface exposed by the OpenAEV helper.""" + + events: list[RecordedLog] = field(default_factory=list) + + def __post_init__(self) -> None: + """Attach the direct ERROR surface to the shared event stream.""" + self.local_logger = _RecordingLocalLogger(self.events) + + def debug(self, message: str, metadata: dict[str, object]) -> None: + """Record one DEBUG lifecycle event.""" + self.events.append(RecordedLog("debug", message, metadata)) + + def info(self, message: str, metadata: dict[str, object] | None = None) -> None: + """Record one INFO lifecycle event.""" + self.events.append(RecordedLog("info", message, metadata)) + + def warning(self, message: str) -> None: + """Record one WARNING lifecycle event.""" + self.events.append(RecordedLog("warning", message)) + + +@pytest.fixture +def provider_forms() -> dict[str, dict[str, object]]: + """Return valid placeholder-only forms for all four providers.""" + return { + "aws": { + "aws_access_key_id": "CANARY-AWS-ACCESS", + "aws_secret_access_key": "CANARY-AWS-SECRET", + "aws_account_id": "123456789012", + "aws_region": "eu-west-1", + "aws_session_token": "CANARY-AWS-SESSION", + }, + "azure": { + "azure_tenant_id": "CANARY-AZURE-TENANT", + "azure_client_id": "CANARY-AZURE-CLIENT", + "azure_client_secret": "CANARY-AZURE-SECRET", + "azure_subscription_id": "subscription-123", + "azure_provider": "Microsoft.Compute", + }, + "gcp": { + "gcp_service_account_json": "CANARY-GCP-SERVICE-ACCOUNT", + "gcp_project_id": "acme-prod", + }, + "kubernetes": { + "kubernetes_kubeconfig": "CANARY-KUBECONFIG-CONTENT", + "kubernetes_context": "acme-prod-cluster", + }, + } + + +@pytest.fixture +def compliance_ocsf_record_factory() -> Any: + """Build a complete OCSF record with caller-controlled compliance values.""" + + def build( + title: str, + *, + provider: str = "aws", + status: str = "FAIL", + compliance: object = ("1.1", "1.1", "2.2"), + ) -> dict[str, Any]: + if status == "PASS": + record_status, status_code = "New", "PASS" + elif status == "MUTED": + record_status, status_code = "Suppressed", "FAIL" + else: + record_status, status_code = "New", status + resources = [{"uid": f"asset-{title}", "name": f"Asset {title}"}] + unmapped = {"compliance": compliance} + record = { + "finding_info": { + "uid": f"check-{title}", + "title": title, + "desc": f"Description {title}", + }, + "status": record_status, + "status_code": status_code, + "severity": "High", + "resources": resources, + "unmapped": unmapped, + "remediation": { + "desc": f"Remediate {title}", + "references": ["https://example.invalid/remediation"], + }, + } + if provider.casefold() == "kubernetes": + resources[0]["namespace"] = "default" + unmapped.update(provider=provider, provider_uid="acme-prod-cluster") + else: + record["cloud"] = { + "provider": provider, + "region": "eu-west-1", + "account": {"uid": "account-123"}, + } + return record + + return build diff --git a/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/test_chk015_compliance_nis2_iso27001_bdd.py b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/test_chk015_compliance_nis2_iso27001_bdd.py new file mode 100644 index 00000000..78e6b1d2 --- /dev/null +++ b/prowler/tests/behaviour/chk015_compliance_nis2_iso27001/test_chk015_compliance_nis2_iso27001_bdd.py @@ -0,0 +1,740 @@ +"""Raw pytest executable contract for CHK.015.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, cast, get_args + +import pytest +from pydantic import SecretStr +from pyoaev.configuration import ConfigLoaderOAEV + +from prowler._core.cli_engine import ( + CommandResult, + ExecutionSpecification, + OutputSpecification, + ValidatedCommandRequest, +) +from prowler._core.prowler_client import ( + OUTPUT_ARTIFACT_FILENAME, + ProwlerClientFactory, +) +from prowler.contracts import DEFAULT_PROWLER_CONTRACTS, stable_contract_id +from prowler.models.configs.config_loader import ( + ConfigLoader, + InjectorConfig, + ProwlerConfig, +) +from prowler.models.provider_inputs import AwsProviderInput + +from .conftest import RecordingLogger + +_ROUTES = ( + ("nis2/aws", "aws", "nis2_aws"), + ("nis2/azure", "azure", "nis2_azure"), + ("nis2/gcp", "gcp", "nis2_gcp"), + ("iso27001/aws", "aws", "iso27001_2022_aws"), + ("iso27001/azure", "azure", "iso27001_2022_azure"), + ("iso27001/gcp", "gcp", "iso27001_2022_gcp"), + ("iso27001/kubernetes", "kubernetes", "iso27001_2022_kubernetes"), +) +_EXISTING_COMPLIANCE = ( + ("cis/aws", "aws", "cis_3.0_aws"), + ("cis/azure", "azure", "cis_3.0_azure"), + ("cis/gcp", "gcp", "cis_3.0_gcp"), + ("cis/kubernetes", "kubernetes", "cis_1.12_kubernetes"), +) +_TEMP_PATHS = { + "gcp": Path("/tmp/CANARY-GCP-CREDENTIAL.json"), # noqa: S108 + "kubernetes": Path("/tmp/CANARY-KUBE-CREDENTIAL.yaml"), # noqa: S108 +} +_ASSESSMENT_RECEIVED = "[PROWLER_INJECTOR] - Assessment received" +_RECEPTION_ACKNOWLEDGED = "[PROWLER_INJECTOR] - Reception acknowledged" +_CONTRACT_RESOLVED = "[PROWLER_INJECTOR] - Contract resolved" +_ASSESSMENT_VALIDATED = "[PROWLER_INJECTOR] - Assessment input validated" +_EXECUTION_STARTED = "[PROWLER_INJECTOR] - Assessment execution starting" +_ASSESSMENT_SUCCEEDED = "[PROWLER_INJECTOR] - Assessment completed" +_ASSESSMENT_FAILED = "[PROWLER_INJECTOR] - Assessment failed" +_CALLBACK_COMPLETED = "[PROWLER_INJECTOR] - Assessment callback completed" + + +def _specification(arguments: tuple[str, ...] = ()) -> ExecutionSpecification: + return ExecutionSpecification( + executable="/fake/prowler", + arguments=arguments, + environment=(), + working_directory=None, + input_bytes=b"", + output=OutputSpecification(parser="raw"), + timeout_seconds=1.0, + maximum_accepted_output_bytes=1024, + ) + + +@dataclass +class _ClientFactory: + result: CommandResult + calls: list[tuple[Any, Any, tuple[str, ...], object, object]] = field( + default_factory=list + ) + + def run( + self, + config: Any, + provider: Any, + *, + check_filters: Any = (), + service_selector: object = None, + compliance_selector: object = None, + ) -> CommandResult: + self.calls.append( + ( + config, + provider, + tuple(check_filters), + service_selector, + compliance_selector, + ) + ) + return self.result + + +def _contract(route: str) -> Any: + return DEFAULT_PROWLER_CONTRACTS.resolve(str(stable_contract_id(route))) + + +@pytest.mark.parametrize(("route", "provider_name", "compliance"), _ROUTES) +def test_route_selects_exact_typed_compliance_once( + route: str, + provider_name: str, + compliance: str, + provider_forms: dict[str, dict[str, object]], +) -> None: + """A route-owned selector crosses the client seam once without other selectors.""" + factory = _ClientFactory(CommandResult(specification=_specification())) + contract = _contract(route) + contract._client_factory = factory + + contract.execute( + ProwlerConfig(), contract.parse_input(provider_forms[provider_name]) + ) + + assert len(factory.calls) == 1 + assert factory.calls[0][2:] == ((), None, compliance) + + +def test_compliance_selector_type_contains_exact_supported_values() -> None: + """The internal typed seam admits the existing and seven new exact values.""" + import prowler._core.prowler_client as client_api + + selector_type = client_api.__dict__.get("ComplianceSelector") + + assert selector_type is not None + assert get_args(selector_type) == tuple( + item[2] for item in (*_EXISTING_COMPLIANCE, *_ROUTES) + ) + + +def test_registry_has_22_canonical_contracts_without_selector_fields() -> None: + """The executable public surface is stable, ordered, and not user-selectable.""" + serialized = DEFAULT_PROWLER_CONTRACTS.contracts() + routes = ( + "aws", + "azure", + "gcp", + "kubernetes", + "aws/iam", + "aws/s3", + "aws/ec2", + "azure/iam", + "azure/storage", + "gcp/iam", + "gcp/compute", + *(item[0] for item in _EXISTING_COMPLIANCE), + *(item[0] for item in _ROUTES), + ) + + assert [item["contract_id"] for item in serialized] == [ + str(stable_contract_id(route)) for route in routes + ] + for item, (route, provider_name, _) in zip(serialized[15:], _ROUTES, strict=True): + content = json.loads(item["contract_content"]) + assert route.split("/", maxsplit=1)[0] in content["label"]["en"].casefold() + keys = tuple(field["key"] for field in content["fields"]) + assert keys + assert all("compliance" not in key and "framework" not in key for key in keys) + assert all(route in output["labels"] for output in content["outputs"]) + assert keys == tuple( + field.key for field in _contract(provider_name).build_provider_fields() + ) + + +def test_unsupported_contract_selector_is_rejected_pre_client( + provider_forms: dict[str, dict[str, object]], +) -> None: + """Invalid internal route metadata cannot consume the CHK.004 seam.""" + import prowler.contracts as contract_api + + nis2_contract = contract_api.__dict__.get("Nis2ComplianceContract") + assert nis2_contract is not None + + class InvalidNis2Contract(nis2_contract): + contract_id = str(stable_contract_id("nis2/aws")) + external_id = "prowler:nis2/aws" + route_name = "nis2/aws" + provider = "aws" + label = "Invalid" + compliance_selector = "nis2_azure" + + factory = _ClientFactory(CommandResult(specification=_specification())) + contract = InvalidNis2Contract(factory) + + with pytest.raises(ValueError, match="unsupported NIS2 compliance selection"): + contract.execute(ProwlerConfig(), contract.parse_input(provider_forms["aws"])) + + assert factory.calls == [] + + +def test_mapping_preserves_model_duplicates_compliance_outputs_and_trace( + provider_forms: dict[str, dict[str, object]], compliance_ocsf_record_factory: Any +) -> None: + """Keep CHK.005 findings unchanged across all existing presentation channels.""" + duplicate = compliance_ocsf_record_factory( + "duplicate", compliance={"NIS2": ["21.1", "21.1", "21.2"]} + ) + records = [ + compliance_ocsf_record_factory("first", status="PASS", compliance=["a", "b"]), + duplicate, + duplicate.copy(), + compliance_ocsf_record_factory("excluded", provider="azure"), + compliance_ocsf_record_factory("last", status="PASS", compliance=["z"]), + ] + artifact = json.dumps(records).encode() + factory = _ClientFactory( + CommandResult( + specification=_specification(), + return_code=0, + stdout=b"\x1b[32mconsole output is not OCSF JSON\x1b[0m", + parsed=artifact, + ) + ) + contract = _contract("nis2/aws") + contract._client_factory = factory + provider = contract.parse_input(provider_forms["aws"]) + + outcome = contract.execute(ProwlerConfig(), provider) + payload = contract.output_payload(outcome.findings) + trace = contract.render_trace( + provider, + outcome.findings, + 1, + raw_record_count=outcome.raw_record_count, + raw_output_bytes=outcome.raw_output_bytes, + raw_preview=outcome.raw_preview, + ) + + expected = ("first", "duplicate", "duplicate", "last") + assert tuple(item.value for item in outcome.findings) == expected + assert tuple(item.compliance_tags for item in outcome.findings) == ( + ("a", "b"), + ("NIS2:21.1", "NIS2:21.1", "NIS2:21.2"), + ("NIS2:21.1", "NIS2:21.1", "NIS2:21.2"), + ("z",), + ) + assert tuple(json.loads(item)["value"] for item in payload["findings"]) == expected + assert tuple(item["name"] for item in payload["vulnerabilities"]) == ( + "duplicate", + "duplicate", + ) + raw_section_index = trace.index("[PROWLER] Raw OCSF evidence (bounded preview)") + rendered_rows = tuple( + line + for line in trace[:raw_section_index].splitlines() + if any(name in line for name in ("first", "duplicate", "last")) + ) + assert ( + tuple(next(name for name in expected if name in line) for line in rendered_rows) + == expected + ) + assert "excluded" not in trace[:raw_section_index] + assert trace.index("Prowler Findings") < raw_section_index + assert outcome.raw_record_count == 5 + assert outcome.raw_output_bytes == len(artifact) + assert tuple(payload) == ("findings", "vulnerabilities") + assert "nis2/aws" in trace + assert "compliance=nis2_aws" in trace + + +@dataclass +class _Engine: + payload: bytes + lifecycle: list[str] + requests: list[ValidatedCommandRequest] = field(default_factory=list) + + def run(self, request: ValidatedCommandRequest) -> CommandResult: + self.lifecycle.append("engine") + self.requests.append(request) + arguments = tuple(request.arguments) + output_directory = Path(arguments[arguments.index("--output-directory") + 1]) + (output_directory / OUTPUT_ARTIFACT_FILENAME).write_bytes(self.payload) + specification = ExecutionSpecification.from_request(request) + return CommandResult( + specification=ExecutionSpecification( + executable=specification.executable, + arguments=(*specification.arguments, "PROCESS-ARGV-CANARY"), + environment=( + *specification.environment, + ("PROCESS-ENV-NAME-CANARY", "PROCESS-ENV-VALUE-CANARY"), + ), + working_directory="/tmp/PROCESS-WORKDIR-CANARY", # noqa: S108 + input_bytes=specification.input_bytes, + output=specification.output, + timeout_seconds=specification.timeout_seconds, + maximum_accepted_output_bytes=( + specification.maximum_accepted_output_bytes + ), + ), + return_code=0, + stdout=b"\x1b[31mPROCESS-STDOUT-CANARY\x1b[0m", + stderr=b"PROCESS-STDERR-CANARY", + ) + + +@dataclass +class _EngineFactory: + engine: _Engine + calls: int = 0 + + def create(self) -> _Engine: + self.calls += 1 + return self.engine + + +@dataclass +class _CredentialLease: + lifecycle: list[str] + path: Path + + def cleanup(self) -> None: + self.lifecycle.append("cleanup") + + +@dataclass +class _CredentialLeaseFactory: + lifecycle: list[str] + calls: list[tuple[SecretStr, str]] = field(default_factory=list) + + def create(self, secret: SecretStr, *, suffix: str) -> _CredentialLease: + self.calls.append((secret, suffix)) + self.lifecycle.append(f"create:{suffix}") + provider_name = "gcp" if suffix == ".json" else "kubernetes" + return _CredentialLease(self.lifecycle, _TEMP_PATHS[provider_name]) + + +class _InjectApi: + def __init__(self) -> None: + self.events: list[tuple[str, str, dict[str, Any]]] = [] + + def execution_reception(self, *, inject_id: str, data: dict[str, Any]) -> None: + self.events.append(("reception", inject_id, data)) + + def execution_callback(self, *, inject_id: str, data: dict[str, Any]) -> None: + self.events.append(("callback", inject_id, data)) + + +class _Helper: + def __init__(self) -> None: + self.api = type("Api", (), {})() + self.api.inject = _InjectApi() + self.injector_logger = RecordingLogger() + + +def _config() -> ConfigLoader: + return ConfigLoader.model_construct( + openaev=ConfigLoaderOAEV( + url="http://127.0.0.1:8080", token="runtime-placeholder" + ), + injector=InjectorConfig(id="injector-test"), + prowler=ProwlerConfig(executable_path="/fake/prowler"), + ) + + +def _message( + route: str, provider_name: str, content: dict[str, object] +) -> dict[str, object]: + return { + "injection": { + "inject_id": f"INJECT-ID-CANARY-{provider_name}", + "injector_contract_id": str(stable_contract_id(route)), + "inject_content": content, + } + } + + +@pytest.mark.parametrize(("route", "provider_name", "compliance"), _ROUTES) +def test_runtime_one_call_exact_compliance_argv_lifecycle_and_canaries( + route: str, + provider_name: str, + compliance: str, + provider_forms: dict[str, dict[str, object]], + compliance_ocsf_record_factory: Any, +) -> None: + """Runtime uses one fake request and preserves provider resource handling.""" + from prowler.injector import ProwlerInjector + + callback_name = f"CALLBACK-CANARY-{provider_name}" + finding_name = f"FINDING-CANARY-{provider_name}" + records = [ + compliance_ocsf_record_factory( + callback_name, provider=provider_name, status="PASS" + ), + compliance_ocsf_record_factory(finding_name, provider=provider_name), + compliance_ocsf_record_factory("excluded", provider="unsupported"), + ] + lifecycle: list[str] = [] + engine = _Engine(json.dumps(records).encode(), lifecycle) + engine_factory = _EngineFactory(engine) + leases = _CredentialLeaseFactory(lifecycle) + contract = _contract(route) + contract._client_factory = ProwlerClientFactory(engine_factory, leases) + helper = _Helper() + + ProwlerInjector(_config(), helper).process_message( + _message(route, provider_name, provider_forms[provider_name]) + ) + + assert engine_factory.calls == 1 + assert len(engine.requests) == 1 + arguments = tuple(engine.requests[0].arguments) + compliance_index = arguments.index("--compliance") + assert arguments[compliance_index : compliance_index + 2] == ( + "--compliance", + compliance, + ) + assert "-c" not in arguments + assert "--services" not in arguments + if provider_name in _TEMP_PATHS: + suffix = ".json" if provider_name == "gcp" else ".yaml" + assert lifecycle == [f"create:{suffix}", "engine", "cleanup"] + assert len(leases.calls) == 1 + else: + assert lifecycle == ["engine"] + assert leases.calls == [] + assert tuple(event[0] for event in helper.api.inject.events) == ( + "reception", + "callback", + ) + callback = helper.api.inject.events[1][2] + assert callback["execution_status"] == "SUCCESS" + structured = json.loads(callback["execution_output_structured"]) + assert tuple(json.loads(item)["value"] for item in structured["findings"]) == ( + callback_name, + finding_name, + ) + assert tuple(item["name"] for item in structured["vulnerabilities"]) == ( + finding_name, + ) + assert callback_name in callback["execution_message"] + assert finding_name in callback["execution_message"] + raw_section_index = callback["execution_message"].index( + "[PROWLER] Raw OCSF evidence (bounded preview)" + ) + assert "excluded" not in callback["execution_message"][:raw_section_index] + trace_identifiers = { + "aws": ("123456789012", "eu-west-1"), + "azure": ("subscription-123", "Microsoft.Compute"), + "gcp": ("acme-prod",), + "kubernetes": ("acme-prod-cluster",), + } + assert all( + marker in callback["execution_message"] + for marker in trace_identifiers[provider_name] + ) + assert route in callback["execution_message"] + assert f"compliance={compliance}" in callback["execution_message"] + credential_fields = { + "aws": ( + "aws_access_key_id", + "aws_secret_access_key", + "aws_session_token", + ), + "azure": ( + "azure_tenant_id", + "azure_client_id", + "azure_client_secret", + ), + "gcp": ("gcp_service_account_json",), + "kubernetes": ("kubernetes_kubeconfig",), + } + callback_excluded_canaries = tuple( + str(provider_forms[provider_name][field_name]) + for field_name in credential_fields[provider_name] + ) + ( + "PROCESS-ARGV-CANARY", + "PROCESS-ENV-NAME-CANARY", + "PROCESS-ENV-VALUE-CANARY", + "PROCESS-STDOUT-CANARY", + "PROCESS-STDERR-CANARY", + "/tmp/PROCESS-WORKDIR-CANARY", # noqa: S108 - deliberate leak canary + *(str(path) for path in _TEMP_PATHS.values()), + "EXCEPTION-FIELD-CANARY", + "EXCEPTION-VALUE-CANARY", + ) + assert all( + marker not in json.dumps(callback) for marker in callback_excluded_canaries + ) + + logs = helper.injector_logger.events + assert tuple((event.level, event.message) for event in logs) == ( + ("info", _ASSESSMENT_RECEIVED), + ("debug", _RECEPTION_ACKNOWLEDGED), + ("debug", _CONTRACT_RESOLVED), + ("debug", _ASSESSMENT_VALIDATED), + ("info", _EXECUTION_STARTED), + ("info", _ASSESSMENT_SUCCEEDED), + ("debug", _CALLBACK_COMPLETED), + ) + assert [event.metadata["stage"] for event in logs if event.metadata] == [ + "message_reception", + "reception_acknowledged", + "contract_resolution", + "input_validation", + "assessment_execution", + "assessment_completion", + "callback", + ] + assert all(event.metadata is not None for event in logs) + inject_id = f"INJECT-ID-CANARY-{provider_name}" + assert all( + event.metadata["inject_id"] == inject_id for event in logs if event.metadata + ) + assert all( + 0 <= event.metadata["elapsed_ms"] <= 86_400_000 + for event in logs + if event.metadata + ) + identifier = str(stable_contract_id(route)) + for event in logs[2:]: + assert event.metadata is not None + assert event.metadata["contract_id"] == identifier + assert event.metadata["route"] == route + assert event.metadata["provider"] == provider_name + safe_context = { + "aws": { + "aws_account_id": "123456789012", + "aws_region": "eu-west-1", + "aws_session_token_present": True, + "aws_endpoint_override_present": False, + }, + "azure": { + "azure_tenant_id_present": True, + "azure_client_id_present": True, + "azure_client_secret_present": True, + "azure_subscription_id": "subscription-123", + "azure_provider": "Microsoft.Compute", + }, + "gcp": { + "gcp_project_id": "acme-prod", + "gcp_credentials_present": True, + }, + "kubernetes": { + "kubernetes_context": "acme-prod-cluster", + "kubernetes_credentials_present": True, + }, + } + for event in logs[3:]: + assert event.metadata is not None + assert all( + event.metadata[key] == value + for key, value in safe_context[provider_name].items() + ) + success_metadata = logs[5].metadata + assert success_metadata is not None + assert success_metadata["status"] == "SUCCESS" + assert success_metadata["finding_count"] == 2 + assert success_metadata["vulnerability_count"] == 1 + assert success_metadata["raw_record_count"] == 3 + assert success_metadata["raw_output_bytes"] == len(engine.payload) + callback_metadata = logs[6].metadata + assert callback_metadata is not None + assert callback_metadata["assessment_status"] == "SUCCESS" + assert callback_metadata["delivery_status"] == "SUCCESS" + assert "status" not in callback_metadata + assert "attempted_status" not in callback_metadata + log_excluded_canaries = ( + *callback_excluded_canaries, + callback_name, + finding_name, + ) + assert all(marker not in repr(logs) for marker in log_excluded_canaries) + + invalid_fields = { + "aws": "aws_account_id", + "azure": "azure_subscription_id", + "gcp": "gcp_project_id", + "kubernetes": "kubernetes_context", + } + invalid_field = invalid_fields[provider_name] + invalid_helper = _Helper() + ProwlerInjector(_config(), invalid_helper).process_message( + _message( + route, + provider_name, + { + **provider_forms[provider_name], + invalid_field: " ", + "EXCEPTION-FIELD-CANARY": "EXCEPTION-VALUE-CANARY", + }, + ) + ) + + assert tuple(event[0] for event in invalid_helper.api.inject.events) == ( + "reception", + "callback", + ) + invalid_callback = invalid_helper.api.inject.events[1][2] + assert invalid_callback["execution_status"] == "ERROR" + assert all( + marker not in json.dumps(invalid_callback) for marker in log_excluded_canaries + ) + assert engine_factory.calls == 1 + expected_lifecycle = ( + ["engine"] + if provider_name not in _TEMP_PATHS + else [ + "create:.json" if provider_name == "gcp" else "create:.yaml", + "engine", + "cleanup", + ] + ) + assert lifecycle == expected_lifecycle + + invalid_logs = invalid_helper.injector_logger.events + assert tuple((event.level, event.message) for event in invalid_logs) == ( + ("info", _ASSESSMENT_RECEIVED), + ("debug", _RECEPTION_ACKNOWLEDGED), + ("debug", _CONTRACT_RESOLVED), + ("error", _ASSESSMENT_FAILED), + ("debug", _CALLBACK_COMPLETED), + ) + assert [event.metadata["stage"] for event in invalid_logs if event.metadata] == [ + "message_reception", + "reception_acknowledged", + "contract_resolution", + "input_validation", + "callback", + ] + assert all(event.metadata is not None for event in invalid_logs) + assert all( + event.metadata["inject_id"] == inject_id + for event in invalid_logs + if event.metadata + ) + assert all( + 0 <= event.metadata["elapsed_ms"] <= 86_400_000 + for event in invalid_logs + if event.metadata + ) + for event in invalid_logs[2:]: + assert event.metadata is not None + assert event.metadata["contract_id"] == identifier + assert event.metadata["route"] == route + assert event.metadata["provider"] == provider_name + failure_metadata = invalid_logs[3].metadata + assert failure_metadata is not None + assert failure_metadata["status"] == "ERROR" + assert failure_metadata["stage"] == "input_validation" + assert failure_metadata["failure_kind"] == "invalid_input" + assert failure_metadata["failure_summary"] == "The assessment input was invalid." + assert failure_metadata["operator_guidance"] == ( + "Correct the listed assessment fields and retry." + ) + assert failure_metadata["issues"] == [ + {"location": [provider_name, invalid_field], "type": "value_error"}, + { + "location": [provider_name, "unrecognized_field"], + "type": "extra_forbidden", + }, + ] + assert failure_metadata["issues_truncated"] is True + assert invalid_logs[3].exc_info is False + invalid_callback_metadata = invalid_logs[4].metadata + assert invalid_callback_metadata is not None + assert invalid_callback_metadata["assessment_status"] == "ERROR" + assert invalid_callback_metadata["delivery_status"] == "SUCCESS" + assert "status" not in invalid_callback_metadata + assert "attempted_status" not in invalid_callback_metadata + assert all(marker not in repr(invalid_logs) for marker in log_excluded_canaries) + + +def test_existing_check_and_service_argv_are_unchanged( + provider_forms: dict[str, dict[str, object]], +) -> None: + """Compliance selection does not repurpose either existing selector channel.""" + lifecycle: list[str] = [] + engine = _Engine(b"[]", lifecycle) + factory = ProwlerClientFactory( + _EngineFactory(engine), _CredentialLeaseFactory(lifecycle) + ) + provider = _contract("aws").parse_input(provider_forms["aws"]) + + factory.run( + ProwlerConfig(executable_path="/fake/prowler"), + provider, + check_filters=("check-one",), + ) + check_index = engine.requests[0].arguments.index("-c") + assert tuple(engine.requests[0].arguments[check_index : check_index + 2]) == ( + "-c", + "check-one", + ) + assert "--services" not in engine.requests[0].arguments + assert "--compliance" not in engine.requests[0].arguments + + second_engine = _Engine(b"[]", []) + second_factory = ProwlerClientFactory( + _EngineFactory(second_engine), _CredentialLeaseFactory([]) + ) + second_provider = _contract("aws").parse_input(provider_forms["aws"]) + second_factory.run( + ProwlerConfig(executable_path="/fake/prowler"), + second_provider, + service_selector="iam", + ) + service_index = second_engine.requests[0].arguments.index("--services") + assert tuple( + second_engine.requests[0].arguments[service_index : service_index + 2] + ) == ("--services", "iam") + assert "-c" not in second_engine.requests[0].arguments + assert "--compliance" not in second_engine.requests[0].arguments + + +@pytest.mark.parametrize("compliance", ("nis2_azure", "iso27001_2022_kubernetes")) +def test_cross_provider_compliance_rejected_before_adapter_or_engine( + compliance: str, +) -> None: + """Provider/selector pairs are validated before credentials or CLI use.""" + lifecycle: list[str] = [] + engine = _Engine(b"[]", lifecycle) + leases = _CredentialLeaseFactory(lifecycle) + factory = ProwlerClientFactory(_EngineFactory(engine), leases) + provider = AwsProviderInput( + provider="aws", + aws_access_key_id="access", + aws_secret_access_key="secret", + aws_account_id="123456789012", + aws_region="eu-west-1", + ) + + with pytest.raises(ValueError, match="compliance selector"): + factory.run( + ProwlerConfig(), + provider, + compliance_selector=cast(Any, compliance), + ) + + assert leases.calls == [] + assert engine.requests == [] + assert lifecycle == [] diff --git a/prowler/tests/unit/chk006_executable_base/test_outputs_registry_runtime.py b/prowler/tests/unit/chk006_executable_base/test_outputs_registry_runtime.py index d6e1f696..d7355589 100644 --- a/prowler/tests/unit/chk006_executable_base/test_outputs_registry_runtime.py +++ b/prowler/tests/unit/chk006_executable_base/test_outputs_registry_runtime.py @@ -398,7 +398,7 @@ def test_runtime_start_logs_one_fixed_listener_event() -> None: assert message == _LISTENER_START assert metadata["injector_id"] == "injector-test" assert metadata["injector_name"] == "Prowler" - assert metadata["registered_contract_count"] == 15 + assert metadata["registered_contract_count"] == 22 assert metadata["configured_executable_path"] == "/usr/local/bin/prowler" assert set(metadata) == { "injector_id", @@ -1317,7 +1317,7 @@ def test_runtime_resolved_contract_uses_renderer_for_safe_error( def test_default_registry_and_daemon_config_register_executable_routes() -> None: - """CHK.014 adds four CIS routes after the CHK.013 executable routes.""" + """The daemon registers all executable routes through CHK.015.""" subject = _subject() contracts = subject.DEFAULT_PROWLER_CONTRACTS.contracts() assert [item["contract_id"] for item in contracts] == [ @@ -1336,6 +1336,13 @@ def test_default_registry_and_daemon_config_register_executable_routes() -> None str(subject.stable_contract_id("cis/azure")), str(subject.stable_contract_id("cis/gcp")), str(subject.stable_contract_id("cis/kubernetes")), + str(subject.stable_contract_id("nis2/aws")), + str(subject.stable_contract_id("nis2/azure")), + str(subject.stable_contract_id("nis2/gcp")), + str(subject.stable_contract_id("iso27001/aws")), + str(subject.stable_contract_id("iso27001/azure")), + str(subject.stable_contract_id("iso27001/gcp")), + str(subject.stable_contract_id("iso27001/kubernetes")), ] daemon = _config().to_daemon_config() assert daemon.get("injector_contracts") == contracts From 6122ede804945fd8a7d8e3b158997deee23cbd87 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Fri, 28 Aug 2026 14:14:32 +0200 Subject: [PATCH 3/4] feat(prowler): execute CHK.015 compliance routes (#422) --- .../prowler/_core/prowler_client/client.py | 7 + .../prowler/_core/prowler_client/contracts.py | 7 + prowler/prowler/contracts/__init__.py | 20 +++ prowler/prowler/contracts/nis2_iso27001.py | 143 ++++++++++++++++++ prowler/prowler/contracts/registry.py | 16 ++ .../test_chk001_catalog_scaffold_bdd.py | 7 + .../test_chk007_base_aws_provider_bdd.py | 7 + .../test_chk008_base_azure_provider_bdd.py | 9 +- .../test_chk009_base_gcp_provider_bdd.py | 9 +- ...est_chk010_base_kubernetes_provider_bdd.py | 9 +- .../test_chk011_service_aws_assessment_bdd.py | 11 +- ...est_chk012_service_azure_assessment_bdd.py | 11 +- .../test_chk013_service_gcp_assessment_bdd.py | 11 +- ...est_chk014_compliance_cis_benchmark_bdd.py | 24 ++- 14 files changed, 272 insertions(+), 19 deletions(-) create mode 100644 prowler/prowler/contracts/nis2_iso27001.py diff --git a/prowler/prowler/_core/prowler_client/client.py b/prowler/prowler/_core/prowler_client/client.py index 34ecd8ec..a770df4c 100644 --- a/prowler/prowler/_core/prowler_client/client.py +++ b/prowler/prowler/_core/prowler_client/client.py @@ -87,6 +87,13 @@ def _safe_log(level: int, message: str, **metadata: object) -> None: "cis_3.0_azure": AzureProviderInput, "cis_3.0_gcp": GcpProviderInput, "cis_1.12_kubernetes": KubernetesProviderInput, + "nis2_aws": AwsProviderInput, + "nis2_azure": AzureProviderInput, + "nis2_gcp": GcpProviderInput, + "iso27001_2022_aws": AwsProviderInput, + "iso27001_2022_azure": AzureProviderInput, + "iso27001_2022_gcp": GcpProviderInput, + "iso27001_2022_kubernetes": KubernetesProviderInput, } diff --git a/prowler/prowler/_core/prowler_client/contracts.py b/prowler/prowler/_core/prowler_client/contracts.py index 49745651..071d0497 100644 --- a/prowler/prowler/_core/prowler_client/contracts.py +++ b/prowler/prowler/_core/prowler_client/contracts.py @@ -18,6 +18,13 @@ "cis_3.0_azure", "cis_3.0_gcp", "cis_1.12_kubernetes", + "nis2_aws", + "nis2_azure", + "nis2_gcp", + "iso27001_2022_aws", + "iso27001_2022_azure", + "iso27001_2022_gcp", + "iso27001_2022_kubernetes", ] diff --git a/prowler/prowler/contracts/__init__.py b/prowler/prowler/contracts/__init__.py index b5e3bd86..a395207c 100644 --- a/prowler/prowler/contracts/__init__.py +++ b/prowler/prowler/contracts/__init__.py @@ -32,6 +32,17 @@ from .dispatcher import ContractDispatcher, RouteHandler, RouteNotFoundError from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract, GcpServiceContract from .kubernetes import KubernetesBaseContract +from .nis2_iso27001 import ( + AwsIso27001Contract, + AwsNis2Contract, + AzureIso27001Contract, + AzureNis2Contract, + GcpIso27001Contract, + GcpNis2Contract, + Iso27001ComplianceContract, + KubernetesIso27001Contract, + Nis2ComplianceContract, +) from .registry import ( DEFAULT_PROWLER_CONTRACTS, PROWLER_CONTRACT_NAMESPACE, @@ -51,15 +62,24 @@ "AzureServiceContract", "AzureStorageContract", "AwsCisContract", + "AwsIso27001Contract", + "AwsNis2Contract", "AzureCisContract", + "AzureIso27001Contract", + "AzureNis2Contract", "CisComplianceContract", "GcpBaseContract", "GcpComputeContract", "GcpIamContract", "GcpServiceContract", "GcpCisContract", + "GcpIso27001Contract", + "GcpNis2Contract", + "Iso27001ComplianceContract", "KubernetesBaseContract", "KubernetesCisContract", + "KubernetesIso27001Contract", + "Nis2ComplianceContract", "ContractDispatcher", "ContractExecutionOutcome", "ContractInputError", diff --git a/prowler/prowler/contracts/nis2_iso27001.py b/prowler/prowler/contracts/nis2_iso27001.py new file mode 100644 index 00000000..93ba3603 --- /dev/null +++ b/prowler/prowler/contracts/nis2_iso27001.py @@ -0,0 +1,143 @@ +"""Executable NIS2 and ISO 27001 compliance contracts.""" + +from typing import ClassVar + +from prowler._core.prowler_client import ComplianceSelector +from prowler.models.configs.config_loader import ProwlerConfig +from prowler.models.provider_inputs import ProviderInput + +from .base import ( + BaseProwlerContract, + ContractExecutionOutcome, + ProviderName, + RouteFamily, +) + +_NIS2_BY_PROVIDER: dict[ProviderName, ComplianceSelector] = { + "aws": "nis2_aws", + "azure": "nis2_azure", + "gcp": "nis2_gcp", +} +_ISO27001_BY_PROVIDER: dict[ProviderName, ComplianceSelector] = { + "aws": "iso27001_2022_aws", + "azure": "iso27001_2022_azure", + "gcp": "iso27001_2022_gcp", + "kubernetes": "iso27001_2022_kubernetes", +} + + +class _FrameworkComplianceContract(BaseProwlerContract): + """Share fixed framework routing across the CHK.015 contract families.""" + + family: ClassVar[RouteFamily] = "compliance" + compliance_selector: ClassVar[ComplianceSelector] + compliance_by_provider: ClassVar[dict[ProviderName, ComplianceSelector]] + framework_name: ClassVar[str] + + def safe_request_info(self, provider: ProviderInput | None) -> dict[str, object]: + """Identify framework selection through safe route metadata only.""" + info = super().safe_request_info(provider) + info["filters"] = f"compliance={self.compliance_selector}" + return info + + def execute( + self, config: ProwlerConfig, provider: ProviderInput + ) -> ContractExecutionOutcome: + """Reject unsupported route metadata before one compliance client call.""" + if ( + self.compliance_by_provider.get(self.provider) != self.compliance_selector + or provider.provider != self.provider + ): + raise ValueError(f"unsupported {self.framework_name} compliance selection") + return self._execute_compliance(config, provider, self.compliance_selector) + + +class Nis2ComplianceContract(_FrameworkComplianceContract): + """Execute one provider-owned NIS2 selector through the CHK.004 seam.""" + + compliance_by_provider = _NIS2_BY_PROVIDER + framework_name = "NIS2" + + +class Iso27001ComplianceContract(_FrameworkComplianceContract): + """Execute one provider-owned ISO 27001 selector through the CHK.004 seam.""" + + compliance_by_provider = _ISO27001_BY_PROVIDER + framework_name = "ISO27001" + + +class AwsNis2Contract(Nis2ComplianceContract): + """Run the Prowler 5.36 AWS NIS2 framework.""" + + contract_id = "eff788bc-44ef-5381-840a-e56ecc34db99" + external_id = "prowler:nis2/aws" + route_name = "nis2/aws" + provider = "aws" + label = "Prowler AWS NIS2" + compliance_selector = "nis2_aws" + + +class AzureNis2Contract(Nis2ComplianceContract): + """Run the Prowler 5.36 Azure NIS2 framework.""" + + contract_id = "7799ecf6-98db-5045-8154-65c085c6ded7" + external_id = "prowler:nis2/azure" + route_name = "nis2/azure" + provider = "azure" + label = "Prowler Azure NIS2" + compliance_selector = "nis2_azure" + + +class GcpNis2Contract(Nis2ComplianceContract): + """Run the Prowler 5.36 GCP NIS2 framework.""" + + contract_id = "12139fd2-48d5-51ba-ba60-9bd072a8720b" + external_id = "prowler:nis2/gcp" + route_name = "nis2/gcp" + provider = "gcp" + label = "Prowler GCP NIS2" + compliance_selector = "nis2_gcp" + + +class AwsIso27001Contract(Iso27001ComplianceContract): + """Run the Prowler 5.36 AWS ISO 27001:2022 framework.""" + + contract_id = "3adfd3d1-8a80-5bb8-b9f8-d28d1d54a2a5" + external_id = "prowler:iso27001/aws" + route_name = "iso27001/aws" + provider = "aws" + label = "Prowler AWS ISO27001" + compliance_selector = "iso27001_2022_aws" + + +class AzureIso27001Contract(Iso27001ComplianceContract): + """Run the Prowler 5.36 Azure ISO 27001:2022 framework.""" + + contract_id = "51f5a42a-fed2-50d7-a438-4c9d38139fce" + external_id = "prowler:iso27001/azure" + route_name = "iso27001/azure" + provider = "azure" + label = "Prowler Azure ISO27001" + compliance_selector = "iso27001_2022_azure" + + +class GcpIso27001Contract(Iso27001ComplianceContract): + """Run the Prowler 5.36 GCP ISO 27001:2022 framework.""" + + contract_id = "0de81cd3-21fc-5225-ba37-7ccc708bd633" + external_id = "prowler:iso27001/gcp" + route_name = "iso27001/gcp" + provider = "gcp" + label = "Prowler GCP ISO27001" + compliance_selector = "iso27001_2022_gcp" + + +class KubernetesIso27001Contract(Iso27001ComplianceContract): + """Run the Prowler 5.36 Kubernetes ISO 27001:2022 framework.""" + + contract_id = "a52206f3-c3da-5120-9f1f-965ec9efdcc6" + external_id = "prowler:iso27001/kubernetes" + route_name = "iso27001/kubernetes" + provider = "kubernetes" + label = "Prowler Kubernetes ISO27001" + compliance_selector = "iso27001_2022_kubernetes" diff --git a/prowler/prowler/contracts/registry.py b/prowler/prowler/contracts/registry.py index 73643461..cbc44efa 100644 --- a/prowler/prowler/contracts/registry.py +++ b/prowler/prowler/contracts/registry.py @@ -14,6 +14,15 @@ from .cis import AwsCisContract, AzureCisContract, GcpCisContract, KubernetesCisContract from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract from .kubernetes import KubernetesBaseContract +from .nis2_iso27001 import ( + AwsIso27001Contract, + AwsNis2Contract, + AzureIso27001Contract, + AzureNis2Contract, + GcpIso27001Contract, + GcpNis2Contract, + KubernetesIso27001Contract, +) # Committed project namespace: changing it would break stable platform identities. PROWLER_CONTRACT_NAMESPACE = UUID("ee49522d-80b9-5d71-b164-569ee61a75bd") @@ -99,5 +108,12 @@ def contracts(self) -> list[dict[str, object]]: AzureCisContract, GcpCisContract, KubernetesCisContract, + AwsNis2Contract, + AzureNis2Contract, + GcpNis2Contract, + AwsIso27001Contract, + AzureIso27001Contract, + GcpIso27001Contract, + KubernetesIso27001Contract, ) ) 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 5fe58027..d4cc392d 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 @@ -94,6 +94,13 @@ def _then_base_contracts_are_registered(config: ConfigLoader, helper: Mock) -> N str(stable_contract_id("cis/azure")), str(stable_contract_id("cis/gcp")), str(stable_contract_id("cis/kubernetes")), + str(stable_contract_id("nis2/aws")), + str(stable_contract_id("nis2/azure")), + str(stable_contract_id("nis2/gcp")), + str(stable_contract_id("iso27001/aws")), + str(stable_contract_id("iso27001/azure")), + str(stable_contract_id("iso27001/gcp")), + str(stable_contract_id("iso27001/kubernetes")), ] callback = helper.listen.call_args.kwargs["message_callback"] assert callable(callback) diff --git a/prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py b/prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py index 5aca02cd..ee9df9fc 100644 --- a/prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py +++ b/prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py @@ -106,6 +106,13 @@ def test_default_registration_identity_fields_and_outputs() -> None: str(stable_contract_id("cis/azure")), str(stable_contract_id("cis/gcp")), str(stable_contract_id("cis/kubernetes")), + str(stable_contract_id("nis2/aws")), + str(stable_contract_id("nis2/azure")), + str(stable_contract_id("nis2/gcp")), + str(stable_contract_id("iso27001/aws")), + str(stable_contract_id("iso27001/azure")), + str(stable_contract_id("iso27001/gcp")), + str(stable_contract_id("iso27001/kubernetes")), ] assert UUID(serialized[0]["contract_id"]) == expected_id assert expected_id.version == 5 diff --git a/prowler/tests/behaviour/chk008_base_azure_provider/test_chk008_base_azure_provider_bdd.py b/prowler/tests/behaviour/chk008_base_azure_provider/test_chk008_base_azure_provider_bdd.py index 28631358..1bf6a432 100644 --- a/prowler/tests/behaviour/chk008_base_azure_provider/test_chk008_base_azure_provider_bdd.py +++ b/prowler/tests/behaviour/chk008_base_azure_provider/test_chk008_base_azure_provider_bdd.py @@ -89,7 +89,7 @@ def test_default_registration_identity_fields_and_outputs() -> None: serialized = DEFAULT_PROWLER_CONTRACTS.contracts() expected_id = stable_contract_id("azure") - assert len(serialized) == 15 + assert len(serialized) == 22 assert [item["contract_id"] for item in serialized] == [ str(stable_contract_id("aws")), str(expected_id), @@ -106,6 +106,13 @@ def test_default_registration_identity_fields_and_outputs() -> None: str(stable_contract_id("cis/azure")), str(stable_contract_id("cis/gcp")), str(stable_contract_id("cis/kubernetes")), + str(stable_contract_id("nis2/aws")), + str(stable_contract_id("nis2/azure")), + str(stable_contract_id("nis2/gcp")), + str(stable_contract_id("iso27001/aws")), + str(stable_contract_id("iso27001/azure")), + str(stable_contract_id("iso27001/gcp")), + str(stable_contract_id("iso27001/kubernetes")), ] assert UUID(serialized[1]["contract_id"]) == expected_id assert expected_id.version == 5 diff --git a/prowler/tests/behaviour/chk009_base_gcp_provider/test_chk009_base_gcp_provider_bdd.py b/prowler/tests/behaviour/chk009_base_gcp_provider/test_chk009_base_gcp_provider_bdd.py index 4e07cdc1..353a9e12 100644 --- a/prowler/tests/behaviour/chk009_base_gcp_provider/test_chk009_base_gcp_provider_bdd.py +++ b/prowler/tests/behaviour/chk009_base_gcp_provider/test_chk009_base_gcp_provider_bdd.py @@ -93,7 +93,7 @@ def test_default_registration_identity_fields_and_outputs() -> None: serialized = DEFAULT_PROWLER_CONTRACTS.contracts() expected_id = stable_contract_id("gcp") - assert len(serialized) == 15 + assert len(serialized) == 22 assert [item["contract_id"] for item in serialized] == [ str(stable_contract_id("aws")), str(stable_contract_id("azure")), @@ -110,6 +110,13 @@ def test_default_registration_identity_fields_and_outputs() -> None: str(stable_contract_id("cis/azure")), str(stable_contract_id("cis/gcp")), str(stable_contract_id("cis/kubernetes")), + str(stable_contract_id("nis2/aws")), + str(stable_contract_id("nis2/azure")), + str(stable_contract_id("nis2/gcp")), + str(stable_contract_id("iso27001/aws")), + str(stable_contract_id("iso27001/azure")), + str(stable_contract_id("iso27001/gcp")), + str(stable_contract_id("iso27001/kubernetes")), ] assert UUID(serialized[2]["contract_id"]) == expected_id assert expected_id.version == 5 diff --git a/prowler/tests/behaviour/chk010_base_kubernetes_provider/test_chk010_base_kubernetes_provider_bdd.py b/prowler/tests/behaviour/chk010_base_kubernetes_provider/test_chk010_base_kubernetes_provider_bdd.py index f0f679a2..a3c16e59 100644 --- a/prowler/tests/behaviour/chk010_base_kubernetes_provider/test_chk010_base_kubernetes_provider_bdd.py +++ b/prowler/tests/behaviour/chk010_base_kubernetes_provider/test_chk010_base_kubernetes_provider_bdd.py @@ -93,7 +93,7 @@ def test_default_registration_identity_fields_and_outputs() -> None: serialized = DEFAULT_PROWLER_CONTRACTS.contracts() expected_id = stable_contract_id("kubernetes") - assert len(serialized) == 15 + assert len(serialized) == 22 assert [item["contract_id"] for item in serialized] == [ str(stable_contract_id("aws")), str(stable_contract_id("azure")), @@ -110,6 +110,13 @@ def test_default_registration_identity_fields_and_outputs() -> None: str(stable_contract_id("cis/azure")), str(stable_contract_id("cis/gcp")), str(stable_contract_id("cis/kubernetes")), + str(stable_contract_id("nis2/aws")), + str(stable_contract_id("nis2/azure")), + str(stable_contract_id("nis2/gcp")), + str(stable_contract_id("iso27001/aws")), + str(stable_contract_id("iso27001/azure")), + str(stable_contract_id("iso27001/gcp")), + str(stable_contract_id("iso27001/kubernetes")), ] assert UUID(serialized[3]["contract_id"]) == expected_id assert expected_id.version == 5 diff --git a/prowler/tests/behaviour/chk011_service_aws_assessment/test_chk011_service_aws_assessment_bdd.py b/prowler/tests/behaviour/chk011_service_aws_assessment/test_chk011_service_aws_assessment_bdd.py index d6820825..4abfcadc 100644 --- a/prowler/tests/behaviour/chk011_service_aws_assessment/test_chk011_service_aws_assessment_bdd.py +++ b/prowler/tests/behaviour/chk011_service_aws_assessment/test_chk011_service_aws_assessment_bdd.py @@ -95,9 +95,7 @@ def test_route_selects_exact_service_once( assert factory.calls[0][2:] == ((), service) -def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields() -> ( - None -): +def test_registry_has_22_canonical_contracts_without_selector_fields() -> None: """The public surface is ordered, stable, labelled, and not user-selectable.""" serialized = DEFAULT_PROWLER_CONTRACTS.contracts() routes = ( @@ -114,6 +112,13 @@ def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields( "cis/azure", "cis/gcp", "cis/kubernetes", + "nis2/aws", + "nis2/azure", + "nis2/gcp", + "iso27001/aws", + "iso27001/azure", + "iso27001/gcp", + "iso27001/kubernetes", ) assert [item["contract_id"] for item in serialized] == [ diff --git a/prowler/tests/behaviour/chk012_service_azure_assessment/test_chk012_service_azure_assessment_bdd.py b/prowler/tests/behaviour/chk012_service_azure_assessment/test_chk012_service_azure_assessment_bdd.py index 2d1ecb71..dff5949b 100644 --- a/prowler/tests/behaviour/chk012_service_azure_assessment/test_chk012_service_azure_assessment_bdd.py +++ b/prowler/tests/behaviour/chk012_service_azure_assessment/test_chk012_service_azure_assessment_bdd.py @@ -90,9 +90,7 @@ def test_route_selects_exact_service_once( assert factory.calls[0][2:] == ((), service) -def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields() -> ( - None -): +def test_registry_has_22_canonical_contracts_without_selector_fields() -> None: """The public surface is ordered, stable, labelled, and not user-selectable.""" serialized = DEFAULT_PROWLER_CONTRACTS.contracts() routes = ( @@ -110,6 +108,13 @@ def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields( "cis/azure", "cis/gcp", "cis/kubernetes", + "nis2/aws", + "nis2/azure", + "nis2/gcp", + "iso27001/aws", + "iso27001/azure", + "iso27001/gcp", + "iso27001/kubernetes", ) assert [item["contract_id"] for item in serialized] == [ diff --git a/prowler/tests/behaviour/chk013_service_gcp_assessment/test_chk013_service_gcp_assessment_bdd.py b/prowler/tests/behaviour/chk013_service_gcp_assessment/test_chk013_service_gcp_assessment_bdd.py index bf8d5ab2..997d855b 100644 --- a/prowler/tests/behaviour/chk013_service_gcp_assessment/test_chk013_service_gcp_assessment_bdd.py +++ b/prowler/tests/behaviour/chk013_service_gcp_assessment/test_chk013_service_gcp_assessment_bdd.py @@ -103,9 +103,7 @@ def test_selector_type_is_exactly_iam_or_compute() -> None: assert get_args(selector_type) == ("iam", "compute") -def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields() -> ( - None -): +def test_registry_has_22_canonical_contracts_without_selector_fields() -> None: """The public surface is ordered, stable, labelled, and not user-selectable.""" serialized = DEFAULT_PROWLER_CONTRACTS.contracts() routes = ( @@ -123,6 +121,13 @@ def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields( "cis/azure", "cis/gcp", "cis/kubernetes", + "nis2/aws", + "nis2/azure", + "nis2/gcp", + "iso27001/aws", + "iso27001/azure", + "iso27001/gcp", + "iso27001/kubernetes", ) assert [item["contract_id"] for item in serialized] == [ diff --git a/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py b/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py index 050c0c68..b5040bb1 100644 --- a/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py +++ b/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py @@ -37,6 +37,15 @@ ("cis/gcp", "gcp", "cis_3.0_gcp"), ("cis/kubernetes", "kubernetes", "cis_1.12_kubernetes"), ) +_CHK015_ROUTES = ( + "nis2/aws", + "nis2/azure", + "nis2/gcp", + "iso27001/aws", + "iso27001/azure", + "iso27001/gcp", + "iso27001/kubernetes", +) _TEMP_PATHS = { "gcp": Path("/tmp/CANARY-GCP-CREDENTIAL.json"), # noqa: S108 "kubernetes": Path("/tmp/CANARY-KUBE-CREDENTIAL.yaml"), # noqa: S108 @@ -116,19 +125,17 @@ def test_route_selects_exact_typed_compliance_once( assert factory.calls[0][2:] == ((), None, compliance) -def test_compliance_selector_type_contains_only_supported_cis_values() -> None: - """The internal typed seam admits only the four evidenced Prowler values.""" +def test_compliance_selector_type_retains_exact_supported_cis_values() -> None: + """The expanded internal typed seam retains all four exact CIS values.""" import prowler._core.prowler_client as client_api selector_type = client_api.__dict__.get("ComplianceSelector") assert selector_type is not None - assert get_args(selector_type) == tuple(item[2] for item in _ROUTES) + assert get_args(selector_type)[:4] == tuple(item[2] for item in _ROUTES) -def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields() -> ( - None -): +def test_registry_retains_cis_contracts_without_selector_fields() -> None: """The executable public surface is stable, ordered, and not user-selectable.""" serialized = DEFAULT_PROWLER_CONTRACTS.contracts() routes = ( @@ -144,12 +151,15 @@ def test_registry_has_exact_fifteen_canonical_contracts_without_selector_fields( "gcp/iam", "gcp/compute", *(item[0] for item in _ROUTES), + *_CHK015_ROUTES, ) assert [item["contract_id"] for item in serialized] == [ str(stable_contract_id(route)) for route in routes ] - for item, (route, provider_name, _) in zip(serialized[11:], _ROUTES, strict=True): + for item, (route, provider_name, _) in zip( + serialized[11:15], _ROUTES, strict=True + ): content = json.loads(item["contract_content"]) assert "cis" in content["label"]["en"].casefold() keys = tuple(field["key"] for field in content["fields"]) From 9a5dedbaa6f4d8f353c84cd6da97bd2918e590b3 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Fri, 28 Aug 2026 14:30:56 +0200 Subject: [PATCH 4/4] style(prowler): format CHK.015 registry regression (#422) --- .../test_chk014_compliance_cis_benchmark_bdd.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py b/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py index b5040bb1..88e1f469 100644 --- a/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py +++ b/prowler/tests/behaviour/chk014_compliance_cis_benchmark/test_chk014_compliance_cis_benchmark_bdd.py @@ -157,9 +157,7 @@ def test_registry_retains_cis_contracts_without_selector_fields() -> None: assert [item["contract_id"] for item in serialized] == [ str(stable_contract_id(route)) for route in routes ] - for item, (route, provider_name, _) in zip( - serialized[11:15], _ROUTES, strict=True - ): + for item, (route, provider_name, _) in zip(serialized[11:15], _ROUTES, strict=True): content = json.loads(item["contract_content"]) assert "cis" in content["label"]["en"].casefold() keys = tuple(field["key"] for field in content["fields"])