From daf7c644a62a9768bbd396a3e7fd23cd943fc501 Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Tue, 31 Mar 2026 16:59:02 +0800 Subject: [PATCH] =?UTF-8?q?=E6=89=AC=E4=BA=86=20sys=20=E6=8E=A7=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pytests/test_plugin_runtime.py | 134 +--------------- src/plugin_runtime/runner/runner_main.py | 188 ----------------------- 2 files changed, 2 insertions(+), 320 deletions(-) diff --git a/pytests/test_plugin_runtime.py b/pytests/test_plugin_runtime.py index 5c9f39b0..b866d3ff 100644 --- a/pytests/test_plugin_runtime.py +++ b/pytests/test_plugin_runtime.py @@ -1298,142 +1298,12 @@ class TestDependencyResolution: assert "test.demo-plugin" in loader.failed_plugins assert "on_unload" in loader.failed_plugins["test.demo-plugin"] - def test_isolate_sys_path_preserves_plugin_dirs(self): - import builtins - import importlib - - from src.plugin_runtime.runner import runner_main - - plugin_root = os.path.normpath("/tmp/maibot-plugin-root") - original_import = builtins.__import__ - original_import_module = importlib.import_module - original_path = list(sys.path) - original_meta_path = list(sys.meta_path) - - try: - if plugin_root in sys.path: - sys.path.remove(plugin_root) - - runner_main._isolate_sys_path([plugin_root]) - - assert plugin_root in sys.path - finally: - builtins.__import__ = original_import - importlib.import_module = original_import_module - sys.path[:] = original_path - sys.meta_path[:] = original_meta_path - - def test_isolate_sys_path_blocks_disallowed_src_imports(self): - import builtins - import importlib - - from src.plugin_runtime.runner import runner_main - - original_import = builtins.__import__ - original_import_module = importlib.import_module - original_path = list(sys.path) - original_meta_path = list(sys.meta_path) - sys.modules.pop("src.forbidden_demo", None) - - try: - runner_main._isolate_sys_path([]) - plugin_globals = { - "__name__": "_maibot_plugin_demo", - "__package__": "_maibot_plugin_demo", - "importlib": importlib, - } - - with pytest.raises(ImportError, match="不允许导入主程序模块"): - exec('importlib.import_module("src.forbidden_demo")', plugin_globals) - finally: - builtins.__import__ = original_import - importlib.import_module = original_import_module - sys.path[:] = original_path - sys.meta_path[:] = original_meta_path - sys.modules.pop("src.forbidden_demo", None) - - def test_isolate_sys_path_blocks_preloaded_runtime_modules(self): - import builtins - import importlib - - from src.plugin_runtime.runner import runner_main - - original_import = builtins.__import__ - original_import_module = importlib.import_module - original_path = list(sys.path) - original_meta_path = list(sys.meta_path) - - try: - runner_main._isolate_sys_path([]) - plugin_globals = { - "__name__": "_maibot_plugin_demo", - "__package__": "_maibot_plugin_demo", - "importlib": importlib, - } - - with pytest.raises(ImportError, match="rpc_client"): - exec('importlib.import_module("src.plugin_runtime.runner.rpc_client")', plugin_globals) - finally: - builtins.__import__ = original_import - importlib.import_module = original_import_module - sys.path[:] = original_path - sys.meta_path[:] = original_meta_path - - def test_isolate_sys_path_keeps_legacy_logger_import_available(self): - import builtins - import importlib - - from src.plugin_runtime.runner import runner_main - - original_import = builtins.__import__ - original_import_module = importlib.import_module - original_path = list(sys.path) - original_meta_path = list(sys.meta_path) - - try: - runner_main._isolate_sys_path([]) - plugin_globals = { - "__name__": "_maibot_plugin_demo", - "__package__": "_maibot_plugin_demo", - "importlib": importlib, - } - - exec('logger_module = importlib.import_module("src.common.logger")', plugin_globals) - logger_module = plugin_globals["logger_module"] - assert callable(logger_module.get_logger) - finally: - builtins.__import__ = original_import - importlib.import_module = original_import_module - sys.path[:] = original_path - sys.meta_path[:] = original_meta_path - - def test_isolate_sys_path_keeps_runtime_imports_working(self): - import builtins - import importlib - - from src.plugin_runtime.runner import runner_main - - original_import = builtins.__import__ - original_import_module = importlib.import_module - original_path = list(sys.path) - original_meta_path = list(sys.meta_path) - - try: - runner_main._isolate_sys_path([]) - - uds_module = importlib.import_module("src.plugin_runtime.transport.uds") - assert hasattr(uds_module, "UDSTransportClient") - finally: - builtins.__import__ = original_import - importlib.import_module = original_import_module - sys.path[:] = original_path - sys.meta_path[:] = original_meta_path - @pytest.mark.asyncio async def test_async_main_removes_sensitive_runtime_env_vars(self, monkeypatch): from src.plugin_runtime.runner import runner_main captured = {} + original_path = list(sys.path) class FakeRunner: def __init__( @@ -1457,7 +1327,6 @@ class TestDependencyResolution: monkeypatch.setenv(runner_main.ENV_PLUGIN_DIRS, "/tmp/plugins") monkeypatch.setenv(runner_main.ENV_EXTERNAL_PLUGIN_IDS, '{"demo.plugin":"1.0.0"}') monkeypatch.setattr(runner_main, "_install_shutdown_signal_handlers", lambda callback: None) - monkeypatch.setattr(runner_main, "_isolate_sys_path", lambda plugin_dirs: None) monkeypatch.setattr(runner_main, "PluginRunner", FakeRunner) await runner_main._async_main() @@ -1466,6 +1335,7 @@ class TestDependencyResolution: assert captured["session_token"] == "secret-token" assert captured["plugin_dirs"] == ["/tmp/plugins"] assert captured["external_available_plugins"] == {"demo.plugin": "1.0.0"} + assert sys.path == original_path # ─── Host-side ComponentRegistry 测试 ────────────────────── diff --git a/src/plugin_runtime/runner/runner_main.py b/src/plugin_runtime/runner/runner_main.py index 9de5d977..a1dc56fa 100644 --- a/src/plugin_runtime/runner/runner_main.py +++ b/src/plugin_runtime/runner/runner_main.py @@ -1183,191 +1183,6 @@ class PluginRunner: return self._rpc_client -# ─── sys.path 隔离 ──────────────────────────────────────── - - -def _isolate_sys_path(plugin_dirs: List[str]) -> None: - """清理 sys.path，限制 Runner 子进程只能访问标准库、SDK 和插件目录。 - - 同时阻止插件代码直接导入主程序内部 ``src.*`` 模块，并清理可直接从 - ``sys.modules`` 摸到的高权限叶子模块，避免绕过 SDK / capability 边界。 - """ - from importlib import util as importlib_util - from types import ModuleType - - import builtins - import importlib - import sysconfig - - # 保留: 标准库路径 + site-packages（含 SDK 和依赖） - stdlib_paths = set() - for key in ("stdlib", "platstdlib", "purelib", "platlib"): - if path := sysconfig.get_path(key): - stdlib_paths.add(os.path.normpath(path)) - - runtime_paths = set(stdlib_paths) - if os.name == "nt": - # Windows 的部分平台扩展模块和依赖会通过 /DLLs 暴露在 sys.path 中。 - for prefix in {sys.prefix, sys.exec_prefix, sys.base_prefix, sys.base_exec_prefix}: - if prefix: - runtime_paths.add(os.path.normpath(os.path.join(prefix, "DLLs"))) - - allowed = set() - for p in sys.path: - norm = os.path.normpath(p) - # 保留标准库和 site-packages - if any(norm.startswith(runtime_path) for runtime_path in runtime_paths): - allowed.add(p) - # 保留 site-packages（第三方库 + SDK） - if "site-packages" in norm or "dist-packages" in norm: - allowed.add(p) - - # 添加插件目录 - plugin_dir_paths = [os.path.normpath(d) for d in plugin_dirs] - for d in plugin_dir_paths: - allowed.add(d) - - preserved_paths = [p for p in sys.path if p in allowed] - for extra_path in plugin_dir_paths: - if extra_path not in preserved_paths: - preserved_paths.append(extra_path) - sys.path[:] = preserved_paths - - # 仅为旧版插件兼容层保留极小的 src.* 可见面： - # - src.plugin_system.*: 通过 maibot_sdk.compat 导入钩子重定向 - # - src.common.logger: 仓库内仍有少量旧插件沿用该日志入口 - allowed_src_exact_modules = frozenset( - { - "src", - "src.common", - "src.common.logger", - "src.common.logger_color_and_mapping", - } - ) - allowed_src_prefixes = ("src.plugin_system",) - plugin_module_prefix = "_maibot_plugin_" - - def _is_allowed_src_module(fullname: str) -> bool: - """判断给定 src.* 模块是否在 Runner 允许列表中。""" - if fullname in allowed_src_exact_modules: - return True - return any(fullname == prefix or fullname.startswith(f"{prefix}.") for prefix in allowed_src_prefixes) - - def _resolve_requester_name(import_globals: Any = None) -> str: - """解析当前导入请求的发起模块名。""" - if isinstance(import_globals, dict): - for key in ("__name__", "__package__"): - value = import_globals.get(key) - if isinstance(value, str) and value: - return value - - frame = inspect.currentframe() - try: - current = frame.f_back if frame is not None else None - while current is not None: - module_name = current.f_globals.get("__name__", "") - if not isinstance(module_name, str) or not module_name: - current = current.f_back - continue - if module_name == __name__ or module_name.startswith("importlib"): - current = current.f_back - continue - return module_name - return "" - finally: - del frame - - def _is_plugin_import_request(import_globals: Any = None) -> bool: - """判断当前导入是否由插件模块直接发起。""" - requester_name = _resolve_requester_name(import_globals) - return requester_name.startswith(plugin_module_prefix) - - def _format_block_message(fullname: str) -> str: - """构造统一的拒绝导入错误信息。""" - return ( - f"Runner 子进程不允许导入主程序模块: {fullname}。" - "请改用 maibot_sdk 或 src.plugin_system 兼容层提供的接口。" - ) - - def _iter_requested_src_modules(name: str, fromlist: Any) -> List[str]: - """展开本次导入请求涉及的 src.* 模块名。""" - requested_modules = [name] - if not name.startswith("src") or not fromlist: - return requested_modules - - for item in fromlist: - if not isinstance(item, str) or not item or item == "*": - continue - requested_modules.append(f"{name}.{item}") - return requested_modules - - def _assert_plugin_import_allowed(name: str, import_globals: Any = None, fromlist: Any = ()) -> None: - """在插件发起导入时校验目标 src.* 模块是否允许访问。""" - if not _is_plugin_import_request(import_globals): - return - - for requested_module in _iter_requested_src_modules(name, fromlist): - if not requested_module.startswith("src"): - continue - if _is_allowed_src_module(requested_module): - continue - raise ImportError(_format_block_message(requested_module)) - - def _detach_module_from_parent(fullname: str, module: ModuleType) -> None: - """从父模块上移除已清理模块的属性引用。""" - parent_name, _, child_name = fullname.rpartition(".") - if not parent_name or not child_name: - return - - parent_module = sys.modules.get(parent_name) - if parent_module is None: - return - if getattr(parent_module, child_name, None) is module: - with contextlib.suppress(AttributeError): - delattr(parent_module, child_name) - - # 仅清理已加载的叶子模块，保留包对象给 Runner 自己的延迟导入和相对导入使用。 - existing_src_modules = sorted( - ( - (module_name, module) - for module_name, module in list(sys.modules.items()) - if module_name == "src" or module_name.startswith("src.") - ), - key=lambda item: item[0].count("."), - reverse=True, - ) - for module_name, module in existing_src_modules: - if _is_allowed_src_module(module_name) or hasattr(module, "__path__"): - continue - _detach_module_from_parent(module_name, module) - sys.modules.pop(module_name, None) - - # ``import`` 语句与 ``importlib.import_module`` 走的是不同入口，因此两边都需要兜底。 - builtins_module = cast(Any, builtins) - original_import = getattr(builtins_module, "__maibot_runner_original_import__", builtins.__import__) - builtins_module.__maibot_runner_original_import__ = original_import - - def _guarded_import(name: str, globals: Any = None, locals: Any = None, fromlist: Any = (), level: int = 0) -> Any: - if level == 0: - _assert_plugin_import_allowed(name, import_globals=globals, fromlist=fromlist) - return original_import(name, globals, locals, fromlist, level) - - cast(Any, _guarded_import).__maibot_runner_plugin_import_guard__ = True - builtins.__import__ = _guarded_import - - importlib_module = cast(Any, importlib) - original_import_module = getattr(importlib_module, "__maibot_runner_original_import_module__", importlib.import_module) - importlib_module.__maibot_runner_original_import_module__ = original_import_module - - def _guarded_import_module(name: str, package: Optional[str] = None) -> Any: - resolved_name = importlib_util.resolve_name(name, package) if name.startswith(".") else name - _assert_plugin_import_allowed(resolved_name) - return original_import_module(name, package) - - cast(Any, _guarded_import_module).__maibot_runner_plugin_import_guard__ = True - importlib.import_module = _guarded_import_module - - # ─── 进程入口 ────────────────────────────────────────────── @@ -1392,9 +1207,6 @@ async def _async_main() -> None: logger.warning("外部依赖插件版本映射格式非法，已回退为空映射") external_plugin_ids = {} - # sys.path 隔离: 只保留标准库、SDK 包、插件目录 - _isolate_sys_path(plugin_dirs) - runner = PluginRunner( host_address, session_token,