feat(prowler): chunk9 add the GCP base contract (#422)

prowler/prowler/contracts/__init__.py

@@ -12,6 +12,7 @@
 )
 from .catalog import ROUTE_CATALOG, RouteDescriptor
 from .dispatcher import ContractDispatcher, RouteHandler, RouteNotFoundError
+from .gcp import GcpBaseContract
 from .registry import (
     DEFAULT_PROWLER_CONTRACTS,
     PROWLER_CONTRACT_NAMESPACE,
@@ -23,6 +24,7 @@
     "BaseProwlerContract",
     "AwsBaseContract",
     "AzureBaseContract",
+    "GcpBaseContract",
     "ContractDispatcher",
     "ContractExecutionOutcome",
     "ContractInputError",

prowler/prowler/contracts/gcp.py

@@ -0,0 +1,35 @@
+"""Executable CHK.009 complete-scope GCP 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 GcpBaseContract(BaseProwlerContract):
+    """Run the complete GCP provider scope and retain mapped GCP findings."""
+
+    contract_id: ClassVar[str] = "91344897-632c-518d-ab75-818941b43ae7"
+    external_id: ClassVar[str] = "prowler:gcp"
+    route_name: ClassVar[str] = "gcp"
+    provider = "gcp"
+    family = "base"
+    label = "Prowler GCP"
+    check_filters = ()
+
+    def execute(
+        self, config: ProwlerConfig, provider: ProviderInput
+    ) -> ContractExecutionOutcome:
+        """Map one complete-scope result, preserving ordered GCP 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": "gcp"})
+            for finding in outcome.findings
+            if finding.cloud_provider.casefold() == "gcp"
+        )
+        return replace(outcome, findings=findings)

prowler/prowler/contracts/registry.py

@@ -11,6 +11,7 @@
 from .azure import AzureBaseContract
 from .base import BaseProwlerContract
 from .catalog import ROUTE_CATALOG
+from .gcp import GcpBaseContract
 
 # Committed project namespace: changing it would break stable platform identities.
 PROWLER_CONTRACT_NAMESPACE = UUID("ee49522d-80b9-5d71-b164-569ee61a75bd")
@@ -79,4 +80,6 @@ def contracts(self) -> list[dict[str, object]]:
         )
 
 
-DEFAULT_PROWLER_CONTRACTS = ProwlerContracts((AwsBaseContract, AzureBaseContract))
+DEFAULT_PROWLER_CONTRACTS = ProwlerContracts(
+    (AwsBaseContract, AzureBaseContract, GcpBaseContract)
+)

prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py

@@ -81,6 +81,7 @@ def _then_base_contracts_are_registered(config: ConfigLoader, helper: Mock) -> N
     assert [item["contract_id"] for item in contracts] == [
         str(stable_contract_id("aws")),
         str(stable_contract_id("azure")),
+        str(stable_contract_id("gcp")),
     ]
     callback = helper.listen.call_args.kwargs["message_callback"]
     assert callable(callback)
@@ -147,7 +148,7 @@ def test_foundation_configuration_excludes_future_provider_settings() -> None:
 def test_startup_registers_the_base_assessment_contracts(
     standard_injector_environment: None,
 ) -> None:
-    """CHK.008 starts its listener with canonical AWS and Azure contracts."""
+    """CHK.009 starts its listener with three canonical base contracts."""
     _given_the_prowler_project()
     config, helper = _when_injector_starts()
     _then_base_contracts_are_registered(config, helper)

prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py

@@ -93,6 +93,7 @@ def test_default_registration_identity_fields_and_outputs() -> None:
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(stable_contract_id("azure")),
+        str(stable_contract_id("gcp")),
     ]
     assert UUID(serialized[0]["contract_id"]) == expected_id
     assert expected_id.version == 5

prowler/tests/behaviour/chk008_base_azure_provider/test_chk008_base_azure_provider_bdd.py

@@ -85,14 +85,15 @@ def run(
 
 
 def test_default_registration_identity_fields_and_outputs() -> None:
-    """The default surface is exactly canonical AWS then Azure base routes."""
+    """Azure remains second as the canonical registry grows through GCP."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("azure")
 
-    assert len(serialized) == 2
+    assert len(serialized) == 3
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(expected_id),
+        str(stable_contract_id("gcp")),
     ]
     assert UUID(serialized[1]["contract_id"]) == expected_id
     assert expected_id.version == 5

prowler/tests/behaviour/chk009_base_gcp_provider/chk009_base_gcp_provider.feature

@@ -0,0 +1,43 @@
+Feature: CHK.009 GCP base provider
+  The registered GCP route validates local input, requests the complete GCP
+  provider scope once, and emits deterministic mapped GCP findings.
+
+  Scenario: The default registry exposes the stable GCP contract
+    Given the canonical routes "aws", "azure", and "gcp"
+    When the default contracts are serialized
+    Then exactly three contracts are registered in canonical order
+    And the GCP UUIDv5 and external ID are stable
+    And it inherits the GCP fields and shared outputs
+
+  Scenario: Invalid GCP input is rejected locally
+    Given a blank service-account or project value
+    When the GCP contract parses the form
+    Then it rejects the request without authenticating or dispatching Prowler
+
+  Scenario: Valid GCP input requests the complete provider scope
+    Given complete GCP service-account credentials and a project
+    When the GCP contract executes
+    Then the client is invoked exactly once
+    And one temporary JSON credential lease exists during fake engine execution
+    And the lease is cleaned after execution
+    And no service, check, or compliance narrowing is present
+
+  Scenario: Only normalized GCP findings are emitted
+    Given ordered OCSF records for GCP and other cloud providers
+    When a successful command result is mapped
+    Then only case-normalized GCP findings are retained in source order
+    And every retained finding preserves all 14 mapped fields
+    And text output contains every finding while vulnerability output contains only failures
+
+  Scenario: OpenAEV runtime success is completed end to end
+    Given a fake Prowler client and a valid GCP injection
+    When the injector receives the message
+    Then reception occurs before one SUCCESS callback
+    And structured output and the dynamic Rich trace contain the same mapped findings
+    And credentials and process internals are absent
+
+  Scenario: OpenAEV runtime failure is safe
+    Given an invalid GCP 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/chk009_base_gcp_provider/conftest.py

@@ -0,0 +1,110 @@
+"""Local deterministic CHK.009 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 gcp_form() -> dict[str, object]:
+    """Return structurally valid placeholder-only GCP form input."""
+    return {
+        "gcp_service_account_json": (
+            '{"private_key":"SERVICE-ACCOUNT-JSON-CANARY",' '"form":"FORM-CANARY"}'
+        ),
+        "gcp_project_id": "PROJECT-ID-CANARY",
+    }
+
+
+@pytest.fixture
+def gcp_ocsf_record_factory() -> Any:
+    """Build one complete minimal OCSF record for projection tests."""
+
+    def build(
+        title: str,
+        *,
+        provider: str = "gcp",
+        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": "us-central1",
+                "account": {"uid": "acme-prod"},
+            },
+            "unmapped": {"compliance": ["cis", "nis2"]},
+            "remediation": {
+                "desc": f"Remediate {title}",
+                "references": ["https://example.invalid/remediation"],
+            },
+        }
+
+    return build