feat(4.0-P7): 签名目录来源：LLM 安装器泛化为通用插件安装器

- SignedCatalogInstaller 提供目录 / 插件包验签、sha256、防 zip-slip 与落盘；PluginInstaller 从 plugins tag 安装 plugin.toml 插件，LLM 安装器改为其子类，行为不变
- 加载器新增 catalog 来源（builtin > catalog > local > pip），视为已签名
- 新增 GET /plugins/catalog、POST /plugins/{id}/install、DELETE /plugins/{id}；安装前拒绝保留 id、非法 id 与内置同名，卸载只删除安装器装入的目录
- scripts/build_plugin_catalog.py 生成并签名 catalog.json

backend/src/module/api/plugins.py

@@ -7,9 +7,11 @@ from fastapi.responses import FileResponse
 from pydantic import BaseModel, ValidationError
 
 from ab_sdk import points
-from module.conf import settings
+from module.conf import VERSION, settings
 from module.core import AppContext
 from module.plugin.host import plugin_provider_ids
+from module.plugin.installer import PluginInstaller
+from module.plugin.loader import CATALOG_ROOT, installed_version
 from module.plugin.manifest import UiSlot
 from module.plugin.secrets import mask_options, restore_options
 from module.security.api import get_current_user
@@ -51,6 +53,19 @@ class PluginsSettingsUpdate(BaseModel):
     allow_unsigned: bool
 
 
+class CatalogEntry(BaseModel):
+    """签名目录里的一个插件，以及本机已安装的版本。"""
+
+    id: str
+    name: str = ""
+    version: str
+    kind: str = "plugin"
+    extension_points: list[str] = []
+    description: str = ""
+    min_ab_version: str = "0.0.0"
+    installed_version: str | None
+
+
 class PluginUiSlot(BaseModel):
     plugin_id: str
     slot: UiSlot
@@ -93,6 +108,24 @@ async def list_plugins(ctx: AppContext = Depends(get_context)):
     return _overview(ctx)
 
 
+@router.get("/catalog", response_model=list[CatalogEntry])
+async def get_catalog():
+    """签名目录（GitHub release ``plugins``）里可安装的插件。"""
+    try:
+        entries = await PluginInstaller(app_version=VERSION).fetch_catalog()
+    except Exception as e:  # noqa: BLE001 - 网络或验签失败统一报 502
+        logger.warning("Plugin catalog unavailable: %s", e)
+        raise HTTPException(
+            status_code=502, detail=f"Plugin catalog unavailable: {e}"
+        ) from None
+    return [
+        CatalogEntry(
+            **entry, installed_version=installed_version(CATALOG_ROOT, entry["id"])
+        )
+        for entry in entries
+    ]
+
+
 @router.get("/ui", response_model=list[PluginUiSlot])
 async def list_plugin_ui(ctx: AppContext = Depends(get_context)):
     """已启用插件声明的前端挂载点（清单中的 ``[[plugin.ui]]``）。"""
