feat(compliance): add DORA framework for AWS (#11131)

api/src/backend/tasks/jobs/export.py

@@ -39,11 +39,6 @@ from prowler.lib.outputs.compliance.cis.cis_oraclecloud import OracleCloudCIS
 from prowler.lib.outputs.compliance.cisa_scuba.cisa_scuba_googleworkspace import (
     GoogleWorkspaceCISASCuBA,
 )
-from prowler.lib.outputs.compliance.csa.csa_alibabacloud import AlibabaCloudCSA
-from prowler.lib.outputs.compliance.csa.csa_aws import AWSCSA
-from prowler.lib.outputs.compliance.csa.csa_azure import AzureCSA
-from prowler.lib.outputs.compliance.csa.csa_gcp import GCPCSA
-from prowler.lib.outputs.compliance.csa.csa_oraclecloud import OracleCloudCSA
 from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS
 from prowler.lib.outputs.compliance.ens.ens_azure import AzureENS
 from prowler.lib.outputs.compliance.ens.ens_gcp import GCPENS
@@ -102,7 +97,6 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name == "prowler_threatscore_aws", ProwlerThreatScoreAWS),
         (lambda name: name.startswith("ccc_"), CCC_AWS),
         (lambda name: name.startswith("c5_"), AWSC5),
-        (lambda name: name.startswith("csa_"), AWSCSA),
         (lambda name: name == "asd_essential_eight_aws", ASDEssentialEightAWS),
     ],
     "azure": [
@@ -113,7 +107,6 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name.startswith("ccc_"), CCC_Azure),
         (lambda name: name == "prowler_threatscore_azure", ProwlerThreatScoreAzure),
         (lambda name: name == "c5_azure", AzureC5),
