feat(prowler): chunk14 add the CIS benchmark contract (#422)

prowler/prowler/_core/prowler_client/__init__.py

@@ -10,6 +10,7 @@
 from .contracts import (
     AwsServiceSelector,
     AzureServiceSelector,
+    ComplianceSelector,
     GcpServiceSelector,
     ServiceSelector,
 )
@@ -43,6 +44,7 @@
     "OutputArtifactError",
     "OutputWorkspaceCleanupError",
     "OutputWorkspacePreparationError",
+    "ComplianceSelector",
     "AwsServiceSelector",
     "AzureServiceSelector",
     "GcpServiceSelector",

prowler/prowler/_core/prowler_client/client.py

@@ -16,11 +16,14 @@
     AwsProviderInput,
     AzureProviderInput,
     GcpProviderInput,
+    ImmutableProviderInput,
+    KubernetesProviderInput,
     ProviderInput,
 )
 
 from .contracts import (
     CliEnginePort,
+    ComplianceSelector,
     OutputWorkspaceFactoryPort,
     ServiceSelector,
 )
@@ -79,6 +82,25 @@ def _safe_log(level: int, message: str, **metadata: object) -> None:
         return
 
 
+_COMPLIANCE_PROVIDER_TYPES: dict[ComplianceSelector, type[ImmutableProviderInput]] = {
+    "cis_3.0_aws": AwsProviderInput,
+    "cis_3.0_azure": AzureProviderInput,
+    "cis_3.0_gcp": GcpProviderInput,
+    "cis_1.12_kubernetes": KubernetesProviderInput,
+}
+
+
+def _validate_compliance_provider(
+    provider: ProviderInput, compliance_selector: ComplianceSelector
+) -> None:
+    """Reject unknown or cross-provider compliance selection before adaptation."""
+    provider_type = _COMPLIANCE_PROVIDER_TYPES.get(compliance_selector)
+    if provider_type is None:
+        raise ValueError("unsupported compliance selector")
+    if not isinstance(provider, provider_type):
+        raise ValueError("compliance selector requires its matching provider")
+
+
 class ProwlerClientConsumedError(RuntimeError):
     """Reject reuse of a client whose provider input was already consumed."""
 
@@ -107,6 +129,7 @@ def run(
         check_filters: Sequence[str] = (),
         *,
         service_selector: ServiceSelector | None = None,
+        compliance_selector: ComplianceSelector | None = None,
     ) -> CommandResult:
         """Run one assessment and capture its controlled OCSF artifact."""
         with self._consumption_lock:
