feat(prowler): chunk15 add the NIS2/ISO27001 compliance contract (#422)

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,
 }
 
 

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",
 ]
 
 

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",

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"

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,
     )
 )

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)

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

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

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

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

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] == [

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] == [