-        (lambda name: name.startswith("csa_"), AzureCSA),
     ],
     "gcp": [
         (lambda name: name.startswith("cis_"), GCPCIS),
@@ -123,7 +116,6 @@ COMPLIANCE_CLASS_MAP = {
         (lambda name: name == "prowler_threatscore_gcp", ProwlerThreatScoreGCP),
         (lambda name: name.startswith("ccc_"), CCC_GCP),
         (lambda name: name == "c5_gcp", GCPC5),
-        (lambda name: name.startswith("csa_"), GCPCSA),
     ],
     "kubernetes": [
         (lambda name: name.startswith("cis_"), KubernetesCIS),
@@ -152,11 +144,9 @@ COMPLIANCE_CLASS_MAP = {
     "image": [],
     "oraclecloud": [
         (lambda name: name.startswith("cis_"), OracleCloudCIS),
-        (lambda name: name.startswith("csa_"), OracleCloudCSA),
     ],
     "alibabacloud": [
         (lambda name: name.startswith("cis_"), AlibabaCloudCIS),
-        (lambda name: name.startswith("csa_"), AlibabaCloudCSA),
         (
             lambda name: name == "prowler_threatscore_alibabacloud",
             ProwlerThreatScoreAlibaba,

api/src/backend/tasks/jobs/report.py

@@ -29,7 +29,10 @@ from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.db_utils import rls_transaction
 from api.models import Provider, Scan, ScanSummary, StateChoices, ThreatScoreSnapshot
 from api.utils import initialize_prowler_provider
-from prowler.lib.check.compliance_models import Compliance
+from prowler.lib.check.compliance_models import (
+    Compliance,
+    get_bulk_compliance_frameworks_universal,
+)
 from prowler.lib.outputs.finding import Finding as FindingOutput
 
 logger = get_task_logger(__name__)
@@ -571,7 +574,7 @@ def generate_csa_report(
     Args:
         tenant_id: The tenant ID for Row-Level Security context.
         scan_id: ID of the scan executed by Prowler.
-        compliance_id: ID of the compliance framework (e.g., "csa_ccm_4.0_aws").
+        compliance_id: ID of the compliance framework (e.g., "csa_ccm_4.0").
         output_path: Output PDF file path.
         provider_id: Provider ID for the scan.
         only_failed: If True, only include failed requirements in detailed section.
@@ -883,9 +886,11 @@ def generate_compliance_reports(
             frameworks_bulk.get(f"nis2_{provider_type}")
         )
     if generate_csa:
-        pending_checks_by_framework["csa"] = _get_compliance_check_ids(
-            frameworks_bulk.get(f"csa_ccm_4.0_{provider_type}")
-        )
+        # csa_ccm_4.0 lives at the top level, not under compliance/{provider}/.
+        csa_framework = frameworks_bulk.get(
+            "csa_ccm_4.0"
+        ) or get_bulk_compliance_frameworks_universal(provider_type).get("csa_ccm_4.0")
+        pending_checks_by_framework["csa"] = _get_compliance_check_ids(csa_framework)
     if generate_cis and latest_cis:
         pending_checks_by_framework["cis"] = _get_compliance_check_ids(
             frameworks_bulk.get(latest_cis)
@@ -1183,7 +1188,7 @@ def generate_compliance_reports(
     if generate_csa:
         generated_report_keys.append("csa")
         csa_path = output_paths["csa"]
-        compliance_id_csa = f"csa_ccm_4.0_{provider_type}"
+        compliance_id_csa = "csa_ccm_4.0"
         pdf_path_csa = f"{csa_path}_csa_report.pdf"
         logger.info("Generating CSA CCM report with compliance %s", compliance_id_csa)
 

api/src/backend/tasks/jobs/reports/base.py

@@ -5,6 +5,7 @@ import time
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
 from dataclasses import dataclass, field
+from types import SimpleNamespace
 from typing import Any
 
 from celery.utils.log import get_task_logger
@@ -26,7 +27,10 @@ from api.db_router import READ_REPLICA_ALIAS
 from api.db_utils import rls_transaction
 from api.models import Provider, StatusChoices
 from api.utils import initialize_prowler_provider
-from prowler.lib.check.compliance_models import Compliance
+from prowler.lib.check.compliance_models import (
+    Compliance,
+    get_bulk_compliance_frameworks_universal,
+)
 from prowler.lib.outputs.finding import Finding as FindingOutput
 
 from .components import (
@@ -222,6 +226,46 @@ def get_requirement_metadata(
     return None
 
 
+def _universal_attributes_to_list(attributes) -> list:
+    """Flatten a universal requirement's ``attributes`` into a list of objects
+    with attribute access. MITRE wraps its list under ``_raw_attributes``."""
+    if isinstance(attributes, dict) and "_raw_attributes" in attributes:
+        entries = attributes.get("_raw_attributes") or []
+        return [
+            SimpleNamespace(**entry) for entry in entries if isinstance(entry, dict)
+        ]
+    if isinstance(attributes, dict):
+        return [SimpleNamespace(**attributes)] if attributes else []
+    return list(attributes or [])
+
+
+def _adapt_universal_to_legacy(framework, provider_type: str) -> SimpleNamespace:
+    """Expose a universal ``ComplianceFramework`` under the legacy ``Compliance``
+    attribute names used by the PDF pipeline."""
+    provider_key = (provider_type or "").lower()
+    requirements = []
+    for requirement in framework.requirements:
+        checks_by_provider = (
+            requirement.checks if isinstance(requirement.checks, dict) else {}
+        )
+        requirements.append(
+            SimpleNamespace(
+                Id=requirement.id,
+                Description=requirement.description or "",
+                Checks=list(checks_by_provider.get(provider_key, [])),
+                Attributes=_universal_attributes_to_list(requirement.attributes),
+            )
+        )
+    return SimpleNamespace(
+        Framework=framework.framework,
+        Name=framework.name,
+        Version=framework.version or "",
+        Description=framework.description or "",
+        Provider=framework.provider or provider_type,
+        Requirements=requirements,
+    )
+
+
 # =============================================================================
 # PDF Styles Cache
 # =============================================================================
@@ -869,9 +913,18 @@ class BaseComplianceReportGenerator(ABC):
                 prowler_provider = initialize_prowler_provider(provider_obj)
             provider_type = provider_obj.provider
 
-            # Load compliance framework
-            frameworks_bulk = Compliance.get_bulk(provider_type)
-            compliance_obj = frameworks_bulk.get(compliance_id)
+            # Load compliance framework — fall back to the universal loader
+            # for top-level JSONs (e.g. csa_ccm_4.0) that Compliance.get_bulk
+            # does not scan.
+            compliance_obj = Compliance.get_bulk(provider_type).get(compliance_id)
+            if not compliance_obj:
+                universal_framework = get_bulk_compliance_frameworks_universal(
+                    provider_type
+                ).get(compliance_id)
+                if universal_framework:
+                    compliance_obj = _adapt_universal_to_legacy(
+                        universal_framework, provider_type
+                    )
 
             if not compliance_obj:
                 raise ValueError(f"Compliance framework not found: {compliance_id}")

api/src/backend/tasks/jobs/threatscore_utils.py

@@ -359,35 +359,40 @@ def _load_findings_for_requirement_checks(
 def _get_compliance_check_ids(compliance_obj) -> set[str]:
     """Return the union of all check_ids referenced by a compliance framework.
 
-    Used by the master report orchestrator to know which checks each
-    framework consumes from the shared ``findings_cache``, so that once a
-    framework finishes the entries no other pending framework needs can be
-    evicted from the cache (PROWLER-1733).
+    Used by the master report orchestrator to evict entries from
+    ``findings_cache`` once no pending framework needs them (PROWLER-1733).
 
-    Args:
-        compliance_obj: A loaded Compliance framework object exposing a
-            ``Requirements`` iterable, each requirement carrying ``Checks``.
-            ``None`` is treated as "no checks" rather than raising, so the
-            caller can pass ``frameworks_bulk.get(...)`` directly without
-            an extra existence check.
-
-    Returns:
-        Set of check_id strings (empty if ``compliance_obj`` is ``None``).
+    Accepts the legacy ``Compliance`` shape (``Requirements`` / ``Checks``
+    lists) and the universal ``ComplianceFramework`` shape (``requirements``
+    / ``checks`` dict keyed by provider). ``None`` returns an empty set so
+    callers can pass ``frameworks_bulk.get(...)`` directly.
     """
     if compliance_obj is None:
         return set()
-    checks: set[str] = set()
-    requirements = getattr(compliance_obj, "Requirements", None) or []
+
+    requirements = getattr(compliance_obj, "Requirements", None) or getattr(
+        compliance_obj, "requirements", None
+    )
+    if not requirements:
+        return set()
+
+    check_ids: set[str] = set()
     try:
-        # Defensive: Mock objects (used in unit tests) return another Mock
-        # for any attribute access, which is truthy but not iterable. Treat
-        # any non-iterable Requirements value as "no checks".
-        for req in requirements:
-            req_checks = getattr(req, "Checks", None) or []
+        # Mock objects in unit tests return another Mock for any attribute
+        # access — truthy but not iterable. Treat that as "no checks".
+        for requirement in requirements:
+            requirement_checks = getattr(requirement, "Checks", None)
+            if requirement_checks is None:
+                checks_by_provider = getattr(requirement, "checks", None) or {}
+                requirement_checks = [
+                    check_id
+                    for check_ids_list in checks_by_provider.values()
+                    for check_id in check_ids_list
+                ]
             try:
-                checks.update(req_checks)
+                check_ids.update(requirement_checks)
             except TypeError:
                 continue
     except TypeError:
         return set()
-    return checks
+    return check_ids

api/src/backend/tasks/tasks.py

@@ -68,7 +68,10 @@ from tasks.utils import (
     get_next_execution_datetime,
 )
 
-from api.compliance import get_compliance_frameworks
+from api.compliance import (
+    get_compliance_frameworks,
+    get_prowler_provider_compliance,
+)
 from api.db_router import READ_REPLICA_ALIAS
 from api.db_utils import delete_related_daily_task, rls_transaction
 from api.decorators import handle_provider_deletion, set_tenant
@@ -76,6 +79,9 @@ from api.models import Finding, Integration, Provider, Scan, ScanSummary, StateC
 from api.utils import initialize_prowler_provider
 from api.v1.serializers import ScanTaskSerializer
 from prowler.lib.check.compliance_models import Compliance
+from prowler.lib.outputs.compliance.compliance import (
+    process_universal_compliance_frameworks,
+)
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
 from prowler.lib.outputs.finding import Finding as FindingOutput
 
@@ -543,7 +549,16 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     provider_uid = provider_obj.uid
     provider_type = provider_obj.provider
 
+    # Per-framework exporters in `COMPLIANCE_CLASS_MAP` consume the legacy bulk.
     frameworks_bulk = Compliance.get_bulk(provider_type)
+    # Universal-only frameworks (top-level JSONs like `dora.json`) are emitted
+    # via `process_universal_compliance_frameworks` below.
+    universal_bulk = get_prowler_provider_compliance(provider_type)
+    universal_only_names = {
+        name
+        for name in universal_bulk
+        if name not in frameworks_bulk and universal_bulk[name].outputs
+    }
     frameworks_avail = get_compliance_frameworks(provider_type)
     out_dir, comp_dir = _generate_output_directory(
         DJANGO_TMP_OUTPUT_DIRECTORY, provider_uid, tenant_id, scan_id
@@ -568,6 +583,10 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
 
     output_writers = {}
     compliance_writers = {}
+    # Shared across batches so universal writers are created once and reused.
+    universal_compliance_state: dict[str, list] = {"compliance": []}
+    universal_base_dir = os.path.dirname(out_dir)
+    universal_output_filename = os.path.basename(out_dir)
 
     scan_summary = FindingOutput._transform_findings_stats(
         ScanSummary.objects.filter(scan_id=scan_id)
@@ -622,8 +641,30 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
                 writer.batch_write_data_to_file(**extra)
                 writer._data.clear()
 
-            # Compliance CSVs
+            # Universal-only frameworks (e.g. `dora.json`).
+            if universal_only_names:
+                process_universal_compliance_frameworks(
+                    input_compliance_frameworks=universal_only_names,
+                    universal_frameworks=universal_bulk,
+                    finding_outputs=fos,
+                    output_directory=universal_base_dir,
+                    output_filename=universal_output_filename,
+                    provider=provider_type,
+                    generated_outputs=universal_compliance_state,
+                    from_cli=False,
+                    is_last=is_last,
+                )
+
+            # Compliance CSVs (per-framework exporters).
             for name in frameworks_avail:
+                if name in universal_only_names:
+                    continue
+                if name not in frameworks_bulk:
+                    logger.warning(
+                        "Compliance framework '%s' missing from bulk; skipping CSV export",
+                        name,
+                    )
+                    continue
                 compliance_obj = frameworks_bulk[name]
 
                 klass = GenericCompliance

api/src/backend/tasks/tests/test_reports_csa.py

@@ -80,7 +80,7 @@ def basic_csa_compliance_data():
         tenant_id="tenant-123",
         scan_id="scan-456",
         provider_id="provider-789",
-        compliance_id="csa_ccm_4.0_aws",
+        compliance_id="csa_ccm_4.0",
         framework="CSA-CCM",
         name="CSA Cloud Controls Matrix v4.0",
         version="4.0",

api/src/backend/tasks/tests/test_tasks.py

@@ -323,6 +323,7 @@ class TestGenerateOutputs:
 
         mock_transformed_stats = {"some": "stats"}
         with (
+            patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch(
                 "tasks.tasks.FindingOutput._transform_findings_stats",
                 return_value=mock_transformed_stats,
@@ -441,6 +442,7 @@ class TestGenerateOutputs:
         mock_provider.uid = "test-provider-uid"
 
         with (
+            patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
             patch("tasks.tasks.Provider.objects.get", return_value=mock_provider),
             patch("tasks.tasks.initialize_prowler_provider"),
@@ -596,6 +598,7 @@ class TestGenerateOutputs:
         ]
 
         with (
+            patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary,
             patch(
                 "tasks.tasks.Provider.objects.get",
@@ -670,6 +673,7 @@ class TestGenerateOutputs:
         mock_provider.uid = "test-provider-uid"
 
         with (
+            patch("tasks.tasks.get_prowler_provider_compliance", return_value={}),
             patch("tasks.tasks.ScanSummary.objects.filter") as mock_filter,
             patch("tasks.tasks.Provider.objects.get", return_value=mock_provider),
             patch("tasks.tasks.initialize_prowler_provider"),
@@ -1113,6 +1117,7 @@ class TestCheckIntegrationsTask:
             enabled=True,
         )
 
+    @patch("tasks.tasks.get_prowler_provider_compliance", return_value={})
     @patch("tasks.tasks.s3_integration_task")
     @patch("tasks.tasks.Integration.objects.filter")
     @patch("tasks.tasks.ScanSummary.objects.filter")
@@ -1145,6 +1150,7 @@ class TestCheckIntegrationsTask:
         mock_scan_summary,
         mock_integration_filter,
         mock_s3_task,
+        mock_get_prowler_compliance,
     ):
         """Test that ASFF output is generated for AWS providers with SecurityHub integration."""
         # Setup
@@ -1241,6 +1247,7 @@ class TestCheckIntegrationsTask:
 
             assert result == {"upload": True}
 
+    @patch("tasks.tasks.get_prowler_provider_compliance", return_value={})
     @patch("tasks.tasks.s3_integration_task")
     @patch("tasks.tasks.Integration.objects.filter")
     @patch("tasks.tasks.ScanSummary.objects.filter")
@@ -1273,6 +1280,7 @@ class TestCheckIntegrationsTask:
         mock_scan_summary,
         mock_integration_filter,
         mock_s3_task,
+        mock_get_prowler_compliance,
     ):
         """Test that ASFF output is NOT generated for AWS providers without SecurityHub integration."""
         # Setup
@@ -1366,6 +1374,7 @@ class TestCheckIntegrationsTask:
 
             assert result == {"upload": True}
 
+    @patch("tasks.tasks.get_prowler_provider_compliance", return_value={})
     @patch("tasks.tasks.ScanSummary.objects.filter")
     @patch("tasks.tasks.Provider.objects.get")
     @patch("tasks.tasks.initialize_prowler_provider")
@@ -1394,6 +1403,7 @@ class TestCheckIntegrationsTask:
         mock_initialize_provider,
         mock_provider_get,
         mock_scan_summary,
+        mock_get_prowler_compliance,
     ):
         """Test that ASFF output is NOT generated for non-AWS providers (e.g., Azure, GCP)."""
         # Setup