"""自动升级包的安全下载、解压和暂存校验。""" from __future__ import annotations import hashlib import ipaddress import json import os import re import shutil import socket 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 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=PACKAGE_FORMAT, updater_protocol=UPDATER_PROTOCOL, min_updater_protocol=UPDATER_PROTOCOL, ) validate_metadata(metadata) return metadata def validate_metadata(metadata): if not re.fullmatch(r"\d+(?:\.\d+)*", metadata.version): raise UpdateInstallError("新版版本号格式不正确") 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: 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 _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(".") 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: _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, resolver): super().__init__() self.resolver = resolver def redirect_request(self, req, fp, code, msg, headers, newurl): validate_download_url(newurl, resolver=self.resolver) 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, *, resolver=None, opener=None, cancelled=None, progress=None, timeout=66, ): 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") 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(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(), resolver=resolver) 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) 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)