From 0883baad7842cf13aeb16bd46d9e68a7deba7981 Mon Sep 17 00:00:00 2001 From: StylusFrost Date: Fri, 24 Apr 2026 20:18:20 +0200 Subject: [PATCH] fix(sdk): external providers with --service and external checks for new services --- prowler/lib/check/utils.py | 40 +++++-- .../external/test_dynamic_provider_loading.py | 108 ++++++++++++++++++ 2 files changed, 138 insertions(+), 10 deletions(-) diff --git a/prowler/lib/check/utils.py b/prowler/lib/check/utils.py index 718daaaa19..5ba915a96a 100644 --- a/prowler/lib/check/utils.py +++ b/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}") diff --git a/tests/providers/external/test_dynamic_provider_loading.py b/tests/providers/external/test_dynamic_provider_loading.py index ca4542e9fd..39662d79c4 100644 --- a/tests/providers/external/test_dynamic_provider_loading.py +++ b/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