fix(plugins): 按 Codex 审查加固插件市场流水线

- 拒绝符号链接形式的条目;PR 检查纳入类型变更(T)
- PR 头部先解析为固定 SHA,作者代码不能借 FETCH_HEAD 换掉被检查的 commit
- 使用已发布的 catalog 前先验签,篡改的 release 资源不会被重新签名
- 全部下架时签发空目录
This commit is contained in:
Estrella Pan
2026-10-09 10:31:55 +02:00
parent c77f437fc7
commit d664122af8
5 changed files with 101 additions and 19 deletions

View File

@@ -44,10 +44,12 @@ jobs:
- name: Fetch PR head
run: |
git fetch -q origin "refs/pull/${{ github.event.number }}/head"
# 固定为 SHA:作者代码在检查过程中可以改写 .git/FETCH_HEAD
echo "HEAD_SHA=$(git rev-parse FETCH_HEAD)" >> "$GITHUB_ENV"
# 引入本脚本的那个 PR 时 base 上还没有它,只能用 PR 自己的版本
if [ ! -f scripts/plugin_registry.py ]; then
echo "::warning::base 分支没有 scripts/plugin_registry.py,使用 PR 中的版本"
git checkout FETCH_HEAD -- scripts/plugin_registry.py
git checkout "$(git rev-parse FETCH_HEAD)" -- scripts/plugin_registry.py
fi
- &python
uses: actions/setup-python@v5
@@ -73,12 +75,12 @@ jobs:
name: Download published catalog
env:
GH_TOKEN: ${{ github.token }}
run: gh release download plugins -p catalog.json -D old || echo "release plugins 还不存在"
run: gh release download plugins -p catalog.json -p catalog.json.sig -D old || echo "release plugins 还不存在"
- name: Check changed entries
run: |
set -o pipefail
uv run --project backend python scripts/plugin_registry.py check \
--base "${{ github.event.pull_request.base.sha }}" --head FETCH_HEAD \
--base "${{ github.event.pull_request.base.sha }}" --head "$HEAD_SHA" \
--catalog old/catalog.json --out dist | tee -a "$GITHUB_STEP_SUMMARY"
plan:
@@ -158,8 +160,10 @@ jobs:
fi
umask 077
printf '%s\n' "$PLUGIN_SIGNING_KEY" > "$RUNNER_TEMP/key.pem"
shopt -s nullglob # 全部下架时没有 zip,签一个空目录
zips=(stage/*.zip)
uv run --no-project --with cryptography python scripts/build_plugin_catalog.py \
--key "$RUNNER_TEMP/key.pem" --sources stage/sources.json --out release-assets stage/*.zip
--key "$RUNNER_TEMP/key.pem" --sources stage/sources.json --out release-assets "${zips[@]}"
rm -f "$RUNNER_TEMP/key.pem"
- name: Publish
env:

View File

@@ -30,6 +30,15 @@ from module.update.signing import DEFAULT_PUBKEY_PATH
PLUGIN_ID = "catalog-demo"
def load_catalog_script():
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)
return module
@pytest.fixture
def keypair(tmp_path):
priv = Ed25519PrivateKey.generate()
@@ -248,18 +257,23 @@ class TestInstall:
class TestReleaseScript:
def test_empty_registry_builds_an_empty_signed_catalog(self, tmp_path, keypair):
module = load_catalog_script()
priv, _ = keypair
key_path = tmp_path / "key.pem"
key_path.write_bytes(
priv.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())
)
catalog = module.build([], tmp_path / "release", key_path)
assert json.loads(catalog.read_text())["plugins"] == []
assert (tmp_path / "release" / "catalog.json.sig").is_file()
# 默认 --min-ab 须放行 4.0 的 beta 宿主(semver 中 4.0.0-beta.N < 4.0.0)
@pytest.mark.parametrize("app_version", ["4.0.0-beta.1", "4.0.0"])
async def test_catalog_built_by_release_script_installs(
self, tmp_path, keypair, app_version
):
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)
module = load_catalog_script()
priv, pubkey_path = keypair
key_path = tmp_path / "key.pem"
key_path.write_bytes(

View File

@@ -1,11 +1,14 @@
"""插件市场索引工具 scripts/plugin_registry.py:条目解析、上架规则、测试门槛与增量构建。"""
import base64
import hashlib
import importlib.util
import json
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
from ab_sdk.cli import main, pack
from ab_sdk.manifest import load_manifest
@@ -249,3 +252,44 @@ def test_diff_lists_registry_changes_between_commits(reg, tmp_path, monkeypatch)
changed = reg._diff("HEAD~1", "HEAD", "--diff-filter=AMR", "--", "plugins/registry")
assert changed == [f"plugins/registry/{PLUGIN_ID}.toml"]
def test_symlinked_registry_entry_is_rejected(reg, tmp_path, monkeypatch):
(tmp_path / "elsewhere.toml").write_text(entry_text())
root = tmp_path / "registry"
root.mkdir()
(root / f"{PLUGIN_ID}.toml").symlink_to(tmp_path / "elsewhere.toml")
monkeypatch.setattr(reg, "REGISTRY", root)
with pytest.raises(reg.Reject, match="符号链接"):
reg.registry_entries()
class TestLoadCatalog:
@pytest.fixture
def signed(self, reg, tmp_path, monkeypatch):
priv = Ed25519PrivateKey.generate()
pub = tmp_path / "pub.pem"
pub.write_bytes(
priv.public_key().public_bytes(
Encoding.PEM, PublicFormat.SubjectPublicKeyInfo
)
)
monkeypatch.setattr(reg, "PLUGIN_PUBKEY", pub)
catalog = tmp_path / "catalog.json"
catalog.write_text(json.dumps({"schema": 2, "plugins": [{"id": PLUGIN_ID}]}))
sig = tmp_path / "catalog.json.sig"
sig.write_text(base64.b64encode(priv.sign(catalog.read_bytes())).decode())
return catalog, sig
def test_valid_signature_loads(self, reg, signed):
assert list(reg.load_catalog(signed[0])) == [PLUGIN_ID]
@pytest.mark.parametrize("tamper", ["catalog", "missing_sig"])
def test_tampered_or_unsigned_catalog_is_refused(self, reg, signed, tamper):
catalog, sig = signed
if tamper == "catalog":
catalog.write_text(json.dumps({"schema": 2, "plugins": []}))
else:
sig.unlink()
with pytest.raises(SystemExit, match="签名"):
reg.load_catalog(catalog)

View File

@@ -86,7 +86,7 @@ def build(
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("archives", nargs="+", type=Path)
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=DEFAULT_MIN_AB)

View File

@@ -20,6 +20,7 @@
"""
import argparse
import base64
import hashlib
import html
import json
@@ -35,6 +36,9 @@ import zipfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from packaging.version import Version
from ab_sdk.cli import _PACK_EXCLUDE_DIRS, pack
@@ -44,6 +48,7 @@ ROOT = Path(__file__).resolve().parent.parent
REGISTRY = ROOT / "plugins" / "registry"
BUILTIN_ROOT = ROOT / "backend" / "src" / "module" / "plugins" / "builtin"
PLUGIN_UI = ROOT / "webui" / "packages" / "plugin-ui"
PLUGIN_PUBKEY = ROOT / "backend" / "src" / "module" / "plugin" / "ab_plugin_pubkey.pem"
RELEASE_BASE = "https://github.com/EstrellaXD/Auto_Bangumi/releases/download/plugins"
SHA_RE = re.compile(r"^[0-9a-f]{40}$")
REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
@@ -95,10 +100,11 @@ class Entry:
def registry_entries() -> list[Entry]:
return [
Entry.parse(p.stem, p.read_text("utf-8"))
for p in sorted(REGISTRY.glob("*.toml"))
]
paths = sorted(REGISTRY.glob("*.toml"))
if links := [p.name for p in paths if p.is_symlink()]:
# 符号链接会把条目指向 registry 之外、未经检查的文件
raise Reject(f"条目不能是符号链接:{', '.join(links)}")
return [Entry.parse(p.stem, p.read_text("utf-8")) for p in paths]
def _run(cmd: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
@@ -231,10 +237,22 @@ def readme(src: Path) -> str:
def load_catalog(path: Path | None) -> dict[str, dict]:
"""已发布的 catalog:id → 条目。release 还不存在时为空。"""
"""已发布的 catalog:id → 条目。release 还不存在时为空。
先验签:复用的 zip 与已上架版本都以它为准,能改 release 资源但没有签名密钥的
人不能借下一次发布让篡改过的 zip 被重新签名。
"""
if path is None or not path.is_file():
return {}
return {p["id"]: p for p in json.loads(path.read_text("utf-8")).get("plugins", [])}
data = path.read_bytes()
sig_path = path.with_name(path.name + ".sig")
pubkey = load_pem_public_key(PLUGIN_PUBKEY.read_bytes())
assert isinstance(pubkey, Ed25519PublicKey)
try:
pubkey.verify(base64.b64decode(sig_path.read_text().strip()), data)
except (FileNotFoundError, ValueError, InvalidSignature):
raise SystemExit(f"{path} 的签名无效或缺失") from None
return {p["id"]: p for p in json.loads(data).get("plugins", [])}
def _git(*args: str) -> subprocess.CompletedProcess[str]:
@@ -262,6 +280,8 @@ def check_entry(
rel = f"plugins/registry/{plugin_id}.toml"
lines: list[tuple[str, str]] = []
try:
if _git("ls-tree", head, rel).stdout.startswith("120000"):
raise Reject("条目不能是符号链接")
entry = Entry.parse(plugin_id, _git("show", f"{head}:{rel}").stdout)
old_text = _git("show", f"{base}:{rel}")
old = (
@@ -342,7 +362,7 @@ def summary(results: list[Result], removed: list[str], others: list[str]) -> str
def cmd_check(base: str, head: str, catalog: Path | None, out: Path) -> int:
changed = sorted(
PurePosixPath(p).stem
for p in _diff(base, head, "--diff-filter=AMR", "--", "plugins/registry")
for p in _diff(base, head, "--diff-filter=AMRT", "--", "plugins/registry")
)
removed = sorted(
PurePosixPath(p).stem