ykxmz/windows/lanremote/updater.py

# -*- coding: utf-8 -*-
r'''被控端**静默自动更新**。

== 用户要的效果 ==

"手表端点击一个更新按钮，电脑端就自动进行静默更新，
 其他什么设置都不变 …… 不弹任何窗口出来。"

== 流程 ==

    1. 问中转：最新是哪版、在哪下、sha256 多少
       GET <中转>/v1/update/latest?platform=windows
    2. 和本机版本比，一样就直接返回"已是最新"
    3. 下载安装包到 %TEMP%\ykxmz-update\
    4. **校验 sha256**（下坏了/被换了都必须拦住）
    5. 写一个 .cmd 帮手，**分离进程**启动它，然后自己退出
    6. 帮手等本进程退出 -> 静默跑安装器 -> 重新拉起被控端

== 为什么要有第 5 步那个 .cmd ==

Windows 上**正在运行的 exe 文件是锁着的**，安装器覆盖不了。
所以必须"先退出、再安装、再起来"，而这三步不能由本进程做
（它一退出就没人执行后面的了）。

用 `DETACHED_PROCESS` 起的 .cmd 不依赖本进程存活，
本进程退出后它继续跑 —— 这是 Windows 自更新的标准做法。

== 配置为什么不会丢 ==

配置在 `%ProgramData%\ykxmz\data\`（见 config.py 的 data_dir），
**不在程序目录里**。安装器升级时只替换 `{app}` 下的文件，
不碰 ProgramData —— 所以中转地址、配对码、各种开关原样保留。

这一点是这个功能能做的前提。要是配置在程序目录里，
每次更新都会被清掉，那"设置不变"就无从谈起。

== 为什么 sha256 必须校验 ==

自更新的最大风险是"下到一半断了"或"下到的不是我们要的东西"，
装上去就是一个坏掉的程序，而且它还会自己再更新 —— 螺旋下去很难救。
sha256 对不上就**直接放弃这一轮**，保持现状（现状至少是能用的）。

== 安全边界（写清楚，免得以后自己忘了）==

这个功能**只能从中转站取包**，而中转站地址是用户自己配的。
所以：
  * 用户自己搭的中转，就只从那里更新 —— 不存在"被第三方投毒"
  * 中转上的包是用户自己上传的
  * 校验和不匹配就拒绝，不做"绕过校验"的降级路径

换句话说：**更新源 = 用户自己的服务器**，不是作者的服务器，
也不是任何写死的公网地址。这一点是刻意的。
'''
from __future__ import annotations

import hashlib
import json
import os
import subprocess
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

# 中转没有这个接口时（老版本服务端）会返回 404/其它，直接当"不支持"处理
TIMEOUT_CHECK = 20
TIMEOUT_DOWNLOAD = 600
CHUNK = 256 * 1024


class UpdateError(Exception):
    """更新过程中出的问题（会被上层翻成中文提示）。"""


def _ver_tuple(v: str) -> tuple:
    """'2.1.5' -> (2,1,5)。解析不了就当 (0,0,0)。"""
    try:
        parts = [int(x) for x in str(v).strip().split(".")]
        while len(parts) < 3:
            parts.append(0)
        return tuple(parts[:3])
    except (ValueError, AttributeError):
        return (0, 0, 0)


def _ua() -> str:
    # 用自家标识。Cloudflare 会把 Python-urllib 判为可疑客户端直接 403
    # （报 error code 1010），这个坑仓库里记过两次了。
    return "LanRemote-updater/1.0"


def check(relay_url: str, current_version: str,
          timeout: int = TIMEOUT_CHECK) -> dict:
    """问中转最新版。返回：

        {"supported": bool,     中转支持不支持这个接口
         "latest": "2.1.5",
         "url": "https://.../download/ykxmz-setup-2.1.5.exe",
         "name": "...", "size": 123, "sha256": "...",
         "needUpdate": bool}
    """
    base = (relay_url or "").strip().rstrip("/")
    if not base:
        raise UpdateError("还没配中转地址 —— 自动更新要从你自己的中转站取包")

    if not base.startswith(("http://", "https://")):
        base = "https://" + base

    url = base + "/v1/update/latest?platform=windows"
    req = urllib.request.Request(url)
    req.add_header("User-Agent", _ua())
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            return {"supported": False,
                    "error": "中转站版本太旧，不支持自动更新（先手动升级一次服务端）"}
        raise UpdateError("查询更新失败：HTTP %s" % exc.code)
    except Exception as exc:                              # noqa: BLE001
        raise UpdateError("连不上中转站：%s" % exc)

    if not data.get("ok"):
        return {"supported": True,
                "error": data.get("error") or "中转站上还没有可下载的安装包"}

    latest = str(data.get("version") or "")
    url_path = str(data.get("url") or "")
    if not latest or not url_path:
        raise UpdateError("中转返回的信息不完整")

    full = url_path if url_path.startswith("http") else base + url_path
    return {
        "supported": True,
        "latest": latest,
        "current": current_version,
        "url": full,
        "name": data.get("name") or "",
        "size": int(data.get("size") or 0),
        "sha256": (data.get("sha256") or "").lower(),
        "needUpdate": _ver_tuple(latest) > _ver_tuple(current_version),
    }


