From 74b98cd12375eb7b6d6b5d6fd7466be621a077f9 Mon Sep 17 00:00:00 2001 From: chengma Date: Mon, 13 Jul 2026 16:19:03 +0800 Subject: [PATCH] feat(update): support verified public HTTP downloads --- app/gui/update_dialog.py | 7 ++ app/update_installer.py | 85 ++++++++++++++++------ docs/04-architecture.md | 2 +- docs/packaging.md | 4 +- docs/tasks/T-623.md | 7 +- docs/update-check.md | 9 +-- tests/test_gui.py | 31 ++++++++ tests/test_update_installer.py | 126 ++++++++++++++++++++++++++++++--- 8 files changed, 234 insertions(+), 37 deletions(-) diff --git a/app/gui/update_dialog.py b/app/gui/update_dialog.py index 253ff9a..90583d9 100644 --- a/app/gui/update_dialog.py +++ b/app/gui/update_dialog.py @@ -117,6 +117,13 @@ class ForcedUpdateDialog(QDialog): self.details_label = QLabel(details) self.details_label.setWordWrap(True) layout.addWidget(self.details_label) + self.http_warning_label = QLabel("当前使用临时 HTTP 通道下载，安装前仍会校验完整性") + self.http_warning_label.setWordWrap(True) + self.http_warning_label.setStyleSheet("color: #b45309;") + self.http_warning_label.setHidden( + not update_installer.uses_plain_http(self.result.download_url) + ) + layout.addWidget(self.http_warning_label) self.stage_label = QLabel("发现必须升级的新版本") layout.addWidget(self.stage_label) self.progress_bar = QProgressBar() diff --git a/app/update_installer.py b/app/update_installer.py index 7353091..b731fb2 100644 --- a/app/update_installer.py +++ b/app/update_installer.py @@ -8,6 +8,7 @@ import json import os import re import shutil +import socket import stat import urllib.parse import urllib.request @@ -30,7 +31,6 @@ 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", @@ -83,10 +83,10 @@ def metadata_from_update_info(info): return metadata -def validate_metadata(metadata, trusted_hosts=DEFAULT_TRUSTED_HOSTS): +def validate_metadata(metadata): if not re.fullmatch(r"\d+(?:\.\d+)*", metadata.version): raise UpdateInstallError("新版版本号格式不正确") - validate_download_url(metadata.download_url, trusted_hosts) + validate_download_url(metadata.download_url) 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: @@ -99,32 +99,74 @@ def validate_metadata(metadata, trusted_hosts=DEFAULT_TRUSTED_HOSTS): raise UpdateInstallError("当前更新器版本过低") -def validate_download_url(url, trusted_hosts=DEFAULT_TRUSTED_HOSTS): - parsed = urllib.parse.urlsplit(str(url or "")) +def _system_resolve_host(host, port): + try: + answers = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except OSError as exc: + raise UpdateInstallError("新版下载地址无法解析") from exc + return [answer[4][0] for answer in answers if answer[4]] + + +def _validate_public_addresses(addresses): + addresses = list(addresses or ()) + if not addresses: + raise UpdateInstallError("新版下载地址无法解析") + for value in addresses: + try: + address = ipaddress.ip_address(str(value).split("%", 1)[0]) + except ValueError as exc: + raise UpdateInstallError("新版下载地址解析结果无效") from exc + if not address.is_global: + raise UpdateInstallError("新版下载地址不能指向本机或内网") + + +def uses_plain_http(url): + try: + return urllib.parse.urlsplit(str(url or "")).scheme.lower() == "http" + except ValueError: + return False + + +def validate_download_url(url, *, resolver=None): + try: + parsed = urllib.parse.urlsplit(str(url or "")) + except ValueError as exc: + raise UpdateInstallError("新版下载地址格式不正确") from exc 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.scheme.lower() not in {"http", "https"} or not host: + raise UpdateInstallError("新版下载地址必须使用 HTTP 或 HTTPS") if parsed.username or parsed.password: raise UpdateInstallError("新版下载地址不能包含账号信息") + if host == "localhost" or host.endswith(".localhost") or host == "localhost.localdomain": + raise UpdateInstallError("新版下载地址不能指向本机或内网") + try: + port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80) + except ValueError as exc: + raise UpdateInstallError("新版下载地址端口无效") from exc 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("新版下载地址不在受信任域名内") + if address is not None: + _validate_public_addresses((address,)) + elif resolver is not None: + try: + addresses = resolver(host, port) + except UpdateInstallError: + raise + except Exception as exc: + raise UpdateInstallError("新版下载地址无法解析") from exc + _validate_public_addresses(addresses) return parsed class _SafeRedirectHandler(urllib.request.HTTPRedirectHandler): - def __init__(self, trusted_hosts): + def __init__(self, resolver): super().__init__() - self.trusted_hosts = trusted_hosts + self.resolver = resolver def redirect_request(self, req, fp, code, msg, headers, newurl): - validate_download_url(newurl, self.trusted_hosts) + validate_download_url(newurl, resolver=self.resolver) return super().redirect_request(req, fp, code, msg, headers, newurl) @@ -143,14 +185,15 @@ def download_package( metadata, install_root, *, - trusted_hosts=DEFAULT_TRUSTED_HOSTS, + resolver=None, opener=None, cancelled=None, progress=None, timeout=66, ): - validate_metadata(metadata, trusted_hosts) - validate_download_url(metadata.download_url, trusted_hosts) + validate_metadata(metadata) + resolver = resolver or _system_resolve_host + validate_download_url(metadata.download_url, resolver=resolver) downloads = _update_root(install_root) / "downloads" downloads.mkdir(parents=True, exist_ok=True) part_path = downloads / (metadata.version + ".zip.part") @@ -165,13 +208,13 @@ def download_package( headers={"Accept": "application/zip", "User-Agent": "cmshopee-updater/1"}, ) if opener is None: - opener = urllib.request.build_opener(_SafeRedirectHandler(trusted_hosts)) + opener = urllib.request.build_opener(_SafeRedirectHandler(resolver)) 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) + validate_download_url(response.geturl(), resolver=resolver) while True: if cancelled and cancelled(): raise UpdateCancelled("已取消下载新版") @@ -384,7 +427,7 @@ def load_verified_pending(install_root, metadata): def prepare_update(metadata, install_root, stage_callback=None, **download_kwargs): - validate_metadata(metadata, download_kwargs.get("trusted_hosts", DEFAULT_TRUSTED_HOSTS)) + validate_metadata(metadata) if stage_callback: stage_callback("正在检查已下载的新版") reusable = load_verified_pending(install_root, metadata) diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 9b3ae6a..58d90b4 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -531,7 +531,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`；此阶段不替换任何运行中程序文件。 +- T-616 的下载暂存根固定为安装目录下 `.cmshopee-update/`，与 `data/` 完全隔离。T-623 为兼容对象存储/CDN外链，临时允许任意公网 HTTP/HTTPS 域名或公网 IP 及跨域、跨协议重定向，但版本接口继续使用受信任 HTTPS；初始地址、每次跳转和最终地址均拒绝本机、内网及非 HTTP(S) 目标。远程zip必须通过声明大小、整包SHA-256、安全zip路径和包内manifest逐文件校验，才写 `pending.json`；HTTP 不降低任何安装校验，也不替换运行中程序文件。 - T-617 的独立 `cmshopee-updater.exe` 必须先复制到系统临时目录运行，并等待主程序退出。替换粒度是manifest允许的程序根项目，旧根先整体移动到同盘backup，新根再整体移入；事务锁防止并发更新，journal记录每次移动，任一步失败逆序恢复。`data/`、更新管理目录与未知安装根项目永不进入替换清单。 - T-618 在创建 `MainWindow` 前显示强制升级 `QDialog`；下载与校验只能在 `QObject + QThread` worker中执行，线程结束前保留引用，取消时等待part清理。独立更新器进程成功创建后才退出旧主程序；强制版本后续失败保持阻断，只有版本接口本身不可达/非法继续失败放行。 - T-619 要求新版按事务写 `process_started`、`main_window_ready`、`environment_blocked` 健康标记。更新器只在主窗口就绪后认定升级成功并清理成功备份；无标记/早期退出/超时则事务回滚并按目标版本+包hash熔断。环境阻断保留新版，不把本地数据目录、配置或Chrome问题误判成坏发布包。 diff --git a/docs/packaging.md b/docs/packaging.md index aa8d643..2fd55cb 100644 --- a/docs/packaging.md +++ b/docs/packaging.md @@ -176,7 +176,7 @@ 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/` 不变。 +T-616 提供无Qt依赖的安全下载暂存层 `app/update_installer.py`。安装包下载到 `<安装目录>/.cmshopee-update/downloads/`，完成大小与zip SHA-256校验后安全解压到同盘 `staging/`；路径穿越、符号链接、大小写重复、Windows保留名、ADS、异常压缩比、文件数/解压总量超限，以及manifest缺失或逐文件hash不一致都会拒绝。完整验证后才原子写 `pending.json`，当前程序目录和 `data/` 不变。 T-617 增加独立 `cmshopee-updater.exe`。构建脚本用 `cmshopee-updater.spec` 生成无控制台单文件更新器并放进release和manifest。执行更新前，主程序把它复制到系统临时目录；更新器有上限地等待主程序退出，再按manifest白名单将旧程序根项目整体移动到 `.cmshopee-update/backup/`，把已验证暂存根项目移入安装目录。每一步写事务journal，移动或新版启动失败时逆序恢复；`data/` 与安装根未知文件不扫描、不移动、不删除。 @@ -184,6 +184,8 @@ T-618 已把安全暂存和独立更新器接回启动门禁。强制升级窗 T-619 增加启动健康确认与失败熔断。新版带事务参数启动，在 `.cmshopee-update/transactions/<事务>/health.json` 原子写 `process_started`、`main_window_ready` 或 `environment_blocked`。更新器在主窗口就绪前保留旧根项目；早期退出/超时会回滚并按版本+zip hash写 `failed-versions.json`，防止同一坏包无限循环。主窗口就绪后清理pending、zip、staging和本次成功备份；环境阻断保留新版与备份。备份清理只处理 `.cmshopee-update/backup/` 且保留最近两份，不扫描 `data/`。 +T-623 为外部对象存储/CDN下载链接增加临时 HTTP 兼容。版本检查接口仍固定使用受信任 HTTPS；安装包 `download_url` 不限制域名，可以使用任意公网 HTTP/HTTPS 域名或公网 IP，并允许跨域、跨协议重定向。初始地址、每次跳转和最终响应都会拒绝本机、内网、账号信息和非 HTTP(S) 协议。HTTP 不会降低安装校验标准：声明大小、整包 SHA-256、安全解压、包内manifest、事务替换和健康回滚全部照常执行；强制升级窗口会显示临时 HTTP 通道提示。服务端外链支持 HTTPS 后应删除 HTTP 分支。 + ## 四、绝不打包的本地数据 发布包里不能包含以下本地数据、密钥、业务数据或登录态： diff --git a/docs/tasks/T-623.md b/docs/tasks/T-623.md index 6ae9a6c..6ca63f9 100644 --- a/docs/tasks/T-623.md +++ b/docs/tasks/T-623.md @@ -3,7 +3,7 @@ id: T-623 title: 自动升级受限 HTTP 临时兼容 phase: 8 deps: [T-619] -status: TODO +status: DONE created: 2026-07-13 --- @@ -46,4 +46,7 @@ created: 2026-07-13 ## 执行记录 -- 尚未执行。 +- 2026-07-13：自动升级安装包地址由固定 HTTPS 域名白名单改为任意公网 HTTP/HTTPS 外链；初始地址、跨域/跨协议重定向和最终响应统一解析并拒绝本机、内网、保留地址、账号信息及非 HTTP(S) 协议。 +- 2026-07-13：保留声明大小、整包 SHA-256、安全解压、包内manifest、事务替换和健康回滚全部校验；强制升级窗口只在初始地址为 HTTP 时显示中文临时通道提示。 +- 2026-07-13：补齐公网域名、公网 IP、HTTP 完整暂存、跨域/跨协议跳转、最终地址复核、内网拒绝、hash 篡改和 GUI 提示测试，并同步架构、版本接口与打包文档。 +- 2026-07-13：为隔离工作区既有提示词/版本号改动，在独立干净 worktree 验证通过：Ruff、Python 3.10 compileall、完整 unittest 436 项及 `git diff --check` 全绿；未连接真实发布外链执行在线升级。 diff --git a/docs/update-check.md b/docs/update-check.md index 37fe8c0..74a3efc 100644 --- a/docs/update-check.md +++ b/docs/update-check.md @@ -22,7 +22,7 @@ | `version` / `latest_version` | string | 最新版本号。客户端取 `payload.latest_version` 或 `release.version`。**`latest_version` 与 `min_supported_version` 至少要有一个**，否则客户端判为"接口缺字段"、按失败放行 | | `force_update` | bool | 是否强制升级。`true`/`1`/`yes`/`y`/`on`（大小写不敏感）都识别为真 | | `min_supported_version` | string | 最低支持版本；本地低于它一律强制 | -| `download_url` | string(URL) | 下载地址；自动安装时必须是 HTTPS | +| `download_url` | string(URL) | 安装包外链；优先 HTTPS，T-623 临时兼容任意公网 HTTP/HTTPS 地址 | | `sha256` | string | zip 的 SHA-256；自动安装时必须是 64 位十六进制且不可为空 | | `size_bytes` | integer | zip 的准确字节数；自动安装时必须大于 0 | | `package_format` | string | 自动安装固定为 `cmshopee-portable-v1` | @@ -49,13 +49,13 @@ - **非强制**：**不弹任何提示**，直接进主界面（当前无"温和可跳过提示"分支；如需另立任务）。 - **失败放行**：接口断网、超时、返回非法 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完成。 +T-616 已实现安全暂存层，T-623 为兼容对象存储、CDN 和文件分发服务外链，临时允许任意公网 HTTP/HTTPS 下载地址及跨域、跨协议重定向。初始地址、每次跳转和最终响应都会拒绝本机、内网和非 HTTP(S) 目标；下载仍流式写入安装目录 `.cmshopee-update/`，并严格校验zip声明大小、SHA-256和包内manifest。任一步失败都不修改当前程序或 `data/`。 T-617 已提供独立无控制台更新器和事务回滚能力：更新器从系统临时目录运行，旧主程序退出后才切换程序根项目，并在新版进程无法创建时恢复旧版。T-618 已将其接入启动强制升级进度窗口。 ## 五、发版约定（服务端据此控制） -自动安装使用的全部字段必须放在同一个 `release` 对象内，不得把版本取自一个对象、hash 取自另一个对象。构建脚本生成的 `release/release-metadata.json` 是服务端录入模板；发布人员只补 HTTPS `download_url`、强制策略和中文发布说明，不得手工改写 hash、大小、包格式或协议版本。 +自动安装使用的全部字段必须放在同一个 `release` 对象内，不得把版本取自一个对象、hash 取自另一个对象。构建脚本生成的 `release/release-metadata.json` 是服务端录入模板；发布人员补公网 HTTP/HTTPS `download_url`、强制策略和中文发布说明，不得手工改写 hash、大小、包格式或协议版本。正式发布仍应优先使用 HTTPS；HTTP 只用于外链暂时无法提供 HTTPS 的兼容期。 | 想要的效果 | 服务端返回 | | --- | --- | @@ -119,8 +119,9 @@ T-617 已提供独立无控制台更新器和事务回滚能力：更新器从 - 服务端返回 `force_update`/`min_supported_version` 属**本仓库外的服务端配置**；客户端代码（`app/update_check.py`/`app/version.py`/启动流程）已按本契约实现，**发强制版不需要改客户端**。 - `download_url` 里的中文可用 URL 编码（如 `%E8%9D%A6...`），系统浏览器能正确打开。 +- 版本检查接口 `APP_UPDATE_CHECK_URL` 必须保持受信任 HTTPS；只有它返回的 `download_url` 可以在 T-623 兼容期使用公网 HTTP 外链。客户端不提供手工输入下载地址或关闭校验的入口。 - 若将来要"非强制也温和提示（可跳过、不阻断）"，需在客户端加一个非强制分支，另立任务。 -- 空 `sha256` 只兼容 T-544 的人工下载引导，绝不能进入自动安装。自动安装还必须同时校验 HTTPS、`size_bytes`、包格式和更新器协议。 +- 空 `sha256` 只兼容 T-544 的人工下载引导，绝不能进入自动安装。自动安装无论使用 HTTP 还是 HTTPS，都必须同时校验 `size_bytes`、整包 SHA-256、包格式、更新器协议和包内manifest。 - `manifest_signature` 与 `signature_algorithm` 是预留字段；当前未启用数字签名，不能将 SHA-256 描述为发布者身份认证。 ## 八、引导版本与灰度发布 diff --git a/tests/test_gui.py b/tests/test_gui.py index a8e472f..e4c734a 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -1739,6 +1739,37 @@ class GuiTests(TempDirMixin, unittest.TestCase): self.assertEqual("重试", dialog.action_button.text()) self.assertIsNone(dialog.thread) + def test_forced_update_dialog_only_warns_for_plain_http_download(self): + common = { + "current_version": "1.0.0", + "checked": True, + "forced": True, + "latest_version": "1.2.0", + "sha256": "a" * 64, + "size_bytes": 100, + "package_format": "cmshopee-portable-v1", + "updater_protocol": 1, + "min_updater_protocol": 1, + } + http_dialog = ForcedUpdateDialog( + update_check.UpdateCheckResult( + download_url="http://external.example.test/cmshopee.zip", + **common, + ) + ) + https_dialog = ForcedUpdateDialog( + update_check.UpdateCheckResult( + download_url="https://external.example.test/cmshopee.zip", + **common, + ) + ) + self.addCleanup(http_dialog.close) + self.addCleanup(https_dialog.close) + + self.assertFalse(http_dialog.http_warning_label.isHidden()) + self.assertIn("临时 HTTP 通道", http_dialog.http_warning_label.text()) + self.assertTrue(https_dialog.http_warning_label.isHidden()) + def test_startup_update_gate_fuses_a_previously_failed_release(self): captured = [] diff --git a/tests/test_update_installer.py b/tests/test_update_installer.py index 5d5a507..33cfedb 100644 --- a/tests/test_update_installer.py +++ b/tests/test_update_installer.py @@ -3,6 +3,7 @@ import io import json import tempfile import unittest +import urllib.request import zipfile from pathlib import Path @@ -37,7 +38,9 @@ class FakeOpener: class UpdateInstallerTests(unittest.TestCase): - trusted_hosts = {"updates.example.test"} + @staticmethod + def public_resolver(_host, _port): + return ["93.184.216.34"] def make_zip(self, root, version="1.2.3"): package = Path(root) / "package" @@ -80,13 +83,13 @@ class UpdateInstallerTests(unittest.TestCase): staged = update_installer.prepare_update( self.metadata(payload), install_root, - trusted_hosts=self.trusted_hosts, + resolver=self.public_resolver, opener=opener, ) reused = update_installer.prepare_update( self.metadata(payload), install_root, - trusted_hosts=self.trusted_hosts, + resolver=self.public_resolver, opener=opener, ) @@ -97,15 +100,41 @@ class UpdateInstallerTests(unittest.TestCase): 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): + def test_prepare_update_accepts_public_http_and_writes_verified_pending(self): + with tempfile.TemporaryDirectory() as temp_dir: + payload = self.make_zip(temp_dir) + install_root = Path(temp_dir) / "installed" + install_root.mkdir() + metadata = self.metadata( + payload, + download_url="http://downloads.external.test/releases/app.zip", + ) + + staged = update_installer.prepare_update( + metadata, + install_root, + resolver=self.public_resolver, + opener=FakeOpener(payload, url=metadata.download_url), + ) + + pending = json.loads(staged.pending_path.read_text(encoding="utf-8")) + self.assertEqual("verified", pending["stage"]) + self.assertEqual(metadata.sha256, pending["sha256"]) + + def test_download_rejects_invalid_metadata_and_unsafe_urls(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"), + self.metadata(payload, download_url="file:///tmp/app.zip"), + self.metadata(payload, download_url="http://user:secret@example.test/app.zip"), + self.metadata(payload, download_url="http://127.0.0.1/app.zip"), + self.metadata(payload, download_url="http://10.0.0.8/app.zip"), + self.metadata(payload, download_url="http://[::1]/app.zip"), + self.metadata(payload, download_url="http://localhost/app.zip"), + self.metadata(payload, download_url="http://[invalid/app.zip"), ) for metadata in cases: with self.subTest(metadata=metadata): @@ -113,10 +142,91 @@ class UpdateInstallerTests(unittest.TestCase): update_installer.download_package( metadata, root, - trusted_hosts=self.trusted_hosts, + resolver=self.public_resolver, opener=FakeOpener(payload), ) + def test_download_allows_public_http_https_domains_and_public_ip(self): + payload = b"zip" + urls = ( + "http://cdn-one.example.test:8080/releases/app.zip", + "https://storage.other.test/app.zip?token=hidden", + "http://8.8.8.8/app.zip", + ) + with tempfile.TemporaryDirectory() as temp_dir: + for index, url in enumerate(urls): + with self.subTest(url=url): + root = Path(temp_dir) / str(index) + downloaded = update_installer.download_package( + self.metadata(payload, download_url=url), + root, + resolver=self.public_resolver, + opener=FakeOpener(payload, url=url), + ) + self.assertEqual(payload, downloaded.read_bytes()) + + def test_download_rejects_domain_resolving_to_non_public_address(self): + payload = b"zip" + with tempfile.TemporaryDirectory() as temp_dir: + with self.assertRaisesRegex(update_installer.UpdateInstallError, "本机或内网"): + update_installer.download_package( + self.metadata(payload, download_url="http://files.example.test/app.zip"), + temp_dir, + resolver=lambda _host, _port: ["192.168.1.20"], + opener=FakeOpener(payload, url="http://files.example.test/app.zip"), + ) + + def test_redirect_handler_allows_public_cross_domain_and_protocol_redirects(self): + handler = update_installer._SafeRedirectHandler(self.public_resolver) + request = urllib.request.Request("https://origin.example.test/app.zip") + for target in ( + "http://cdn.example.test:8080/releases/app.zip", + "https://objects.other.test/app.zip", + ): + with self.subTest(target=target): + redirected = handler.redirect_request(request, None, 302, "Found", {}, target) + self.assertEqual(target, redirected.full_url) + + with self.assertRaisesRegex(update_installer.UpdateInstallError, "本机或内网"): + handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "http://169.254.169.254/latest/meta-data", + ) + + def test_download_revalidates_final_response_url(self): + payload = b"zip" + with tempfile.TemporaryDirectory() as temp_dir: + with self.assertRaisesRegex(update_installer.UpdateInstallError, "本机或内网"): + update_installer.download_package( + self.metadata(payload, download_url="http://cdn.example.test/app.zip"), + temp_dir, + resolver=self.public_resolver, + opener=FakeOpener(payload, url="http://127.0.0.1/app.zip"), + ) + + def test_http_download_still_requires_matching_hash(self): + payload = b"zip" + metadata = self.metadata( + payload, + download_url="http://cdn.example.test/app.zip", + sha256="0" * 64, + ) + with tempfile.TemporaryDirectory() as temp_dir: + with self.assertRaisesRegex(update_installer.UpdateInstallError, "完整性校验失败"): + update_installer.download_package( + metadata, + temp_dir, + resolver=self.public_resolver, + opener=FakeOpener(payload, url=metadata.download_url), + ) + self.assertFalse( + (Path(temp_dir) / ".cmshopee-update/downloads/1.2.3.zip").exists() + ) + def test_cancel_removes_partial_download(self): with tempfile.TemporaryDirectory() as temp_dir: payload = b"zip-content" @@ -124,7 +234,7 @@ class UpdateInstallerTests(unittest.TestCase): update_installer.download_package( self.metadata(payload), temp_dir, - trusted_hosts=self.trusted_hosts, + resolver=self.public_resolver, opener=FakeOpener(payload), cancelled=lambda: True, )