feat(prowler): chunk13 add the GCP service routes (#422)

prowler/prowler/_core/prowler_client/__init__.py

@@ -7,7 +7,12 @@
     ProwlerClient,
     ProwlerClientConsumedError,
 )
-from .contracts import AwsServiceSelector, AzureServiceSelector, ServiceSelector
+from .contracts import (
+    AwsServiceSelector,
+    AzureServiceSelector,
+    GcpServiceSelector,
+    ServiceSelector,
+)
 from .credentials import (
     CredentialCleanupError,
     TemporaryCredentialLease,
@@ -40,6 +45,7 @@
     "OutputWorkspacePreparationError",
     "AwsServiceSelector",
     "AzureServiceSelector",
+    "GcpServiceSelector",
     "ProwlerClient",
     "ProwlerClientConsumedError",
     "ProwlerClientFactory",

prowler/prowler/_core/prowler_client/client.py

@@ -15,6 +15,7 @@
 from prowler.models.provider_inputs import (
     AwsProviderInput,
     AzureProviderInput,
+    GcpProviderInput,
     ProviderInput,
 )
 
@@ -124,7 +125,14 @@ def run(
             filters = tuple(check_filters)
             if any(not isinstance(item, str) or not item.strip() for item in filters):
                 raise ValueError("check filters must be nonblank strings")
-            if service_selector not in (None, "iam", "s3", "ec2", "storage"):
+            if service_selector not in (
+                None,
+                "iam",
+                "s3",
+                "ec2",
+                "storage",
+                "compute",
+            ):
                 raise ValueError("unsupported service selector")
             if service_selector in ("s3", "ec2") and not isinstance(
                 provider, AwsProviderInput
@@ -134,11 +142,15 @@ def run(
                 provider, AzureProviderInput
             ):
                 raise ValueError("Azure service selector requires an Azure provider")
+            if service_selector == "compute" and not isinstance(
+                provider, GcpProviderInput
+            ):
+                raise ValueError("GCP service selector requires a GCP provider")
             if service_selector == "iam" and not isinstance(
-                provider, AwsProviderInput | AzureProviderInput
+                provider, AwsProviderInput | AzureProviderInput | GcpProviderInput
             ):
                 raise ValueError(
-                    "IAM service selector requires an AWS or Azure provider"
+                    "IAM service selector requires an AWS, Azure, or GCP provider"
                 )
 
             _safe_log(logging.INFO, "Preparing Prowler output workspace")

prowler/prowler/_core/prowler_client/contracts.py

@@ -11,7 +11,8 @@
 
 AwsServiceSelector = Literal["iam", "s3", "ec2"]
 AzureServiceSelector = Literal["iam", "storage"]
-ServiceSelector = AwsServiceSelector | AzureServiceSelector
+GcpServiceSelector = Literal["iam", "compute"]
+ServiceSelector = AwsServiceSelector | AzureServiceSelector | GcpServiceSelector
 
 
 class CliEnginePort(Protocol):

prowler/prowler/contracts/__init__.py

@@ -23,7 +23,7 @@
 )
 from .catalog import ROUTE_CATALOG, RouteDescriptor
 from .dispatcher import ContractDispatcher, RouteHandler, RouteNotFoundError
-from .gcp import GcpBaseContract
+from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract, GcpServiceContract
 from .kubernetes import KubernetesBaseContract
 from .registry import (
     DEFAULT_PROWLER_CONTRACTS,
@@ -44,6 +44,9 @@
     "AzureServiceContract",
     "AzureStorageContract",
     "GcpBaseContract",
+    "GcpComputeContract",
+    "GcpIamContract",
+    "GcpServiceContract",
     "KubernetesBaseContract",
     "ContractDispatcher",
     "ContractExecutionOutcome",

prowler/prowler/contracts/aws.py

@@ -5,11 +5,6 @@
 
 from prowler._core.prowler_client import AwsServiceSelector
 from prowler.models.configs.config_loader import ProwlerConfig
-from prowler.models.findings import (
-    OcsfDecodeError,
-    OcsfMappingError,
-    map_command_result_with_evidence,
-)
 from prowler.models.provider_inputs import AwsProviderInput, ProviderInput
 
 from .base import BaseProwlerContract, ContractExecutionOutcome, RouteFamily
@@ -56,26 +51,7 @@ def execute(
             raise ValueError("unsupported AWS service selector")
         if not isinstance(provider, AwsProviderInput):
             raise ValueError("AWS service selector requires an AWS provider")
-        result = self._client_factory.run(
-            config,
-            provider,
-            check_filters=self.check_filters,
-            service_selector=self.service_selector,
-        )
-        if result.error is not None or result.return_code != 0:
-            return ContractExecutionOutcome(command_result=result, error=result.error)
-        try:
-            mapping = map_command_result_with_evidence(result)
-        except (OcsfDecodeError, OcsfMappingError) as error:
-            return ContractExecutionOutcome(command_result=result, error=error)
-        outcome = ContractExecutionOutcome(
-            command_result=result,
-            findings=mapping.findings,
-            raw_record_count=mapping.raw_record_count,
-            raw_output_bytes=mapping.raw_output_bytes,
-            raw_preview=mapping.raw_preview,
-        )
-        return replace(outcome, findings=self._provider_findings(outcome.findings))
+        return self._execute_service(config, provider, self.service_selector)
 
 
 class AwsIamContract(AwsServiceContract):

prowler/prowler/contracts/azure.py

@@ -5,11 +5,6 @@
 
 from prowler._core.prowler_client import AzureServiceSelector
 from prowler.models.configs.config_loader import ProwlerConfig
-from prowler.models.findings import (
-    OcsfDecodeError,
-    OcsfMappingError,
-    map_command_result_with_evidence,
-)
 from prowler.models.provider_inputs import AzureProviderInput, ProviderInput
 
 from .base import BaseProwlerContract, ContractExecutionOutcome, RouteFamily
@@ -56,26 +51,7 @@ def execute(
             raise ValueError("unsupported Azure service selector")
         if not isinstance(provider, AzureProviderInput):
             raise ValueError("Azure service selector requires an Azure provider")
-        result = self._client_factory.run(
-            config,
-            provider,
-            check_filters=self.check_filters,
-            service_selector=self.service_selector,
-        )
-        if result.error is not None or result.return_code != 0:
-            return ContractExecutionOutcome(command_result=result, error=result.error)
-        try:
-            mapping = map_command_result_with_evidence(result)
-        except (OcsfDecodeError, OcsfMappingError) as error:
-            return ContractExecutionOutcome(command_result=result, error=error)
-        outcome = ContractExecutionOutcome(
-            command_result=result,
-            findings=mapping.findings,
-            raw_record_count=mapping.raw_record_count,
-            raw_output_bytes=mapping.raw_output_bytes,
-            raw_preview=mapping.raw_preview,
-        )
-        return replace(outcome, findings=self._provider_findings(outcome.findings))
+        return self._execute_service(config, provider, self.service_selector)
 
 
 class AzureIamContract(AzureServiceContract):

prowler/prowler/contracts/base.py

@@ -218,6 +218,33 @@ def _provider_findings(
             if finding.cloud_provider.casefold() == self.provider
         )
 
+    def _execute_service(
+        self,
+        config: ProwlerConfig,
+        provider: ProviderInput,
+        service_selector: ServiceSelector,
+    ) -> ContractExecutionOutcome:
+        """Run one validated service selector and retain its provider findings."""
+        result = self._client_factory.run(
+            config,
+            provider,
+            check_filters=self.check_filters,
+            service_selector=service_selector,
+        )
+        if result.error is not None or result.return_code != 0:
+            return ContractExecutionOutcome(command_result=result, error=result.error)
+        try:
+            mapping = map_command_result_with_evidence(result)
+        except (OcsfDecodeError, OcsfMappingError) as error:
+            return ContractExecutionOutcome(command_result=result, error=error)
+        return ContractExecutionOutcome(
+            command_result=result,
+            findings=self._provider_findings(mapping.findings),
+            raw_record_count=mapping.raw_record_count,
+            raw_output_bytes=mapping.raw_output_bytes,
+            raw_preview=mapping.raw_preview,
+        )
+
     @staticmethod
     def output_trace_config() -> dict[str, object]:
         """Return the common flattened-field trace contract."""

prowler/prowler/contracts/gcp.py

@@ -1,12 +1,13 @@
-"""Executable CHK.009 complete-scope GCP base contract."""
+"""Executable complete-scope and service-specific GCP contracts."""
 
 from dataclasses import replace
 from typing import ClassVar
 
+from prowler._core.prowler_client import GcpServiceSelector
 from prowler.models.configs.config_loader import ProwlerConfig
-from prowler.models.provider_inputs import ProviderInput
+from prowler.models.provider_inputs import GcpProviderInput, ProviderInput
 
-from .base import BaseProwlerContract, ContractExecutionOutcome
+from .base import BaseProwlerContract, ContractExecutionOutcome, RouteFamily
 
 
 class GcpBaseContract(BaseProwlerContract):
@@ -16,7 +17,7 @@ class GcpBaseContract(BaseProwlerContract):
     external_id: ClassVar[str] = "prowler:gcp"
     route_name: ClassVar[str] = "gcp"
     provider = "gcp"
-    family = "base"
+    family: ClassVar[RouteFamily] = "base"
     label = "Prowler GCP"
     check_filters = ()
 
@@ -28,3 +29,46 @@ def execute(
         if outcome.error is not None or outcome.command_result.return_code != 0:
             return outcome
         return replace(outcome, findings=self._provider_findings(outcome.findings))
+
+
+class GcpServiceContract(GcpBaseContract):
+    """Execute one route-owned GCP service selector through the CHK.004 seam."""
+
+    family = "service"
+    service_selector: ClassVar[GcpServiceSelector]
+
+    def safe_request_info(self, provider: ProviderInput | None) -> dict[str, object]:
+        """Identify the service through safe route metadata, never form input."""
+        info = super().safe_request_info(provider)
+        info["filters"] = f"service={self.service_selector}"
+        return info
+
+    def execute(
+        self, config: ProwlerConfig, provider: ProviderInput
+    ) -> ContractExecutionOutcome:
+        """Reject invalid route combinations before one service-specific client call."""
+        if self.service_selector not in ("iam", "compute"):
+            raise ValueError("unsupported GCP service selector")
+        if not isinstance(provider, GcpProviderInput):
+            raise ValueError("GCP service selector requires a GCP provider")
+        return self._execute_service(config, provider, self.service_selector)
+
+
+class GcpIamContract(GcpServiceContract):
+    """Run only Prowler GCP IAM checks."""
+
+    contract_id = "0684cf36-b1fc-5cdf-926e-e4d0c9f90aea"
+    external_id = "prowler:gcp/iam"
+    route_name = "gcp/iam"
+    label = "Prowler GCP IAM"
+    service_selector = "iam"
+
+
+class GcpComputeContract(GcpServiceContract):
+    """Run only Prowler GCP Compute checks."""
+
+    contract_id = "b50c4b61-1c03-55ef-80eb-2a45c000264d"
+    external_id = "prowler:gcp/compute"
+    route_name = "gcp/compute"
+    label = "Prowler GCP Compute"
+    service_selector = "compute"

prowler/prowler/contracts/registry.py

@@ -11,7 +11,7 @@
 from .azure import AzureBaseContract, AzureIamContract, AzureStorageContract
 from .base import BaseProwlerContract
 from .catalog import ROUTE_CATALOG
-from .gcp import GcpBaseContract
+from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract
 from .kubernetes import KubernetesBaseContract
 
 # Committed project namespace: changing it would break stable platform identities.
@@ -92,5 +92,7 @@ def contracts(self) -> list[dict[str, object]]:
         AwsEc2Contract,
         AzureIamContract,
         AzureStorageContract,
+        GcpIamContract,
+        GcpComputeContract,
     )
 )

prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py

@@ -88,6 +88,8 @@ def _then_base_contracts_are_registered(config: ConfigLoader, helper: Mock) -> N
         str(stable_contract_id("aws/ec2")),
         str(stable_contract_id("azure/iam")),
         str(stable_contract_id("azure/storage")),
+        str(stable_contract_id("gcp/iam")),
+        str(stable_contract_id("gcp/compute")),
     ]
     callback = helper.listen.call_args.kwargs["message_callback"]
     assert callable(callback)

prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py

@@ -100,6 +100,8 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("aws/ec2")),
         str(stable_contract_id("azure/iam")),
         str(stable_contract_id("azure/storage")),
+        str(stable_contract_id("gcp/iam")),
+        str(stable_contract_id("gcp/compute")),
     ]
     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,11 +85,11 @@ def run(
 
 
 def test_default_registration_identity_fields_and_outputs() -> None:
-    """Azure remains second as the canonical registry grows through CHK.012."""
+    """Azure remains second as the canonical registry grows through CHK.013."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("azure")
 
-    assert len(serialized) == 9
+    assert len(serialized) == 11
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(expected_id),
@@ -100,6 +100,8 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("aws/ec2")),
         str(stable_contract_id("azure/iam")),
         str(stable_contract_id("azure/storage")),
+        str(stable_contract_id("gcp/iam")),
+        str(stable_contract_id("gcp/compute")),
     ]
     assert UUID(serialized[1]["contract_id"]) == expected_id
     assert expected_id.version == 5

prowler/tests/behaviour/chk009_base_gcp_provider/test_chk009_base_gcp_provider_bdd.py

@@ -89,11 +89,11 @@ def run(
 
 
 def test_default_registration_identity_fields_and_outputs() -> None:
-    """GCP remains third as the canonical registry grows through CHK.012."""
+    """GCP remains third as the canonical registry grows through CHK.013."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("gcp")
 
-    assert len(serialized) == 9
+    assert len(serialized) == 11
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(stable_contract_id("azure")),
@@ -104,6 +104,8 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("aws/ec2")),
         str(stable_contract_id("azure/iam")),
         str(stable_contract_id("azure/storage")),
+        str(stable_contract_id("gcp/iam")),
+        str(stable_contract_id("gcp/compute")),
     ]
     assert UUID(serialized[2]["contract_id"]) == expected_id
     assert expected_id.version == 5

prowler/tests/behaviour/chk010_base_kubernetes_provider/test_chk010_base_kubernetes_provider_bdd.py

@@ -89,11 +89,11 @@ def run(
 
 
 def test_default_registration_identity_fields_and_outputs() -> None:
-    """The four base routes remain first in the CHK.012 executable surface."""
+    """The four base routes remain first in the CHK.013 executable surface."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("kubernetes")
 
-    assert len(serialized) == 9
+    assert len(serialized) == 11
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(stable_contract_id("azure")),
@@ -104,6 +104,8 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("aws/ec2")),
         str(stable_contract_id("azure/iam")),
         str(stable_contract_id("azure/storage")),
+        str(stable_contract_id("gcp/iam")),
+        str(stable_contract_id("gcp/compute")),
     ]
     assert UUID(serialized[3]["contract_id"]) == expected_id
     assert expected_id.version == 5