From ab5995a609562e08152d37587e7186dd2189a600 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sun, 5 Jul 2026 09:26:09 +0800 Subject: [PATCH] fix(security): contain plugin zip extraction (#6051) --- app/helper/plugin.py | 146 ++++++++++++++++++++++++++---------- tests/test_plugin_helper.py | 115 ++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+), 38 deletions(-) diff --git a/app/helper/plugin.py b/app/helper/plugin.py index c2681d9d..6c419e70 100644 --- a/app/helper/plugin.py +++ b/app/helper/plugin.py @@ -5,13 +5,14 @@ import io import json import shutil import site +import stat import sys import tempfile import threading import time import traceback import zipfile -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Dict, List, Optional, Tuple, Set, Callable, Awaitable from urllib.parse import parse_qs, quote, unquote, urlsplit @@ -1723,6 +1724,95 @@ class PluginHelper(metaclass=WeakSingleton): self.refresh_persistent_plugin_backup(pid) return True, "" + @staticmethod + def __validate_release_zip_name(name: str) -> None: + """ + 校验 release zip 成员名在 POSIX 与 Windows 语义下都只能表示相对路径。 + """ + if not name: + raise ValueError("非法 Release 压缩包成员：成员名为空") + if "\x00" in name: + raise ValueError(f"非法 Release 压缩包成员：{name}") + if "\\" in name: + raise ValueError(f"非法 Release 压缩包成员：{name}") + + posix_path = PurePosixPath(name) + windows_path = PureWindowsPath(name) + if ( + name.startswith("//") + or posix_path.is_absolute() + or windows_path.is_absolute() + or windows_path.drive + ): + raise ValueError(f"非法 Release 压缩包成员：{name}") + + parts = [part for part in posix_path.parts if part not in ("", ".")] + if not parts: + raise ValueError(f"非法 Release 压缩包成员：{name}") + if ".." in parts: + raise ValueError(f"非法 Release 压缩包成员：{name}") + + @staticmethod + def __validate_release_zip_type(info: zipfile.ZipInfo) -> None: + """ + release zip 只接受普通文件和目录，避免归档内的符号链接或设备文件影响安装边界。 + """ + mode = info.external_attr >> 16 + file_type = stat.S_IFMT(mode) + if not file_type: + return + if stat.S_ISREG(mode) or stat.S_ISDIR(mode): + return + raise ValueError(f"非法 Release 压缩包成员：{info.filename}") + + @staticmethod + def __get_release_zip_base_prefix(infos: List[zipfile.ZipInfo]) -> str: + """ + 识别 release zip 的单一顶层目录，用于保持插件包根目录剥离行为。 + """ + names = [info.filename for info in infos] + names_with_slash = [name for name in names if "/" in name] + if names_with_slash and len(names_with_slash) == len(names): + first_seg = names_with_slash[0].split("/", 1)[0] + if first_seg and all(name.startswith(first_seg + "/") for name in names): + return first_seg + "/" + return "" + + @classmethod + def __iter_release_zip_targets( + cls, zf: zipfile.ZipFile, dest_base: Path + ) -> List[Tuple[zipfile.ZipInfo, Path, bool]]: + """ + 将 release zip 成员解析为安装目标路径，并保证目标路径不会逃逸插件目录。 + """ + infos = zf.infolist() + for info in infos: + cls.__validate_release_zip_type(info) + cls.__validate_release_zip_name(info.filename) + + base_prefix = cls.__get_release_zip_base_prefix(infos) + dest_root = dest_base.resolve() + targets = [] + for info in infos: + raw_name = info.filename + rel_name = raw_name[len(base_prefix):] if base_prefix else raw_name + if not rel_name: + if base_prefix and raw_name == base_prefix: + continue + raise ValueError(f"非法 Release 压缩包成员：{raw_name}") + + cls.__validate_release_zip_name(rel_name) + rel_parts = [part for part in PurePosixPath(rel_name).parts if part not in ("", ".")] + if not rel_parts: + raise ValueError(f"非法 Release 压缩包成员：{raw_name}") + dest_path = (dest_root / Path(*rel_parts)).resolve() + try: + dest_path.relative_to(dest_root) + except ValueError as exc: + raise ValueError(f"非法 Release 压缩包成员：{raw_name}") from exc + targets.append((info, dest_path, info.is_dir())) + return targets + def __install_from_release(self, pid: str, user_repo: str, release_tag: str) -> Tuple[bool, str]: """ 通过 GitHub Release 资产文件安装插件。 @@ -1766,29 +1856,18 @@ class PluginHelper(metaclass=WeakSingleton): try: with zipfile.ZipFile(io.BytesIO(res.content)) as zf: - namelist = zf.namelist() - if not namelist: + infos = zf.infolist() + if not infos: return False, "压缩包内容为空" - # 若所有条目均在同一顶层目录下（如 pid/），则剥离这一层，避免出现双层目录 - names_with_slash = [n for n in namelist if '/' in n] - base_prefix = '' - if names_with_slash and len(names_with_slash) == len(namelist): - first_seg = names_with_slash[0].split('/')[0] - if all(n.startswith(first_seg + '/') for n in namelist): - base_prefix = first_seg + '/' - dest_base = Path(settings.ROOT_PATH) / "app" / "plugins" / pid.lower() + targets = self.__iter_release_zip_targets(zf, dest_base) wrote_any = False - for name in namelist: - rel_path = name[len(base_prefix):] - if not rel_path: + for info, dest_path, is_dir in targets: + if is_dir: + dest_path.mkdir(parents=True, exist_ok=True) continue - if rel_path.endswith('/'): - (dest_base / rel_path.rstrip('/')).mkdir(parents=True, exist_ok=True) - continue - dest_path = dest_base / rel_path dest_path.parent.mkdir(parents=True, exist_ok=True) - with zf.open(name, 'r') as src, open(dest_path, 'wb') as dst: + with zf.open(info, 'r') as src, open(dest_path, 'wb') as dst: dst.write(src.read()) wrote_any = True if not wrote_any: @@ -2783,28 +2862,19 @@ class PluginHelper(metaclass=WeakSingleton): try: with zipfile.ZipFile(io.BytesIO(res.content)) as zf: - namelist = zf.namelist() - if not namelist: + infos = zf.infolist() + if not infos: return False, "压缩包内容为空" - names_with_slash = [n for n in namelist if '/' in n] - base_prefix = '' - if names_with_slash and len(names_with_slash) == len(namelist): - first_seg = names_with_slash[0].split('/')[0] - if all(n.startswith(first_seg + '/') for n in namelist): - base_prefix = first_seg + '/' - - dest_base = AsyncPath(settings.ROOT_PATH) / "app" / "plugins" / pid.lower() + dest_base = Path(settings.ROOT_PATH) / "app" / "plugins" / pid.lower() + targets = self.__iter_release_zip_targets(zf, dest_base) wrote_any = False - for name in namelist: - rel_path = name[len(base_prefix):] - if not rel_path: + for info, dest_path, is_dir in targets: + async_dest_path = AsyncPath(dest_path) + if is_dir: + await async_dest_path.mkdir(parents=True, exist_ok=True) continue - if rel_path.endswith('/'): - await (dest_base / rel_path.rstrip('/')).mkdir(parents=True, exist_ok=True) - continue - dest_path = dest_base / rel_path - await dest_path.parent.mkdir(parents=True, exist_ok=True) - with zf.open(name, 'r') as src: + await async_dest_path.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info, 'r') as src: data = src.read() async with aiofiles.open(dest_path, 'wb') as dst: await dst.write(data) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index c566808c..09acb137 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -1,5 +1,6 @@ import asyncio import io +import stat import sys import tempfile import threading @@ -60,6 +61,40 @@ def _build_zip(entries: dict[str, bytes]) -> bytes: return buffer.getvalue() +def _build_release_zip_member(name: str, *, symlink: bool = False) -> bytes: + """构造单成员 release zip，用于覆盖成员路径与文件类型校验。""" + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + if symlink: + info = zipfile.ZipInfo(name) + info.create_system = 3 + info.external_attr = (stat.S_IFLNK | 0o777) << 16 + zf.writestr(info, b"target") + else: + zf.writestr(name, b"evil") + return buffer.getvalue() + + +def _patch_release_install_settings(monkeypatch, tmp_path: Path) -> None: + """隔离 release 安装根目录，并阻止测试误触真实根路径。""" + monkeypatch.setattr("app.helper.plugin.settings", SimpleNamespace( + ROOT_PATH=tmp_path, + REPO_GITHUB_HEADERS=lambda repo=None: {}, + )) + + original_mkdir = Path.mkdir + safe_root = tmp_path.resolve() + + def guarded_mkdir(path: Path, *args, **kwargs): + try: + path.resolve().relative_to(safe_root) + except ValueError as exc: + raise AssertionError(f"unsafe mkdir attempted: {path}") from exc + return original_mkdir(path, *args, **kwargs) + + monkeypatch.setattr(Path, "mkdir", guarded_mkdir) + + def _patch_sync_remote_install(helper, monkeypatch, meta: dict, release_result: tuple[bool, str], filelist_result: tuple[bool, str] = (True, "")): @@ -1752,6 +1787,43 @@ class TestPluginHelper: assert not success assert "下载资产失败：502" == message + @pytest.mark.parametrize( + "member_name, symlink", + [ + ("../evil.py", False), + ("/tmp/evil.py", False), + ("..\\evil.py", False), + ("C:/evil.py", False), + ("//server/share/evil.py", False), + ("demoplugin/link.py", True), + ], + ) + def test_install_from_release_rejects_unsafe_zip_member(self, monkeypatch, tmp_path, member_name, symlink): + """ + release zip 成员必须限制在插件运行目录内，且不能是符号链接或特殊文件。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, _build_release_zip_member(member_name, symlink=symlink)), + ]) + _patch_release_install_settings(monkeypatch, tmp_path) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "非法 Release 压缩包成员" in message + assert not (tmp_path / "app" / "plugins" / "evil.py").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "..\\evil.py").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "C:").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "link.py").exists() + def test_install_from_release_extracts_zip_with_top_level_directory(self, monkeypatch, tmp_path): """ release zip 带顶层插件目录时剥离该层后写入运行目录。 @@ -2287,6 +2359,49 @@ class TestPluginHelper: assert not success assert "下载资产失败：502" == message + @pytest.mark.parametrize( + "member_name, symlink", + [ + ("../evil.py", False), + ("/tmp/evil.py", False), + ("..\\evil.py", False), + ("C:/evil.py", False), + ("//server/share/evil.py", False), + ("demoplugin/link.py", True), + ], + ) + def test_async_install_from_release_rejects_unsafe_zip_member(self, monkeypatch, tmp_path, member_name, symlink): + """ + 异步 release zip 成员使用同步路径相同的边界与文件类型规则。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, _build_release_zip_member(member_name, symlink=symlink)), + ]) + + async def fake_request(*_args, **_kwargs): + return next(responses) + + _patch_release_install_settings(monkeypatch, tmp_path) + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "非法 Release 压缩包成员" in message + assert not (tmp_path / "app" / "plugins" / "evil.py").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "..\\evil.py").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "C:").exists() + assert not (tmp_path / "app" / "plugins" / "demoplugin" / "link.py").exists() + def test_async_install_from_release_extracts_zip_with_top_level_directory(self, monkeypatch, tmp_path): """ 异步 release zip 带顶层插件目录时剥离该层后写入运行目录。