From 3b051f43d8755adc58b1fbd510a5ad13cb577fc9 Mon Sep 17 00:00:00 2001 From: wangzhizhou <824219521@qq.com> Date: Tue, 20 Oct 2026 19:53:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(auth):=20auth.json=20=E8=B7=A8=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E5=86=99=E9=94=81=20+=20runtime=20=E7=9C=9F=E8=BA=AB?= =?UTF-8?q?=E6=8D=9F=E5=9D=8F=E5=9B=9E=E6=BB=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两件共享契约加固（规范见父仓库 `frontends.md` F4.1 与 `install-update.md` §10.1）。 **1. `auth.json` 跨进程写锁（F4.1）** 刷新令牌是「读 → 联网换新 → 写」的读改写，两个前端同时刷会让后者的 refresh token 立刻失效 （用户被登出）。新增 `orzmc/infra/filelock.py`：`/auth.lock` **目录锁** （`mkdir` 在 POSIX/Windows 都原子；陈旧 30s 自动破除；等 10s 拿不到 → `E_AUTH_LOCKED`）。 算法与 Swift 侧**逐字一致**（用 flock 不行：`flock` 与 `fcntl` 锁跨语言互不可见，Windows 又没有 flock）。 `Authenticator.ensure()` 改为：**持锁 → 锁内重读 → 需要才刷新**。锁内重读到另一个前端刚写好的有效会话时 直接复用，不再刷一次（否则把它的 refresh token 轮换掉）。 **2. `runtime` 真身损坏回滚（§10.1）** - 安装时把旧真身改名为 `.bak` 再原子替换，替换失败还原； - 运行前对**自己的副本**做魔数预检，坏了且 `.bak` 完好 → 回滚并继续（记 warn）； - `exec` 报不可执行（EACCES/ENOEXEC/…）→ 用 `.bak` 回滚后**重试一次**，仍失败报 `E_RUNTIME` 并提示重装； - 顺手修：`runtime install` 之前对 `channel=wrapper` 自己的副本也误判为"可复用"→ 永远无法升级； 现在只对**别人装的**原生 Swift 安装 no-op。 测试：新增 `orzmc/tests/test_auth_lock.py`（7 例：获取/owner/争用超时/陈旧破除/新鲜不抢/锁路径/ 锁内重读复用/超时映射 E_AUTH_LOCKED）与 `TestPayloadRollback`（3 例：备份生成/坏文件回滚/无备份不静默删）。 `ruff`/`mypy`/`pytest`（全量）与共享验收 25/25 全绿。 --- orzmc/orzmc/infra/auth_store.py | 9 ++ orzmc/orzmc/infra/filelock.py | 154 +++++++++++++++++++++++ orzmc/orzmc/services/auth.py | 22 +++- orzmc/orzmc/services/runtime.py | 89 ++++++++++++-- orzmc/tests/test_auth_lock.py | 211 ++++++++++++++++++++++++++++++++ orzmc/tests/test_runtime.py | 58 +++++++++ 6 files changed, 530 insertions(+), 13 deletions(-) create mode 100644 orzmc/orzmc/infra/filelock.py create mode 100644 orzmc/tests/test_auth_lock.py diff --git a/orzmc/orzmc/infra/auth_store.py b/orzmc/orzmc/infra/auth_store.py index 3080130..ffdc705 100644 --- a/orzmc/orzmc/infra/auth_store.py +++ b/orzmc/orzmc/infra/auth_store.py @@ -10,6 +10,7 @@ import os from orzmc.domain.auth import AuthSession +from orzmc.infra.filelock import DEFAULT_TIMEOUT, STALE_AFTER_SECONDS, FileLock, lock_path from orzmc.infra.fs import FileStore, default_state_dir FILENAME = "auth.json" @@ -46,3 +47,11 @@ def save(self, session: AuthSession, client_id: str) -> None: def clear(self) -> None: self._fs.remove(self._path) + + def lock(self, *, timeout: float = DEFAULT_TIMEOUT, stale_after: float = STALE_AFTER_SECONDS) -> FileLock: + """跨进程锁 (``frontends.md`` F4.1): 刷新令牌的整个「读->换新->写」要持它. + + 与 Swift 侧算法逐字一致 (``/auth.lock`` 目录锁); 拿不到锁会抛 + :class:`~orzmc.infra.filelock.LockTimeout` (``E_AUTH_LOCKED``). + """ + return FileLock(lock_path(os.path.dirname(self._path)), timeout=timeout, stale_after=stale_after) diff --git a/orzmc/orzmc/infra/filelock.py b/orzmc/orzmc/infra/filelock.py new file mode 100644 index 0000000..76e89d2 --- /dev/null +++ b/orzmc/orzmc/infra/filelock.py @@ -0,0 +1,154 @@ +"""跨进程文件锁(`frontends.md` F4.1):`/auth.lock` 目录锁。 + +**算法必须与 Swift 侧逐字一致**,否则两边互不加锁。核心只有三点: + +1. **获取**:`os.mkdir(lock)` —— 在 POSIX 与 Windows 上都是原子的; +2. **陈旧检测**:目录 `mtime` 超过 `stale_after`(默认 30s)视为持锁进程已崩,删掉重试; +3. **超时**:最多等 `timeout`(默认 10s),拿不到就报 ``E_AUTH_LOCKED``—— + **绝不**在没拿到锁的情况下进临界区(令牌刷新会在网络请求期间持有它)。 + +为什么不用 `flock`:`flock` 与 `fcntl` 锁在 Linux/macOS 上属于不同命名空间,跨语言极易"看起来加了锁, +其实互不可见";Windows 又没有 `flock`。`mkdir` 三平台语义一致,代价是崩溃后要陈旧检测兜底。 +""" + +from __future__ import annotations + +import os +import time +from types import TracebackType + +STALE_AFTER_SECONDS = 30.0 +DEFAULT_TIMEOUT = 10.0 +POLL_INTERVAL = 0.05 + + +class LockTimeout(RuntimeError): + """拿不到锁(规范错误码 ``E_AUTH_LOCKED``)。""" + + code = "E_AUTH_LOCKED" + + +def lock_path(state_dir: str) -> str: + """锁目录 `/auth.lock`(**目录**,靠 `mkdir` 原子性)。""" + return os.path.join(state_dir, "auth.lock") + + +def _age_seconds(path: str, now: float | None = None) -> float: + try: + mtime = os.stat(path).st_mtime + except OSError: + return 0.0 + return (now if now is not None else time.time()) - mtime + + +def _is_stale(path: str, stale_after: float, now: float | None = None) -> bool: + return _age_seconds(path, now) > stale_after + + +def _break_stale(path: str) -> None: + """清掉陈旧锁目录(先删 owner,再 rmdir;失败就算了,下轮还会重试)。""" + try: + owner = os.path.join(path, "owner") + if os.path.exists(owner): + os.remove(owner) + os.rmdir(path) + except OSError: + pass + + +class FileLock: + """目录锁上下文管理器(`with FileLock(path): ...`)。""" + + def __init__( + self, + path: str, + *, + timeout: float = DEFAULT_TIMEOUT, + stale_after: float = STALE_AFTER_SECONDS, + sleep=time.sleep, + clock=time.time, + ) -> None: + self.path = path + self.timeout = timeout + self.stale_after = stale_after + self._sleep = sleep + self._clock = clock + self._held = False + + # ── 获取 / 释放 ──────────────────────────────────────────────────────── + + def acquire(self) -> None: + deadline = self._clock() + self.timeout + while True: + try: + os.mkdir(self.path) + except FileExistsError: + if _is_stale(self.path, self.stale_after, self._clock()): + _break_stale(self.path) + continue + if self._clock() >= deadline: + raise LockTimeout( + f"拿不到认证锁 {self.path}(另一进程正在刷新令牌);稍后重试,或确认没有卡住的 orzmc 进程", + ) from None + self._sleep(POLL_INTERVAL) + continue + except OSError as exc: # 权限 / 磁盘满等:不该静默继续 + raise LockTimeout(f"无法创建认证锁 {self.path}: {exc}") from exc + self._write_owner() + self._held = True + return + + def _write_owner(self) -> None: + """写持有者信息(pid + 起始时间)——仅供排查,不参与判定。""" + try: + with open(os.path.join(self.path, "owner"), "w", encoding="utf-8") as handle: + handle.write(f"pid={os.getpid()}\nsince={_iso_now()}\n") + except OSError: + pass + + def release(self) -> None: + if not self._held: + return + self._held = False + try: + owner = os.path.join(self.path, "owner") + if os.path.exists(owner): + os.remove(owner) + except OSError: + pass + try: + os.rmdir(self.path) + except OSError: + # 释放失败不抛:陈旧检测会在 30s 后兜底。 + pass + + # ── 上下文管理器 ────────────────────────────────────────────────────── + + def __enter__(self) -> FileLock: + self.acquire() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + self.release() + + +def _iso_now() -> str: + """UTC ISO8601(秒精度)——写进 ``owner`` 供排查,不参与判定。""" + from datetime import datetime, timezone + + return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + + +__all__ = [ + "DEFAULT_TIMEOUT", + "POLL_INTERVAL", + "STALE_AFTER_SECONDS", + "FileLock", + "LockTimeout", + "lock_path", +] diff --git a/orzmc/orzmc/services/auth.py b/orzmc/orzmc/services/auth.py index ae3e064..fde8e72 100644 --- a/orzmc/orzmc/services/auth.py +++ b/orzmc/orzmc/services/auth.py @@ -52,16 +52,26 @@ def __init__( # ── 对外入口 ──────────────────────────────────────────────────────────── def ensure(self) -> AuthSession: - """返回一个 MC 令牌有效的会话:缓存 → 刷新 → 设备码登录。""" + """返回一个 MC 令牌有效的会话:缓存 → 刷新 → 设备码登录。 + + **刷新必须持跨进程锁, 并在锁内先重读** (`frontends.md` F4/F4.1): + 刷新会轮换 refresh token,两个前端同时刷会让后者的令牌立刻失效(用户被登出)。 + """ session = self._store.load() if session is not None and auth_map.token_state(session.expires_at, self._clock()).state == "valid": return replace(session, source="cache") if session is not None: - try: - return self.refresh(session.msa) - except AuthError: - # refresh token 被拒(如用户改密/撤销授权)→ 清掉重登,不吞错。 - self._store.clear() + with self._store.lock(): + # 锁内**重读**:另一个前端可能刚刷完,直接用它的结果,别再刷一次。 + current = self._store.load() + if current is not None and auth_map.token_state(current.expires_at, self._clock()).state == "valid": + return replace(current, source="cache") + target = current if current is not None else session + try: + return self.refresh(target.msa) + except AuthError: + # refresh token 被拒(如用户改密/撤销授权)→ 清掉重登,不吞错。 + self._store.clear() return self.login() def login(self) -> AuthSession: diff --git a/orzmc/orzmc/services/runtime.py b/orzmc/orzmc/services/runtime.py index a0331d2..af6aaa8 100644 --- a/orzmc/orzmc/services/runtime.py +++ b/orzmc/orzmc/services/runtime.py @@ -286,13 +286,68 @@ def resolve(self) -> tuple[str, str]: record = self._record() recorded = record.get("binary", "") source = record.get("source", "") - if recorded and _is_swift_payload(source, record.get("channel", "")) and self._runnable(recorded): - return recorded, source own = self.path + if recorded and _is_swift_payload(source, record.get("channel", "")) and self._runnable(recorded): + if recorded != own: + # 别人装的 Swift 原生版:直接用(它的备份与更新由那条通道负责,我们不插手)。 + return recorded, source + # 是**我们自己的副本**:跑之前做魔数预检,坏了就用 .bak 回滚(§10.1)。 + usable = self._verified_payload(recorded) + if usable is not None: + return usable, source + return own, source # 坏且无备份:交给 forward 报 E_RUNTIME 并提示重装 if self._fs.is_file(own): - return own, (source if recorded == own else f"github:{REPO}") + usable = self._verified_payload(own) + if usable is not None: + return usable, (source if recorded == own else f"github:{REPO}") return own, f"github:{REPO}" + @property + def backup_path(self) -> str: + """旧真身的备份路径 (同一目录 rename, 跨盘不会发生).""" + return self.path + ".bak" + + def _replace_payload(self, staged: str) -> None: + """把新真身原子换上去; **先把旧的改名成 .bak**, 失败则还原 (规范 §10.1 回滚).""" + backup = self.backup_path + had_old = self._fs.is_file(self.path) + if had_old: + try: + if self._fs.is_file(backup): + os.remove(backup) + os.replace(self.path, backup) + except OSError as exc: # 备份失败就别动现有的 (安全优先) + raise RuntimeChannelError(f"无法备份旧真身: {exc}", "E_RUNTIME") from exc + try: + os.replace(staged, self.path) + except OSError as exc: + if had_old: + try: + os.replace(backup, self.path) + except OSError: + pass + raise RuntimeChannelError(f"写入真身失败: {exc}", "E_RUNTIME") from exc + + def _recover_from_backup(self, reason: str) -> str | None: + """用 `.bak` 还原真身; 成功返回路径, 否则 ``None`` (规范 §10.1 回滚).""" + backup = self.backup_path + if not (self._fs.is_file(backup) and looks_like_executable(backup, self._windows)): + return None + try: + os.replace(backup, self.path) + except OSError: + return None + self._reporter.warn(f"真身不可用({reason}),已回滚到上一份: {self.path}") + return self.path + + def _verified_payload(self, path: str) -> str | None: + """跑之前的魔数预检: 坏文件 + 好 `.bak` -> 回滚, 返回可用路径.""" + if not self._fs.is_file(path): + return None + if looks_like_executable(path, self._windows): + return path + return self._recover_from_backup("魔数校验失败") + def _runnable(self, path: str) -> bool: if not self._fs.is_file(path): return False @@ -358,7 +413,11 @@ def install(self, version: str | None = None) -> RuntimeStatus: """ recorded = self._record() native = recorded.get("binary", "") - reusable = _is_swift_payload(recorded.get("source", ""), recorded.get("channel", "")) + # 只在「**别人装的** Swift 原生安装」时 no-op(别在同一台机器上再存一份); + # 本通道自己的副本(channel=wrapper)要允许重装/升级,否则永远换不了新版本。 + reusable = recorded.get("channel") != WRAPPER_CHANNEL and _is_swift_payload( + recorded.get("source", ""), recorded.get("channel", "") + ) if native and reusable and self._runnable(native): self._reporter.info(f"复用已有的 Swift 安装: {native}") return self.status() @@ -392,7 +451,7 @@ def install(self, version: str | None = None) -> RuntimeStatus: raise RuntimeChannelError(f"{self._asset} 解出来的不是可执行文件", "E_VERIFY") if not self._windows: os.chmod(staged, 0o755) - os.replace(staged, self.path) + self._replace_payload(staged) finally: shutil.rmtree(staging, ignore_errors=True) @@ -461,13 +520,29 @@ def forward(self, argv: list[str], binary: str | None = None) -> int: """ target = binary or self.ensure() if os.name != "nt": - os.execv(target, [target, *argv]) # pragma: no cover - 进程被替换,after 不可达 + try: + os.execv(target, [target, *argv]) # pragma: no cover - 进程被替换,after 不可达 + except OSError as exc: + # 文件本身不可执行 (被截断 / 权限坏了 / 不是本机架构) -> 用 .bak 回滚后重试一次. + recovered = self._recover_from_backup(f"执行失败: {exc}") + if recovered is None: + raise RuntimeChannelError( + f"无法执行真身 {target}: {exc};可运行 `orzmc runtime install` 重装", + "E_RUNTIME", + ) from exc + os.execv(recovered, [recovered, *argv]) # pragma: no cover try: return subprocess.call([target, *argv]) except KeyboardInterrupt: # pragma: no cover - 交互中断 return 130 except OSError as exc: - raise RuntimeChannelError(f"无法执行 {target}: {exc}", "E_RUNTIME") from exc + recovered = self._recover_from_backup(f"执行失败: {exc}") + if recovered is None: + raise RuntimeChannelError( + f"无法执行真身 {target}: {exc};可运行 `orzmc runtime install` 重装", + "E_RUNTIME", + ) from exc + return subprocess.call([recovered, *argv]) def _stderr_reporter() -> tuple[Reporter, ProgressSink]: diff --git a/orzmc/tests/test_auth_lock.py b/orzmc/tests/test_auth_lock.py new file mode 100644 index 0000000..3000ca0 --- /dev/null +++ b/orzmc/tests/test_auth_lock.py @@ -0,0 +1,211 @@ +"""跨进程锁 (`frontends.md` F4.1) 与「刷新持锁」的单测。 + +锁是**跨语言契约**:算法与 Swift 侧逐字一致(`/auth.lock` 目录锁), +所以这里既测算法本身(获取/争用/陈旧/超时/释放),也测 `Authenticator.ensure()` 的临界区语义 +(锁内重读、别人刚刷完就不重复刷)。 +""" + +from __future__ import annotations + +import os +import time + +import pytest + +from orzmc.infra import auth_store as auth_store_module +from orzmc.infra.auth_store import AuthStore +from orzmc.infra.filelock import FileLock, LockTimeout, lock_path +from orzmc.infra.fs import FileStore + + +def _clock_of(start: float = 1000.0): + """可控时钟:每次调用前进 ``tick``,用来把超时/陈旧判定跑快。""" + state = {"now": start} + + def clock() -> float: + return state["now"] + + def advance(seconds: float) -> None: + state["now"] += seconds + + def sleep(seconds: float) -> None: + state["now"] += max(seconds, 0.001) + + return clock, advance, sleep + + +class TestFileLock: + def test_acquire_creates_directory_and_owner(self, tmp_path) -> None: + path = str(tmp_path / "auth.lock") + with FileLock(path): + assert os.path.isdir(path), "锁用**目录**表示(mkdir 原子)" + owner = (tmp_path / "auth.lock" / "owner").read_text(encoding="utf-8") + assert f"pid={os.getpid()}" in owner + assert "since=" in owner + assert not os.path.exists(path), "退出上下文必须释放" + + def test_second_holder_waits_until_released(self, tmp_path) -> None: + path = str(tmp_path / "auth.lock") + clock, _advance, sleep = _clock_of() + first = FileLock(path, clock=clock, sleep=sleep) + first.acquire() + second = FileLock(path, timeout=0.3, clock=clock, sleep=sleep) + with pytest.raises(LockTimeout) as excinfo: + second.acquire() + assert excinfo.value.code == "E_AUTH_LOCKED" + first.release() + second.acquire() + assert os.path.isdir(path) + second.release() + + def test_stale_lock_is_broken(self, tmp_path) -> None: + path = str(tmp_path / "auth.lock") + os.mkdir(path) # 模拟"上一个进程崩溃,锁留在磁盘上" + (tmp_path / "auth.lock" / "owner").write_text("pid=999999\n", encoding="utf-8") + # 陈旧判定读的是**目录 mtime** (与真实时钟比较), 所以直接把 mtime 拨老 31s. + old = time.time() - 31 + os.utime(path, (old, old)) + lock = FileLock(path, timeout=1) + lock.acquire() + assert os.path.isdir(path) + lock.release() + assert not os.path.exists(path) + + def test_fresh_lock_is_not_broken(self, tmp_path) -> None: + path = str(tmp_path / "auth.lock") + os.mkdir(path) + clock, _advance, sleep = _clock_of() + with pytest.raises(LockTimeout): + FileLock(path, timeout=0.2, clock=clock, sleep=sleep).acquire() + assert os.path.isdir(path), "不是陈旧的锁不能被抢" + + def test_lock_path_is_next_to_auth_json(self, tmp_path) -> None: + store = AuthStore(state_dir=str(tmp_path), fs=FileStore()) + assert store.lock().path == lock_path(str(tmp_path)) + assert store.lock().path.endswith("auth.lock") + assert os.path.dirname(store.path) == os.path.dirname(store.lock().path) + + +class TestRefreshHoldsLock: + """`Authenticator.ensure()` 必须在锁内**重读**,别人刚刷完就不要再刷。""" + + def test_state_dir_lock_blocks_concurrent_refresh(self, tmp_path, monkeypatch) -> None: + from orzmc.domain.auth import AuthSession, MsaTokens + from orzmc.services import auth as auth_service + + store = AuthStore(state_dir=str(tmp_path), fs=FileStore()) + old = AuthSession( + uuid="u", + name="n", + access_token="mc", + expires_at=100, # 已过期 -> 需要刷新 + msa=MsaTokens( + access_token="msa", + refresh_token="refresh", + expires_at=100, + token_type="Bearer", + scope="XboxLive.signin", + ), + ) + store.save(old, client_id="cid") + + refreshed: list[str] = [] + + class FakeHttp: + def get_json(self, *_args, **_kwargs): + return {} + + def fake_refresh(self, msa): + refreshed.append(msa.refresh_token) + return AuthSession( + uuid="u", + name="n", + access_token="new", + expires_at=10_000, + msa=MsaTokens( + access_token="msa2", + refresh_token="refresh2", + expires_at=10_000, + token_type="Bearer", + scope="XboxLive.signin", + ), + ) + + monkeypatch.setattr(auth_service.Authenticator, "refresh", fake_refresh) + authenticator = auth_service.Authenticator( + http=FakeHttp(), # type: ignore[arg-type] + store=store, + client_id="cid", + clock=lambda: 500.0, + ) + + # 模拟"另一个前端刚刷完": 持锁期间把文件改成有效会话. + lock = store.lock() + lock.acquire() + try: + store.save( + AuthSession( + uuid="u", + name="n", + access_token="other", + expires_at=10_000, + msa=MsaTokens( + access_token="msa3", + refresh_token="refresh3", + expires_at=10_000, + token_type="Bearer", + scope="XboxLive.signin", + ), + ), + client_id="cid", + ) + finally: + lock.release() + + session = authenticator.ensure() + + assert session.access_token == "other", "应复用另一个前端刚写好的会话" + assert refreshed == [], "锁内重读发现已有效,不应再刷新(否则 refresh token 会被轮换掉)" + + def test_lock_timeout_surfaces_as_auth_locked(self, tmp_path, monkeypatch) -> None: + from orzmc.domain.auth import AuthSession, MsaTokens + from orzmc.services import auth as auth_service + + store = AuthStore(state_dir=str(tmp_path), fs=FileStore()) + store.save( + AuthSession( + uuid="u", + name="n", + access_token="mc", + expires_at=100, + msa=MsaTokens( + access_token="msa", + refresh_token="refresh", + expires_at=100, + token_type="Bearer", + scope="XboxLive.signin", + ), + ), + client_id="cid", + ) + # 另一个进程持锁不放 (未过期) + held = store.lock() + held.acquire() + try: + authenticator = auth_service.Authenticator( + http=object(), # type: ignore[arg-type] + store=AuthStore(state_dir=str(tmp_path), fs=FileStore()), + client_id="cid", + clock=lambda: 500.0, + ) + monkeypatch.setattr( + auth_store_module, + "DEFAULT_TIMEOUT", + 0.05, + raising=False, + ) + with pytest.raises(LockTimeout) as excinfo: + authenticator.ensure() + assert excinfo.value.code == "E_AUTH_LOCKED" + finally: + held.release() diff --git a/orzmc/tests/test_runtime.py b/orzmc/tests/test_runtime.py index 13f0890..40bb5dd 100644 --- a/orzmc/tests/test_runtime.py +++ b/orzmc/tests/test_runtime.py @@ -318,6 +318,64 @@ def test_dangling_native_record_falls_back_to_own_copy(self, tmp_path) -> None: assert channel.status().to_dict()["reused"] is False +# ── 真身损坏回滚(§10.1) ───────────────────────────────────────────────────── + + +class TestPayloadRollback: + def _install_ok(self, tmp_path, http) -> RuntimeChannel: + http.json_responses["/releases"] = [{"tag_name": "v0.19.0", "draft": False}] + http.text_responses["SHA256SUMS"] = _sums("orzmc-linux-x86_64", SAFE_ELF) + http.canned["orzmc-linux-x86_64"] = SAFE_ELF + channel = _channel(tmp_path, http) + channel.install() + return channel + + def test_install_keeps_previous_payload_as_backup(self, tmp_path) -> None: + http = FakeHttp() + channel = self._install_ok(tmp_path, http) + first = pathlib.Path(channel.path).read_bytes() + + # 第二次安装换一份内容不同的真身 + second = b"\x7fELF" + b"\x11" * 60 + http.canned["orzmc-linux-x86_64"] = second + http.text_responses["SHA256SUMS"] = _sums("orzmc-linux-x86_64", second) + channel.install() + + assert pathlib.Path(channel.path).read_bytes() == second + assert pathlib.Path(channel.backup_path).read_bytes() == first, "旧真身应留作 .bak" + + def test_corrupt_payload_is_replaced_by_backup_before_run(self, tmp_path) -> None: + http = FakeHttp() + channel = self._install_ok(tmp_path, http) + # 再装一次,得到 .bak + second = b"\x7fELF" + b"\x22" * 60 + http.canned["orzmc-linux-x86_64"] = second + http.text_responses["SHA256SUMS"] = _sums("orzmc-linux-x86_64", second) + channel.install() + + # 模拟磁盘损坏:把当前真身写成垃圾 + pathlib.Path(channel.path).write_bytes(b"\x00garbage") + + path, _ = channel.resolve() + + assert path == channel.path + restored = pathlib.Path(channel.path).read_bytes() + assert restored == SAFE_ELF, "应回滚到 .bak(上一版真身)的内容" + assert looks_like_executable(channel.path, is_windows=False), "回滚后必须是可执行文件" + assert not pathlib.Path(channel.backup_path).exists(), ".bak 已被提升(消费掉)" + + def test_corrupt_payload_without_backup_still_reported(self, tmp_path) -> None: + http = FakeHttp() + channel = self._install_ok(tmp_path, http) + pathlib.Path(channel.path).write_bytes(b"\x00garbage") + + path, _ = channel.resolve() + + # 没有 .bak 可回滚:不静默删文件,交给上层报错(forward 会给出 E_RUNTIME) + assert path == channel.path + assert pathlib.Path(channel.path).exists() + + # ── 转发 ──────────────────────────────────────────────────────────────────────