mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 21:11:53 +00:00
fix(sdk): use find_spec to distinguish missing vs broken built-ins
This commit is contained in:
@@ -1,4 +1,6 @@
|
||||
import importlib
|
||||
import importlib.metadata
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -365,17 +367,21 @@ def import_check(check_path: str) -> ModuleType:
|
||||
def _resolve_check_module(
|
||||
provider_type: str, service: str, check_name: str
|
||||
) -> ModuleType:
|
||||
"""Resolve and import a check module — tries built-in path first, then entry points."""
|
||||
"""Resolve and import a check module — tries built-in path first, then entry points.
|
||||
|
||||
Uses find_spec to distinguish "built-in doesn't exist" from "built-in
|
||||
exists but failed to import" (broken transitive dep, etc.). We only fall
|
||||
through to entry points when the built-in module truly isn't there —
|
||||
otherwise the import error propagates and the user sees the real cause,
|
||||
instead of being silently replaced by an entry-point plug-in that
|
||||
happens to share the same check name.
|
||||
"""
|
||||
# Built-in path
|
||||
builtin_path = f"prowler.providers.{provider_type}.services.{service}.{check_name}.{check_name}"
|
||||
try:
|
||||
if importlib.util.find_spec(builtin_path) is not None:
|
||||
return import_check(builtin_path)
|
||||
except ModuleNotFoundError:
|
||||
pass
|
||||
|
||||
# Entry point lookup
|
||||
import importlib.metadata
|
||||
|
||||
# Entry point lookup — only consulted when the built-in truly doesn't exist
|
||||
for ep in importlib.metadata.entry_points(group=f"prowler.checks.{provider_type}"):
|
||||
if ep.name == check_name:
|
||||
return importlib.import_module(ep.value)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import importlib
|
||||
import importlib.metadata
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
from pkgutil import walk_packages
|
||||
@@ -56,8 +57,18 @@ def recover_checks_from_provider(
|
||||
return []
|
||||
|
||||
checks = []
|
||||
# Built-in checks from prowler.providers.{provider}.services
|
||||
try:
|
||||
# Built-in checks from prowler.providers.{provider}.services. Use
|
||||
# find_spec to distinguish "service doesn't exist" from "service
|
||||
# exists but failed to import" (broken transitive dep, etc.). We
|
||||
# only fall through to entry points when the built-in package truly
|
||||
# doesn't exist — otherwise the import error propagates and the
|
||||
# user sees the real cause, instead of being silently replaced by
|
||||
# an entry-point plug-in that happens to share a name.
|
||||
service_path = f"prowler.providers.{provider}.services"
|
||||
if service:
|
||||
service_path += f".{service}"
|
||||
|
||||
if importlib.util.find_spec(service_path) is not None:
|
||||
modules = list_modules(provider, service)
|
||||
for module_name in modules:
|
||||
# Format: "prowler.providers.{provider}.services.{service}.{check_name}.{check_name}"
|
||||
@@ -72,12 +83,6 @@ def recover_checks_from_provider(
|
||||
check_name = check_module_name.split(".")[-1]
|
||||
check_info = (check_name, check_path)
|
||||
checks.append(check_info)
|
||||
except ModuleNotFoundError:
|
||||
# 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 — always consulted, with
|
||||
# optional service filter. Previously gated by `if not service:`, which
|
||||
|
||||
+93
-27
@@ -697,14 +697,19 @@ class TestCheckDiscovery:
|
||||
assert checks == []
|
||||
|
||||
@patch("prowler.lib.check.utils._recover_ep_checks")
|
||||
@patch("prowler.lib.check.utils.list_modules")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_handles_external_provider_without_services(
|
||||
self, mock_list_modules, mock_ep_checks
|
||||
self, mock_find_spec, mock_ep_checks
|
||||
):
|
||||
"""Test 13: recover_checks_from_provider doesn't crash for external providers."""
|
||||
"""Test 13: recover_checks_from_provider doesn't crash for external providers.
|
||||
|
||||
With find_spec returning None (built-in package doesn't exist), discovery
|
||||
falls through to entry points cleanly — no ModuleNotFoundError catch
|
||||
needed.
|
||||
"""
|
||||
from prowler.lib.check.utils import recover_checks_from_provider
|
||||
|
||||
mock_list_modules.side_effect = ModuleNotFoundError("No services")
|
||||
mock_find_spec.return_value = None # not a built-in
|
||||
mock_ep_checks.return_value = [("ext_check", "/path/to/check")]
|
||||
|
||||
checks = recover_checks_from_provider("fakeexternal")
|
||||
@@ -714,12 +719,15 @@ class TestCheckDiscovery:
|
||||
|
||||
@patch("prowler.lib.check.utils._recover_ep_checks")
|
||||
@patch("prowler.lib.check.utils.list_modules")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_combines_builtin_and_entry_points(
|
||||
self, mock_list_modules, mock_ep_checks
|
||||
self, mock_find_spec, mock_list_modules, mock_ep_checks
|
||||
):
|
||||
"""Test 14: recover_checks_from_provider combines built-in and entry point checks."""
|
||||
from prowler.lib.check.utils import recover_checks_from_provider
|
||||
|
||||
mock_find_spec.return_value = MagicMock() # built-in package exists
|
||||
|
||||
# Simulate a built-in module
|
||||
builtin_module = MagicMock()
|
||||
builtin_module.name = "prowler.providers.aws.services.ec2.check_a.check_a"
|
||||
@@ -789,16 +797,16 @@ class TestCheckDiscovery:
|
||||
assert len(checks) == 2
|
||||
|
||||
@patch("prowler.lib.check.utils._recover_ep_checks")
|
||||
@patch("prowler.lib.check.utils.list_modules")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_external_provider_with_service(
|
||||
self, mock_list_modules, mock_ep_checks
|
||||
self, mock_find_spec, 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."""
|
||||
"""External provider with --service: built-in package doesn't exist,
|
||||
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_find_spec.return_value = None # not a built-in
|
||||
mock_ep_checks.return_value = [("container_check", "/ext/path")]
|
||||
|
||||
checks = recover_checks_from_provider("dockerdesktop", service="container")
|
||||
@@ -808,32 +816,32 @@ class TestCheckDiscovery:
|
||||
mock_ep_checks.assert_called_once_with("dockerdesktop", "container")
|
||||
|
||||
@patch("prowler.lib.check.utils._recover_ep_checks")
|
||||
@patch("prowler.lib.check.utils.list_modules")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_unknown_service_fails_cleanly(
|
||||
self, mock_list_modules, mock_ep_checks
|
||||
self, mock_find_spec, 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_find_spec.return_value = None
|
||||
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")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_builtin_with_new_external_service(
|
||||
self, mock_list_modules, mock_ep_checks
|
||||
self, mock_find_spec, 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."""
|
||||
the built-in package for that specific service doesn't exist (find_spec
|
||||
returns None), 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_find_spec.return_value = None # built-in for new_aws_service doesn't exist
|
||||
mock_ep_checks.return_value = [("new_check", "/ext/path")]
|
||||
|
||||
checks = recover_checks_from_provider("aws", service="new_aws_service")
|
||||
@@ -842,6 +850,29 @@ class TestCheckDiscovery:
|
||||
assert checks[0][0] == "new_check"
|
||||
mock_ep_checks.assert_called_once_with("aws", "new_aws_service")
|
||||
|
||||
@patch("prowler.lib.check.utils._recover_ep_checks")
|
||||
@patch("prowler.lib.check.utils.list_modules")
|
||||
@patch("prowler.lib.check.utils.importlib.util.find_spec")
|
||||
def test_recover_checks_surfaces_error_when_builtin_service_import_fails(
|
||||
self, mock_find_spec, mock_list_modules, mock_ep_checks
|
||||
):
|
||||
"""Regression guard: when a built-in service's package exists but one
|
||||
of its modules fails to import (e.g. a broken transitive dependency),
|
||||
the error must surface via the global exception handler — not be
|
||||
silently swallowed and replaced by an entry-point plug-in that happens
|
||||
to share a name. See PR #10700 review (HugoPBrito)."""
|
||||
from prowler.lib.check.utils import recover_checks_from_provider
|
||||
|
||||
mock_find_spec.return_value = MagicMock() # built-in service exists
|
||||
mock_list_modules.side_effect = ImportError("missing transitive dep: foo")
|
||||
|
||||
# Even if a plug-in registers checks for the same service, it must NOT
|
||||
# silently take over — the import error wins.
|
||||
mock_ep_checks.return_value = [("evil_check", "/evil/path")]
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
recover_checks_from_provider("aws", service="ec2")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 4. Check Execution
|
||||
@@ -851,13 +882,15 @@ class TestCheckDiscovery:
|
||||
class TestCheckExecution:
|
||||
"""Tests 15-17: _resolve_check_module."""
|
||||
|
||||
@patch("prowler.lib.check.check.importlib.util.find_spec")
|
||||
@patch("prowler.lib.check.check.import_check")
|
||||
def test_resolve_check_module_builtin_first(self, mock_import):
|
||||
def test_resolve_check_module_builtin_first(self, mock_import, mock_find_spec):
|
||||
"""Test 15: _resolve_check_module resolves built-in checks first."""
|
||||
from prowler.lib.check.check import _resolve_check_module
|
||||
|
||||
mock_module = MagicMock()
|
||||
mock_import.return_value = mock_module
|
||||
mock_find_spec.return_value = MagicMock() # built-in package exists
|
||||
|
||||
result = _resolve_check_module("aws", "ec2", "my_check")
|
||||
|
||||
@@ -866,12 +899,15 @@ class TestCheckExecution:
|
||||
"prowler.providers.aws.services.ec2.my_check.my_check"
|
||||
)
|
||||
|
||||
@patch("prowler.lib.check.check.importlib.util.find_spec")
|
||||
@patch("prowler.lib.check.check.import_check")
|
||||
def test_resolve_check_module_fallback_to_entry_point(self, mock_import_check):
|
||||
"""Test 16: _resolve_check_module falls back to entry point."""
|
||||
def test_resolve_check_module_fallback_to_entry_point(
|
||||
self, mock_import_check, mock_find_spec
|
||||
):
|
||||
"""Test 16: _resolve_check_module falls back to entry point when built-in is absent."""
|
||||
from prowler.lib.check.check import _resolve_check_module
|
||||
|
||||
mock_import_check.side_effect = ModuleNotFoundError("Not built-in")
|
||||
mock_find_spec.return_value = None # built-in does not exist
|
||||
|
||||
mock_ext_module = MagicMock()
|
||||
ep = _make_entry_point(
|
||||
@@ -886,19 +922,49 @@ class TestCheckExecution:
|
||||
|
||||
assert result is mock_ext_module
|
||||
mock_imp.assert_called_with("ext_pkg.checks.my_check")
|
||||
mock_import_check.assert_not_called()
|
||||
|
||||
@patch("prowler.lib.check.check.importlib.metadata.entry_points")
|
||||
@patch("prowler.lib.check.check.import_check")
|
||||
def test_resolve_check_module_raises_when_not_found(self, mock_import, mock_ep):
|
||||
@patch("prowler.lib.check.check.importlib.util.find_spec")
|
||||
def test_resolve_check_module_raises_when_not_found(self, mock_find_spec, mock_ep):
|
||||
"""Test 17: _resolve_check_module raises ModuleNotFoundError when both fail."""
|
||||
from prowler.lib.check.check import _resolve_check_module
|
||||
|
||||
mock_import.side_effect = ModuleNotFoundError("Not built-in")
|
||||
mock_find_spec.return_value = None
|
||||
mock_ep.return_value = []
|
||||
|
||||
with pytest.raises(ModuleNotFoundError, match="not found"):
|
||||
_resolve_check_module("fake", "svc", "nonexistent_check")
|
||||
|
||||
@patch("prowler.lib.check.check.importlib.util.find_spec")
|
||||
@patch("prowler.lib.check.check.import_check")
|
||||
def test_resolve_check_module_surfaces_error_when_builtin_import_fails(
|
||||
self, mock_import_check, mock_find_spec
|
||||
):
|
||||
"""Regression guard: a built-in check whose module exists but fails to
|
||||
import (e.g. broken transitive dependency) MUST surface the real error,
|
||||
not be silently replaced by an entry-point plug-in that happens to
|
||||
share the same check name. See PR #10700 review (HugoPBrito)."""
|
||||
from prowler.lib.check.check import _resolve_check_module
|
||||
|
||||
mock_find_spec.return_value = MagicMock() # built-in module exists
|
||||
mock_import_check.side_effect = ImportError("missing transitive dep: foo")
|
||||
|
||||
# Even if a plug-in registers the same check name, it must NOT hijack
|
||||
ep = _make_entry_point(
|
||||
"ec2_instance_public_ip",
|
||||
"evil_pkg.checks.ec2_instance_public_ip",
|
||||
"prowler.checks.aws",
|
||||
)
|
||||
with (
|
||||
patch("importlib.metadata.entry_points", return_value=[ep]),
|
||||
patch("importlib.import_module") as mock_imp,
|
||||
):
|
||||
with pytest.raises(ImportError, match="missing transitive dep"):
|
||||
_resolve_check_module("aws", "ec2", "ec2_instance_public_ip")
|
||||
# Entry-point fallback must not be reached when built-in exists
|
||||
mock_imp.assert_not_called()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 5. CLI Arguments
|
||||
|
||||
Reference in New Issue
Block a user