diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 2d3a2d803e..7296816033 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -263,32 +263,16 @@ class Provider(ABC): @staticmethod def init_global_provider(arguments: Namespace) -> None: try: - # Discriminate built-in vs external upfront via find_spec, so an - # ImportError from a transitive dependency missing inside a - # built-in's own import chain surfaces clearly instead of being - # silently re-routed to the entry-point path. - provider_class = None - if Provider.is_builtin(arguments.provider): - # Built-in wins on provider-name collision. Plug-ins are - # first-class extenders (they can register new provider - # names) but cannot override existing built-ins — a security - # tool prefers fail-loud predictability over silent - # overrides. Surface the override so the user knows their - # plug-in is being ignored and can rename it. - if Provider._load_ep_provider(arguments.provider) is not None: - logger.warning( - f"Plug-in provider '{arguments.provider}' registered " - f"via entry points is being IGNORED — a built-in with " - f"the same name exists. To use your plug-in, register " - f"it under a different name." - ) - provider_class_path = f"{providers_path}.{arguments.provider}.{arguments.provider}_provider" - provider_class_name = f"{arguments.provider.capitalize()}Provider" - try: - provider_class = getattr( - import_module(provider_class_path), provider_class_name - ) - except ImportError as e: + # Delegate class resolution to the public, side-effect-free + # resolver. init_global_provider owns the CLI-specific error + # handling: a missing transitive dep in a built-in becomes a + # logger.critical + sys.exit(1); a completely unknown provider + # re-raises so the outer try/except can sys.exit too. + try: + provider_class = Provider.get_class(arguments.provider) + except ImportError as e: + if Provider.is_builtin(arguments.provider): + # Built-in's transitive dependency is missing — loud CLI error. logger.critical( f"Failed to load built-in provider '{arguments.provider}'. " f"Missing dependency: {e}. " @@ -296,16 +280,24 @@ class Provider(ABC): ) logger.debug("Full traceback:", exc_info=True) sys.exit(1) - except AttributeError: - # Module exists but doesn't define the expected class — - # treat as external and try entry points. - provider_class = Provider._load_ep_provider(arguments.provider) - else: - provider_class = Provider._load_ep_provider(arguments.provider) + # Unknown or missing external provider — propagate so the + # outer try/except can handle it (sys.exit(1) via generic + # exception handler). + raise - if provider_class is None: - raise ImportError( - f"Provider '{arguments.provider}' not found as built-in or entry point" + # Built-in wins on name collision — warn that a same-named + # plug-in is ignored. This lives here (not in get_class) so + # that `prowler --help` and API callers that resolve a class + # without initialising a global provider do not see spurious + # warnings. + if Provider.is_builtin(arguments.provider) and ( + Provider._load_ep_provider(arguments.provider) is not None + ): + logger.warning( + f"Plug-in provider '{arguments.provider}' registered " + f"via entry points is being IGNORED — a built-in with " + f"the same name exists. To use your plug-in, register " + f"it under a different name." ) # Kept for downstream forks that may extend the dispatch below @@ -673,30 +665,57 @@ class Provider(ABC): Provider._ep_providers[name] = None return None + @staticmethod + def get_class(provider: str) -> type: + """Resolve the provider class for a name (built-in or entry-point). + + Side-effect-free: no ``sys.exit``, no global state. Collision warnings + are emitted by ``init_global_provider``, not here. The caller handles + errors (CLI exits; the API can return HTTP 400). + + Args: + provider: Provider name, e.g. ``"aws"`` or an external plug-in. + + Returns: + The provider class (a subclass of :class:`Provider`). + + Raises: + ImportError: If not found as built-in or entry point, or a + built-in's transitive dependency is missing. + """ + if Provider.is_builtin(provider): + provider_class_path = f"{providers_path}.{provider}.{provider}_provider" + provider_class_name = f"{provider.capitalize()}Provider" + # Let ImportError propagate — the caller decides whether to + # sys.exit (CLI) or return HTTP 400 (API). + module = import_module(provider_class_path) + try: + return getattr(module, provider_class_name) + except AttributeError: + # Module exists but doesn't define the expected class — + # fall through to entry points. + cls = Provider._load_ep_provider(provider) + if cls is not None: + return cls + raise ImportError( + f"Provider '{provider}' not found as built-in or entry point" + ) + + cls = Provider._load_ep_provider(provider) + if cls is None: + raise ImportError( + f"Provider '{provider}' not found as built-in or entry point" + ) + return cls + @staticmethod def get_providers_help_text() -> dict: """Returns a dict of {provider_name: cli_help_text} for all available providers.""" help_text = {} for name in Provider.get_available_providers(): try: - # Try built-in first - module_path = f"{providers_path}.{name}.{name}_provider" - module = import_module(module_path) - cls = None - for attr_name in dir(module): - attr = getattr(module, attr_name) - if ( - isinstance(attr, type) - and issubclass(attr, Provider) - and attr is not Provider - ): - cls = attr - break - help_text[name] = getattr(cls, "_cli_help_text", "") if cls else "" - except ImportError: - # External provider — load via entry point - cls = Provider._load_ep_provider(name) - help_text[name] = getattr(cls, "_cli_help_text", "") if cls else "" + cls = Provider.get_class(name) + help_text[name] = getattr(cls, "_cli_help_text", "") except Exception as error: logger.warning( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" diff --git a/tests/providers/external/test_dynamic_provider_loading.py b/tests/providers/external/test_dynamic_provider_loading.py index 9f45648c94..fd91f8b005 100644 --- a/tests/providers/external/test_dynamic_provider_loading.py +++ b/tests/providers/external/test_dynamic_provider_loading.py @@ -502,18 +502,22 @@ class TestInitGlobalProviderBuiltinDependencyFailure: assert cls is None @patch("prowler.providers.common.provider.import_module") + @patch("prowler.providers.common.provider.Provider.is_builtin") @patch("prowler.providers.common.provider.Provider.get_available_providers") - def test_get_providers_help_text_builtin_path(self, mock_providers, mock_import): + def test_get_providers_help_text_builtin_path( + self, mock_providers, mock_is_builtin, mock_import + ): """get_providers_help_text reads _cli_help_text from a built-in provider module.""" import types mock_providers.return_value = ["fakebuiltin"] + mock_is_builtin.return_value = True mock_cls = type( - "FakeBuiltinProvider", (Provider,), {"_cli_help_text": "Built-in Help"} + "FakebuiltinProvider", (Provider,), {"_cli_help_text": "Built-in Help"} ) mock_module = types.ModuleType("fake_module") - mock_module.FakeBuiltinProvider = mock_cls + mock_module.FakebuiltinProvider = mock_cls mock_import.return_value = mock_module help_text = Provider.get_providers_help_text() @@ -1907,3 +1911,261 @@ class TestComplianceTableDispatch: mock_generic.assert_called_once() Provider._global = None + + +# =========================================================================== +# 12. Provider.get_class — Public side-effect-free class resolver +# =========================================================================== + + +class TestGetClass: + """Tests for Provider.get_class(provider) — the public, side-effect-free + class resolver that unblocks the Django API and other callers that need + a provider class without triggering CLI side-effects (sys.exit, global + provider mutation).""" + + # ----------------------------------------------------------------------- + # T1: Built-in provider resolves to correct class + # ----------------------------------------------------------------------- + + def test_get_class_builtin_returns_correct_class(self): + """get_class('aws') returns AwsProvider — identity check.""" + from prowler.providers.aws.aws_provider import AwsProvider + + cls = Provider.get_class("aws") + + assert cls is AwsProvider + + # ----------------------------------------------------------------------- + # T2: External entry-point provider resolves + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.importlib.metadata.entry_points") + @patch("prowler.providers.common.provider.Provider.is_builtin") + def test_get_class_external_ep_returns_class(self, mock_is_builtin, mock_ep): + """get_class resolves an external entry-point provider and returns that class.""" + mock_is_builtin.return_value = False + mock_ep.return_value = [ + _make_entry_point( + "fakeexternal", "pkg:FakeExternalProvider", "prowler.providers" + ), + ] + mock_ep.return_value[0].load.return_value = FakeExternalProvider + + cls = Provider.get_class("fakeexternal") + + assert cls is FakeExternalProvider + + # ----------------------------------------------------------------------- + # T3: Unknown provider raises, does NOT call sys.exit + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.importlib.metadata.entry_points") + @patch("prowler.providers.common.provider.Provider.is_builtin") + def test_get_class_unknown_raises_and_does_not_sys_exit( + self, mock_is_builtin, mock_ep + ): + """get_class raises for an unknown provider and never calls sys.exit.""" + mock_is_builtin.return_value = False + mock_ep.return_value = [] + + with pytest.raises((ImportError, Exception)) as exc_info: + Provider.get_class("totally_unknown_xyz_provider") + + # Must NOT be a SystemExit — that belongs in init_global_provider's + # wrapper, not in the pure resolver. + assert not isinstance(exc_info.value, SystemExit) + + # ----------------------------------------------------------------------- + # T4: get_class is PURE for built-ins — no collision warning, no EP call + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.logger") + @patch("prowler.providers.common.provider.Provider._load_ep_provider") + @patch("prowler.providers.common.provider.import_module") + @patch("prowler.providers.common.provider.Provider.is_builtin") + def test_get_class_builtin_with_ep_shadow_is_pure( + self, mock_is_builtin, mock_import, mock_load_ep, mock_logger + ): + """get_class for a built-in with a same-named EP is PURE: + - returns the built-in class + - does NOT emit a collision warning + - does NOT call _load_ep_provider (so _ep_providers cache stays empty for + this key, proving no side-effect) + """ + import types + + mock_is_builtin.return_value = True + mock_load_ep.return_value = FakeExternalProvider # plug-in shadow present + + fake_module = types.ModuleType("fake_builtin_module") + fake_builtin_cls = type("AwsProvider", (Provider,), {"_type": "aws"}) + fake_module.AwsProvider = fake_builtin_cls + mock_import.return_value = fake_module + + cls = Provider.get_class("aws") + + # Built-in class returned + assert cls is fake_builtin_cls + # No collision warning emitted — that is now init_global_provider's job + warning_msgs = [ + call.args[0] + for call in mock_logger.warning.call_args_list + if call.args and "IGNORED" in call.args[0] + ] + assert not warning_msgs, ( + "get_class must NOT emit a collision warning; " + "init_global_provider owns that responsibility" + ) + # _load_ep_provider must NOT have been called for the built-in path + mock_load_ep.assert_not_called() + # _ep_providers cache must not contain 'aws' (no side-effect) + assert "aws" not in Provider._ep_providers + + # ----------------------------------------------------------------------- + # T5: Regression — init_global_provider still resolves built-in correctly + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.load_and_validate_config_file") + @patch("prowler.providers.common.provider.Provider._load_ep_provider") + @patch("prowler.providers.common.provider.import_module") + def test_init_global_provider_still_resolves_builtin_via_get_class( + self, mock_import, mock_load_ep, mock_config + ): + """Regression: init_global_provider continues to work for external providers + after the class-resolution block is delegated to get_class. + + This mirrors TestProviderInitialization.test_init_global_provider_fallback_to_entry_point + which tests the entry-point fallback path. Here we verify the FakePureContractProvider + path (pure from_cli_args returning instance) still works — i.e., init_global_provider + correctly wires the returned instance as global provider. + """ + mock_import.side_effect = ImportError("No built-in") + mock_load_ep.return_value = FakePureContractProvider + mock_config.return_value = {} + + args = Namespace( + provider="fakepure", + fixer_config="config.yaml", + config_file="config.yaml", + ) + + Provider._global = None + Provider.init_global_provider(args) + + assert isinstance(Provider._global, FakePureContractProvider) + Provider._global = None + + # ----------------------------------------------------------------------- + # T6: Regression — get_providers_help_text returns same text after refactor + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.Provider._load_ep_provider") + @patch("prowler.providers.common.provider.Provider.get_available_providers") + def test_get_providers_help_text_identical_after_refactor_external( + self, mock_providers, mock_load_ep + ): + """get_providers_help_text returns identical _cli_help_text for an external + provider both before and after the refactor to use get_class internally.""" + mock_providers.return_value = ["fakeexternal"] + mock_load_ep.return_value = FakeExternalProvider + + help_text = Provider.get_providers_help_text() + + # Must match the known _cli_help_text on FakeExternalProvider + assert help_text["fakeexternal"] == "Fake External Provider" + + @patch("prowler.providers.common.provider.import_module") + @patch("prowler.providers.common.provider.Provider.is_builtin") + @patch("prowler.providers.common.provider.Provider.get_available_providers") + def test_get_providers_help_text_identical_after_refactor_builtin( + self, mock_providers, mock_is_builtin, mock_import + ): + """get_providers_help_text returns identical _cli_help_text for a built-in + provider both before and after the refactor to use get_class internally. + is_builtin is mocked to True so get_class takes the built-in import path.""" + import types + + mock_providers.return_value = ["fakebuiltin"] + mock_is_builtin.return_value = True + mock_cls = type( + "FakebuiltinProvider", (Provider,), {"_cli_help_text": "Built-in Help"} + ) + mock_module = types.ModuleType("fake_module") + mock_module.FakebuiltinProvider = mock_cls + mock_import.return_value = mock_module + + help_text = Provider.get_providers_help_text() + + assert help_text["fakebuiltin"] == "Built-in Help" + + # ----------------------------------------------------------------------- + # T7: init_global_provider emits collision warning (not get_class) + # ----------------------------------------------------------------------- + + @patch("prowler.providers.common.provider.load_and_validate_config_file") + @patch("prowler.providers.common.provider.Provider._load_ep_provider") + @patch("prowler.providers.common.provider.import_module") + @patch("prowler.providers.common.provider.Provider.is_builtin") + def test_init_global_provider_emits_collision_warning_for_builtin_ep_shadow( + self, mock_is_builtin, mock_import, mock_load_ep, mock_config, caplog + ): + """init_global_provider (not get_class) emits the collision warning + when a built-in provider has a same-named entry-point plug-in registered. + + This is the counterpart to test_get_class_builtin_with_ep_shadow_is_pure: + the warning responsibility moved OUT of get_class and INTO + init_global_provider, so users still see the message on CLI invocation + but prowler --help and API calls (which never hit init_global_provider) + do not spuriously emit it. + """ + import logging + import types + + mock_is_builtin.return_value = True + mock_load_ep.return_value = FakeExternalProvider # plug-in shadow + + fake_module = types.ModuleType("fake_builtin_module") + fake_module.AwsProvider = MagicMock(side_effect=lambda **_kw: None) + mock_import.return_value = fake_module + mock_config.return_value = {} + + args = Namespace( + provider="aws", + fixer_config="config.yaml", + config_file="config.yaml", + aws_retries_max_attempts=3, + role=None, + session_duration=None, + external_id=None, + role_session_name=None, + mfa=None, + profile=None, + region=None, + excluded_region=None, + organizations_role=None, + scan_unused_services=False, + resource_tag=None, + resource_arn=None, + mutelist_file=None, + ) + + Provider._global = None + with caplog.at_level(logging.WARNING, logger="prowler"): + try: + Provider.init_global_provider(args) + except BaseException: + # AwsProvider mock is fake; dispatch may fail — only the + # warning emitted BEFORE dispatch matters here. + pass + Provider._global = None + + collision_warnings = [ + r.message + for r in caplog.records + if "Plug-in provider 'aws'" in r.message and "IGNORED" in r.message + ] + assert collision_warnings, ( + "init_global_provider must emit the collision warning when a " + "same-named EP plug-in exists for a built-in provider" + )