feat(api): derive report availability from provider frameworks

- Gate threatscore/ens/nis2/csa reports on the provider's available
  compliance frameworks instead of hard-coded provider whitelists
- Reference the ASFF/SecurityHub gate via the provider enum symbol
- Cover external-provider report availability and generic compliance
  fallback with tests

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

@@ -751,43 +751,10 @@ def generate_compliance_reports(
         provider_uid = provider_obj.uid
         provider_type = provider_obj.provider
 
-    # Check provider compatibility
-    if generate_threatscore and provider_type not in [
-        "aws",
-        "azure",
-        "gcp",
-        "m365",
-        "kubernetes",
-        "alibabacloud",
-    ]:
-        logger.info("Provider %s not supported for ThreatScore report", provider_type)
-        results["threatscore"] = {"upload": False, "path": ""}
-        generate_threatscore = False
-
-    if generate_ens and provider_type not in ["aws", "azure", "gcp"]:
-        logger.info("Provider %s not supported for ENS report", provider_type)
-        results["ens"] = {"upload": False, "path": ""}
-        generate_ens = False
-
-    if generate_nis2 and provider_type not in ["aws", "azure", "gcp"]:
-        logger.info("Provider %s not supported for NIS2 report", provider_type)
-        results["nis2"] = {"upload": False, "path": ""}
-        generate_nis2 = False
-
-    if generate_csa and provider_type not in [
-        "aws",
-        "azure",
-        "gcp",
-        "oraclecloud",
-        "alibabacloud",
-    ]:
-        logger.info("Provider %s not supported for CSA CCM report", provider_type)
-        results["csa"] = {"upload": False, "path": ""}
-        generate_csa = False
-
-    # Load the framework definitions for this provider once. We use this map
-    # both to pick the latest CIS variant and to precompute the set of
-    # check_ids each framework consumes (for findings_cache eviction).
+    # Load the framework definitions for this provider once. We use this map to
+    # gate per-report availability, to pick the latest CIS variant, and to
+    # precompute the set of check_ids each framework consumes (for findings_cache
+    # eviction).
     frameworks_bulk: dict = {}
     try:
         frameworks_bulk = Compliance.get_bulk(provider_type)
@@ -796,6 +763,32 @@ def generate_compliance_reports(
         # Fall through; individual frameworks will still try and fail
         # gracefully if their compliance_id is missing.
 
+    # Report availability is derived from the dynamically loaded framework map,
+    # not a hard-coded provider whitelist (which drifts the moment a new
+    # framework JSON ships, and would silently exclude an external provider that
+    # ships the framework). Same approach already used for CIS below.
+    if generate_threatscore and not frameworks_bulk.get(
+        f"prowler_threatscore_{provider_type}"
+    ):
+        logger.info("Provider %s not supported for ThreatScore report", provider_type)
+        results["threatscore"] = {"upload": False, "path": ""}
+        generate_threatscore = False
+
+    if generate_ens and not frameworks_bulk.get(f"ens_rd2022_{provider_type}"):
+        logger.info("Provider %s not supported for ENS report", provider_type)
+        results["ens"] = {"upload": False, "path": ""}
+        generate_ens = False
+
+    if generate_nis2 and not frameworks_bulk.get(f"nis2_{provider_type}"):
+        logger.info("Provider %s not supported for NIS2 report", provider_type)
+        results["nis2"] = {"upload": False, "path": ""}
+        generate_nis2 = False
+
+    if generate_csa and not frameworks_bulk.get(f"csa_ccm_4.0_{provider_type}"):
+        logger.info("Provider %s not supported for CSA CCM report", provider_type)
+        results["csa"] = {"upload": False, "path": ""}
+        generate_csa = False
+
     # For CIS we do NOT pre-check the provider against a hard-coded whitelist
     # (that list drifts the moment a new CIS JSON ships). Instead, we inspect
     # the dynamically loaded framework map and pick the latest available CIS

api/src/backend/tasks/tasks.py

@@ -543,7 +543,7 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
 
     # Check if we need to generate ASFF output for AWS providers with SecurityHub integration
     generate_asff = False
-    if provider_type == "aws":
+    if provider_type == Provider.ProviderChoices.AWS:
         security_hub_integrations = Integration.objects.filter(
             integrationproviderrelationship__provider_id=provider_id,
             integration_type=Integration.IntegrationChoices.AWS_SECURITY_HUB,

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

@@ -539,12 +539,12 @@ class TestLoadFindingsForChecks:
                 total_counts_out=totals,
             )
 
-        assert (
-            len(result[check_id]) == 5
-        ), f"cap=5 should yield exactly 5 loaded findings, got {len(result[check_id])}"
-        assert (
-            totals[check_id] == 12
-        ), f"total_counts_out should report the pre-cap total (12), got {totals[check_id]}"
+        assert len(result[check_id]) == 5, (
+            f"cap=5 should yield exactly 5 loaded findings, got {len(result[check_id])}"
+        )
+        assert totals[check_id] == 12, (
+            f"total_counts_out should report the pre-cap total (12), got {totals[check_id]}"
+        )
 
     def test_only_failed_findings_pushes_down_to_sql(
         self, tenants_fixture, scans_fixture
@@ -616,13 +616,13 @@ class TestLoadFindingsForChecks:
         loaded = result[check_id]
         assert len(loaded) == 3, f"expected 3 FAIL findings, got {len(loaded)}"
         statuses = {getattr(f, "status", None) for f in loaded}
-        assert statuses == {
-            StatusChoices.FAIL
-        }, f"expected all loaded findings to be FAIL; got statuses {statuses}"
+        assert statuses == {StatusChoices.FAIL}, (
+            f"expected all loaded findings to be FAIL; got statuses {statuses}"
+        )
         # total_counts must reflect the FAIL-only total, not the global total.
-        assert (
-            totals[check_id] == 3
-        ), f"total_counts should be FAIL-only (3), got {totals[check_id]}"
+        assert totals[check_id] == 3, (
+            f"total_counts should be FAIL-only (3), got {totals[check_id]}"
+        )
 
     def test_max_findings_per_check_disabled(self, tenants_fixture, scans_fixture):
         """``MAX_FINDINGS_PER_CHECK=0`` disables the cap; load all rows."""
@@ -1259,12 +1259,12 @@ class TestGenerateComplianceReportsOptimized:
 
         # ``tsc_only`` was exclusive to ThreatScore → evicted before ENS ran.
         # ``shared`` is still pending for ENS → must remain.
-        assert (
-            "tsc_only" not in observed_state["cache_keys_when_ens_runs"]
-        ), "tsc_only should have been evicted before ENS ran"
-        assert (
-            "shared" in observed_state["cache_keys_when_ens_runs"]
-        ), "shared must remain in cache because ENS still needs it"
+        assert "tsc_only" not in observed_state["cache_keys_when_ens_runs"], (
+            "tsc_only should have been evicted before ENS ran"
+        )
+        assert "shared" in observed_state["cache_keys_when_ens_runs"], (
+            "shared must remain in cache because ENS still needs it"
+        )
 
     @patch("tasks.jobs.report.initialize_prowler_provider")
     @patch("tasks.jobs.report.rmtree")
@@ -1301,8 +1301,15 @@ class TestGenerateComplianceReportsOptimized:
         """
         mock_scan_summary_filter.return_value.exists.return_value = True
         mock_provider_get.return_value = Mock(uid="provider-uid", provider="aws")
-        # CIS variant discovery needs at least one cis_* key.
-        mock_get_bulk.return_value = {"cis_6.0_aws": Mock()}
+        # All five report frameworks exist for AWS; report availability is now
+        # derived from the framework map (CIS variant discovery needs a cis_* key).
+        mock_get_bulk.return_value = {
+            "prowler_threatscore_aws": Mock(),
+            "ens_rd2022_aws": Mock(),
+            "nis2_aws": Mock(),
+            "csa_ccm_4.0_aws": Mock(),
+            "cis_6.0_aws": Mock(),
+        }
         mock_aggregate_stats.return_value = {}
         mock_generate_output_dir.return_value = "/tmp/tenant/scan/x/prowler-out"
         mock_upload_to_s3.return_value = "s3://bucket/tenant/scan/x/report.pdf"
@@ -1365,7 +1372,9 @@ class TestGenerateComplianceReportsOptimized:
         """Cleanup must run when all generated (supported) reports are uploaded."""
         mock_scan_summary_filter.return_value.exists.return_value = True
         mock_provider_get.return_value = Mock(uid="provider-uid", provider="m365")
-        mock_get_bulk.return_value = {}
+        # Real ``Compliance.get_bulk("m365")`` exposes the ThreatScore framework
+        # but not ENS/NIS2/CSA; report availability is derived from this map.
+        mock_get_bulk.return_value = {"prowler_threatscore_m365": Mock()}
         mock_aggregate_stats.return_value = {}
         mock_generate_output_dir.return_value = (
             "/tmp/tenant/scan/threatscore/prowler-output-provider-20240101000000"
@@ -1416,7 +1425,9 @@ class TestGenerateComplianceReportsOptimized:
         """Cleanup must not run when a generated report upload fails."""
         mock_scan_summary_filter.return_value.exists.return_value = True
         mock_provider_get.return_value = Mock(uid="provider-uid", provider="m365")
-        mock_get_bulk.return_value = {}
+        # Real ``Compliance.get_bulk("m365")`` exposes the ThreatScore framework
+        # but not ENS/NIS2/CSA; report availability is derived from this map.
+        mock_get_bulk.return_value = {"prowler_threatscore_m365": Mock()}
         mock_aggregate_stats.return_value = {}
         mock_generate_output_dir.return_value = (
             "/tmp/tenant/scan/threatscore/prowler-output-provider-20240101000000"
@@ -1440,6 +1451,111 @@ class TestGenerateComplianceReportsOptimized:
         mock_threatscore.assert_called_once()
         mock_rmtree.assert_not_called()
 
+    @patch("tasks.jobs.report.generate_csa_report")
+    @patch("tasks.jobs.report.generate_nis2_report")
+    @patch("tasks.jobs.report.generate_ens_report")
+    @patch("tasks.jobs.report.generate_threatscore_report")
+    @patch("tasks.jobs.report._aggregate_requirement_statistics_from_database")
+    @patch("tasks.jobs.report.Compliance.get_bulk")
+    @patch("tasks.jobs.report.Provider.objects.get")
+    @patch("tasks.jobs.report.ScanSummary.objects.filter")
+    def test_external_provider_without_frameworks_disables_all_reports(
+        self,
+        mock_scan_summary_filter,
+        mock_provider_get,
+        mock_get_bulk,
+        mock_aggregate_stats,
+        mock_threatscore,
+        mock_ens,
+        mock_nis2,
+        mock_csa,
+    ):
+        """An externally-registered provider with no compliance frameworks gets
+        every report disabled (empty output set) without a code change."""
+        mock_scan_summary_filter.return_value.exists.return_value = True
+        mock_provider_get.return_value = Mock(uid="acme-001", provider="local-template")
+        mock_get_bulk.return_value = {}
+        mock_aggregate_stats.return_value = {}
+
+        result = generate_compliance_reports(
+            tenant_id=str(uuid.uuid4()),
+            scan_id=str(uuid.uuid4()),
+            provider_id=str(uuid.uuid4()),
+            generate_threatscore=True,
+            generate_ens=True,
+            generate_nis2=True,
+            generate_csa=True,
+            generate_cis=False,
+        )
+
+        assert result["threatscore"]["upload"] is False
+        assert result["ens"]["upload"] is False
+        assert result["nis2"]["upload"] is False
+        assert result["csa"]["upload"] is False
+        mock_threatscore.assert_not_called()
+        mock_ens.assert_not_called()
+        mock_nis2.assert_not_called()
+        mock_csa.assert_not_called()
+
+    @patch("tasks.jobs.report.initialize_prowler_provider")
+    @patch("tasks.jobs.report.rmtree")
+    @patch("tasks.jobs.report._upload_to_s3")
+    @patch("tasks.jobs.report.generate_csa_report")
+    @patch("tasks.jobs.report.generate_nis2_report")
+    @patch("tasks.jobs.report.generate_ens_report")
+    @patch("tasks.jobs.report.generate_threatscore_report")
+    @patch("tasks.jobs.report._generate_compliance_output_directory")
+    @patch("tasks.jobs.report._aggregate_requirement_statistics_from_database")
+    @patch("tasks.jobs.report.Compliance.get_bulk")
+    @patch("tasks.jobs.report.Provider.objects.get")
+    @patch("tasks.jobs.report.ScanSummary.objects.filter")
+    def test_external_provider_with_threatscore_framework_enables_it(
+        self,
+        mock_scan_summary_filter,
+        mock_provider_get,
+        mock_get_bulk,
+        mock_aggregate_stats,
+        mock_generate_output_dir,
+        mock_threatscore,
+        mock_ens,
+        mock_nis2,
+        mock_csa,
+        mock_upload_to_s3,
+        mock_rmtree,
+        mock_init_provider,
+    ):
+        """An externally-registered provider that ships its own ThreatScore
+        framework gets the ThreatScore report generated with no API code change
+        — exactly what the old hard-coded provider whitelist silently blocked.
+        Frameworks it does not ship (ENS/NIS2/CSA) stay disabled."""
+        mock_scan_summary_filter.return_value.exists.return_value = True
+        mock_provider_get.return_value = Mock(uid="acme-001", provider="local-template")
+        mock_get_bulk.return_value = {"prowler_threatscore_local-template": Mock()}
+        mock_aggregate_stats.return_value = {}
+        mock_generate_output_dir.return_value = "/tmp/tenant/scan/threatscore/out"
+        mock_upload_to_s3.return_value = "s3://bucket/report.pdf"
+        mock_init_provider.return_value = Mock(name="prowler_provider")
+
+        result = generate_compliance_reports(
+            tenant_id=str(uuid.uuid4()),
+            scan_id=str(uuid.uuid4()),
+            provider_id=str(uuid.uuid4()),
+            generate_threatscore=True,
+            generate_ens=True,
+            generate_nis2=True,
+            generate_csa=True,
+            generate_cis=False,
+        )
+
+        # The provider ships the ThreatScore framework -> report generated.
+        mock_threatscore.assert_called_once()
+        assert result["threatscore"]["upload"] is True
+        # It does not ship the others -> they stay disabled.
+        mock_ens.assert_not_called()
+        mock_nis2.assert_not_called()
+        mock_csa.assert_not_called()
+        assert result["ens"]["upload"] is False
+
 
 @pytest.mark.django_db
 class TestGenerateComplianceReportsCIS:

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

@@ -643,6 +643,92 @@ class TestGenerateOutputs:
         assert writer.transform_calls == [([raw2], compliance_obj, "cis")]
         assert result == {"upload": True}
 
+    def test_external_provider_falls_back_to_generic_compliance(self):
+        """A provider absent from COMPLIANCE_CLASS_MAP (e.g. an externally
+        registered one) gets GenericCompliance for every framework — the
+        provider-specific writers are never used and nothing errors."""
+        raw = MagicMock()
+        compliance_obj = MagicMock()
+        generic_instances = []
+        specific_instances = []
+
+        class TrackingGeneric:
+            def __init__(self, *args, **kwargs):
+                self._data = []
+                self.close_file = False
+                generic_instances.append(self)
+
+            def transform(self, *args, **kwargs):
+                pass
+
+            def batch_write_data_to_file(self):
+                pass
+
+        class AwsSpecificWriter:
+            def __init__(self, *args, **kwargs):
+                specific_instances.append(self)
+
+            def transform(self, *args, **kwargs):
+                pass
+
+            def batch_write_data_to_file(self):
+                pass
+
+        one_batch = [([raw], True)]
+
+        with (
+            patch("tasks.tasks.ScanSummary.objects.filter") as mock_summary,
+            patch(
+                "tasks.tasks.Provider.objects.get",
+                return_value=MagicMock(uid="acme-001", provider="local-template"),
+            ),
+            patch("tasks.tasks.initialize_prowler_provider"),
+            patch(
+                "tasks.tasks.Compliance.get_bulk",
+                return_value={"prowlerbaseline_1.0_local-template": compliance_obj},
+            ),
+            patch(
+                "tasks.tasks.get_compliance_frameworks",
+                return_value=["prowlerbaseline_1.0_local-template"],
+            ),
+            patch(
+                "tasks.tasks._generate_output_directory",
+                return_value=("/tmp/test/outdir", "/tmp/test/compdir"),
+            ),
+            patch("tasks.tasks.FindingOutput._transform_findings_stats"),
+            patch(
+                "tasks.tasks.FindingOutput.transform_api_finding",
+                side_effect=lambda f, _prov: f,
+            ),
+            patch("tasks.tasks._compress_output_files", return_value="outdir.zip"),
+            patch("tasks.tasks._upload_to_s3", return_value="s3://bucket/outdir.zip"),
+            patch(
+                "tasks.tasks.Scan.all_objects.filter",
+                return_value=MagicMock(update=lambda **_kw: None),
+            ),
+            patch("tasks.tasks.batched", return_value=one_batch),
+            patch("tasks.tasks.OUTPUT_FORMATS_MAPPING", {}),
+            patch("tasks.tasks.rmtree"),
+            patch("tasks.tasks.GenericCompliance", TrackingGeneric),
+            patch(
+                "tasks.tasks.COMPLIANCE_CLASS_MAP",
+                {"aws": [(lambda name: True, AwsSpecificWriter)]},
+            ),
+        ):
+            mock_summary.return_value.exists.return_value = True
+
+            result = generate_outputs_task(
+                scan_id=self.scan_id,
+                provider_id=self.provider_id,
+                tenant_id=self.tenant_id,
+            )
+
+        assert result == {"upload": True}
+        # The external provider's framework used the generic writer…
+        assert len(generic_instances) == 1
+        # …and the AWS-specific writer was never instantiated.
+        assert specific_instances == []
+
     # TODO: We need to add a periodic task to delete old output files
     def test_generate_outputs_logs_rmtree_exception(self, caplog):
         mock_finding_output = MagicMock()
@@ -1434,9 +1520,9 @@ class TestCheckIntegrationsTask:
             )
 
             # Verify ASFF was NOT created for non-AWS provider
-            assert (
-                "asff" not in created_writers
-            ), "ASFF writer should NOT be created for non-AWS providers"
+            assert "asff" not in created_writers, (
+                "ASFF writer should NOT be created for non-AWS providers"
+            )
             assert "csv" in created_writers, "CSV writer should be created"
             assert "ocsf" in created_writers, "OCSF writer should be created"