feat(prowler): chunk7 add the AWS base contract (#422)

prowler/prowler/contracts/__init__.py

@@ -1,5 +1,6 @@
-"""Reusable CHK.006 contract declarations; no contracts are registered here."""
+"""Prowler contract declarations and the executable default registry."""
 
+from .aws import AwsBaseContract
 from .base import (
     BaseProwlerContract,
     ContractExecutionOutcome,
@@ -19,6 +20,7 @@
 
 __all__ = [
     "BaseProwlerContract",
+    "AwsBaseContract",
     "ContractDispatcher",
     "ContractExecutionOutcome",
     "ContractInputError",

prowler/prowler/contracts/aws.py

@@ -0,0 +1,35 @@
+"""Executable CHK.007 complete-scope AWS base contract."""
+
+from dataclasses import replace
+from typing import ClassVar
+
+from prowler.models.configs.config_loader import ProwlerConfig
+from prowler.models.provider_inputs import ProviderInput
+
+from .base import BaseProwlerContract, ContractExecutionOutcome
+
+
+class AwsBaseContract(BaseProwlerContract):
+    """Run the complete AWS provider scope and retain only mapped AWS findings."""
+
+    contract_id: ClassVar[str] = "a0464aa7-9451-54ea-bc00-3e89019a315a"
+    external_id: ClassVar[str] = "prowler:aws"
+    route_name: ClassVar[str] = "aws"
+    provider = "aws"
+    family = "base"
+    label = "Prowler AWS"
+    check_filters = ()
+
+    def execute(
+        self, config: ProwlerConfig, provider: ProviderInput
+    ) -> ContractExecutionOutcome:
+        """Map one complete-scope result, preserving ordered AWS findings only."""
+        outcome = super().execute(config, provider)
+        if outcome.error is not None or outcome.command_result.return_code != 0:
+            return outcome
+        findings = tuple(
+            finding.model_copy(update={"cloud_provider": "aws"})
+            for finding in outcome.findings
+            if finding.cloud_provider.casefold() == "aws"
+        )
+        return replace(outcome, findings=findings)

prowler/prowler/contracts/registry.py

@@ -7,6 +7,7 @@
 
 from pyoaev.contracts.contract_config import prepare_contracts
 
+from .aws import AwsBaseContract
 from .base import BaseProwlerContract
 from .catalog import ROUTE_CATALOG
 
@@ -77,4 +78,4 @@ def contracts(self) -> list[dict[str, object]]:
         )
 
 
-DEFAULT_PROWLER_CONTRACTS = ProwlerContracts()
+DEFAULT_PROWLER_CONTRACTS = ProwlerContracts((AwsBaseContract,))

prowler/prowler/injector/openaev_prowler.py

@@ -362,7 +362,7 @@ class _FailurePresentation:
 
 
 class ProwlerInjector:
-    """Register the foundation injector without assessment contracts."""
+    """Register and execute the available Prowler assessment contracts."""
 
     def __init__(
         self,

prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py

@@ -10,6 +10,7 @@
 from pydantic import BaseModel, ValidationError
 
 from prowler import __main__ as prowler_main
+from prowler.contracts import stable_contract_id
 from prowler.injector.openaev_prowler import ProwlerInjector
 from prowler.models.configs.config_loader import ConfigLoader
 
@@ -74,8 +75,10 @@ def _when_injector_starts() -> tuple[ConfigLoader, Mock]:
     return config, helper
 
 
-def _then_zero_contracts_are_registered(config: ConfigLoader, helper: Mock) -> None:
-    assert config.to_daemon_config().get("injector_contracts") == []
+def _then_aws_contract_is_registered(config: ConfigLoader, helper: Mock) -> None:
+    contracts = config.to_daemon_config().get("injector_contracts")
+    assert isinstance(contracts, list) and len(contracts) == 1
+    assert contracts[0]["contract_id"] == str(stable_contract_id("aws"))
     callback = helper.listen.call_args.kwargs["message_callback"]
     assert callable(callback)
 
@@ -138,13 +141,13 @@ def test_foundation_configuration_excludes_future_provider_settings() -> None:
     _then_only_standard_and_prowler_settings_are_available(settings)
 
 
-def test_foundation_startup_registers_no_assessment_contracts(
+def test_startup_registers_the_aws_assessment_contract(
     standard_injector_environment: None,
 ) -> None:
-    """The foundation starts its listener with an empty contract catalog."""
+    """CHK.007 starts its listener with the AWS base contract."""
     _given_the_prowler_project()
     config, helper = _when_injector_starts()
-    _then_zero_contracts_are_registered(config, helper)
+    _then_aws_contract_is_registered(config, helper)
 
 
 @pytest.mark.parametrize(

prowler/tests/behaviour/chk007_base_aws_provider/__init__.py

@@ -0,0 +1 @@
+"""CHK.007 AWS base-provider behaviour tests."""

prowler/tests/behaviour/chk007_base_aws_provider/chk007_base_aws_provider.feature

@@ -0,0 +1,41 @@
+Feature: CHK.007 AWS base provider
+  The registered AWS route validates local input, requests the complete AWS
+  provider scope once, and emits deterministic mapped AWS findings.
+
+  Scenario: The default registry exposes one stable AWS contract
+    Given the canonical route "aws"
+    When the default contracts are serialized
+    Then exactly one contract is registered in canonical order
+    And its UUIDv5 and external ID are stable
+    And it inherits the AWS fields and shared outputs
+
+  Scenario: Invalid AWS input is rejected locally
+    Given missing credentials, a blank region, or a non-12-digit account ID
+    When the AWS contract parses the form
+    Then it rejects the request without authenticating or dispatching Prowler
+
+  Scenario: Valid AWS input requests the complete provider scope
+    Given complete AWS credentials and target details
+    When the AWS contract executes
+    Then the client is invoked exactly once
+    And the command contains the AWS provider, region, and OCSF output arguments
+    And no service, check, or compliance narrowing is present
+
+  Scenario: Only normalized AWS findings are emitted
+    Given ordered OCSF records for AWS and other cloud providers
+    When a successful command result is mapped
+    Then only case-normalized AWS findings are retained in source order
+    And every retained finding preserves all 14 mapped fields
+
+  Scenario: OpenAEV runtime success is completed end to end
+    Given a fake Prowler client and a valid AWS injection
+    When the injector receives the message
+    Then reception occurs before one SUCCESS callback
+    And structured output and the Rich trace contain the same mapped findings
+    And credentials and process internals are absent
+
+  Scenario: OpenAEV runtime failure is safe
+    Given an invalid AWS injection containing secret canaries
+    When the injector receives the message
+    Then no Prowler request is made
+    And one ERROR callback contains no secret canary

prowler/tests/behaviour/chk007_base_aws_provider/conftest.py

@@ -0,0 +1,112 @@
+"""Local deterministic CHK.007 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 aws_form() -> dict[str, object]:
+    """Return structurally valid placeholder-only AWS form input."""
+    return {
+        "aws_access_key_id": "CANARY-ACCESS-KEY",
+        "aws_secret_access_key": "CANARY-SECRET-KEY",
+        "aws_account_id": "123456789012",
+        "aws_region": "eu-west-1",
+        "aws_endpoint_url": "https://aws.internal.example:8443",
+        "aws_session_token": "CANARY-SESSION-TOKEN",
+    }
+
+
+@pytest.fixture
+def ocsf_record_factory() -> Any:
+    """Build one complete minimal OCSF record for projection tests."""
+
+    def build(
+        title: str,
+        *,
+        provider: str = "aws",
+        status: str = "FAIL",
+    ) -> 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
+        return {
+            "finding_info": {
+                "uid": f"check-{title}",
+                "title": title,
+                "desc": f"Description {title}",
+            },
+            "status": record_status,
+            "status_code": status_code,
+            "severity": "High",
+            "resources": [{"uid": f"asset-{title}", "name": f"Asset {title}"}],
+            "cloud": {
+                "provider": provider,
+                "region": "eu-west-1",
+                "account": {"uid": "123456789012"},
+            },
+            "unmapped": {"compliance": ["cis", "nis2"]},
+            "remediation": {
+                "desc": f"Remediate {title}",
+                "references": ["https://example.invalid/remediation"],
+            },
+        }
+
+    return build