diff --git a/backend/src/module/llm_plugins/installer.py b/backend/src/module/llm_plugins/installer.py index a60ec0dd..409ce202 100644 --- a/backend/src/module/llm_plugins/installer.py +++ b/backend/src/module/llm_plugins/installer.py @@ -24,6 +24,7 @@ _MANIFEST_SCHEMA = 1 class PluginInstaller(SignedCatalogInstaller): tag = "llm-plugins" catalog_schema = _MANIFEST_SCHEMA + manifest_name = "plugin.json" def _reject_reason(self, plugin_id: str) -> Optional[str]: from module.parser.analyser.providers.builtin import BUILTIN diff --git a/backend/src/module/plugin/installer.py b/backend/src/module/plugin/installer.py index 4b165aad..4e23437b 100644 --- a/backend/src/module/plugin/installer.py +++ b/backend/src/module/plugin/installer.py @@ -77,6 +77,8 @@ def safe_extract(zip_path: Path, dest: Path) -> None: class SignedCatalogInstaller: tag: ClassVar[str] catalog_schema: ClassVar[int] + # 版本目录里的清单文件名,卸载时据此确认目录归本安装器所有 + manifest_name: ClassVar[str] def __init__( self, @@ -169,7 +171,7 @@ class SignedCatalogInstaller: return InstallResult(success=False, message=str(e)) async def _install(self, plugin_id: str) -> InstallResult: - if reason := self._reject_reason(plugin_id): + if reason := self._check_id(plugin_id): return InstallResult(success=False, message=reason) catalog = await self.fetch_catalog() entry = next((p for p in catalog if p.get("id") == plugin_id), None) @@ -221,7 +223,7 @@ class SignedCatalogInstaller: async def uninstall(self, plugin_id: str) -> InstallResult: async with self._lock: - if reason := self._reject_reason(plugin_id): + if reason := self._check_id(plugin_id): return InstallResult(success=False, message=reason) if reason := self._uninstall_reject_reason(plugin_id): return InstallResult(success=False, message=reason) @@ -230,8 +232,22 @@ class SignedCatalogInstaller: self._reload(plugin_id) return InstallResult(success=True) + def _check_id(self, plugin_id: str) -> Optional[str]: + """id 不合法或不允许经安装器安装 / 卸载的原因;可以时为 None。""" + # 同时挡住 "local"(本地插件目录)与路径穿越:id 会拼进文件系统路径 + if not ID_RE.match(plugin_id) or plugin_id in RESERVED_IDS: + return f"Invalid plugin id: {plugin_id}" + return self._reject_reason(plugin_id) + def _uninstall_reject_reason(self, plugin_id: str) -> Optional[str]: - """卸载前的额外检查;默认没有。""" + # 只删经本安装器装入的目录(installed.json 指向含本类清单的版本目录), + # 绝不碰 config/plugins/ 下的其它内容:本地插件、另一类安装器装入的插件 + version = installed_version(self.root, plugin_id) + if ( + not version + or not (self.root / plugin_id / version / self.manifest_name).is_file() + ): + return f"Plugin not installed: {plugin_id}" return None async def _after_uninstall(self, plugin_id: str) -> None: @@ -248,26 +264,13 @@ class PluginInstaller(SignedCatalogInstaller): tag = "plugins" catalog_schema = 2 + manifest_name = MANIFEST_NAME def _reject_reason(self, plugin_id: str) -> Optional[str]: - # 同时挡住 "local"(本地插件目录)与路径穿越:id 会拼进文件系统路径 - if not ID_RE.match(plugin_id) or plugin_id in RESERVED_IDS: - return f"Invalid plugin id: {plugin_id}" if (BUILTIN_ROOT / plugin_id).is_dir(): return f"{plugin_id} is a builtin plugin" return None - def _uninstall_reject_reason(self, plugin_id: str) -> Optional[str]: - # 只删经本安装器装入的目录(installed.json 指向含 plugin.toml 的版本目录), - # 绝不碰 config/plugins/ 下的其它内容,包括同样带 installed.json 的 LLM 插件 - version = installed_version(self.root, plugin_id) - if ( - not version - or not (self.root / plugin_id / version / MANIFEST_NAME).is_file() - ): - return f"Plugin not installed: {plugin_id}" - return None - def _validate_manifest( self, unpacked: Path, entry: dict[str, Any] ) -> Optional[str]: diff --git a/backend/src/test/test_llm_plugin_installer.py b/backend/src/test/test_llm_plugin_installer.py index 898cbfef..e03a5c1f 100644 --- a/backend/src/test/test_llm_plugin_installer.py +++ b/backend/src/test/test_llm_plugin_installer.py @@ -206,6 +206,29 @@ class TestInstallRejections: assert result.success is False +class TestUninstallScope: + @pytest.mark.parametrize("plugin_id", ["..", "local", "generic-plugin"]) + async def test_uninstall_foreign_or_unsafe_id_deletes_nothing( + self, tmp_path, keypair, plugin_id + ): + installer = _installer(tmp_path, keypair, _make_plugin_zip()) + root = tmp_path / "plugins" + # 本地插件源码、经通用安装器装入的插件,以及插件根之外的配置 + (root / "local" / "mine").mkdir(parents=True) + generic = root / "generic-plugin" + (generic / "0.1.0").mkdir(parents=True) + (generic / "0.1.0" / "plugin.toml").write_text("") + (generic / "installed.json").write_text('{"version": "0.1.0"}') + (tmp_path / "data.db").write_text("") + + result = await installer.uninstall(plugin_id) + + assert result.success is False + assert (root / "local" / "mine").is_dir() + assert (generic / "installed.json").is_file() + assert (tmp_path / "data.db").is_file() + + class TestUninstallClearsCredential: async def test_uninstall_deletes_credential_row(self, tmp_path, keypair): from unittest.mock import AsyncMock, patch