From 3ad9ffd239742d6f74272261d99ba64c73fed716 Mon Sep 17 00:00:00 2001 From: chengma Date: Mon, 13 Jul 2026 12:02:47 +0800 Subject: [PATCH] feat(update): securely stage verified packages --- app/update_check.py | 24 ++ app/update_installer.py | 398 +++++++++++++++++++++++++++++++++ docs/04-architecture.md | 2 + docs/packaging.md | 2 + docs/tasks/T-616.md | 6 +- docs/update-check.md | 4 +- tests/test_update_check.py | 5 + tests/test_update_installer.py | 178 +++++++++++++++ 8 files changed, 616 insertions(+), 3 deletions(-) create mode 100644 app/update_installer.py create mode 100644 tests/test_update_installer.py diff --git a/app/update_check.py b/app/update_check.py index a48cd58..57879a3 100644 --- a/app/update_check.py +++ b/app/update_check.py @@ -27,6 +27,10 @@ class UpdateInfo: force_update: bool = False download_url: str = "" sha256: str = "" + size_bytes: int = 0 + package_format: str = "" + updater_protocol: int = 0 + min_updater_protocol: int = 0 message: str = "" @@ -39,6 +43,10 @@ class UpdateCheckResult: min_supported_version: str = "" download_url: str = "" sha256: str = "" + size_bytes: int = 0 + package_format: str = "" + updater_protocol: int = 0 + min_updater_protocol: int = 0 message: str = "" error: str = "" @@ -111,6 +119,18 @@ def parse_update_info(payload) -> UpdateInfo: force_update=_as_bool(payload.get("force_update", release.get("force_update"))), download_url=str(payload.get("download_url") or release.get("download_url") or "").strip(), sha256=str(payload.get("sha256") or release.get("sha256") or "").strip(), + size_bytes=int(payload.get("size_bytes") or release.get("size_bytes") or 0), + package_format=str( + payload.get("package_format") or release.get("package_format") or "" + ).strip(), + updater_protocol=int( + payload.get("updater_protocol") or release.get("updater_protocol") or 0 + ), + min_updater_protocol=int( + payload.get("min_updater_protocol") + or release.get("min_updater_protocol") + or 0 + ), message=str( payload.get("message") or payload.get("release_notes") @@ -178,6 +198,10 @@ def check_for_update( min_supported_version=info.min_supported_version, download_url=info.download_url, sha256=info.sha256, + size_bytes=info.size_bytes, + package_format=info.package_format, + updater_protocol=info.updater_protocol, + min_updater_protocol=info.min_updater_protocol, message=info.message, ) except Exception as exc: diff --git a/app/update_installer.py b/app/update_installer.py new file mode 100644 index 0000000..57f6c35 --- /dev/null +++ b/app/update_installer.py @@ -0,0 +1,398 @@ +"""自动升级包的安全下载、解压和暂存校验。""" + +from __future__ import annotations + +import hashlib +import ipaddress +import json +import os +import re +import shutil +import stat +import urllib.parse +import urllib.request +import uuid +import zipfile +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + +from .release_manifest import ( + ALLOWED_ROOTS, + ENTRY_POINT, + MANIFEST_FILENAME, + PACKAGE_FORMAT, + UPDATER_PROTOCOL, + sha256_file, +) + + +MAX_DOWNLOAD_BYTES = 512 * 1024 * 1024 +MAX_EXTRACTED_BYTES = 2 * 1024 * 1024 * 1024 +MAX_FILE_COUNT = 20000 +MAX_COMPRESSION_RATIO = 200 +DEFAULT_TRUSTED_HOSTS = frozenset({"cm.833729.com"}) +RESERVED_NAMES = { + "CON", + "PRN", + "AUX", + "NUL", + *("COM%d" % number for number in range(1, 10)), + *("LPT%d" % number for number in range(1, 10)), +} + + +class UpdateInstallError(RuntimeError): + """升级包不能安全下载或暂存。""" + + +class UpdateCancelled(UpdateInstallError): + """用户取消升级包下载。""" + + +@dataclass(frozen=True) +class UpdatePackageMetadata: + version: str + download_url: str + sha256: str + size_bytes: int + package_format: str + updater_protocol: int + min_updater_protocol: int = 1 + + +@dataclass(frozen=True) +class StagedUpdate: + version: str + zip_path: Path + staging_dir: Path + pending_path: Path + sha256: str + + +def metadata_from_update_info(info): + metadata = UpdatePackageMetadata( + version=str(getattr(info, "latest_version", "") or "").strip(), + download_url=str(getattr(info, "download_url", "") or "").strip(), + sha256=str(getattr(info, "sha256", "") or "").strip().lower(), + size_bytes=int(getattr(info, "size_bytes", 0) or 0), + package_format=str(getattr(info, "package_format", "") or "").strip(), + updater_protocol=int(getattr(info, "updater_protocol", 0) or 0), + min_updater_protocol=int(getattr(info, "min_updater_protocol", 0) or 1), + ) + validate_metadata(metadata) + return metadata + + +def validate_metadata(metadata, trusted_hosts=DEFAULT_TRUSTED_HOSTS): + if not re.fullmatch(r"\d+(?:\.\d+)*", metadata.version): + raise UpdateInstallError("新版版本号格式不正确") + validate_download_url(metadata.download_url, trusted_hosts) + if not re.fullmatch(r"[0-9a-fA-F]{64}", metadata.sha256): + raise UpdateInstallError("新版安装包校验值缺失或格式不正确") + if metadata.size_bytes <= 0 or metadata.size_bytes > MAX_DOWNLOAD_BYTES: + raise UpdateInstallError("新版安装包大小不正确或超过限制") + if metadata.package_format != PACKAGE_FORMAT: + raise UpdateInstallError("新版安装包格式不受支持") + if metadata.updater_protocol != UPDATER_PROTOCOL: + raise UpdateInstallError("新版安装包更新协议不受支持") + if metadata.min_updater_protocol > UPDATER_PROTOCOL: + raise UpdateInstallError("当前更新器版本过低") + + +def validate_download_url(url, trusted_hosts=DEFAULT_TRUSTED_HOSTS): + parsed = urllib.parse.urlsplit(str(url or "")) + host = (parsed.hostname or "").lower().rstrip(".") + trusted = {value.lower().rstrip(".") for value in trusted_hosts} + if parsed.scheme.lower() != "https" or not host: + raise UpdateInstallError("新版下载地址必须使用 HTTPS") + if parsed.username or parsed.password: + raise UpdateInstallError("新版下载地址不能包含账号信息") + try: + address = ipaddress.ip_address(host) + except ValueError: + address = None + if address is not None and not address.is_global: + raise UpdateInstallError("新版下载地址不能指向内网") + if host not in trusted: + raise UpdateInstallError("新版下载地址不在受信任域名内") + return parsed + + +class _SafeRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, trusted_hosts): + super().__init__() + self.trusted_hosts = trusted_hosts + + def redirect_request(self, req, fp, code, msg, headers, newurl): + validate_download_url(newurl, self.trusted_hosts) + return super().redirect_request(req, fp, code, msg, headers, newurl) + + +def _update_root(install_root): + return Path(install_root).resolve() / ".cmshopee-update" + + +def _safe_unlink(path): + try: + Path(path).unlink() + except FileNotFoundError: + pass + + +def download_package( + metadata, + install_root, + *, + trusted_hosts=DEFAULT_TRUSTED_HOSTS, + opener=None, + cancelled=None, + progress=None, + timeout=66, +): + validate_metadata(metadata, trusted_hosts) + validate_download_url(metadata.download_url, trusted_hosts) + downloads = _update_root(install_root) / "downloads" + downloads.mkdir(parents=True, exist_ok=True) + part_path = downloads / (metadata.version + ".zip.part") + zip_path = downloads / (metadata.version + ".zip") + _safe_unlink(part_path) + + if shutil.disk_usage(str(downloads)).free < metadata.size_bytes: + raise UpdateInstallError("磁盘空间不足，无法下载新版") + + request = urllib.request.Request( + metadata.download_url, + headers={"Accept": "application/zip", "User-Agent": "cmshopee-updater/1"}, + ) + if opener is None: + opener = urllib.request.build_opener(_SafeRedirectHandler(trusted_hosts)) + + digest = hashlib.sha256() + downloaded = 0 + try: + with opener.open(request, timeout=timeout) as response, part_path.open("wb") as output: + validate_download_url(response.geturl(), trusted_hosts) + while True: + if cancelled and cancelled(): + raise UpdateCancelled("已取消下载新版") + chunk = response.read(1024 * 1024) + if not chunk: + break + downloaded += len(chunk) + if downloaded > metadata.size_bytes or downloaded > MAX_DOWNLOAD_BYTES: + raise UpdateInstallError("新版安装包大小超过接口声明") + output.write(chunk) + digest.update(chunk) + if progress: + progress(downloaded, metadata.size_bytes) + if downloaded != metadata.size_bytes: + raise UpdateInstallError("新版安装包大小校验失败") + if digest.hexdigest().lower() != metadata.sha256.lower(): + raise UpdateInstallError("新版安装包完整性校验失败") + os.replace(str(part_path), str(zip_path)) + return zip_path + except UpdateInstallError: + _safe_unlink(part_path) + raise + except Exception as exc: + _safe_unlink(part_path) + raise UpdateInstallError("下载新版失败，请检查网络后重试") from exc + + +def _validate_member_name(name): + if not name or "\\" in name or name.startswith(("/", "\\")): + raise UpdateInstallError("新版安装包包含不安全路径") + if re.match(r"^[A-Za-z]:", name) or ":" in name: + raise UpdateInstallError("新版安装包包含不安全路径") + path = PurePosixPath(name) + if any(part in {"", ".", ".."} for part in path.parts): + raise UpdateInstallError("新版安装包包含路径穿越") + for part in path.parts: + if part.endswith((".", " ")): + raise UpdateInstallError("新版安装包包含 Windows 不支持的路径") + stem = part.split(".", 1)[0].upper() + if stem in RESERVED_NAMES: + raise UpdateInstallError("新版安装包包含 Windows 保留名称") + return path + + +def _is_zip_symlink(info): + mode = (info.external_attr >> 16) & 0xFFFF + return stat.S_ISLNK(mode) + + +def safe_extract(zip_path, target_dir): + target_dir = Path(target_dir) + seen = set() + total_size = 0 + with zipfile.ZipFile(str(zip_path), "r") as archive: + infos = archive.infolist() + if len(infos) > MAX_FILE_COUNT: + raise UpdateInstallError("新版安装包文件数量超过限制") + validated = [] + for info in infos: + path = _validate_member_name(info.filename.rstrip("/") if info.is_dir() else info.filename) + key = path.as_posix().casefold() + if key in seen: + raise UpdateInstallError("新版安装包包含重复路径") + seen.add(key) + if _is_zip_symlink(info): + raise UpdateInstallError("新版安装包不允许符号链接") + total_size += info.file_size + if total_size > MAX_EXTRACTED_BYTES: + raise UpdateInstallError("新版安装包解压大小超过限制") + if info.file_size and ( + info.compress_size == 0 + or info.file_size / max(info.compress_size, 1) > MAX_COMPRESSION_RATIO + ): + raise UpdateInstallError("新版安装包压缩比异常") + validated.append((info, path)) + + target_dir.mkdir(parents=True, exist_ok=False) + for info, relative in validated: + destination = target_dir.joinpath(*relative.parts) + if info.is_dir(): + destination.mkdir(parents=True, exist_ok=True) + continue + destination.parent.mkdir(parents=True, exist_ok=True) + with archive.open(info, "r") as source, destination.open("wb") as output: + shutil.copyfileobj(source, output, length=1024 * 1024) + return target_dir + + +def validate_staging(staging_dir, metadata): + staging_dir = Path(staging_dir) + for name in (ENTRY_POINT, "_internal", "version.txt", MANIFEST_FILENAME): + if not (staging_dir / name).exists(): + raise UpdateInstallError("新版安装包缺少必要程序文件") + if not (staging_dir / "_internal").is_dir(): + raise UpdateInstallError("新版安装包依赖目录无效") + if (staging_dir / "version.txt").read_text(encoding="utf-8-sig").strip() != metadata.version: + raise UpdateInstallError("新版安装包版本不一致") + + try: + manifest = json.loads((staging_dir / MANIFEST_FILENAME).read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise UpdateInstallError("新版安装包清单无效") from exc + if ( + manifest.get("app_version") != metadata.version + or manifest.get("package_format") != metadata.package_format + or manifest.get("entry_point") != ENTRY_POINT + or int(manifest.get("updater_protocol") or 0) != metadata.updater_protocol + or int(manifest.get("min_updater_protocol") or 0) > UPDATER_PROTOCOL + ): + raise UpdateInstallError("新版安装包清单与发布信息不一致") + + replace_roots = manifest.get("replace_roots") + if not isinstance(replace_roots, list) or not replace_roots: + raise UpdateInstallError("新版安装包替换范围无效") + if any(str(root) not in ALLOWED_ROOTS for root in replace_roots): + raise UpdateInstallError("新版安装包包含未授权替换范围") + + declared = {} + for item in manifest.get("files") or []: + if not isinstance(item, dict): + raise UpdateInstallError("新版安装包清单文件项无效") + relative = _validate_member_name(str(item.get("path") or "")) + if relative.parts[0] not in ALLOWED_ROOTS or relative.name == MANIFEST_FILENAME: + raise UpdateInstallError("新版安装包清单包含未授权文件") + key = relative.as_posix().casefold() + if key in declared: + raise UpdateInstallError("新版安装包清单包含重复路径") + declared[key] = (relative, int(item.get("size_bytes") or -1), str(item.get("sha256") or "")) + + actual = {} + for path in staging_dir.rglob("*"): + if path.is_symlink(): + raise UpdateInstallError("新版暂存目录不允许符号链接") + if not path.is_file() or path.name == MANIFEST_FILENAME: + continue + relative = PurePosixPath(path.relative_to(staging_dir).as_posix()) + actual[relative.as_posix().casefold()] = relative + if set(actual) != set(declared): + raise UpdateInstallError("新版安装包文件与清单不一致") + for key, (relative, size_bytes, expected_hash) in declared.items(): + path = staging_dir.joinpath(*relative.parts) + if path.stat().st_size != size_bytes or sha256_file(path).lower() != expected_hash.lower(): + raise UpdateInstallError("新版安装包文件校验失败") + return manifest + + +def _write_pending(update_root, metadata, staging_dir, zip_path): + pending_path = update_root / "pending.json" + payload = { + "schema_version": 1, + "stage": "verified", + "version": metadata.version, + "install_root": str(update_root.parent), + "staging_dir": str(Path(staging_dir).relative_to(update_root).as_posix()), + "zip_path": str(Path(zip_path).relative_to(update_root).as_posix()), + "sha256": metadata.sha256, + "package_format": metadata.package_format, + "updater_protocol": metadata.updater_protocol, + } + temporary = pending_path.with_suffix(".json.tmp") + temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(str(temporary), str(pending_path)) + return pending_path + + +def _discard_pending(update_root, payload=None): + if isinstance(payload, dict): + relative = payload.get("staging_dir") + if relative: + try: + staging_dir = (update_root / relative).resolve() + if update_root in staging_dir.parents and staging_dir.exists(): + shutil.rmtree(str(staging_dir), ignore_errors=True) + except (OSError, ValueError): + pass + _safe_unlink(update_root / "pending.json") + + +def load_verified_pending(install_root, metadata): + update_root = _update_root(install_root) + pending_path = update_root / "pending.json" + payload = None + try: + payload = json.loads(pending_path.read_text(encoding="utf-8")) + if ( + payload.get("stage") != "verified" + or payload.get("version") != metadata.version + or payload.get("sha256") != metadata.sha256 + ): + raise UpdateInstallError("待安装更新记录已失效") + staging_dir = (update_root / payload["staging_dir"]).resolve() + zip_path = (update_root / payload["zip_path"]).resolve() + if update_root not in staging_dir.parents or update_root not in zip_path.parents: + raise UpdateInstallError("待安装更新路径无效") + validate_staging(staging_dir, metadata) + if not zip_path.is_file() or sha256_file(zip_path) != metadata.sha256: + raise UpdateInstallError("待安装更新压缩包已失效") + return StagedUpdate(metadata.version, zip_path, staging_dir, pending_path, metadata.sha256) + except FileNotFoundError: + return None + except (OSError, ValueError, KeyError, UpdateInstallError): + _discard_pending(update_root, payload) + return None + + +def prepare_update(metadata, install_root, **download_kwargs): + validate_metadata(metadata, download_kwargs.get("trusted_hosts", DEFAULT_TRUSTED_HOSTS)) + reusable = load_verified_pending(install_root, metadata) + if reusable is not None: + return reusable + update_root = _update_root(install_root) + zip_path = download_package(metadata, install_root, **download_kwargs) + staging_parent = update_root / "staging" + staging_parent.mkdir(parents=True, exist_ok=True) + staging_dir = staging_parent / (metadata.version + "-" + uuid.uuid4().hex[:12]) + try: + safe_extract(zip_path, staging_dir) + validate_staging(staging_dir, metadata) + pending_path = _write_pending(update_root, metadata, staging_dir, zip_path) + except Exception: + if staging_dir.exists(): + shutil.rmtree(str(staging_dir), ignore_errors=True) + raise + return StagedUpdate(metadata.version, zip_path, staging_dir, pending_path, metadata.sha256) diff --git a/docs/04-architecture.md b/docs/04-architecture.md index d76c318..1657f44 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -499,6 +499,7 @@ cmshopee/ │ ├── appconfig.py / db.py / excel.py / config.py / accounts.py / chrome.py │ ├── editor.py / ai.py / prompts.py / gui.py / workers.py │ ├── release_manifest.py # 发布包文件清单、zip哈希与服务端元数据模板 +│ ├── update_installer.py # 自动升级安全下载、解压、manifest校验与同盘暂存 ├── main.py # GUI 启动入口：from app.gui import main ├── shopee待处理任务模板.xlsx # 标准空 Excel 模板，可提交；业务填写后的副本不提交 ├── data/ # 用户本地数据根（整体 gitignore；打包更新时保留） @@ -527,6 +528,7 @@ cmshopee/ - 发布包格式固定为 `cmshopee-portable-v1`，入口为 `cmshopee.exe`，程序依赖集中在 `_internal/`。manifest 覆盖除自身外的所有程序文件，并记录规范化相对路径、字节数和 SHA-256。 - 发布包只能包含程序根项目，`data/` 和 `.cmshopee-update/` 永远在替换边界之外。第一阶段预留签名字段，但 SHA-256 只负责传输完整性，不等同于发布者身份认证。 - T-615 只提供可验证发布契约；启动门禁仍保持 T-544 的人工下载行为，直到后续下载、独立更新器、事务替换和失败熔断任务全部接入。 +- T-616 的下载暂存根固定为安装目录下 `.cmshopee-update/`，与 `data/` 完全隔离。远程zip必须经过HTTPS/受信任域名、声明大小、整包SHA-256、安全zip路径和包内manifest逐文件校验，才写 `pending.json`；此阶段不替换任何运行中程序文件。 - CDP 交互事实变化同步第七节。 - 正式代码只放 `app/` 包；根目录只保留 `main.py`、配置/数据目录、文档和原型目录，不新增正式业务模块。 diff --git a/docs/packaging.md b/docs/packaging.md index 8eff000..5de29a7 100644 --- a/docs/packaging.md +++ b/docs/packaging.md @@ -176,6 +176,8 @@ T-544 第一版目标是**启动时检查是否必须升级**，但仍不做自 T-615 不改变客户端行为：T-544 仍只打开浏览器下载。客户端下载、解压校验、独立进程替换、回滚和重启分别由 T-616 至 T-619 实现。清单预留签名字段，但当前只验证完整性，不宣称已验证发布者身份。 +T-616 提供无Qt依赖的安全下载暂存层 `app/update_installer.py`。它只接受受信任域名的HTTPS，下载到 `<安装目录>/.cmshopee-update/downloads/`，完成大小与zip SHA-256校验后安全解压到同盘 `staging/`；路径穿越、符号链接、大小写重复、Windows保留名、ADS、异常压缩比、文件数/解压总量超限，以及manifest缺失或逐文件hash不一致都会拒绝。完整验证后才原子写 `pending.json`，当前程序目录和 `data/` 不变。 + ## 四、绝不打包的本地数据 发布包里不能包含以下本地数据、密钥、业务数据或登录态： diff --git a/docs/tasks/T-616.md b/docs/tasks/T-616.md index 9ce2375..cf737bf 100644 --- a/docs/tasks/T-616.md +++ b/docs/tasks/T-616.md @@ -3,7 +3,7 @@ id: T-616 title: 自动升级安全下载校验与同盘暂存 phase: 8 deps: [T-615] -status: TODO +status: DONE created: 2026-07-13 --- @@ -44,4 +44,6 @@ created: 2026-07-13 ## 执行记录 -（完成后记录实现、验证命令与结果。） +- 2026-07-13：新增无Qt依赖的 `app/update_installer.py`，实现受信任HTTPS下载、同盘part暂存、大小/hash校验、安全zip解压、manifest逐文件复核及可复用 `pending.json`。 +- 2026-07-13：补齐版本接口自动安装元数据透传；所有失败均发生在当前程序替换前，取消与无效pending会清理不完整状态，`data/` 不参与更新。 +- 2026-07-13：干净worktree验证通过：ruff、compileall、完整unittest（400项）和 `git diff --check`。 diff --git a/docs/update-check.md b/docs/update-check.md index d2697f4..28891af 100644 --- a/docs/update-check.md +++ b/docs/update-check.md @@ -15,7 +15,7 @@ ## 二、客户端读取的字段（`update_check.parse_update_info`） -响应是一个 JSON 对象。以下字段**顶层 payload 或 `release` 子对象里都认**（客户端按 `payload.get(x) or release.get(x)` 取，顶层优先）： +响应是一个 JSON 对象。T-544 兼容字段仍可位于顶层或 `release` 子对象；自动安装元数据由 `app/update_installer.py` 做第二层严格校验，生产服务应把同一版本的全部字段放进同一个 `release` 对象。 | 字段 | 类型 | 说明 | | --- | --- | --- | @@ -47,6 +47,8 @@ - **非强制**：**不弹任何提示**，直接进主界面（当前无"温和可跳过提示"分支；如需另立任务）。 - **失败放行**：接口断网、超时、返回非法 JSON、缺 `latest_version`/`min_supported_version` 时，客户端记诊断日志（`data/logs/cmshopee.log`，`step=startup_update_check`「已允许继续使用」）并**放行**，不因服务器故障导致全员打不开。 +T-616 已实现但尚未接入GUI的安全暂存层：仅接受受信任域名的 HTTPS 地址，流式下载到安装目录 `.cmshopee-update/`，校验zip大小和SHA-256，安全解压后再按包内manifest逐文件校验。任一步失败都不修改当前程序或 `data/`；GUI接入由T-618完成。 + ## 五、发版约定（服务端据此控制） 自动安装使用的全部字段必须放在同一个 `release` 对象内，不得把版本取自一个对象、hash 取自另一个对象。构建脚本生成的 `release/release-metadata.json` 是服务端录入模板；发布人员只补 HTTPS `download_url`、强制策略和中文发布说明，不得手工改写 hash、大小、包格式或协议版本。 diff --git a/tests/test_update_check.py b/tests/test_update_check.py index 5638e44..20ca02e 100644 --- a/tests/test_update_check.py +++ b/tests/test_update_check.py @@ -43,6 +43,9 @@ class UpdateCheckTests(unittest.TestCase): "force_update": True, "download_url": "https://example.test/cmshopee.zip", "sha256": "abc", + "size_bytes": 123, + "package_format": "cmshopee-portable-v1", + "updater_protocol": 1, "message": "请升级后继续使用", } @@ -57,6 +60,8 @@ class UpdateCheckTests(unittest.TestCase): self.assertFalse(result.can_enter) self.assertEqual("1.2.0", result.latest_version) self.assertEqual("https://example.test/cmshopee.zip", result.download_url) + self.assertEqual(123, result.size_bytes) + self.assertEqual("cmshopee-portable-v1", result.package_format) def test_check_for_update_accepts_release_wrapper_response(self): def fetcher(_url, _timeout): diff --git a/tests/test_update_installer.py b/tests/test_update_installer.py new file mode 100644 index 0000000..bb36da8 --- /dev/null +++ b/tests/test_update_installer.py @@ -0,0 +1,178 @@ +import hashlib +import io +import json +import tempfile +import unittest +import zipfile +from pathlib import Path + +from app import release_manifest, update_installer + + +class FakeResponse(io.BytesIO): + def __init__(self, payload, url): + super().__init__(payload) + self.url = url + + def geturl(self): + return self.url + + def __enter__(self): + return self + + def __exit__(self, *_args): + self.close() + + +class FakeOpener: + def __init__(self, payload, url="https://updates.example.test/app.zip"): + self.payload = payload + self.url = url + self.calls = 0 + + def open(self, _request, timeout=None): + assert timeout + self.calls += 1 + return FakeResponse(self.payload, self.url) + + +class UpdateInstallerTests(unittest.TestCase): + trusted_hosts = {"updates.example.test"} + + def make_zip(self, root, version="1.2.3"): + package = Path(root) / "package" + (package / "_internal").mkdir(parents=True) + (package / "cmshopee.exe").write_bytes(b"new-exe") + (package / "_internal" / "runtime.dll").write_bytes(b"runtime") + (package / "version.txt").write_text(version, encoding="ascii") + (package / "README.txt").write_text("说明", encoding="utf-8") + release_manifest.write_package_manifest(package, version) + zip_path = Path(root) / "package.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive: + for path in package.rglob("*"): + if path.is_file(): + archive.write(path, path.relative_to(package).as_posix()) + return zip_path.read_bytes() + + def metadata(self, payload, **changes): + values = { + "version": "1.2.3", + "download_url": "https://updates.example.test/app.zip", + "sha256": hashlib.sha256(payload).hexdigest(), + "size_bytes": len(payload), + "package_format": release_manifest.PACKAGE_FORMAT, + "updater_protocol": release_manifest.UPDATER_PROTOCOL, + "min_updater_protocol": 1, + } + values.update(changes) + return update_installer.UpdatePackageMetadata(**values) + + def test_prepare_update_verifies_and_reuses_pending(self): + with tempfile.TemporaryDirectory() as temp_dir: + payload = self.make_zip(temp_dir) + install_root = Path(temp_dir) / "installed" + install_root.mkdir() + (install_root / "cmshopee.exe").write_bytes(b"old-exe") + (install_root / "data").mkdir() + opener = FakeOpener(payload) + + staged = update_installer.prepare_update( + self.metadata(payload), + install_root, + trusted_hosts=self.trusted_hosts, + opener=opener, + ) + reused = update_installer.prepare_update( + self.metadata(payload), + install_root, + trusted_hosts=self.trusted_hosts, + opener=opener, + ) + + self.assertEqual(1, opener.calls) + self.assertEqual(staged.staging_dir, reused.staging_dir) + self.assertEqual(b"old-exe", (install_root / "cmshopee.exe").read_bytes()) + self.assertTrue((install_root / "data").is_dir()) + pending = json.loads(staged.pending_path.read_text(encoding="utf-8")) + self.assertNotIn("download_url", pending) + + def test_download_rejects_hash_size_http_and_untrusted_host(self): + with tempfile.TemporaryDirectory() as temp_dir: + payload = b"zip" + root = Path(temp_dir) + cases = ( + self.metadata(payload, sha256=""), + self.metadata(payload, size_bytes=0), + self.metadata(payload, download_url="http://updates.example.test/app.zip"), + self.metadata(payload, download_url="https://other.example.test/app.zip"), + ) + for metadata in cases: + with self.subTest(metadata=metadata): + with self.assertRaises(update_installer.UpdateInstallError): + update_installer.download_package( + metadata, + root, + trusted_hosts=self.trusted_hosts, + opener=FakeOpener(payload), + ) + + def test_cancel_removes_partial_download(self): + with tempfile.TemporaryDirectory() as temp_dir: + payload = b"zip-content" + with self.assertRaises(update_installer.UpdateCancelled): + update_installer.download_package( + self.metadata(payload), + temp_dir, + trusted_hosts=self.trusted_hosts, + opener=FakeOpener(payload), + cancelled=lambda: True, + ) + self.assertFalse( + (Path(temp_dir) / ".cmshopee-update/downloads/1.2.3.zip.part").exists() + ) + + def test_safe_extract_rejects_path_traversal_case_duplicates_and_symlink(self): + with tempfile.TemporaryDirectory() as temp_dir: + for index, entries in enumerate( + ( + [("../escape.txt", b"x", None)], + [("same.txt", b"a", None), ("SAME.txt", b"b", None)], + [("link", b"target", (stat_mode := (0o120777 << 16)))], + ) + ): + zip_path = Path(temp_dir) / ("bad-%d.zip" % index) + with zipfile.ZipFile(zip_path, "w") as archive: + for name, content, external_attr in entries: + info = zipfile.ZipInfo(name) + if external_attr is not None: + info.create_system = 3 + info.external_attr = stat_mode + archive.writestr(info, content) + with self.assertRaises(update_installer.UpdateInstallError): + update_installer.safe_extract(zip_path, Path(temp_dir) / ("out-%d" % index)) + + def test_staging_rejects_manifest_tampering_and_extra_files(self): + with tempfile.TemporaryDirectory() as temp_dir: + payload = self.make_zip(temp_dir) + metadata = self.metadata(payload) + package = Path(temp_dir) / "package" + (package / "cmshopee.exe").write_bytes(b"tampered") + with self.assertRaises(update_installer.UpdateInstallError): + update_installer.validate_staging(package, metadata) + + (package / "cmshopee.exe").write_bytes(b"new-exe") + (package / "extra.bin").write_bytes(b"extra") + with self.assertRaises(update_installer.UpdateInstallError): + update_installer.validate_staging(package, metadata) + + def test_zip_bomb_ratio_is_rejected(self): + with tempfile.TemporaryDirectory() as temp_dir: + zip_path = Path(temp_dir) / "bomb.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive: + archive.writestr("large.bin", b"0" * 1024 * 1024) + with self.assertRaises(update_installer.UpdateInstallError): + update_installer.safe_extract(zip_path, Path(temp_dir) / "out") + + +if __name__ == "__main__": + unittest.main()