fix(4.0): 安装器统一校验插件 id,卸载只删除本类安装器装入的目录

LLM 安装器此前不校验 id,DELETE /llm/providers/.. 会删除整个 config 目录,
local 会删除本地插件源码,通用插件的 id 会删除正在运行的通用插件。
id 校验与卸载归属检查移到 SignedCatalogInstaller,两类安装器共用。
This commit is contained in:
Estrella Pan
2026-10-08 16:04:47 +02:00
parent ee5704eef0
commit 83527c342b
3 changed files with 44 additions and 17 deletions

View File

@@ -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

View File

@@ -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]:

View File

@@ -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