@@ -152,6 +175,13 @@ def run(
                 raise ValueError(
                     "IAM service selector requires an AWS, Azure, or GCP provider"
                 )
+            if compliance_selector is not None:
+                _validate_compliance_provider(provider, compliance_selector)
+                if filters or service_selector is not None:
+                    raise ValueError(
+                        "compliance selector cannot be combined with check or "
+                        "service selectors"
+                    )
 
             _safe_log(logging.INFO, "Preparing Prowler output workspace")
             try:
@@ -176,10 +206,16 @@ def run(
             service_arguments = (
                 ("--services", service_selector) if service_selector is not None else ()
             )
+            compliance_arguments = (
+                ("--compliance", compliance_selector)
+                if compliance_selector is not None
+                else ()
+            )
             provider_and_selectors = (
                 *invocation.arguments,
                 *filter_arguments,
                 *service_arguments,
+                *compliance_arguments,
             )
             narrowed = any(
                 argument in _NARROWING_OPTIONS for argument in provider_and_selectors

prowler/prowler/_core/prowler_client/contracts.py

@@ -13,6 +13,12 @@
 AzureServiceSelector = Literal["iam", "storage"]
 GcpServiceSelector = Literal["iam", "compute"]
 ServiceSelector = AwsServiceSelector | AzureServiceSelector | GcpServiceSelector
+ComplianceSelector = Literal[
+    "cis_3.0_aws",
+    "cis_3.0_azure",
+    "cis_3.0_gcp",
+    "cis_1.12_kubernetes",
+]
 
 
 class CliEnginePort(Protocol):

prowler/prowler/_core/prowler_client/factory.py

@@ -10,6 +10,7 @@
 from .client import ProwlerClient
 from .contracts import (
     CliEngineFactoryPort,
+    ComplianceSelector,
     CredentialLeaseFactoryPort,
     OutputWorkspaceFactoryPort,
     ServiceSelector,
@@ -48,8 +49,11 @@ def run(
         *,
         check_filters: Sequence[str] = (),
         service_selector: ServiceSelector | None = None,
+        compliance_selector: ComplianceSelector | None = None,
     ) -> CommandResult:
         """Create a client and synchronously run one assessment."""
         return self.create(config, provider).run(
-            check_filters, service_selector=service_selector
+            check_filters,
+            service_selector=service_selector,
+            compliance_selector=compliance_selector,
         )

prowler/prowler/contracts/__init__.py

@@ -22,6 +22,13 @@
     RouteFamily,
 )
 from .catalog import ROUTE_CATALOG, RouteDescriptor
+from .cis import (
+    AwsCisContract,
+    AzureCisContract,
+    CisComplianceContract,
+    GcpCisContract,
+    KubernetesCisContract,
+)
 from .dispatcher import ContractDispatcher, RouteHandler, RouteNotFoundError
 from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract, GcpServiceContract
 from .kubernetes import KubernetesBaseContract
@@ -43,11 +50,16 @@
     "AzureIamContract",
     "AzureServiceContract",
     "AzureStorageContract",
+    "AwsCisContract",
+    "AzureCisContract",
+    "CisComplianceContract",
     "GcpBaseContract",
     "GcpComputeContract",
     "GcpIamContract",
     "GcpServiceContract",
+    "GcpCisContract",
     "KubernetesBaseContract",
+    "KubernetesCisContract",
     "ContractDispatcher",
     "ContractExecutionOutcome",
     "ContractInputError",

prowler/prowler/contracts/base.py

@@ -21,7 +21,11 @@
 )
 
 from prowler._core.cli_engine import CommandResult
-from prowler._core.prowler_client import ProwlerClientFactory, ServiceSelector
+from prowler._core.prowler_client import (
+    ComplianceSelector,
+    ProwlerClientFactory,
+    ServiceSelector,
+)
 from prowler.models.configs.config_loader import ProwlerConfig
 from prowler.models.findings import (
     OcsfDecodeError,
@@ -66,6 +70,7 @@ def run(
         *,
         check_filters: Sequence[str] = (),
         service_selector: ServiceSelector | None = None,
+        compliance_selector: ComplianceSelector | None = None,
     ) -> CommandResult:
         """Run one assessment and return the exact command result."""
 
@@ -245,6 +250,33 @@ def _execute_service(
             raw_preview=mapping.raw_preview,
         )
 
+    def _execute_compliance(
+        self,
+        config: ProwlerConfig,
+        provider: ProviderInput,
+        compliance_selector: ComplianceSelector,
+    ) -> ContractExecutionOutcome:
+        """Run one validated compliance selector and retain provider findings."""
+        result = self._client_factory.run(
+            config,
+            provider,
+            check_filters=self.check_filters,
+            compliance_selector=compliance_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/cis.py

@@ -0,0 +1,89 @@
+"""Executable CIS compliance contracts for the four supported providers."""
+
+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,
+)
+
+_CIS_BY_PROVIDER: dict[ProviderName, ComplianceSelector] = {
+    "aws": "cis_3.0_aws",
+    "azure": "cis_3.0_azure",
+    "gcp": "cis_3.0_gcp",
+    "kubernetes": "cis_1.12_kubernetes",
+}
+
+
+class CisComplianceContract(BaseProwlerContract):
+    """Execute one provider-owned CIS selector through the CHK.004 seam."""
+
+    family: ClassVar[RouteFamily] = "compliance"
+    compliance_selector: ClassVar[ComplianceSelector]
+
+    def safe_request_info(self, provider: ProviderInput | None) -> dict[str, object]:
+        """Identify CIS 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 (
+            _CIS_BY_PROVIDER.get(self.provider) != self.compliance_selector
+            or provider.provider != self.provider
+        ):
+            raise ValueError("unsupported CIS compliance selection")
+        return self._execute_compliance(config, provider, self.compliance_selector)
+
+
+class AwsCisContract(CisComplianceContract):
+    """Run the Prowler 5.36 AWS CIS 3.0 framework."""
+
+    contract_id = "f0766dbc-b04f-5b4b-b762-0aee15884ead"
+    external_id = "prowler:cis/aws"
+    route_name = "cis/aws"
+    provider = "aws"
+    label = "Prowler AWS CIS"
+    compliance_selector = "cis_3.0_aws"
+
+
+class AzureCisContract(CisComplianceContract):
+    """Run the Prowler 5.36 Azure CIS 3.0 framework."""
+
+    contract_id = "3acf796f-71a9-523a-b6dc-fca321ad3bac"
+    external_id = "prowler:cis/azure"
+    route_name = "cis/azure"
+    provider = "azure"
+    label = "Prowler Azure CIS"
+    compliance_selector = "cis_3.0_azure"
+
+
+class GcpCisContract(CisComplianceContract):
+    """Run the Prowler 5.36 GCP CIS 3.0 framework."""
+
+    contract_id = "aec2729b-60c5-59e9-9383-78c5b34507cb"
+    external_id = "prowler:cis/gcp"
+    route_name = "cis/gcp"
+    provider = "gcp"
+    label = "Prowler GCP CIS"
+    compliance_selector = "cis_3.0_gcp"
+
+
+class KubernetesCisContract(CisComplianceContract):
+    """Run the Prowler 5.36 Kubernetes CIS 1.12 framework."""
+
+    contract_id = "a846a50c-16b3-5df5-87cf-de279637a2e9"
+    external_id = "prowler:cis/kubernetes"
+    route_name = "cis/kubernetes"
+    provider = "kubernetes"
+    label = "Prowler Kubernetes CIS"
+    compliance_selector = "cis_1.12_kubernetes"

prowler/prowler/contracts/registry.py

@@ -11,6 +11,7 @@
 from .azure import AzureBaseContract, AzureIamContract, AzureStorageContract
 from .base import BaseProwlerContract
 from .catalog import ROUTE_CATALOG
+from .cis import AwsCisContract, AzureCisContract, GcpCisContract, KubernetesCisContract
 from .gcp import GcpBaseContract, GcpComputeContract, GcpIamContract
 from .kubernetes import KubernetesBaseContract
 
@@ -94,5 +95,9 @@ def contracts(self) -> list[dict[str, object]]:
         AzureStorageContract,
         GcpIamContract,
         GcpComputeContract,
+        AwsCisContract,
+        AzureCisContract,
+        GcpCisContract,
+        KubernetesCisContract,
     )
 )

prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py

@@ -90,6 +90,10 @@ def _then_base_contracts_are_registered(config: ConfigLoader, helper: Mock) -> N
         str(stable_contract_id("azure/storage")),
         str(stable_contract_id("gcp/iam")),
         str(stable_contract_id("gcp/compute")),
+        str(stable_contract_id("cis/aws")),
+        str(stable_contract_id("cis/azure")),
+        str(stable_contract_id("cis/gcp")),
+        str(stable_contract_id("cis/kubernetes")),
     ]
     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

@@ -102,6 +102,10 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("azure/storage")),
         str(stable_contract_id("gcp/iam")),
         str(stable_contract_id("gcp/compute")),
+        str(stable_contract_id("cis/aws")),
+        str(stable_contract_id("cis/azure")),
+        str(stable_contract_id("cis/gcp")),
+        str(stable_contract_id("cis/kubernetes")),
     ]
     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.013."""
+    """Azure remains second as the canonical registry grows through CHK.014."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("azure")
 
-    assert len(serialized) == 11
+    assert len(serialized) == 15
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(expected_id),
@@ -102,6 +102,10 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("azure/storage")),
         str(stable_contract_id("gcp/iam")),
         str(stable_contract_id("gcp/compute")),
+        str(stable_contract_id("cis/aws")),
+        str(stable_contract_id("cis/azure")),
+        str(stable_contract_id("cis/gcp")),
+        str(stable_contract_id("cis/kubernetes")),
     ]
     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.013."""
+    """GCP remains third as the canonical registry grows through CHK.014."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("gcp")
 
-    assert len(serialized) == 11
+    assert len(serialized) == 15
     assert [item["contract_id"] for item in serialized] == [
         str(stable_contract_id("aws")),
         str(stable_contract_id("azure")),
@@ -106,6 +106,10 @@ def test_default_registration_identity_fields_and_outputs() -> None:
         str(stable_contract_id("azure/storage")),
         str(stable_contract_id("gcp/iam")),
         str(stable_contract_id("gcp/compute")),
+        str(stable_contract_id("cis/aws")),
+        str(stable_contract_id("cis/azure")),
+        str(stable_contract_id("cis/gcp")),
+        str(stable_contract_id("cis/kubernetes")),
     ]
     assert UUID(serialized[2]["contract_id"]) == expected_id
     assert expected_id.version == 5