fix(security): harden plugin entry point loading #7785
@@ -249,6 +249,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
e2e test files and removed `tdd_expected_fail` from 4 context assembly e2e tests where
|
||||
bugs were already fixed.
|
||||
|
||||
- **PluginLoader entry point prefix validation** (#7476): Parse entry point targets before
|
||||
import, enforce the module allowlist ahead of loading, and add Behave plus Robot Framework
|
||||
regression coverage to ensure disallowed prefixes never execute untrusted module-level code
|
||||
in either unit or integration flows.
|
||||
|
||||
- **`issue-state-updater` Bash Script Errors**: Removed problematic bash script examples
|
||||
that tried to invoke `task forgejo-label-manager` as a bash command (the Task tool cannot
|
||||
be invoked from bash). Replaced with clear step-by-step operational instructions and
|
||||
|
||||
@@ -18,5 +18,6 @@ Below are some of the specific details of various contributions.
|
||||
* HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
|
||||
* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution.
|
||||
* HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
|
||||
* HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading.
|
||||
* This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
|
||||
* HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system.
|
||||
|
||||
@@ -192,6 +192,15 @@ Feature: Plugin Architecture Framework with module:ClassName resolution
|
||||
Then the discovered plugin list should have 1 entry
|
||||
And the first descriptor name should be "test-ep"
|
||||
|
||||
@entry_points @security @tdd_issue @tdd_issue_7476
|
||||
Scenario: Entry point with disallowed prefix is skipped without loading module
|
||||
Given a PluginLoader with default prefixes
|
||||
And a mocked entry point group "cleveragents.plugins" with raw entry "malicious=maliciouspkg.attack:Exploit"
|
||||
When I discover plugins from the mocked entry point group
|
||||
Then the discovered plugin list should be empty
|
||||
And the mocked entry point load should not be called
|
||||
And a security warning should be emitted for disallowed entry point "malicious"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PluginManager — lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
import contextlib
|
||||
import threading
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import ANY, MagicMock, patch
|
||||
|
||||
from behave import given, then, when # type: ignore[import-untyped]
|
||||
from behave.runner import Context # type: ignore[import-untyped]
|
||||
@@ -429,6 +429,37 @@ def step_first_descriptor_name(context: Context, name: str) -> None:
|
||||
assert context.discovered[0].name == name
|
||||
|
||||
|
||||
@given('a mocked entry point group "{group}" with raw entry "{entry_spec}"')
|
||||
def step_mock_raw_entry(context: Context, group: str, entry_spec: str) -> None:
|
||||
name, value = entry_spec.split("=", 1)
|
||||
mock_ep = MagicMock()
|
||||
mock_ep.name = name
|
||||
mock_ep.value = value
|
||||
mock_ep.load = MagicMock(name="load")
|
||||
|
||||
context.mock_group = group
|
||||
context.mock_eps = [mock_ep]
|
||||
context.mock_ep_last = mock_ep
|
||||
context.loader._logger = MagicMock()
|
||||
|
||||
|
||||
@then("the mocked entry point load should not be called")
|
||||
def step_entry_point_not_loaded(context: Context) -> None:
|
||||
context.mock_ep_last.load.assert_not_called()
|
||||
|
||||
|
||||
@then('a security warning should be emitted for disallowed entry point "{name}"')
|
||||
def step_security_warning_emitted(context: Context, name: str) -> None:
|
||||
logger_mock = context.loader._logger
|
||||
logger_mock.warning.assert_any_call(
|
||||
"plugin.entry_point_disallowed_prefix",
|
||||
name=name,
|
||||
value=context.mock_ep_last.value,
|
||||
group=context.mock_group,
|
||||
error=ANY,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PluginManager — lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -39,7 +39,7 @@ def step_mock_failing_entry_point(context):
|
||||
"""Create a mock entry point whose load() raises an exception."""
|
||||
failing_ep = MagicMock()
|
||||
failing_ep.name = "broken_plugin"
|
||||
failing_ep.value = "some.module:BrokenClass"
|
||||
failing_ep.value = "cleveragents.domain.models.acms.nonexistent:BrokenClass"
|
||||
failing_ep.load.side_effect = ImportError("Simulated import failure")
|
||||
|
||||
# Mock the entry_points() call to return a selectable object
|
||||
|
||||
@@ -284,6 +284,49 @@ def main() -> int:
|
||||
print(f"plugin-extension-points-fail: {exc}")
|
||||
return 1
|
||||
|
||||
if command == "entry-point-security":
|
||||
# Security regression test for issue #7476.
|
||||
# Verifies that load_from_entry_points does NOT call ep.load() for
|
||||
# entry points whose module prefix is not in the allowlist.
|
||||
# Uses importlib.metadata.EntryPoint mocking via a real PluginLoader
|
||||
# instance with a controlled entry_points() override.
|
||||
try:
|
||||
import importlib.metadata
|
||||
import unittest.mock as mock
|
||||
|
||||
loader = PluginLoader()
|
||||
|
||||
# Build a fake entry point that references a disallowed module.
|
||||
fake_ep = mock.MagicMock()
|
||||
fake_ep.name = "malicious"
|
||||
fake_ep.value = "maliciouspkg.attack:Exploit"
|
||||
fake_ep.group = "cleveragents.plugins"
|
||||
|
||||
# Patch importlib.metadata.entry_points to return our fake entry point.
|
||||
fake_eps = mock.MagicMock()
|
||||
fake_eps.select.return_value = [fake_ep]
|
||||
|
||||
with mock.patch.object(
|
||||
importlib.metadata,
|
||||
"entry_points",
|
||||
return_value=fake_eps,
|
||||
):
|
||||
descriptors = loader.load_from_entry_points("cleveragents.plugins")
|
||||
|
||||
# The disallowed entry point must be skipped — no descriptors returned.
|
||||
assert descriptors == [], (
|
||||
f"Expected empty descriptor list, got: {descriptors}"
|
||||
)
|
||||
|
||||
# ep.load() must NEVER have been called for the disallowed entry point.
|
||||
fake_ep.load.assert_not_called()
|
||||
|
||||
print("plugin-entry-point-security-ok")
|
||||
return 0
|
||||
except Exception as exc:
|
||||
print(f"plugin-entry-point-security-fail: {exc}")
|
||||
return 1
|
||||
|
||||
print(f"Unknown command: {command}")
|
||||
return 1
|
||||
|
||||
|
||||
@@ -87,3 +87,16 @@ PluginManager Extension Points
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} plugin-extension-points-ok
|
||||
|
||||
PluginLoader Entry Point Security - Disallowed Prefix Not Loaded
|
||||
[Documentation] Security regression test for issue #7476.
|
||||
...
|
||||
... Verifies that load_from_entry_points does NOT call ep.load()
|
||||
... for entry points whose module prefix is not in the allowlist.
|
||||
... This ensures arbitrary code execution is prevented even when
|
||||
... a malicious package registers an entry point.
|
||||
${result}= Run Process ${PYTHON} ${HELPER} entry-point-security cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} plugin-entry-point-security-ok
|
||||
|
||||
@@ -182,11 +182,33 @@ class PluginLoader:
|
||||
eps = entry_points.select(group=group)
|
||||
|
||||
for ep in eps:
|
||||
raw_value = ep.value
|
||||
try:
|
||||
module_path, class_name = self._parse_entry_point_value(raw_value)
|
||||
except PluginLoadError as exc:
|
||||
self._logger.warning(
|
||||
"plugin.entry_point_invalid_value",
|
||||
name=ep.name,
|
||||
value=raw_value,
|
||||
group=group,
|
||||
error=str(exc),
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
self._validate_module_prefix(module_path)
|
||||
except PluginLoadError as exc:
|
||||
self._logger.warning(
|
||||
"plugin.entry_point_disallowed_prefix",
|
||||
name=ep.name,
|
||||
value=raw_value,
|
||||
group=group,
|
||||
error=str(exc),
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
ep.load()
|
||||
module_path = ep.value.rsplit(":", 1)[0] if ":" in ep.value else ""
|
||||
class_name = ep.value.rsplit(":", 1)[1] if ":" in ep.value else ep.value
|
||||
|
||||
descriptor = PluginDescriptor(
|
||||
name=ep.name,
|
||||
module_path=module_path,
|
||||
@@ -197,14 +219,14 @@ class PluginLoader:
|
||||
self._logger.info(
|
||||
"plugin.discovered_entry_point",
|
||||
name=ep.name,
|
||||
value=ep.value,
|
||||
value=raw_value,
|
||||
group=group,
|
||||
)
|
||||
except Exception as exc:
|
||||
self._logger.warning(
|
||||
"plugin.entry_point_failed",
|
||||
name=ep.name,
|
||||
value=ep.value,
|
||||
value=raw_value,
|
||||
group=group,
|
||||
error=str(exc),
|
||||
)
|
||||
@@ -270,3 +292,28 @@ class PluginLoader:
|
||||
f"may be dynamically imported."
|
||||
)
|
||||
raise PluginLoadError(msg)
|
||||
|
||||
@staticmethod
|
||||
def _parse_entry_point_value(value: str) -> tuple[str, str]:
|
||||
"""Split an entry point value into module and class components.
|
||||
|
||||
The returned ``module_path`` is intentionally not validated beyond
|
||||
basic shape checks; prefix allowlist enforcement and any stricter
|
||||
validation must be performed by :meth:`_validate_module_prefix` to
|
||||
preserve the layered defense order (parse → validate prefix → load).
|
||||
"""
|
||||
normalized_value = value.strip()
|
||||
if ":" not in normalized_value:
|
||||
raise PluginLoadError(
|
||||
"Entry point value must be in 'module:ClassName' format."
|
||||
)
|
||||
|
||||
module_path, class_name = (
|
||||
part.strip() for part in normalized_value.split(":", 1)
|
||||
)
|
||||
if not module_path or not class_name:
|
||||
raise PluginLoadError(
|
||||
"Entry point value must include both module path and class name."
|
||||
)
|
||||
|
||||
return module_path, class_name
|
||||
|
||||
Reference in New Issue
Block a user