feat(sdk): discover external universal compliance frameworks via entry points

prowler/CHANGELOG.md

@@ -14,6 +14,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - Support for external/custom providers, checks, and compliance frameworks without modifying core code [(#10700)](https://github.com/prowler-cloud/prowler/pull/10700)
 - `elbv2_alb_drop_invalid_header_fields_enabled` check for AWS provider, verifying Application Load Balancers have `routing.http.drop_invalid_header_fields.enabled` set to `true` to mitigate HTTP desync attacks (AWS FSBP ELB.4) [(#11471)](https://github.com/prowler-cloud/prowler/pull/11471)
 - `user`, `systemlog` and `idp` service for Okta provider with `user_inactivity_automation_35d_enabled`, `systemlog_streaming_enabled` and `idp_smart_card_dod_approved_ca` checks [(#11496)](https://github.com/prowler-cloud/prowler/pull/11496)
+- External multi-provider compliance frameworks can be registered via the `prowler.compliance.universal` entry point group [(#11490)](https://github.com/prowler-cloud/prowler/pull/11490)
 - AWS AI Security Framework support in the CLI dashboard [(#11475)](https://github.com/prowler-cloud/prowler/pull/11475)
 - `entra_service_principal_privileged_role_no_owners` check for M365 provider, failing when a service principal with a permanent Tier 0 directory role has owners on the service principal or its parent app registration [(#11070)](https://github.com/prowler-cloud/prowler/issues/11070)
 

prowler/config/config.py

@@ -144,8 +144,7 @@ def get_available_compliance_frameworks(provider=None):
                             continue
                     if name not in available_compliance_frameworks:
                         available_compliance_frameworks.append(name)
-    # External compliance via entry points.
-    # Multi-provider support for external plug-ins is tracked in PROWLER-1444.
+    # External per-provider compliance via entry points.
     ep_dirs = _get_ep_compliance_dirs()
     for prov, path in ep_dirs.items():
         if provider and prov != provider:
@@ -156,6 +155,32 @@ def get_available_compliance_frameworks(provider=None):
                     name = file.name.removesuffix(".json")
                     if name not in available_compliance_frameworks:
                         available_compliance_frameworks.append(name)
+    # External multi-provider frameworks via the dedicated universal group;
+    # filtered by supports_provider when a provider is given.
+    for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
+        try:
+            module = ep.load()
+            path = (
+                module.__path__[0]
+                if hasattr(module, "__path__")
+                else os.path.dirname(module.__file__)
+            )
+        except Exception as error:
+            logger.warning(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            continue
+        if not os.path.isdir(path):
+            continue
+        for file in os.scandir(path):
+            if file.is_file() and file.name.endswith(".json"):
+                name = file.name.removesuffix(".json")
+                if provider:
+                    framework = load_compliance_framework_universal(file.path)
+                    if framework is None or not framework.supports_provider(provider):
+                        continue
+                if name not in available_compliance_frameworks:
+                    available_compliance_frameworks.append(name)
     return available_compliance_frameworks
 
 

prowler/lib/check/compliance_models.py

@@ -478,9 +478,15 @@ def get_bulk(provider: str) -> dict:
                                         compliance_framework_name
                                         not in bulk_compliance_frameworks
                                     ):
-                                        bulk_compliance_frameworks[
-                                            compliance_framework_name
-                                        ] = load_compliance_framework(file_path)
+                                        # External JSON: tolerate non-legacy
+                                        # schemas (skip + warn) instead of aborting.
+                                        framework = load_compliance_framework(
+                                            file_path, fatal=False
+                                        )
+                                        if framework is not None:
+                                            bulk_compliance_frameworks[
+                                                compliance_framework_name
+                                            ] = framework
                     except Exception as error:
                         logger.warning(
                             f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -494,18 +500,26 @@ def get_bulk(provider: str) -> dict:
 
 # Testing Pending
 def load_compliance_framework(
-    compliance_specification_file: str,
-) -> Compliance:
-    """load_compliance_framework loads and parse a Compliance Framework Specification"""
+    compliance_specification_file: str, fatal: bool = True
+) -> Optional[Compliance]:
+    """load_compliance_framework loads and parse a Compliance Framework Specification.
+
+    With ``fatal=True`` (built-in JSONs) an invalid file aborts the run; with
+    ``fatal=False`` (external JSONs) it is skipped with a warning and ``None``
+    is returned.
+    """
     try:
-        compliance_framework = Compliance.parse_file(compliance_specification_file)
+        return Compliance.parse_file(compliance_specification_file)
     except ValidationError as error:
-        logger.critical(
-            f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
+        if fatal:
+            logger.critical(
+                f"Compliance Framework Specification from {compliance_specification_file} is not valid: {error}"
+            )
+            sys.exit(1)
+        logger.warning(
+            f"Skipping invalid compliance framework {compliance_specification_file}: {error}"
         )
-        sys.exit(1)
-    else:
-        return compliance_framework
+        return None
 
 
 # ─── Universal Compliance Schema Models (Phase 1-3) ─────────────────────────
@@ -982,6 +996,25 @@ def get_bulk_compliance_frameworks_universal(provider: str) -> dict:
         if compliance_root and os.path.isdir(compliance_root):
             _load_jsons_from_dir(compliance_root, provider, bulk)
 
+        # External multi-provider frameworks via the dedicated universal entry
+        # point group, kept separate from the per-provider `prowler.compliance`
+        # group so the legacy loader never parses a universal JSON. Built-ins
+        # (already in bulk) win on a name collision.
+        for ep in importlib.metadata.entry_points(group="prowler.compliance.universal"):
+            try:
+                module = ep.load()
+                ep_dir = (
+                    module.__path__[0]
+                    if hasattr(module, "__path__")
+                    else os.path.dirname(module.__file__)
+                )
+                if os.path.isdir(ep_dir):
+                    _load_jsons_from_dir(ep_dir, provider, bulk)
+            except Exception as error:
+                logger.warning(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
     except Exception as e:
         logger.error(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}] -- {e}")
     return bulk

tests/lib/check/universal_compliance_models_test.py

@@ -1,5 +1,7 @@
 import json
 import os
+import tempfile
+from unittest.mock import MagicMock, patch
 
 import pytest
 from pydantic.v1 import ValidationError
@@ -23,6 +25,7 @@
     TableLabels,
     UniversalComplianceRequirement,
     adapt_legacy_to_universal,
+    get_bulk_compliance_frameworks_universal,
     load_compliance_framework_universal,
 )
 from tests.lib.outputs.compliance.fixtures import (
@@ -1116,3 +1119,121 @@ def test_multiple_errors_reported(self):
                 ],
                 attributes_metadata=self._metadata(enum=["high", "low"]),
             )
+
+
+class TestGetBulkUniversalEntryPoints:
+    """Entry-point discovery for universal (multi-provider) compliance frameworks."""
+
+    @staticmethod
+    def _write_universal_json(directory, filename, framework, display_name):
+        data = {
+            "framework": framework,
+            "name": display_name,
+            "version": "1.0",
+            "description": "External multi-provider framework",
+            "requirements": [
+                {
+                    "id": "1",
+                    "name": "Requirement 1",
+                    "description": "desc",
+                    "checks": {"fakeexternal": ["check_a"]},
+                }
+            ],
+        }
+        with open(os.path.join(directory, filename), "w") as f:
+            json.dump(data, f)
+
+    @staticmethod
+    def _entry_point(path):
+        module = MagicMock()
+        module.__path__ = [path]
+        ep = MagicMock()
+        ep.name = "fakeexternal"
+        ep.group = "prowler.compliance.universal"
+        ep.load.return_value = module
+        return ep
+
+    @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+    def test_includes_external_universal_framework(self, mock_list_modules, mock_ep):
+        mock_list_modules.return_value = []
+        with tempfile.TemporaryDirectory() as ep_dir:
+            self._write_universal_json(
+                ep_dir, "customuniversal_1.0.json", "CustomUniversal", "Custom"
+            )
+            mock_ep.return_value = [self._entry_point(ep_dir)]
+
+            bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
+
+        mock_ep.assert_called_with(group="prowler.compliance.universal")
+        assert "customuniversal_1.0" in bulk
+        assert bulk["customuniversal_1.0"].framework == "CustomUniversal"
+
+    @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+    def test_builtin_wins_over_external_on_name_collision(
+        self, mock_list_modules, mock_ep
+    ):
+        with (
+            tempfile.TemporaryDirectory() as root,
+            tempfile.TemporaryDirectory() as ep_dir,
+        ):
+            builtin_sub = os.path.join(root, "builtinprov")
+            os.makedirs(builtin_sub)
+            self._write_universal_json(
+                builtin_sub, "shared_1.0.json", "SharedFramework", "Built-in"
+            )
+            builtin_module = MagicMock()
+            builtin_module.module_finder.path = root
+            builtin_module.name = "prowler.compliance.builtinprov"
+            mock_list_modules.return_value = [builtin_module]
+
+            self._write_universal_json(
+                ep_dir, "shared_1.0.json", "SharedFramework", "External"
+            )
+            mock_ep.return_value = [self._entry_point(ep_dir)]
+
+            bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
+
+        assert "shared_1.0" in bulk
+        assert bulk["shared_1.0"].name == "Built-in"
+
+    @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+    def test_loads_all_frameworks_in_a_single_entry_point_path(
+        self, mock_list_modules, mock_ep
+    ):
+        """All JSONs in one entry-point directory are added, not collapsed to one."""
+        mock_list_modules.return_value = []
+        with tempfile.TemporaryDirectory() as ep_dir:
+            self._write_universal_json(ep_dir, "fw_a_1.0.json", "FwA", "Framework A")
+            self._write_universal_json(ep_dir, "fw_b_1.0.json", "FwB", "Framework B")
+            mock_ep.return_value = [self._entry_point(ep_dir)]
+
+            bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
+
+        assert "fw_a_1.0" in bulk
+        assert "fw_b_1.0" in bulk
+
+    @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+    def test_merges_frameworks_from_multiple_packages_same_provider(
+        self, mock_list_modules, mock_ep
+    ):
+        """Two packages under the same provider name are both discovered."""
+        mock_list_modules.return_value = []
+        with (
+            tempfile.TemporaryDirectory() as dir_a,
+            tempfile.TemporaryDirectory() as dir_b,
+        ):
+            self._write_universal_json(dir_a, "pkg_a_1.0.json", "PkgA", "Package A")
+            self._write_universal_json(dir_b, "pkg_b_1.0.json", "PkgB", "Package B")
+            mock_ep.return_value = [
+                self._entry_point(dir_a),
+                self._entry_point(dir_b),
+            ]
+
+            bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
+
+        assert "pkg_a_1.0" in bulk
+        assert "pkg_b_1.0" in bulk

tests/providers/external/test_dynamic_provider_loading.py

@@ -1218,6 +1218,48 @@ def test_get_available_compliance_includes_external(self, mock_dirs):
 
             assert "custom_1.0_ext" in frameworks
 
+    @patch("prowler.config.config.importlib.metadata.entry_points")
+    def test_get_available_compliance_includes_external_universal(self, mock_ep):
+        """External universal frameworks under prowler.compliance.universal are
+        listed, for a provider and for the provider=None case that feeds
+        --compliance choices."""
+        import json
+        import os
+        import tempfile
+
+        from prowler.config.config import get_available_compliance_frameworks
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            framework = {
+                "framework": "CustomUniversal",
+                "name": "Custom Universal",
+                "version": "1.0",
+                "description": "Multi-provider",
+                "requirements": [
+                    {
+                        "id": "1",
+                        "name": "r",
+                        "description": "d",
+                        "checks": {"aws": ["c"]},
+                    }
+                ],
+            }
+            with open(os.path.join(tmpdir, "customuniversal_1.0.json"), "w") as f:
+                json.dump(framework, f)
+
+            module = MagicMock()
+            module.__path__ = [tmpdir]
+            ep = _make_entry_point(
+                "anyname", "pkg.compliance", "prowler.compliance.universal"
+            )
+            ep.load.return_value = module
+            mock_ep.side_effect = lambda group: (
+                [ep] if group == "prowler.compliance.universal" else []
+            )
+
+            assert "customuniversal_1.0" in get_available_compliance_frameworks("aws")
+            assert "customuniversal_1.0" in get_available_compliance_frameworks(None)
+
     @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
     @patch("prowler.lib.check.compliance_models.list_compliance_modules")
     def test_compliance_get_bulk_loads_external(self, mock_list_modules, mock_ep):
@@ -1257,6 +1299,49 @@ def test_compliance_get_bulk_loads_external(self, mock_list_modules, mock_ep):
             assert "custom_1.0_fakeexternal" in bulk
             assert bulk["custom_1.0_fakeexternal"].Framework == "Custom"
 
+    @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.compliance_models.list_compliance_modules")
+    def test_compliance_get_bulk_skips_non_legacy_external_json(
+        self, mock_list_modules, mock_ep
+    ):
+        """A universal-schema JSON registered under prowler.compliance is skipped,
+        not aborting the run via sys.exit."""
+        import json
+        import os
+        import tempfile
+
+        from prowler.lib.check.compliance_models import Compliance
+
+        mock_list_modules.return_value = []
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            json_data = {
+                "framework": "Universal",
+                "name": "Universal Framework",
+                "version": "1.0",
+                "description": "Multi-provider",
+                "requirements": [
+                    {
+                        "id": "1",
+                        "name": "r",
+                        "description": "d",
+                        "checks": {"aws": ["c"]},
+                    }
+                ],
+            }
+            with open(os.path.join(tmpdir, "universal_1.0.json"), "w") as f:
+                json.dump(json_data, f)
+
+            mock_module = MagicMock()
+            mock_module.__path__ = [tmpdir]
+            ep = _make_entry_point("aws", "pkg.compliance", "prowler.compliance")
+            ep.load.return_value = mock_module
+            mock_ep.return_value = [ep]
+
+            bulk = Compliance.get_bulk("aws")
+
+        assert "universal_1.0" not in bulk
+
     @patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
     @patch("prowler.lib.check.compliance_models.list_compliance_modules")
     def test_compliance_get_bulk_file_fallback(self, mock_list_modules, mock_ep):