def download(url: str, dest_dir: Path, expect_sha256: str = "",
             expect_size: int = 0,
             progress=None) -> Path:
    """下载安装包并校验。返回下载好的路径。

    progress: 可选回调 (已下载字节, 总字节)
    """
    dest_dir.mkdir(parents=True, exist_ok=True)
    name = url.rsplit("/", 1)[-1].split("?")[0] or "ykxmz-update.exe"
    # 只留文件名，别让 URL 里的东西跑到路径里（防目录穿越）
    name = os.path.basename(urllib.parse.unquote(name))
    if not name or name.startswith("."):
        name = "ykxmz-update.exe"
    dest = dest_dir / name

    req = urllib.request.Request(url)
    req.add_header("User-Agent", _ua())
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT_DOWNLOAD) as resp:
            total = int(resp.headers.get("Content-Length") or 0)
            h = hashlib.sha256()
            got = 0
            with dest.open("wb") as f:
                while True:
                    chunk = resp.read(CHUNK)
                    if not chunk:
                        break
                    f.write(chunk)
                    h.update(chunk)
                    got += len(chunk)
                    if progress:
                        try:
                            progress(got, total)
                        except Exception:                 # noqa: BLE001
                            pass
    except Exception as exc:                              # noqa: BLE001
        try:
            dest.unlink()
        except OSError:
            pass
        raise UpdateError("下载失败：%s" % exc)

    if expect_size and got != expect_size:
        dest.unlink(missing_ok=True)
        raise UpdateError("下载大小对不上（期望 %d，实际 %d）"
                          % (expect_size, got))
    if expect_sha256 and h.hexdigest() != expect_sha256:
        dest.unlink(missing_ok=True)
        raise UpdateError("校验和不匹配 —— 包可能下坏了，已丢弃（保持当前版本）")

    return dest


WAITER_TEMPLATE = r'''@echo off
rem ===================================================================
rem  ykxmz 自动更新的"帮手"脚本
rem
rem  被控端自己要退出了，没法再做后面的事，所以把这三步交给这个脚本：
rem      1. 等被控端真的退出（不然安装器覆盖不了正在运行的 exe）
rem      2. 静默跑安装器
rem      3. 重新拉起被控端
rem
rem  它由被控端用 DETACHED_PROCESS 启动，不受被控端退出影响。
rem  跑完自己删掉。
rem ===================================================================
setlocal
set "PID={pid}"
set "SETUP={setup}"
set "APPEXE={appexe}"
set "LOGFILE={log}"

echo [%date% %time%] 等待进程 %PID% 退出... >> "%LOGFILE%"
set /a TRIES=0
:waitloop
tasklist /FI "PID eq %PID%" 2>nul | find "%PID%" >nul
if errorlevel 1 goto gone
set /a TRIES+=1
if %TRIES% GEQ 150 goto gone
timeout /t 2 /nobreak >nul
goto waitloop

:gone
echo [%date% %time%] 进程已退出，开始静默安装 >> "%LOGFILE%"
rem /VERYSILENT 全静默、不弹任何窗口
rem /SUPPRESSMSGBOXES 连错误框也不弹
rem /NORESTART 不自动重启电脑
rem 不带 /DIR，让它按注册表里记的上次安装位置原地升级
start "" /wait "%SETUP%" /VERYSILENT /SUPPRESSMSGBOXES /NORESTART
echo [%date% %time%] 安装器退出码 %ERRORLEVEL% >> "%LOGFILE%"

rem 给安装器一点时间把文件落完
timeout /t 3 /nobreak >nul

echo [%date% %time%] 重新拉起被控端 >> "%LOGFILE%"
if exist "%APPEXE%" start "" "%APPEXE%" --silent
timeout /t 3 /nobreak >nul

rem 清理：删掉安装包和自己
del /f /q "%SETUP%" >nul 2>&1
(goto) 2>nul & del /f /q "%~f0"
'''


def apply_update(installer: Path, app_exe: Path, pid: int = None) -> str:
    """写帮手脚本、分离启动它。调用方**紧接着必须自己退出**。

    返回日志文件路径（写进被控端日志，方便事后查）。
    """
    pid = pid or os.getpid()
    work = Path(tempfile.gettempdir()) / "ykxmz-update"
    work.mkdir(parents=True, exist_ok=True)
    log = work / "update.log"
    script = work / "do-update.cmd"

    body = (WAITER_TEMPLATE
            .replace("{pid}", str(pid))
            .replace("{setup}", str(installer))
            .replace("{appexe}", str(app_exe))
            .replace("{log}", str(log)))
    # .bat/.cmd 用系统 ANSI 编码（GBK），不能让中文变乱码
    script.write_bytes(body.replace("\n", "\r\n").encode("mbcs", "replace"))

    # DETACHED_PROCESS: 不继承控制台、不随父进程退出
    # CREATE_NEW_PROCESS_GROUP: 不受父进程的 Ctrl 事件影响
    flags = 0
    flags |= getattr(subprocess, "DETACHED_PROCESS", 0x00000008)
    flags |= getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200)
    subprocess.Popen(["cmd.exe", "/c", str(script)],
                     creationflags=flags,
                     stdin=subprocess.DEVNULL,
                     stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL,
                     close_fds=True)
    return str(log)


def cleanup_old_downloads(keep_days: int = 3) -> None:
    """清掉临时目录里过期的更新包（失败的残留）。"""
    work = Path(tempfile.gettempdir()) / "ykxmz-update"
    if not work.is_dir():
        return
    cutoff = time.time() - keep_days * 86400
    try:
        for f in work.iterdir():
            if f.is_file() and f.stat().st_mtime < cutoff:
                f.unlink()
    except OSError:
        pass