"""自动升级包的安全下载、解压和暂存校验。""" 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("新版安装包清单包含重复路径") try: size_bytes = int(item["size_bytes"]) except (KeyError, TypeError, ValueError) as exc: raise UpdateInstallError("新版安装包清单文件大小无效") from exc if size_bytes < 0: raise UpdateInstallError("新版安装包清单文件大小无效") declared[key] = (relative, size_bytes, 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, stage_callback=None, **download_kwargs): validate_metadata(metadata, download_kwargs.get("trusted_hosts", DEFAULT_TRUSTED_HOSTS)) if stage_callback: stage_callback("正在检查已下载的新版") reusable = load_verified_pending(install_root, metadata) if reusable is not None: if stage_callback: stage_callback("新版已经完成校验") return reusable update_root = _update_root(install_root) if stage_callback: stage_callback("正在下载新版") 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: if stage_callback: stage_callback("正在校验并准备新版") 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 if stage_callback: stage_callback("新版已经完成校验") return StagedUpdate(metadata.version, zip_path, staging_dir, pending_path, metadata.sha256)