feat(prowler): chunk8 add the Azure base contract (#422)

prowler/prowler/contracts/__init__.py

@@ -1,6 +1,7 @@
 """Prowler contract declarations and the executable default registry."""
 
 from .aws import AwsBaseContract
+from .azure import AzureBaseContract
 from .base import (
     BaseProwlerContract,
     ContractExecutionOutcome,
@@ -21,6 +22,7 @@
 __all__ = [
     "BaseProwlerContract",
     "AwsBaseContract",
+    "AzureBaseContract",
     "ContractDispatcher",
     "ContractExecutionOutcome",
     "ContractInputError",

prowler/prowler/contracts/azure.py

@@ -0,0 +1,35 @@
+"""Executable CHK.008 complete-scope Azure 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 AzureBaseContract(BaseProwlerContract):
+    """Run the complete Azure provider scope and retain mapped Azure findings."""
+
+    contract_id: ClassVar[str] = "00558d49-06ee-5a6f-80e6-4dae205be992"
+    external_id: ClassVar[str] = "prowler:azure"
+    route_name: ClassVar[str] = "azure"
+    provider = "azure"
+    family = "base"
+    label = "Prowler Azure"
+    check_filters = ()
+
+    def execute(
+        self, config: ProwlerConfig, provider: ProviderInput
+    ) -> ContractExecutionOutcome:
+        """Map one complete-scope result, preserving ordered Azure 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": "azure"})
+            for finding in outcome.findings
+            if finding.cloud_provider.casefold() == "azure"
+        )
+        return replace(outcome, findings=findings)

prowler/prowler/contracts/registry.py

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

prowler/tests/behaviour/chk001_catalog_scaffold/test_chk001_catalog_scaffold_bdd.py

@@ -75,10 +75,13 @@ def _when_injector_starts() -> tuple[ConfigLoader, Mock]:
     return config, helper
 
 
-def _then_aws_contract_is_registered(config: ConfigLoader, helper: Mock) -> None:
+def _then_base_contracts_are_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"))
+    assert isinstance(contracts, list)
+    assert [item["contract_id"] for item in contracts] == [
+        str(stable_contract_id("aws")),
+        str(stable_contract_id("azure")),
+    ]
     callback = helper.listen.call_args.kwargs["message_callback"]
     assert callable(callback)
 
@@ -141,13 +144,13 @@ def test_foundation_configuration_excludes_future_provider_settings() -> None:
     _then_only_standard_and_prowler_settings_are_available(settings)
 
 
-def test_startup_registers_the_aws_assessment_contract(
+def test_startup_registers_the_base_assessment_contracts(
     standard_injector_environment: None,
 ) -> None:
-    """CHK.007 starts its listener with the AWS base contract."""
+    """CHK.008 starts its listener with canonical AWS and Azure contracts."""
     _given_the_prowler_project()
     config, helper = _when_injector_starts()
-    _then_aws_contract_is_registered(config, helper)
+    _then_base_contracts_are_registered(config, helper)
 
 
 @pytest.mark.parametrize(

prowler/tests/behaviour/chk007_base_aws_provider/test_chk007_base_aws_provider_bdd.py

@@ -86,11 +86,14 @@ def run(
 
 
 def test_default_registration_identity_fields_and_outputs() -> None:
-    """The executable default surface is exactly the canonical AWS base route."""
+    """The canonical AWS base route remains first in the executable surface."""
     serialized = DEFAULT_PROWLER_CONTRACTS.contracts()
     expected_id = stable_contract_id("aws")
 
-    assert len(serialized) == 1
+    assert [item["contract_id"] for item in serialized] == [
+        str(stable_contract_id("aws")),
+        str(stable_contract_id("azure")),
+    ]
     assert UUID(serialized[0]["contract_id"]) == expected_id
     assert expected_id.version == 5
     content = json.loads(serialized[0]["contract_content"])

prowler/tests/behaviour/chk008_base_azure_provider/__init__.py

@@ -0,0 +1 @@
+"""CHK.008 Azure base-provider behaviour tests."""

prowler/tests/behaviour/chk008_base_azure_provider/chk008_base_azure_provider.feature

@@ -0,0 +1,42 @@
+Feature: CHK.008 Azure base provider
+  The registered Azure route validates local input, requests the complete Azure
+  provider scope once, and emits deterministic mapped Azure findings.
+
+  Scenario: The default registry exposes the stable Azure contract
+    Given the canonical routes "aws" and "azure"
+    When the default contracts are serialized
+    Then exactly two contracts are registered in canonical order
+    And the Azure UUIDv5 and external ID are stable
+    And it inherits the Azure fields and shared outputs
+
+  Scenario: Invalid Azure input is rejected locally
+    Given a blank tenant, client, secret, subscription, or provider value
+    When the Azure contract parses the form
+    Then it rejects the request without authenticating or dispatching Prowler
+
+  Scenario: Valid Azure input requests the complete provider scope
+    Given complete Azure service-principal credentials and target details
+    When the Azure contract executes
+    Then the client is invoked exactly once
+    And the command contains the Azure authentication, subscription, provider, and OCSF arguments
+    And no service, check, or compliance narrowing is present
+
+  Scenario: Only normalized Azure findings are emitted
+    Given ordered OCSF records for Azure and other cloud providers
+    When a successful command result is mapped
+    Then only case-normalized Azure 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 Azure 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 Azure 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/chk008_base_azure_provider/conftest.py

@@ -0,0 +1,111 @@
+"""Local deterministic CHK.008 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 azure_form() -> dict[str, object]:
+    """Return structurally valid placeholder-only Azure form input."""
+    return {
+        "azure_tenant_id": "CANARY-TENANT",
+        "azure_client_id": "CANARY-CLIENT",
+        "azure_client_secret": "CANARY-SECRET",
+        "azure_subscription_id": "subscription-123",
+        "azure_provider": "Microsoft.Compute",
+    }
+
+
+@pytest.fixture
+def azure_ocsf_record_factory() -> Any:
+    """Build one complete minimal OCSF record for projection tests."""
+
+    def build(
+        title: str,
+        *,
+        provider: str = "azure",
+        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": "westeurope",
+                "account": {"uid": "subscription-123"},
+            },
+            "unmapped": {"compliance": ["cis", "nis2"]},
+            "remediation": {
+                "desc": f"Remediate {title}",
+                "references": ["https://example.invalid/remediation"],
+            },
+        }
+
+    return build