fix(sdk): external providers with --service and external checks for new services

prowler/lib/check/utils.py

@@ -7,15 +7,25 @@ from pkgutil import walk_packages
 from prowler.lib.logger import logger
 
 
-def _recover_ep_checks(provider: str) -> list[tuple]:
+def _recover_ep_checks(provider: str, service: str = None) -> list[tuple]:
     """Discover external checks registered via entry points for a provider.
 
+    External plugins follow the same layout as built-ins:
+    `{plugin_root}.services.{service}.{check}.{check}`
+
+    When `service` is provided, only entry points whose dotted path contains
+    `.services.{service}.` are included — mirroring how built-in discovery
+    filters by the `prowler.providers.{provider}.services.{service}` package.
+
     Uses find_spec to locate the check module without importing it,
     avoiding service client initialization at discovery time.
     """
     checks = []
     for ep in importlib.metadata.entry_points(group=f"prowler.checks.{provider}"):
         try:
+            if service and f".services.{service}." not in ep.value:
+                continue
+
             spec = importlib.util.find_spec(ep.value)
             if spec and spec.origin:
                 check_path = os.path.dirname(spec.origin)
@@ -58,16 +68,26 @@ def recover_checks_from_provider(
                     check_info = (check_name, check_path)
                     checks.append(check_info)
         except ModuleNotFoundError:
-            if service:
-                logger.critical(
-                    f"Service {service} was not found for the {provider} provider."
-                )
-                sys.exit(1)
-            # No built-in services for this provider (e.g., external provider)
+            # Not a built-in provider (or the requested service is not built-in).
+            # Fall through to entry points — external providers/services may be
+            # registered there. If nothing matches in either source, we fail
+            # with a clear message below.
+            pass
 
-        # External checks registered via entry points
-        if not service:
-            checks.extend(_recover_ep_checks(provider))
+        # External checks registered via entry points — always consulted, with
+        # optional service filter. Previously gated by `if not service:`, which
+        # prevented external providers from being usable with --service.
+        checks.extend(_recover_ep_checks(provider, service))
+
+        # A service was requested but nothing matched in either built-ins or
+        # entry points — surface this as a clear error instead of silently
+        # returning an empty list.
+        if service and not checks:
+            logger.critical(
+                f"Service '{service}' was not found for the '{provider}' provider "
+                f"(neither as a built-in nor via external entry points)."
+            )
+            sys.exit(1)
 
     except Exception as e:
         logger.critical(f"{e.__class__.__name__}[{e.__traceback__.tb_lineno}]: {e}")

tests/providers/external/test_dynamic_provider_loading.py

@@ -591,6 +591,114 @@ class TestCheckDiscovery:
         assert "check_a" in check_names
         assert "check_b" in check_names
 
+    @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.utils.importlib.util.find_spec")
+    def test_recover_ep_checks_filters_by_service(self, mock_spec, mock_ep):
+        """Service filter keeps only entry points whose dotted path includes
+        `.services.{service}.` — mirroring the built-in package filter."""
+        from prowler.lib.check.utils import _recover_ep_checks
+
+        mock_ep.return_value = [
+            _make_entry_point(
+                "container_has_no_root_user",
+                "prowler_artifacts_dockerdesktop.services.container.container_has_no_root_user.container_has_no_root_user",
+                "prowler.checks.dockerdesktop",
+            ),
+            _make_entry_point(
+                "image_is_signed",
+                "prowler_artifacts_dockerdesktop.services.image.image_is_signed.image_is_signed",
+                "prowler.checks.dockerdesktop",
+            ),
+        ]
+        mock_spec_obj = MagicMock()
+        mock_spec_obj.origin = "/some/path/check.py"
+        mock_spec.return_value = mock_spec_obj
+
+        checks = _recover_ep_checks("dockerdesktop", service="container")
+
+        assert len(checks) == 1
+        assert checks[0][0] == "container_has_no_root_user"
+
+    @patch("prowler.lib.check.utils.importlib.metadata.entry_points")
+    @patch("prowler.lib.check.utils.importlib.util.find_spec")
+    def test_recover_ep_checks_without_service_returns_all(self, mock_spec, mock_ep):
+        """Without a service filter, all entry points for the provider are returned."""
+        from prowler.lib.check.utils import _recover_ep_checks
+
+        mock_ep.return_value = [
+            _make_entry_point(
+                "container_has_no_root_user",
+                "prowler_artifacts_dockerdesktop.services.container.container_has_no_root_user.container_has_no_root_user",
+                "prowler.checks.dockerdesktop",
+            ),
+            _make_entry_point(
+                "image_is_signed",
+                "prowler_artifacts_dockerdesktop.services.image.image_is_signed.image_is_signed",
+                "prowler.checks.dockerdesktop",
+            ),
+        ]
+        mock_spec_obj = MagicMock()
+        mock_spec_obj.origin = "/some/path/check.py"
+        mock_spec.return_value = mock_spec_obj
+
+        checks = _recover_ep_checks("dockerdesktop")
+
+        assert len(checks) == 2
+
+    @patch("prowler.lib.check.utils._recover_ep_checks")
+    @patch("prowler.lib.check.utils.list_modules")
+    def test_recover_checks_external_provider_with_service(
+        self, mock_list_modules, mock_ep_checks
+    ):
+        """External provider with --service: built-in lookup fails with
+        ModuleNotFoundError, but entry points are still consulted and return
+        the requested service's checks. No premature sys.exit."""
+        from prowler.lib.check.utils import recover_checks_from_provider
+
+        mock_list_modules.side_effect = ModuleNotFoundError("No built-in")
+        mock_ep_checks.return_value = [("container_check", "/ext/path")]
+
+        checks = recover_checks_from_provider("dockerdesktop", service="container")
+
+        assert len(checks) == 1
+        assert checks[0][0] == "container_check"
+        mock_ep_checks.assert_called_once_with("dockerdesktop", "container")
+
+    @patch("prowler.lib.check.utils._recover_ep_checks")
+    @patch("prowler.lib.check.utils.list_modules")
+    def test_recover_checks_unknown_service_fails_cleanly(
+        self, mock_list_modules, mock_ep_checks
+    ):
+        """A typo or unknown service (not in built-ins nor in entry points)
+        fails with a clear error message, not a silent empty result."""
+        from prowler.lib.check.utils import recover_checks_from_provider
+
+        mock_list_modules.side_effect = ModuleNotFoundError("No built-in")
+        mock_ep_checks.return_value = []
+
+        with pytest.raises(SystemExit):
+            recover_checks_from_provider("aws", service="typo_service")
+
+    @patch("prowler.lib.check.utils._recover_ep_checks")
+    @patch("prowler.lib.check.utils.list_modules")
+    def test_recover_checks_builtin_with_new_external_service(
+        self, mock_list_modules, mock_ep_checks
+    ):
+        """Built-in provider with a new service added via entry points:
+        built-in discovery raises ModuleNotFoundError for the unknown service,
+        but entry points pick it up. The gate `if not service:` that previously
+        skipped entry points when --service was passed is removed."""
+        from prowler.lib.check.utils import recover_checks_from_provider
+
+        mock_list_modules.side_effect = ModuleNotFoundError("No built-in service")
+        mock_ep_checks.return_value = [("new_check", "/ext/path")]
+
+        checks = recover_checks_from_provider("aws", service="new_aws_service")
+
+        assert len(checks) == 1
+        assert checks[0][0] == "new_check"
+        mock_ep_checks.assert_called_once_with("aws", "new_aws_service")
+
 
 # ===========================================================================
 # 4. Check Execution