@@ -163,6 +196,29 @@ async def update_plugin(
     return _overview(ctx)
 
 
+@router.post("/{plugin_id}/install", response_model=PluginsOverview)
+async def install_plugin(plugin_id: str, ctx: AppContext = Depends(get_context)):
+    """从签名目录安装（或升级）插件并启用：用户点安装即同意其运行。"""
+    result = await PluginInstaller(app_version=VERSION).install(plugin_id)
+    if not result.success:
+        raise HTTPException(status_code=400, detail=result.message)
+    settings.plugins.enabled[plugin_id] = True
+    await asyncio.to_thread(settings.save)
+    # 升级时配置快照没有变化，apply_settings 不会重新加载，所以按 id 重载
+    await ctx.plugins.reload(plugin_id)
+    return _overview(ctx)
+
+
+@router.delete("/{plugin_id}", response_model=PluginsOverview)
+async def uninstall_plugin(plugin_id: str, ctx: AppContext = Depends(get_context)):
+    """卸载经签名目录安装的插件；内置与本地插件不受影响。"""
+    result = await PluginInstaller(app_version=VERSION).uninstall(plugin_id)
+    if not result.success:
+        raise HTTPException(status_code=400, detail=result.message)
+    await ctx.plugins.reload(plugin_id)
+    return _overview(ctx)
+
+
 @router.get("/providers", response_model=dict[str, list[str]])
 async def list_plugin_providers():
     """插件提供的 Provider id（按扩展点），设置页把它们并入下拉候选。"""

backend/src/module/llm_plugins/installer.py

@@ -1,221 +1,56 @@
-"""LLM 提供商插件安装管线（克隆 update.updater 的下载/验签/解包模式）。
+"""LLM 提供商插件安装管线（下载 / 验签 / 解包由 ``module.plugin.installer`` 提供）。
 
-安全边界与更新系统一致：
-1. catalog.json 本身也验签（防索引篡改/降级）；
-2. 每个插件 zip 校验 sha256 + ed25519 签名，拒绝未签名/坏签名；
-3. _safe_extract 防 zip-slip；
-4. manifest schema/id/entry/min_ab_version 逐项校验。
-插件依赖仅限 stdlib + httpx + pydantic（应用已有），无 pip 安装。
+LLM 插件仍使用 ``plugin.json`` 清单，发布在 ``llm-plugins`` tag；通用插件走
+``module.plugin.installer.PluginInstaller``。
 """
 
-import asyncio
-import hashlib
 import json
-import logging
-import shutil
-from dataclasses import dataclass
 from pathlib import Path
-from typing import Optional
-
-import httpx
-import semver
+from typing import Any, Optional
 
 from module.database import Database
-from module.update.signing import DEFAULT_PUBKEY_PATH, verify_bundle_signature
-
-logger = logging.getLogger(__name__)
-
-# 插件目录与 catalog 托管位置。
-GITHUB_OWNER = "EstrellaXD"
-GITHUB_REPO = "Auto_Bangumi"
-_CATALOG_TAG = "llm-plugins"
-_RELEASE_BASE = (
-    f"https://github.com/{GITHUB_OWNER}/{GITHUB_REPO}/releases/download/{_CATALOG_TAG}"
+from module.plugin.installer import (
+    DEFAULT_PLUGINS_ROOT,
+    InstallResult,
+    SignedCatalogInstaller,
+    meets_min_version,
 )
-_DL_HEADERS = {"User-Agent": "AutoBangumi-Plugins"}
-DEFAULT_PLUGINS_ROOT = Path("config") / "plugins"
+
+__all__ = ["DEFAULT_PLUGINS_ROOT", "InstallResult", "PluginInstaller"]
 
 _MANIFEST_SCHEMA = 1
 
 
-def _meets_min_version(app_version: str, min_ab: str) -> bool:
-    """app_version 是否满足 min_ab。非 semver 的开发版（DEV_VERSION）放行。"""
-    try:
-        return semver.VersionInfo.parse(app_version).compare(min_ab) >= 0
-    except ValueError:
-        return True
+class PluginInstaller(SignedCatalogInstaller):
+    tag = "llm-plugins"
+    catalog_schema = _MANIFEST_SCHEMA
 
-
-@dataclass
-class InstallResult:
-    success: bool
-    version: str = ""
-    message: str = ""
-
-
-def _safe_extract(zip_path: Path, dest: Path) -> None:
-    """解压 zip 到 dest，拒绝绝对路径/`..` 越界成员（防 zip-slip）。"""
-    import zipfile
-
-    if dest.exists():
-        shutil.rmtree(dest)
-    dest.mkdir(parents=True, exist_ok=True)
-    dest_resolved = dest.resolve()
-    with zipfile.ZipFile(zip_path) as zf:
-        for member in zf.namelist():
-            target = (dest / member).resolve()
-            if not target.is_relative_to(dest_resolved):
-                raise ValueError(f"Unsafe path in plugin: {member}")
-        zf.extractall(dest)
-
-
-def _sha256_bytes(data: bytes) -> str:
-    return hashlib.sha256(data).hexdigest()
-
-
-class PluginInstaller:
-    def __init__(
-        self,
-        *,
-        root: Optional[Path] = None,
-        pubkey_path: Optional[Path] = None,
-        client: Optional[httpx.AsyncClient] = None,
-        app_version: str = "0.0.0",
-    ) -> None:
-        self.root = root if root is not None else DEFAULT_PLUGINS_ROOT
-        self.pubkey_path = pubkey_path or DEFAULT_PUBKEY_PATH
-        self._client = client
-        self.app_version = app_version
-        self._lock = asyncio.Lock()
-
-    async def _get_client(self) -> httpx.AsyncClient:
-        if self._client is not None:
-            return self._client
-        from module.network.request_url import get_shared_client
-
-        return await get_shared_client()
-
-    async def _download_bytes(self, url: str) -> bytes:
-        client = await self._get_client()
-        resp = await client.get(url, headers=_DL_HEADERS)
-        resp.raise_for_status()
-        return resp.content
-
-    async def _download_text(self, url: str) -> str:
-        client = await self._get_client()
-        resp = await client.get(url, headers=_DL_HEADERS)
-        resp.raise_for_status()
-        return resp.text
-
-    async def fetch_catalog(self) -> list[dict]:
-        """拉取并验签 catalog.json，返回插件条目列表。"""
-        catalog_bytes = await self._download_bytes(f"{_RELEASE_BASE}/catalog.json")
-        sig = await self._download_text(f"{_RELEASE_BASE}/catalog.json.sig")
-        tmp = self.root / ".catalog.json"
-        self.root.mkdir(parents=True, exist_ok=True)
-        tmp.write_bytes(catalog_bytes)
-        try:
-            if not verify_bundle_signature(tmp, sig, self.pubkey_path):
-                raise ValueError("catalog signature verification failed")
-        finally:
-            tmp.unlink(missing_ok=True)
-        catalog = json.loads(catalog_bytes)
-        if not isinstance(catalog, dict) or catalog.get("schema") != _MANIFEST_SCHEMA:
-            raise ValueError("Unsupported catalog schema")
-        plugins = catalog.get("plugins", [])
-        return plugins if isinstance(plugins, list) else []
-
-    def _is_builtin(self, plugin_id: str) -> bool:
+    def _reject_reason(self, plugin_id: str) -> Optional[str]:
         from module.parser.analyser.providers.builtin import BUILTIN
         from module.parser.analyser.providers.presets import PRESET_ADAPTERS
 
-        return plugin_id in BUILTIN or plugin_id in PRESET_ADAPTERS
+        if plugin_id in BUILTIN or plugin_id in PRESET_ADAPTERS:
+            return f"{plugin_id} is a builtin provider"
+        return None
 
-    async def install(self, plugin_id: str) -> InstallResult:
-        async with self._lock:
-            try:
-                return await self._install(plugin_id)
-            except Exception as e:  # noqa: BLE001 - 统一转成失败结果
-                logger.warning("Plugin install failed for %s: %s", plugin_id, e)
-                return InstallResult(success=False, message=str(e))
-
-    async def _install(self, plugin_id: str) -> InstallResult:
-        if self._is_builtin(plugin_id):
-            return InstallResult(
-                success=False, message=f"{plugin_id} is a builtin provider"
-            )
-        catalog = await self.fetch_catalog()
-        entry = next((p for p in catalog if p.get("id") == plugin_id), None)
-        if entry is None:
-            return InstallResult(
-                success=False, message=f"Plugin not found in catalog: {plugin_id}"
-            )
-        version = entry.get("version", "")
-        min_ab = entry.get("min_ab_version", "0.0.0")
-        if not _meets_min_version(self.app_version, min_ab):
-            return InstallResult(
-                success=False,
-                message=f"Plugin requires AutoBangumi >= {min_ab}",
-            )
-
-        asset = entry.get("asset") or f"{plugin_id}-{version}.zip"
-        zip_bytes = await self._download_bytes(f"{_RELEASE_BASE}/{asset}")
-        if _sha256_bytes(zip_bytes) != entry.get("sha256"):
-            return InstallResult(
-                success=False, message="Plugin sha256 checksum mismatch"
-            )
-
-        staging = self.root / ".staging" / plugin_id
-        staging.parent.mkdir(parents=True, exist_ok=True)
-        zip_path = self.root / ".staging" / f"{plugin_id}.zip"
-        zip_path.write_bytes(zip_bytes)
-        sig = await self._download_text(f"{_RELEASE_BASE}/{asset}.sig")
-        if not verify_bundle_signature(zip_path, sig, self.pubkey_path):
-            zip_path.unlink(missing_ok=True)
-            return InstallResult(success=False, message="Plugin signature invalid")
-
-        _safe_extract(zip_path, staging)
-        zip_path.unlink(missing_ok=True)
-
-        manifest = self._read_manifest(staging)
-        problem = self._validate_manifest(manifest, plugin_id)
-        if problem:
-            shutil.rmtree(staging, ignore_errors=True)
-            return InstallResult(success=False, message=problem)
-
-        # promote：config/plugins/<id>/<version>/ + installed.json
-        target = self.root / plugin_id / version
-        if target.exists():
-            shutil.rmtree(target)
-        target.parent.mkdir(parents=True, exist_ok=True)
-        shutil.move(str(staging), str(target))
-        (self.root / plugin_id / "installed.json").write_text(
-            json.dumps({"version": version}), encoding="utf-8"
-        )
-
-        self._reload_registry(plugin_id)
-        return InstallResult(success=True, version=version)
-
-    def _read_manifest(self, unpacked: Path) -> dict:
-        data = json.loads((unpacked / "plugin.json").read_text(encoding="utf-8"))
-        if not isinstance(data, dict):
+    def _validate_manifest(
+        self, unpacked: Path, entry: dict[str, Any]
+    ) -> Optional[str]:
+        manifest = json.loads((unpacked / "plugin.json").read_text(encoding="utf-8"))
+        if not isinstance(manifest, dict):
             raise ValueError("plugin.json is not an object")
-        return data
-
-    def _validate_manifest(self, manifest: dict, plugin_id: str) -> Optional[str]:
         if manifest.get("schema") != _MANIFEST_SCHEMA:
             return "Unsupported plugin manifest schema"
-        if manifest.get("id") != plugin_id:
+        if manifest.get("id") != entry.get("id"):
             return "Plugin id mismatch between catalog and manifest"
-        entry = manifest.get("entry", "")
-        if ":" not in entry:
+        if ":" not in manifest.get("entry", ""):
             return "Invalid plugin entry"
         min_ab = manifest.get("min_ab_version", "0.0.0")
-        if not _meets_min_version(self.app_version, min_ab):
+        if not meets_min_version(self.app_version, min_ab):
             return f"Plugin requires AutoBangumi >= {min_ab}"
         return None
 
-    def _reload_registry(self, plugin_id: str) -> None:
+    def _reload(self, plugin_id: str) -> None:
         from module.parser import title_parser
         from module.parser.analyser.providers.registry import registry
 
@@ -225,16 +60,6 @@ class PluginInstaller:
         registry.invalidate(plugin_id)
         title_parser.reset_cache()
 
-    async def uninstall(self, plugin_id: str) -> InstallResult:
-        async with self._lock:
-            if self._is_builtin(plugin_id):
-                return InstallResult(
-                    success=False, message=f"{plugin_id} is a builtin provider"
-                )
-            plugin_dir = self.root / plugin_id
-            if plugin_dir.exists():
-                shutil.rmtree(plugin_dir, ignore_errors=True)
-            async with Database() as db:
-                await db.llm_credential.delete(plugin_id)
-            self._reload_registry(plugin_id)
-            return InstallResult(success=True)
+    async def _after_uninstall(self, plugin_id: str) -> None:
+        async with Database() as db:
+            await db.llm_credential.delete(plugin_id)

backend/src/module/plugin/installer.py

@@ -0,0 +1,276 @@
+"""签名目录来源的安装管线（设计文档第 2.4 节），LLM 插件与通用插件共用。
+
+安全边界与更新系统一致：
+
+1. ``catalog.json`` 本身也验签（防索引篡改 / 降级）；
+2. 每个插件 zip 校验 sha256 + ed25519 签名，拒绝未签名 / 坏签名；
+3. ``_safe_extract`` 防 zip-slip；
+4. 清单逐项校验后才落到 ``<root>/<id>/<version>/``，并写入版本指针
+   ``installed.json``。
+
+插件依赖限于 stdlib 与宿主已有的包，不做 pip 安装。
+
+``SignedCatalogInstaller`` 是下载 / 验签 / 解包 / 落盘的共用部分，子类决定发布
+tag、清单格式与安装后的刷新：``PluginInstaller`` 安装通用插件（``plugins`` tag，
+``plugin.toml`` 清单），LLM 提供商插件的安装器在 ``module.llm_plugins``。
+"""
+
+import asyncio
+import hashlib
+import json
+import logging
+import shutil
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, ClassVar, Optional
+
+import httpx
+import semver
+
+from ab_sdk.manifest import ID_RE, RESERVED_IDS, check
+from module.plugin.loader import BUILTIN_ROOT, INSTALLED_FILE
+from module.update.signing import DEFAULT_PUBKEY_PATH, verify_bundle_signature
+
+logger = logging.getLogger(__name__)
+
+# 插件目录与 catalog 托管位置。
+GITHUB_OWNER = "EstrellaXD"
+GITHUB_REPO = "Auto_Bangumi"
+_DL_HEADERS = {"User-Agent": "AutoBangumi-Plugins"}
+DEFAULT_PLUGINS_ROOT = Path("config") / "plugins"
+
+
+def release_base(tag: str) -> str:
+    return f"https://github.com/{GITHUB_OWNER}/{GITHUB_REPO}/releases/download/{tag}"
+
+
+def meets_min_version(app_version: str, min_ab: str) -> bool:
+    """app_version 是否满足 min_ab。非 semver 的开发版（DEV_VERSION）放行。"""
+    try:
+        return semver.VersionInfo.parse(app_version).compare(min_ab) >= 0
+    except ValueError:
+        return True
+
+
+@dataclass
+class InstallResult:
+    success: bool
+    version: str = ""
+    message: str = ""
+
+
+def safe_extract(zip_path: Path, dest: Path) -> None:
+    """解压 zip 到 dest，拒绝绝对路径/`..` 越界成员（防 zip-slip）。"""
+    if dest.exists():
+        shutil.rmtree(dest)
+    dest.mkdir(parents=True, exist_ok=True)
+    dest_resolved = dest.resolve()
+    with zipfile.ZipFile(zip_path) as zf:
+        for member in zf.namelist():
+            target = (dest / member).resolve()
+            if not target.is_relative_to(dest_resolved):
+                raise ValueError(f"Unsafe path in plugin: {member}")
+        zf.extractall(dest)
+
+
+class SignedCatalogInstaller:
+    tag: ClassVar[str]
+    catalog_schema: ClassVar[int]
+
+    def __init__(
+        self,
+        *,
+        root: Optional[Path] = None,
+        pubkey_path: Optional[Path] = None,
+        client: Optional[httpx.AsyncClient] = None,
+        app_version: str = "0.0.0",
+    ) -> None:
+        self.root = root if root is not None else DEFAULT_PLUGINS_ROOT
+        self.pubkey_path = pubkey_path or DEFAULT_PUBKEY_PATH
+        self._client = client
+        self.app_version = app_version
+        self._lock = asyncio.Lock()
+
+    # ------------------------------------------------------------ 子类实现
+
+    def _reject_reason(self, plugin_id: str) -> Optional[str]:
+        """该 id 不允许经安装器安装 / 卸载的原因（如与内置同名）。"""
+        raise NotImplementedError
+
+    def _validate_manifest(
+        self, unpacked: Path, entry: dict[str, Any]
+    ) -> Optional[str]:
+        """解包目录里的清单是否与 catalog 条目一致且可用；返回问题描述，通过时
+        为 None。"""
+        raise NotImplementedError
+
+    def _check_entry(self, entry: dict[str, Any]) -> Optional[str]:
+        """下载前对 catalog 条目的检查，默认只看宿主版本要求。"""
+        min_ab = entry.get("min_ab_version", "0.0.0")
+        if not meets_min_version(self.app_version, min_ab):
+            return f"Plugin requires AutoBangumi >= {min_ab}"
+        return None
+
+    def _reload(self, plugin_id: str) -> None:
+        """安装 / 卸载后让宿主看到变化（刷新缓存等）；默认没有。"""
+
+    # ------------------------------------------------------------ 下载
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is not None:
+            return self._client
+        from module.network.request_url import get_shared_client
+
+        return await get_shared_client()
+
+    async def _download_bytes(self, url: str) -> bytes:
+        client = await self._get_client()
+        resp = await client.get(url, headers=_DL_HEADERS)
+        resp.raise_for_status()
+        return resp.content
+
+    async def _download_text(self, url: str) -> str:
+        client = await self._get_client()
+        resp = await client.get(url, headers=_DL_HEADERS)
+        resp.raise_for_status()
+        return resp.text
+
+    async def fetch_catalog(self) -> list[dict]:
+        """拉取并验签 catalog.json，返回插件条目列表。"""
+        base = release_base(self.tag)
+        catalog_bytes = await self._download_bytes(f"{base}/catalog.json")
+        sig = await self._download_text(f"{base}/catalog.json.sig")
+        tmp = self.root / ".catalog.json"
+        self.root.mkdir(parents=True, exist_ok=True)
+        tmp.write_bytes(catalog_bytes)
+        try:
+            if not verify_bundle_signature(tmp, sig, self.pubkey_path):
+                raise ValueError("catalog signature verification failed")
+        finally:
+            tmp.unlink(missing_ok=True)
+        catalog = json.loads(catalog_bytes)
+        if (
+            not isinstance(catalog, dict)
+            or catalog.get("schema") != self.catalog_schema
+        ):
+            raise ValueError("Unsupported catalog schema")
+        plugins = catalog.get("plugins", [])
+        return plugins if isinstance(plugins, list) else []
+
+    # ------------------------------------------------------------ 安装
+
+    async def install(self, plugin_id: str) -> InstallResult:
+        async with self._lock:
+            try:
+                return await self._install(plugin_id)
+            except Exception as e:  # noqa: BLE001 - 统一转成失败结果
+                logger.warning("Plugin install failed for %s: %s", plugin_id, e)
+                return InstallResult(success=False, message=str(e))
+
+    async def _install(self, plugin_id: str) -> InstallResult:
+        if reason := self._reject_reason(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)
+        if entry is None:
+            return InstallResult(
+                success=False, message=f"Plugin not found in catalog: {plugin_id}"
+            )
+        if problem := self._check_entry(entry):
+            return InstallResult(success=False, message=problem)
+        version = entry.get("version", "")
+
+        base = release_base(self.tag)
+        asset = entry.get("asset") or f"{plugin_id}-{version}.zip"
+        zip_bytes = await self._download_bytes(f"{base}/{asset}")
+        if hashlib.sha256(zip_bytes).hexdigest() != entry.get("sha256"):
+            return InstallResult(
+                success=False, message="Plugin sha256 checksum mismatch"
+            )
+
+        staging = self.root / ".staging" / plugin_id
+        staging.parent.mkdir(parents=True, exist_ok=True)
+        zip_path = self.root / ".staging" / f"{plugin_id}.zip"
+        zip_path.write_bytes(zip_bytes)
+        sig = await self._download_text(f"{base}/{asset}.sig")
+        if not verify_bundle_signature(zip_path, sig, self.pubkey_path):
+            zip_path.unlink(missing_ok=True)
+            return InstallResult(success=False, message="Plugin signature invalid")
+
+        safe_extract(zip_path, staging)
+        zip_path.unlink(missing_ok=True)
+
+        problem = self._validate_manifest(staging, entry)
+        if problem:
+            shutil.rmtree(staging, ignore_errors=True)
+            return InstallResult(success=False, message=problem)
+
+        # promote：<root>/<id>/<version>/ + installed.json
+        target = self.root / plugin_id / version
+        if target.exists():
+            shutil.rmtree(target)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.move(str(staging), str(target))
+        (self.root / plugin_id / INSTALLED_FILE).write_text(
+            json.dumps({"version": version}), encoding="utf-8"
+        )
+
+        self._reload(plugin_id)
+        return InstallResult(success=True, version=version)
+
+    async def uninstall(self, plugin_id: str) -> InstallResult:
+        async with self._lock:
+            if reason := self._reject_reason(plugin_id):
+                return InstallResult(success=False, message=reason)
+            if reason := self._uninstall_reject_reason(plugin_id):
+                return InstallResult(success=False, message=reason)
+            shutil.rmtree(self.root / plugin_id, ignore_errors=True)
+            await self._after_uninstall(plugin_id)
+            self._reload(plugin_id)
+            return InstallResult(success=True)
+
+    def _uninstall_reject_reason(self, plugin_id: str) -> Optional[str]:
+        """卸载前的额外检查；默认没有。"""
+        return None
+
+    async def _after_uninstall(self, plugin_id: str) -> None:
+        """卸载后清理插件的其它数据（如凭据）；默认没有。"""
+
+
+class PluginInstaller(SignedCatalogInstaller):
+    """通用插件（``plugin.toml``）：从 ``plugins`` tag 安装到 ``config/plugins/<id>/``。
+
+    ``catalog.json`` 条目：``id``、``name``、``version``、``kind``、
+    ``extension_points``、``sdk``、``min_ab_version``、``asset``、``sha256``、
+    ``description``。zip 的内容在 zip 根（``ab-plugin pack`` 的产物）。
+    """
+
+    tag = "plugins"
+    catalog_schema = 2
+
+    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]:
+        # 只删经安装器装入的目录，绝不碰 config/plugins/ 下的其它内容
+        if not (self.root / plugin_id / INSTALLED_FILE).is_file():
+            return f"Plugin not installed: {plugin_id}"
+        return None
+
+    def _validate_manifest(
+        self, unpacked: Path, entry: dict[str, Any]
+    ) -> Optional[str]:
+        parsed, problems = check(unpacked)
+        if parsed is None or problems:
+            return "; ".join(problems)
+        if parsed.id != entry.get("id"):
+            return "Plugin id mismatch between catalog and manifest"
+        if parsed.version != entry.get("version"):
+            return "Plugin version mismatch between catalog and manifest"
+        return None

backend/src/module/plugin/loader.py

@@ -3,18 +3,20 @@
 来源（优先级从高到低，同 id 只取最高优先级来源）：
 
 - ``builtin``：``module/plugins/builtin/<id>/``，随镜像发布
+- ``catalog``：经安装器从签名目录装入的 ``config/plugins/<id>/<version>/``，
+  版本由 ``installed.json`` 指定
 - ``local``：``config/plugins/local/<id>/``，用户二次开发，未签名
 - ``pip``：``autobangumi.plugins`` entry point，未签名
 
 目录型插件以 ``ab_plugin_<id>`` 为包名加载，插件内部可以使用相对导入；
 ``vendor/`` 下的纯 Python 依赖追加到 ``sys.path`` 末尾（宿主依赖优先）。
-签名目录来源（由 LLM 插件安装器泛化）在 P2 接入。
 """
 
 import importlib
 import importlib.metadata
 import importlib.resources
 import importlib.util
+import json
 import logging
 import sys
 import types
@@ -36,10 +38,12 @@ from .manifest import (
 
 logger = logging.getLogger(__name__)
 
-PluginSource = Literal["builtin", "local", "pip"]
+PluginSource = Literal["builtin", "catalog", "local", "pip"]
 
 BUILTIN_ROOT = Path(__file__).resolve().parent.parent / "plugins" / "builtin"
-LOCAL_ROOT = Path("config") / "plugins" / "local"
+CATALOG_ROOT = Path("config") / "plugins"
+LOCAL_ROOT = CATALOG_ROOT / "local"
+INSTALLED_FILE = "installed.json"
 ENTRY_POINT_GROUP = "autobangumi.plugins"
 
 
@@ -58,7 +62,7 @@ class PluginCandidate:
 
     @property
     def signed(self) -> bool:
-        return self.source == "builtin"
+        return self.source in ("builtin", "catalog")
 
 
 @dataclass(frozen=True)
@@ -70,6 +74,7 @@ class DiscoveryError:
 def discover(
     *,
     builtin_root: Path = BUILTIN_ROOT,
+    catalog_root: Path = CATALOG_ROOT,
     local_root: Path = LOCAL_ROOT,
     entry_point_group: str = ENTRY_POINT_GROUP,
 ) -> tuple[list[PluginCandidate], list[DiscoveryError]]:
@@ -91,10 +96,15 @@ def discover(
             return
         found[pid] = candidate
 
-    for source, root in (("builtin", builtin_root), ("local", local_root)):
-        for plugin_dir in _plugin_dirs(root):
+    dirs_by_source: list[tuple[PluginSource, list[Path]]] = [
+        ("builtin", _plugin_dirs(builtin_root)),
+        ("catalog", _installed_dirs(catalog_root)),
+        ("local", _plugin_dirs(local_root)),
+    ]
+    for source, plugin_dirs in dirs_by_source:
+        for plugin_dir in plugin_dirs:
             try:
-                add(_directory_candidate(plugin_dir, source))  # type: ignore[arg-type]
+                add(_directory_candidate(plugin_dir, source))
             except ManifestError as e:
                 errors.append(DiscoveryError(str(plugin_dir), str(e)))
     for ep in importlib.metadata.entry_points(group=entry_point_group):
@@ -113,12 +123,38 @@ def _plugin_dirs(root: Path) -> list[Path]:
     return sorted(p for p in root.iterdir() if (p / MANIFEST_NAME).is_file())
 
 
+def installed_version(root: Path, plugin_id: str) -> str | None:
+    """``<root>/<id>/installed.json`` 指向的版本；未安装或指针损坏时为 None。"""
+    try:
+        data = json.loads((root / plugin_id / INSTALLED_FILE).read_text("utf-8"))
+        return str(data["version"])
+    except (OSError, ValueError, KeyError, TypeError):
+        return None
+
+
+def _installed_dirs(root: Path) -> list[Path]:
+    """签名目录已安装的插件：``installed.json`` 指向的版本目录。
+
+    root 下还有本地插件目录 ``local`` 与 LLM 提供商插件（``plugin.json``，
+    没有 ``plugin.toml``），都不是这里的插件。"""
+    if not root.is_dir():
+        return []
+    dirs = []
+    for plugin_root in sorted(root.iterdir()):
+        version = installed_version(root, plugin_root.name)
+        if version and (plugin_root / version / MANIFEST_NAME).is_file():
+            dirs.append(plugin_root / version)
+    return dirs
+
+
 # ---------------------------------------------------------------- directory
 
 
 def _directory_candidate(plugin_dir: Path, source: PluginSource) -> PluginCandidate:
     manifest = load_manifest(plugin_dir)
-    if plugin_dir.name != manifest.id:
+    # 签名目录的版本目录是 <id>/<version>，id 是上一级目录名
+    dir_id = plugin_dir.parent.name if source == "catalog" else plugin_dir.name
+    if dir_id != manifest.id:
         raise ManifestError(f"{plugin_dir}: 目录名须与清单 id 一致（{manifest.id}）")
     package = "ab_plugin_" + manifest.id.replace("-", "_")
     return PluginCandidate(

backend/src/test/test_plugin_catalog.py

@@ -0,0 +1,388 @@
+"""签名目录来源：通用插件的安装管线、加载器的 catalog 来源与安装 API。"""
+
+import base64
+import hashlib
+import importlib.util
+import io
+import json
+import zipfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import (
+    Encoding,
+    NoEncryption,
+    PrivateFormat,
+    PublicFormat,
+)
+
+from ab_sdk.cli import main, pack
+from ab_sdk.manifest import load_manifest
+from module.models.config import Plugins
+from module.plugin.installer import PluginInstaller
+from module.plugin.loader import discover
+
+PLUGIN_ID = "catalog-demo"
+
+
+@pytest.fixture
+def keypair(tmp_path):
+    priv = Ed25519PrivateKey.generate()
+    pubkey_path = tmp_path / "pub.pem"
+    pubkey_path.write_bytes(
+        priv.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)
+    )
+    return priv, pubkey_path
+
+
+def build_zip(tmp_path: Path, *, edit=None, extra: dict[str, bytes] | None = None):
+    """用 ab-plugin 脚手架与打包生成插件 zip；``edit`` 可在打包前改动目录。"""
+    assert main(["new", PLUGIN_ID, "--dir", str(tmp_path / "src")]) == 0
+    plugin_dir = tmp_path / "src" / PLUGIN_ID
+    if edit:
+        edit(plugin_dir)
+    archive = pack(plugin_dir, tmp_path / "dist", load_manifest(plugin_dir))
+    data = archive.read_bytes()
+    if extra:
+        buf = io.BytesIO(data)
+        with zipfile.ZipFile(buf, "a") as zf:
+            for name, content in extra.items():
+                zf.writestr(name, content)
+        data = buf.getvalue()
+    return data
+
+
+def make_installer(tmp_path, keypair, zip_bytes, *, entry=None, tamper_sig=False):
+    priv, pubkey_path = keypair
+    catalog_entry = {
+        "id": PLUGIN_ID,
+        "name": "Catalog demo",
+        "version": "0.1.0",
+        "kind": "plugin",
+        "extension_points": ["rename_strategy"],
+        "sdk": ">=0.5,<1",
+        "min_ab_version": "4.0.0",
+        "asset": f"{PLUGIN_ID}-0.1.0.zip",
+        "sha256": hashlib.sha256(zip_bytes).hexdigest(),
+    }
+    catalog_entry.update(entry or {})
+    catalog_bytes = json.dumps({"schema": 2, "plugins": [catalog_entry]}).encode()
+
+    def sign(data: bytes) -> str:
+        return base64.b64encode(priv.sign(data)).decode()
+
+    zip_sig = base64.b64encode(b"\x00" * 64).decode() if tamper_sig else sign(zip_bytes)
+    requested: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        url = str(request.url)
+        requested.append(url)
+        assert "/releases/download/plugins/" in url
+        if url.endswith("catalog.json"):
+            return httpx.Response(200, content=catalog_bytes)
+        if url.endswith("catalog.json.sig"):
+            return httpx.Response(200, text=sign(catalog_bytes))
+        if url.endswith(".zip"):
+            return httpx.Response(200, content=zip_bytes)
+        if url.endswith(".zip.sig"):
+            return httpx.Response(200, text=zip_sig)
+        return httpx.Response(404)
+
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    return PluginInstaller(
+        root=tmp_path / "plugins",
+        pubkey_path=pubkey_path,
+        client=client,
+        app_version="4.0.0",
+    )
+
+
+def discover_catalog(root: Path, tmp_path: Path):
+    return discover(
+        builtin_root=tmp_path / "no-builtin",
+        catalog_root=root,
+        local_root=root / "local",
+        entry_point_group="ab-test-none",
+    )
+
+
+class TestInstall:
+    async def test_install_then_discover_load_and_uninstall(self, tmp_path, keypair):
+        installer = make_installer(tmp_path, keypair, build_zip(tmp_path))
+
+        result = await installer.install(PLUGIN_ID)
+
+        assert result.success and result.version == "0.1.0"
+        root = tmp_path / "plugins"
+        assert (root / PLUGIN_ID / "0.1.0" / "plugin.toml").is_file()
+        candidates, errors = discover_catalog(root, tmp_path)
+        assert errors == []
+        [candidate] = candidates
+        assert candidate.source == "catalog" and candidate.signed
+        assert candidate.load().__name__ == "CatalogDemoPlugin"
+
+        assert (await installer.uninstall(PLUGIN_ID)).success
+        assert discover_catalog(root, tmp_path)[0] == []
+
+    async def test_upgrade_switches_to_new_version_directory(self, tmp_path, keypair):
+        root = tmp_path / "plugins"
+        old = await make_installer(tmp_path, keypair, build_zip(tmp_path)).install(
+            PLUGIN_ID
+        )
+        assert old.success
+
+        def bump(plugin_dir: Path) -> None:
+            manifest = plugin_dir / "plugin.toml"
+            manifest.write_text(manifest.read_text().replace("0.1.0", "0.2.0"))
+
+        upgrade = tmp_path / "upgrade"
+        upgrade.mkdir()
+        installer = make_installer(
+            upgrade,
+            keypair,
+            build_zip(upgrade, edit=bump),
+            entry={"version": "0.2.0", "asset": f"{PLUGIN_ID}-0.2.0.zip"},
+        )
+        installer.root = root
+
+        assert (await installer.install(PLUGIN_ID)).success
+        [candidate] = discover_catalog(root, tmp_path)[0]
+        assert candidate.manifest.version == "0.2.0"
+
+    @pytest.mark.parametrize(
+        "case, message",
+        [
+            ("sha", "sha256"),
+            ("signature", "signature"),
+            ("min_ab", "99.0.0"),
+            ("unknown", "not found"),
+            ("id", "id mismatch"),
+            ("version", "version mismatch"),
+            ("sdk", "ab_sdk"),
+            ("native", "原生扩展"),
+            ("zip-slip", "Unsafe path"),
+        ],
+    )
+    async def test_rejected_bundle_installs_nothing(
+        self, tmp_path, keypair, case, message
+    ):
+        def sdk_too_new(plugin_dir: Path) -> None:
+            manifest = plugin_dir / "plugin.toml"
+            manifest.write_text(manifest.read_text().replace(">=0.5", ">=9"))
+
+        def with_native(plugin_dir: Path) -> None:
+            (plugin_dir / "fast.so").write_bytes(b"\x7fELF")
+
+        zip_bytes = build_zip(
+            tmp_path,
+            edit={"sdk": sdk_too_new, "native": with_native}.get(case),
+            extra={"../evil.py": b"x"} if case == "zip-slip" else None,
+        )
+        entry = {
+            "sha": {"sha256": "deadbeef"},
+            "min_ab": {"min_ab_version": "99.0.0"},
+            "id": {"id": "someone-else"},
+            "version": {"version": "9.9.9"},
+        }.get(case)
+        installer = make_installer(
+            tmp_path, keypair, zip_bytes, entry=entry, tamper_sig=case == "signature"
+        )
+
+        plugin_id = {"unknown": "other-plugin", "id": "someone-else"}.get(
+            case, PLUGIN_ID
+        )
+
+        result = await installer.install(plugin_id)
+
+        assert not result.success
+        assert message.lower() in result.message.lower()
+        assert discover_catalog(tmp_path / "plugins", tmp_path)[0] == []
+
+    @pytest.mark.parametrize(
+        "plugin_id", ["local", "core", "../outside", "Bad_Id", "rename"]
+    )
+    async def test_reserved_or_unsafe_id_refused_before_any_download(
+        self, tmp_path, keypair, plugin_id
+    ):
+        installer = make_installer(tmp_path, keypair, build_zip(tmp_path))
+
+        install = await installer.install(plugin_id)
+        uninstall = await installer.uninstall(plugin_id)
+
+        assert not install.success and not uninstall.success
+
+    async def test_uninstall_leaves_unmanaged_directories_alone(
+        self, tmp_path, keypair
+    ):
+        installer = make_installer(tmp_path, keypair, build_zip(tmp_path))
+        # 本地插件目录与没有 installed.json 的目录都不属于安装器
+        unmanaged = tmp_path / "plugins" / "hand-made"
+        unmanaged.mkdir(parents=True)
+
+        result = await installer.uninstall("hand-made")
+
+        assert not result.success and unmanaged.is_dir()
+
+
+class TestReleaseScript:
+    async def test_catalog_built_by_release_script_installs(self, tmp_path, keypair):
+        script = (
+            Path(__file__).resolve().parents[3] / "scripts" / "build_plugin_catalog.py"
+        )
+        spec = importlib.util.spec_from_file_location("build_plugin_catalog", script)
+        assert spec and spec.loader
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        priv, pubkey_path = keypair
+        key_path = tmp_path / "key.pem"
+        key_path.write_bytes(
+            priv.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())
+        )
+        archive = tmp_path / "pack" / f"{PLUGIN_ID}-0.1.0.zip"
+        archive.parent.mkdir()
+        archive.write_bytes(build_zip(tmp_path))
+
+        module.build([archive], tmp_path / "release", key_path, "4.0.0")
+
+        released = tmp_path / "release"
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            name = request.url.path.rsplit("/", 1)[-1]
+            return httpx.Response(200, content=(released / name).read_bytes())
+
+        installer = PluginInstaller(
+            root=tmp_path / "plugins",
+            pubkey_path=pubkey_path,
+            client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+            app_version="4.0.0",
+        )
+        [entry] = await installer.fetch_catalog()
+        assert entry["extension_points"] == ["rename_strategy"]
+        assert (await installer.install(PLUGIN_ID)).success
+
+
+class TestLoaderCatalogSource:
+    def test_priority_builtin_over_catalog_over_local(self, tmp_path):
+        def write(root: Path, version: str) -> None:
+            root.mkdir(parents=True)
+            (root / "plugin.toml").write_text(
+                f'[plugin]\nid = "dup"\nname = "{root.parts[-3]}"\n'
+                f'version = "{version}"\nsdk = ">=0.1"\nentry = "dup:Dup"\n'
+            )
+
+        plugins = tmp_path / "plugins"
+        write(plugins / "dup" / "1.0.0", "1.0.0")
+        (plugins / "dup" / "installed.json").write_text('{"version": "1.0.0"}')
+        write(plugins / "local" / "dup", "2.0.0")
+
+        [candidate] = discover_catalog(plugins, tmp_path)[0]
+
+        assert candidate.source == "catalog"
+
+    def test_llm_plugins_and_broken_pointers_are_ignored(self, tmp_path):
+        plugins = tmp_path / "plugins"
+        (plugins / "github-copilot" / "1.0.0").mkdir(parents=True)
+        (plugins / "github-copilot" / "1.0.0" / "plugin.json").write_text("{}")
+        (plugins / "github-copilot" / "installed.json").write_text(
+            '{"version": "1.0.0"}'
+        )
+        (plugins / "broken").mkdir()
+        (plugins / "broken" / "installed.json").write_text("not json")
+
+        candidates, errors = discover_catalog(plugins, tmp_path)
+
+        assert candidates == [] and errors == []
+
+
+class TestPluginsApi:
+    @pytest.fixture
+    def ctx(self, app, monkeypatch):
+        from module.api import plugins as plugins_api
+        from module.api.deps import get_context
+
+        context = SimpleNamespace(
+            plugins=SimpleNamespace(statuses=lambda: [], reload=AsyncMock()),
+            settings=SimpleNamespace(plugins=Plugins(), save=MagicMock()),
+        )
+        monkeypatch.setattr(plugins_api, "settings", context.settings)
+        app.dependency_overrides[get_context] = lambda: context
+        monkeypatch.setattr(plugins_api, "CATALOG_ROOT", Path("/nonexistent"))
+        return context
+
+    def test_catalog_lists_entries(self, authed_client, ctx, monkeypatch):
+        from module.api import plugins as plugins_api
+
+        entry = {"id": PLUGIN_ID, "name": "Demo", "version": "1.0.0", "x": "ignored"}
+        monkeypatch.setattr(
+            plugins_api.PluginInstaller,
+            "fetch_catalog",
+            AsyncMock(return_value=[entry]),
+        )
+
+        response = authed_client.get("/api/v1/plugins/catalog")
+
+        assert response.status_code == 200
+        [item] = response.json()
+        assert item["id"] == PLUGIN_ID and item["installed_version"] is None
+
+    def test_catalog_unreachable_returns_502(self, authed_client, ctx, monkeypatch):
+        from module.api import plugins as plugins_api
+
+        monkeypatch.setattr(
+            plugins_api.PluginInstaller,
+            "fetch_catalog",
+            AsyncMock(side_effect=httpx.ConnectError("down")),
+        )
+
+        assert authed_client.get("/api/v1/plugins/catalog").status_code == 502
+
+    def test_install_enables_and_reloads_plugin(self, authed_client, ctx, monkeypatch):
+        from module.api import plugins as plugins_api
+        from module.plugin.installer import InstallResult
+
+        monkeypatch.setattr(
+            plugins_api.PluginInstaller,
+            "install",
+            AsyncMock(return_value=InstallResult(success=True, version="1.0.0")),
+        )
+
+        response = authed_client.post(f"/api/v1/plugins/{PLUGIN_ID}/install")
+
+        assert response.status_code == 200
+        assert ctx.settings.plugins.enabled[PLUGIN_ID] is True
+        ctx.plugins.reload.assert_awaited_once_with(PLUGIN_ID)
+
+    def test_failed_install_returns_400_without_enabling(
+        self, authed_client, ctx, monkeypatch
+    ):
+        from module.api import plugins as plugins_api
+        from module.plugin.installer import InstallResult
+
+        monkeypatch.setattr(
+            plugins_api.PluginInstaller,
+            "install",
+            AsyncMock(return_value=InstallResult(success=False, message="bad sig")),
+        )
+
+        response = authed_client.post(f"/api/v1/plugins/{PLUGIN_ID}/install")
+
+        assert response.status_code == 400
+        assert PLUGIN_ID not in ctx.settings.plugins.enabled
+        ctx.plugins.reload.assert_not_awaited()
+
+    def test_uninstall_reloads_plugin(self, authed_client, ctx, monkeypatch):
+        from module.api import plugins as plugins_api
+        from module.plugin.installer import InstallResult
+
+        monkeypatch.setattr(
+            plugins_api.PluginInstaller,
+            "uninstall",
+            AsyncMock(return_value=InstallResult(success=True)),
+        )
+
+        assert authed_client.delete(f"/api/v1/plugins/{PLUGIN_ID}").status_code == 200
+        ctx.plugins.reload.assert_awaited_once_with(PLUGIN_ID)

backend/vulture_whitelist.py

@@ -34,6 +34,8 @@ can_query
 can_rename
 can_manage
 can_rss_rules
+extension_points
+min_ab_version
 
 # --- 协议 / 回调签名要求的参数 ---
 connection_record  # SQLAlchemy connect 事件回调

scripts/build_plugin_catalog.py

@@ -0,0 +1,77 @@
+"""生成并签名插件目录（``plugins`` tag 的发布产物）。
+
+用法（需要更新签名私钥，见 CLAUDE.md 的发布说明）::
+
+    uv run --no-project --with cryptography python scripts/build_plugin_catalog.py \
+        --key ~/.autobangumi/update-signing-key.pem --min-ab 4.0.0 \
+        --out release-assets  dist/*.zip
+
+输入是 ``ab-plugin pack`` 打出的 zip。输出目录里有 ``catalog.json``、每个 zip 以及
+它们的 ``.sig``（对文件全部字节做 ed25519 签名，base64 文本），一并上传到 GitHub
+release ``plugins`` 即可。安装端的校验见 ``backend/src/module/plugin/installer.py``。
+"""
+
+import argparse
+import base64
+import hashlib
+import json
+import shutil
+import tomllib
+import zipfile
+from pathlib import Path
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import load_pem_private_key
+
+CATALOG_SCHEMA = 2
+
+
+def _entry(archive: Path, min_ab: str) -> dict:
+    with zipfile.ZipFile(archive) as zf:
+        plugin = tomllib.loads(zf.read("plugin.toml").decode("utf-8"))["plugin"]
+    return {
+        "id": plugin["id"],
+        "name": plugin["name"],
+        "version": plugin["version"],
+        "kind": "plugin",
+        "extension_points": plugin.get("extension_points", []),
+        "sdk": plugin["sdk"],
+        "min_ab_version": min_ab,
+        "description": plugin.get("description", ""),
+        "asset": archive.name,
+        "sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
+    }
+
+
+def _sign(key: Ed25519PrivateKey, path: Path) -> None:
+    signature = key.sign(path.read_bytes())
+    Path(str(path) + ".sig").write_text(base64.b64encode(signature).decode() + "\n")
+
+
+def build(archives: list[Path], out: Path, key_path: Path, min_ab: str) -> Path:
+    key = load_pem_private_key(key_path.read_bytes(), password=None)
+    if not isinstance(key, Ed25519PrivateKey):
+        raise SystemExit("签名密钥必须是 ed25519")
+    out.mkdir(parents=True, exist_ok=True)
+    entries = []
+    for archive in sorted(archives):
+        entries.append(_entry(archive, min_ab))
+        shutil.copy2(archive, out / archive.name)
+        _sign(key, out / archive.name)
+    catalog = out / "catalog.json"
+    catalog.write_text(
+        json.dumps({"schema": CATALOG_SCHEMA, "plugins": entries}, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    _sign(key, catalog)
+    return catalog
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("archives", nargs="+", type=Path)
+    parser.add_argument("--key", required=True, type=Path)
+    parser.add_argument("--out", required=True, type=Path)
+    parser.add_argument("--min-ab", default="4.0.0")
+    args = parser.parse_args()
+    print(build(args.archives, args.out, args.key, args.min_ab))