扬了 sys 控制

Author: DrSmoothl

pytests/test_plugin_runtime.py

@@ -1298,142 +1298,12 @@ def test_loader_requires_sdk_plugin_to_override_on_unload(self, tmp_path):
         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 @@ async def run(self) -> None:
         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 @@ async def run(self) -> None:
         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 测试 ──────────────────────

src/plugin_runtime/runner/runner_main.py

@@ -1183,191 +1183,6 @@ def request_capability(self) -> RPCClient:
         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 的部分平台扩展模块和依赖会通过 <prefix>/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,