fix(sdk): built-in wins on plug-in collision for providers and checks

This commit is contained in:
StylusFrost
2026-04-30 19:50:19 +02:00
parent e7f23bb13f
commit c7aa536896
5 changed files with 182 additions and 24 deletions
+14 -8
View File
@@ -367,16 +367,22 @@ 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.
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 wins on CheckID collision. Plug-ins are first-class extenders
(they can add new checks under new CheckIDs) but cannot override
existing built-ins — a security tool prefers fail-loud predictability
over silent overrides. CheckMetadata.get_bulk() applies the same
precedence on the metadata side (first-write-wins) and emits a warning
when a plug-in tries to override, so the user knows their plug-in
duplicate is being ignored and can rename it.
Uses find_spec on the built-in path to distinguish "built-in doesn't
exist" from "built-in exists but failed to import" (broken transitive
dep, etc.). When the built-in is present, an import error propagates
instead of being silently swallowed.
"""
# Built-in path
# Built-in first — built-in wins on CheckID collision
builtin_path = f"prowler.providers.{provider_type}.services.{service}.{check_name}.{check_name}"
if importlib.util.find_spec(builtin_path) is not None:
return import_check(builtin_path)
+14
View File
@@ -435,6 +435,20 @@ class CheckMetadata(BaseModel):
metadata_file = f"{check_path}/{check_name}.metadata.json"
# Load metadata
check_metadata = load_check_metadata(metadata_file)
# Built-in wins on CheckID collision. Plug-in entry points are
# appended after built-ins by `recover_checks_from_provider`, so
# a duplicate CheckID here means an entry-point check is trying
# to override a built-in. Ignore the override (the built-in
# metadata stays) and surface it via a warning — matching the
# precedence enforced by `_resolve_check_module`.
if check_metadata.CheckID in bulk_check_metadata:
logger.warning(
f"Plug-in check metadata '{check_metadata.CheckID}' "
f"(loaded from '{metadata_file}') is being IGNORED — "
f"a built-in with the same CheckID exists. To use your "
f"plug-in, register it under a different CheckID."
)
continue
bulk_check_metadata[check_metadata.CheckID] = check_metadata
return bulk_check_metadata
+13
View File
@@ -269,6 +269,19 @@ class Provider(ABC):
# 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:
+32
View File
@@ -95,6 +95,38 @@ class TestCheckMetada:
"/path/to/accessanalyzer_enabled/accessanalyzer_enabled.metadata.json"
)
@mock.patch("prowler.lib.check.models.logger")
@mock.patch("prowler.lib.check.models.load_check_metadata")
@mock.patch("prowler.lib.check.models.recover_checks_from_provider")
def test_get_bulk_builtin_wins_on_check_id_collision(
self, mock_recover_checks, mock_load_metadata, mock_logger
):
"""Regression guard: when an entry-point plug-in re-registers a
built-in CheckID, the BUILT-IN metadata wins (first-write-wins) and
the plug-in is IGNORED. The override is surfaced via a warning so
the user knows their plug-in duplicate is being skipped and can
rename it. Matches the precedence in `_resolve_check_module`. See
PR #10700 review (HugoPBrito)."""
# Built-in first, plug-in last (matches recover_checks_from_provider order)
mock_recover_checks.return_value = [
("accessanalyzer_enabled", "/builtin/accessanalyzer_enabled"),
("accessanalyzer_enabled", "/plugin/accessanalyzer_enabled"),
]
builtin_metadata = mock.MagicMock(CheckID="accessanalyzer_enabled")
plugin_metadata = mock.MagicMock(CheckID="accessanalyzer_enabled")
mock_load_metadata.side_effect = [builtin_metadata, plugin_metadata]
result = CheckMetadata.get_bulk(provider="aws")
# Built-in wins (first-write-wins on CheckID), plug-in is ignored
assert result["accessanalyzer_enabled"] is builtin_metadata
# Override is surfaced via warning naming the plug-in metadata file
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args.args[0]
assert "accessanalyzer_enabled" in warning_msg
assert "/plugin/accessanalyzer_enabled" in warning_msg
@mock.patch("prowler.lib.check.models.load_check_metadata")
@mock.patch("prowler.lib.check.models.recover_checks_from_provider")
def test_list(self, mock_recover_checks, mock_load_metadata):
+109 -16
View File
@@ -642,6 +642,70 @@ class TestProviderInitialization:
assert isinstance(Provider._global, FakeExternalProvider)
Provider._global = None
@patch("prowler.providers.common.provider.logger")
@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_warns_when_plugin_shadowed_by_builtin(
self, mock_is_builtin, mock_import, mock_load_ep, mock_config, mock_logger
):
"""Regression guard: when a plug-in registers a provider name that
collides with a built-in, the BUILT-IN wins and a warning is emitted
naming the shadowed plug-in. Matches the precedence enforced by
`_resolve_check_module` and `CheckMetadata.get_bulk` for checks. See
PR #10700 review (HugoPBrito).
"""
# Simulate a built-in `aws` that exists, AND a plug-in registered
# under the same `aws` name via entry points.
mock_is_builtin.return_value = True
mock_load_ep.return_value = FakeExternalProvider # plug-in shadow
mock_import.return_value = MagicMock(
AwsProvider=MagicMock(side_effect=lambda **kw: None)
)
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
try:
Provider.init_global_provider(args)
except BaseException:
# The AwsProvider mock is fake and the dispatch may sys.exit on
# the simulated failure; we only care about the warning emitted
# before the dispatch happens.
pass
finally:
Provider._global = None
# Warning was emitted naming the shadowed plug-in
warning_msgs = [
call.args[0]
for call in mock_logger.warning.call_args_list
if call.args
and "Plug-in provider 'aws'" in call.args[0]
]
assert warning_msgs, "expected a warning about the shadowed plug-in 'aws'"
assert "IGNORED" in warning_msgs[0]
# ===========================================================================
# 3. Check Discovery
@@ -924,6 +988,44 @@ class TestCheckExecution:
mock_imp.assert_called_with("ext_pkg.checks.my_check")
mock_import_check.assert_not_called()
@patch("prowler.lib.check.check.importlib.util.find_spec")
@patch("prowler.lib.check.check.import_check")
def test_resolve_check_module_builtin_wins_over_entry_point(
self, mock_import_check, mock_find_spec
):
"""Regression guard: when both a built-in and an entry-point check
exist with the same CheckID, the BUILT-IN wins. Plug-ins extend
Prowler with new checks but cannot silently override existing
built-ins a security tool prefers fail-loud predictability over
permissive overrides. CheckMetadata.get_bulk applies the same
precedence (first-write-wins) and emits a warning. See PR #10700
review (HugoPBrito)."""
from prowler.lib.check.check import _resolve_check_module
mock_find_spec.return_value = MagicMock() # built-in exists
builtin_module = MagicMock()
mock_import_check.return_value = builtin_module
# Plug-in registers same CheckID — must NOT take precedence
ep = _make_entry_point(
"ec2_instance_public_ip",
"plug_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,
):
result = _resolve_check_module("aws", "ec2", "ec2_instance_public_ip")
assert result is builtin_module
mock_import_check.assert_called_once_with(
"prowler.providers.aws.services.ec2.ec2_instance_public_ip.ec2_instance_public_ip"
)
# Plug-in must NOT be loaded when a built-in with the same CheckID exists
mock_imp.assert_not_called()
@patch("prowler.lib.check.check.importlib.metadata.entry_points")
@patch("prowler.lib.check.check.importlib.util.find_spec")
def test_resolve_check_module_raises_when_not_found(self, mock_find_spec, mock_ep):
@@ -941,29 +1043,20 @@ class TestCheckExecution:
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)."""
"""Regression guard: when no plug-in entry-point overrides the
check, a built-in whose module exists but fails to import (e.g.
broken transitive dependency) MUST surface the real error instead
of being silently treated as 'not found'. 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,
):
# No plug-in override — the built-in's import failure must propagate
with patch("importlib.metadata.entry_points", return_value=[]):
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()
# ===========================================================================