fix(install): never strand hermes.exe when a Windows update fails
On Windows the updater renames the live `hermes*.exe` shims aside (`hermes.exe.old.<unix-ms>`) so uv can write replacements. When that quarantine succeeds but the install then fails, the recovery path could leave the install with no `hermes` on PATH at all — unrecoverable in place, because the command that would repair it IS `hermes update` (#75584). Restoring a quarantined shim happens at three sites: the updater, the early-recovery installer, and the startup sweep's orphan rescue. Each was a single un-retried rename whose OSError was swallowed in silence, while the OUTBOUND quarantine rename already retried a lock. That is backwards — a failed quarantine merely aborts an update, a failed restore removes `hermes` from PATH — and the two sites that had messages had already drifted apart. - `_early_recovery.restore_quarantined_shims()` is now the single implementation: retry ladder, one recovery message, returns the pairs it could not restore. It lives in the stdlib-only module that both `main` and `_install_repair` already import, so the layers cannot drift again. A pair is not a failure when the original reappeared or the quarantine file vanished — two processes sweeping the same orphan must not produce a spurious error. - `_cleanup_quarantined_exes` unlinked every `*.exe.old.*` on each invocation. When the original shim was already missing, that .old file was the ONLY surviving copy — deleting it converted a one-rename recovery into a full reinstall. It now rescues the orphan through the shared helper instead, and leaves anything inside a 15-minute grace window alone so it cannot destroy a concurrent update's in-flight quarantine. - Ordering is by the PARSED `.old.<unix-ms>` stamp, not the raw filename. Lexicographic ordering only tracks recency while every stamp shares a digit width; a stray `.old.999` sorts above a 13-digit epoch-ms stamp and would be the copy rescued onto the live shim name. - Names whose suffix does not parse as int-ms are not ours: never rescued, never deleted. The sweep should not destroy files whose provenance it cannot establish, and they are not produced by the quarantiner. The stamp is read from the filename rather than st_mtime because `rename` preserves the original shim's mtime, which records when uv wrote the shim — days earlier, in general — not when it was quarantined. A regression test pins that distinction. Messages go to stderr: the sweep runs on EVERY hermes invocation and `hermes acp` speaks JSON-RPC on stdout. Scope note: `_quarantine_running_hermes_exe` is deliberately byte-identical to main here. Why the outbound rename fails in the first place (the launcher holding its own image without FILE_SHARE_DELETE) is #88121's subject; this is the net underneath, covering the case where quarantine SUCCEEDS and the install dies afterwards. The two touch disjoint functions and can merge in either order. Reproduced and verified on Windows 11 (26200), Python 3.11.15: stranded the shims, confirmed a normal `hermes` invocation now rescues the orphan instead of deleting it, and confirmed an exhausted rescue prints the recovery command. 16 new tests; 35 pass across the four quarantine suites.
This commit is contained in:
hsearcy
@@ -56,6 +56,85 @@ LAZY_REFRESH_REPAIR_PACKAGES: dict[str, str] = {
|
||||
"jwt": "PyJWT",
|
||||
}
|
||||
|
||||
|
||||
# --- Windows entry-point shim quarantine -----------------------------------
|
||||
#
|
||||
# ``hermes update`` renames the live ``hermes*.exe`` shims aside
|
||||
# (``hermes.exe.old.<unix-ms>``) so uv can write replacements. Putting them BACK
|
||||
# is the safety-critical direction: losing that rename leaves the install with
|
||||
# no ``hermes`` on PATH, and the command that would repair it IS ``hermes
|
||||
# update`` (#75584).
|
||||
#
|
||||
# Three call sites restore a quarantined shim -- the updater, the
|
||||
# early-recovery installer, and the startup sweep's orphan rescue. They used to
|
||||
# be separate one-shot renames with swallowed errors; the two that had messages
|
||||
# had already drifted apart. The logic lives here, in the one stdlib-only module
|
||||
# all of them can import, so the ladder and the recovery wording stay in
|
||||
# lockstep.
|
||||
|
||||
QUARANTINE_RESTORE_BACKOFF_MS: tuple[int, ...] = (0, 100, 250, 500, 1000)
|
||||
|
||||
|
||||
def restore_quarantined_shims(
|
||||
moved: list[tuple[Path, Path]],
|
||||
*,
|
||||
stream=None,
|
||||
backoff_ms: tuple[int, ...] = QUARANTINE_RESTORE_BACKOFF_MS,
|
||||
) -> list[tuple[Path, Path]]:
|
||||
"""Rename quarantined shims back, retrying a lock instead of giving up.
|
||||
|
||||
``moved`` holds ``(original, quarantined)`` pairs. Returns the pairs that
|
||||
could NOT be restored, and prints an actionable recovery command for each.
|
||||
|
||||
A pair is not a failure when ``original`` already exists or ``quarantined``
|
||||
has gone: the installer wrote a fresh shim, or a concurrent sweep won the
|
||||
race. Both are silent, so two processes sweeping the same orphan cannot
|
||||
produce a spurious error.
|
||||
|
||||
Messages go to stderr by default -- the startup sweep runs on EVERY hermes
|
||||
invocation, and ``hermes acp`` speaks JSON-RPC on stdout.
|
||||
"""
|
||||
if stream is None:
|
||||
stream = sys.stderr
|
||||
|
||||
failed: list[tuple[Path, Path]] = []
|
||||
|
||||
for original, quarantined in moved:
|
||||
last_exc: OSError | None = None
|
||||
|
||||
for delay_ms in backoff_ms:
|
||||
try:
|
||||
if os.path.exists(original) or not os.path.exists(quarantined):
|
||||
last_exc = None
|
||||
break
|
||||
if delay_ms:
|
||||
time.sleep(delay_ms / 1000.0)
|
||||
os.rename(quarantined, original)
|
||||
last_exc = None
|
||||
break
|
||||
except OSError as exc:
|
||||
last_exc = exc
|
||||
continue
|
||||
|
||||
if last_exc is None:
|
||||
continue
|
||||
|
||||
failed.append((original, quarantined))
|
||||
name = os.path.basename(str(original))
|
||||
stem = name[:-4] if name.lower().endswith(".exe") else name
|
||||
print(
|
||||
f" ✖ FAILED to restore {name} "
|
||||
f"({last_exc.__class__.__name__}) — it is still quarantined "
|
||||
f"as {os.path.basename(str(quarantined))}.\n"
|
||||
f" `{stem}` will NOT be on PATH until it is put back. Run this, "
|
||||
f"then re-run the update:\n"
|
||||
f' move "{quarantined}" "{original}"',
|
||||
file=stream,
|
||||
)
|
||||
|
||||
return failed
|
||||
|
||||
|
||||
# Set only when this process successfully finishes a deferred core install for
|
||||
# an ``update`` invocation. The normal CLI import that follows must not resolve
|
||||
# external secret sources: a configured source can map cryptography._rust and
|
||||
|
||||
@@ -527,14 +527,15 @@ def _quarantine_running_hermes_exe(
|
||||
|
||||
|
||||
def _restore_quarantined_exes(moved: list[tuple[Path, Path]]) -> None:
|
||||
"""Put quarantined shims back when the installer did not replace them."""
|
||||
for original, quarantined in moved:
|
||||
if original.exists():
|
||||
continue # installer wrote a fresh shim — the .old one is garbage
|
||||
try:
|
||||
os.rename(quarantined, original)
|
||||
except OSError:
|
||||
pass
|
||||
"""Put quarantined shims back when the installer did not replace them.
|
||||
|
||||
Delegates to the shared helper in the stdlib-only ``_early_recovery``
|
||||
module: one retry ladder and one recovery message for every restore site,
|
||||
instead of the near-identical copies that had already drifted (#75584).
|
||||
Warnings land on stderr — this module runs in the early-recovery path and
|
||||
``hermes acp`` speaks JSON-RPC on stdout.
|
||||
"""
|
||||
_er.restore_quarantined_shims(moved)
|
||||
|
||||
|
||||
def _run_install_cmd(cmd: list[str], *, env: dict | None, root: Path) -> None:
|
||||
|
||||
+88
-18
@@ -9177,13 +9177,18 @@ def _cleanup_pending_shim_renames(scripts_dir: Path) -> int:
|
||||
|
||||
|
||||
def _restore_quarantined_exes(moved: list[tuple[Path, Path]]) -> None:
|
||||
"""Roll back ``_quarantine_running_hermes_exe`` if uv didn't write replacements."""
|
||||
for original, quarantined in moved:
|
||||
try:
|
||||
if not original.exists() and quarantined.exists():
|
||||
quarantined.rename(original)
|
||||
except OSError:
|
||||
pass
|
||||
"""Roll back ``_quarantine_running_hermes_exe`` if uv didn't write replacements.
|
||||
|
||||
This is the safety-critical direction. A failed *quarantine* only aborts an
|
||||
update; a failed *restore* leaves the install with no ``hermes`` on PATH,
|
||||
and therefore no way to run the command that would repair it (#75584). The
|
||||
outbound rename already retries a lock, so this one must too rather than
|
||||
swallow the first ``OSError`` in silence.
|
||||
|
||||
Delegates to the stdlib-only helper that the early-recovery copy in
|
||||
``_install_repair`` also uses, so the two cannot drift apart.
|
||||
"""
|
||||
_early_recovery_mod.restore_quarantined_shims(moved)
|
||||
|
||||
|
||||
class ShimQuarantineError(RuntimeError):
|
||||
@@ -9256,12 +9261,49 @@ def _run_quarantined_install(
|
||||
_restore_quarantined_exes(moved)
|
||||
|
||||
|
||||
def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None:
|
||||
"""Sweep ``hermes.exe.old.*`` and stale reboot renames left by prior updates.
|
||||
# A quarantine file younger than this may belong to an update running RIGHT
|
||||
# NOW in another process, whose restore step still needs it. Deleting one
|
||||
# mid-flight destroys the only copy of that shim.
|
||||
_QUARANTINE_GRACE_SECONDS = 15 * 60
|
||||
|
||||
Called early on every hermes invocation. The .old files are unlocked once
|
||||
their owning process exited, so deletion succeeds the next run. Silent
|
||||
no-op when nothing's there or on file-locked / permission errors.
|
||||
|
||||
def _quarantine_stamp_ms(stale: Path) -> int | None:
|
||||
"""The ``.old.<unix-ms>`` stamp in a quarantine filename, or ``None``.
|
||||
|
||||
``None`` means the name was not produced by
|
||||
:func:`_quarantine_running_hermes_exe`. We neither rescue nor delete those:
|
||||
the sweep should not destroy files whose provenance it cannot establish, and
|
||||
they are not ours to put back.
|
||||
|
||||
Parsed from the NAME rather than ``st_mtime`` because ``rename`` preserves
|
||||
the original shim's mtime, which records when uv wrote the shim — days
|
||||
earlier, in general — not when it was quarantined.
|
||||
"""
|
||||
try:
|
||||
return int(stale.name.rsplit(".old.", 1)[1])
|
||||
except (IndexError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None:
|
||||
"""Sweep — and where necessary RESCUE — ``hermes.exe.old.*`` from updates.
|
||||
|
||||
Called early on every hermes invocation. Two cases the old unconditional
|
||||
``unlink()`` got wrong, both ending with ``hermes`` gone from PATH:
|
||||
|
||||
1. **Orphan rescue.** If ``hermes.exe`` is missing while
|
||||
``hermes.exe.old.*`` is present, that .old file is the ONLY surviving
|
||||
copy of the shim — an update died, or its restore failed, between
|
||||
the rename and uv writing a replacement (#75584). Deleting it converts a
|
||||
one-rename recovery into a full reinstall. Put it back instead, through
|
||||
the same retry-and-report helper the update-time restore uses.
|
||||
2. **Concurrency.** A fresh quarantine file may belong to an update in
|
||||
flight in another process (the desktop update button racing a shell
|
||||
``hermes update`` does exactly this). Leave anything inside the grace
|
||||
window alone; a later run sweeps it.
|
||||
|
||||
Silent no-op on non-Windows, when there is nothing to do, or on
|
||||
file-locked / permission errors.
|
||||
"""
|
||||
if not _is_windows():
|
||||
return
|
||||
@@ -9270,14 +9312,42 @@ def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None:
|
||||
if scripts_dir is None:
|
||||
return
|
||||
_cleanup_pending_shim_renames(scripts_dir)
|
||||
|
||||
now = _time.time()
|
||||
|
||||
try:
|
||||
for stale in scripts_dir.glob("*.exe.old.*"):
|
||||
try:
|
||||
stale.unlink()
|
||||
except OSError:
|
||||
pass # still locked or in use — try again next run
|
||||
candidates = [
|
||||
(stamp, stale)
|
||||
for stale, stamp in (
|
||||
(p, _quarantine_stamp_ms(p)) for p in scripts_dir.glob("*.exe.old.*")
|
||||
)
|
||||
if stamp is not None
|
||||
]
|
||||
except OSError:
|
||||
pass
|
||||
return
|
||||
|
||||
# Newest first by PARSED stamp. Sorting the raw filenames lexicographically
|
||||
# only tracks recency while every stamp shares a digit width: a stray
|
||||
# ``.old.999`` sorts above a 13-digit epoch-ms stamp and would be the copy
|
||||
# rescued onto the live shim name.
|
||||
candidates.sort(key=lambda pair: pair[0], reverse=True)
|
||||
|
||||
for stamp, stale in candidates:
|
||||
try:
|
||||
original = stale.with_name(stale.name.rsplit(".old.", 1)[0])
|
||||
|
||||
if not original.exists():
|
||||
# Orphan rescue: this is the last copy of the shim, so it gets
|
||||
# the retry ladder and the recovery message, not a bare rename.
|
||||
_early_recovery_mod.restore_quarantined_shims([(original, stale)])
|
||||
continue
|
||||
|
||||
if now - stamp / 1000.0 < _QUARANTINE_GRACE_SECONDS:
|
||||
continue # may be a live quarantine from a concurrent update
|
||||
|
||||
stale.unlink()
|
||||
except OSError:
|
||||
pass # still locked or in use — try again next run
|
||||
|
||||
|
||||
# Import probes for venv corruption after a failed lazy ``uv pip install``.
|
||||
|
||||
@@ -0,0 +1,347 @@
|
||||
"""Regression tests: a failed quarantine restore must never strand `hermes`.
|
||||
|
||||
On Windows the updater renames the live ``hermes*.exe`` shims aside
|
||||
(``hermes.exe.old.<unix-ms>``) so uv can write replacements. Gaps in the
|
||||
recovery path ended with ``hermes`` gone from PATH — and, because the command
|
||||
that repairs it IS ``hermes update``, unrecoverable without a manual reinstall
|
||||
(#75584):
|
||||
|
||||
1. Restoring a shim got a single attempt whose ``OSError`` was swallowed in
|
||||
silence, while the outbound quarantine rename already retried a lock.
|
||||
2. The startup sweep unlinked every ``*.exe.old.*``. When the original shim was
|
||||
already missing, that .old file was the ONLY surviving copy — deleting it
|
||||
converted a one-rename recovery into a full reinstall. It also raced a
|
||||
concurrent in-flight update, destroying the quarantine that update's own
|
||||
restore was about to rename back.
|
||||
|
||||
These tests pin the hardened behavior: retry, rescue, report, order by parsed
|
||||
stamp, and leave files we did not create alone.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import _early_recovery as er
|
||||
from hermes_cli import _install_repair as ir
|
||||
from hermes_cli import main as cli_main
|
||||
|
||||
|
||||
def _make_scripts_dir(tmp_path: Path) -> Path:
|
||||
scripts = tmp_path / "venv" / "Scripts"
|
||||
scripts.mkdir(parents=True)
|
||||
return scripts
|
||||
|
||||
|
||||
def _stamp(ms_ago: int = 0) -> int:
|
||||
return int(time.time() * 1000) - ms_ago
|
||||
|
||||
|
||||
def _run_cleanup(scripts: Path):
|
||||
"""Drive the sweep with the Windows gate forced and the registry stubbed.
|
||||
|
||||
``_cleanup_pending_shim_renames`` reaches into PendingFileRenameOperations;
|
||||
it has its own tests and must not run here.
|
||||
"""
|
||||
return patch.multiple(
|
||||
cli_main,
|
||||
_is_windows=lambda: True,
|
||||
_cleanup_pending_shim_renames=lambda _scripts_dir: 0,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# orphan rescue
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cleanup_rescues_orphan_when_original_missing(tmp_path):
|
||||
"""The .old file is the last copy of the shim — put it back, don't delete."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
orphan = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
orphan.write_bytes(b"MZ-orphan")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert (scripts / "hermes.exe").read_bytes() == b"MZ-orphan"
|
||||
assert not orphan.exists()
|
||||
|
||||
|
||||
def test_cleanup_rescue_survives_a_transient_lock(tmp_path, capsys):
|
||||
"""The rescue rename retries a lock instead of stranding on first failure.
|
||||
|
||||
This is the window the sweep runs in: the shim is ALREADY gone from PATH, so
|
||||
giving up here leaves the user stranded exactly as if the sweep had deleted
|
||||
the file.
|
||||
"""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
orphan = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
orphan.write_bytes(b"MZ-orphan")
|
||||
|
||||
real_rename = os.rename
|
||||
calls = {"n": 0}
|
||||
|
||||
def flaky(src, dst):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise PermissionError(32, "being used by another process")
|
||||
return real_rename(src, dst)
|
||||
|
||||
with _run_cleanup(scripts), patch.object(er.os, "rename", flaky):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert (scripts / "hermes.exe").read_bytes() == b"MZ-orphan"
|
||||
assert calls["n"] >= 2, "rescue must retry after a transient lock"
|
||||
assert capsys.readouterr().err == "", "a recovered rescue must stay quiet"
|
||||
|
||||
|
||||
def test_cleanup_rescue_reports_when_it_cannot_recover(tmp_path, capsys):
|
||||
"""A rescue that exhausts its retries must say so, not fail silently."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
orphan = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
orphan.write_bytes(b"MZ-orphan")
|
||||
|
||||
def always_locked(src, dst):
|
||||
raise PermissionError(32, "being used by another process")
|
||||
|
||||
with _run_cleanup(scripts), patch.object(er.os, "rename", always_locked):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "FAILED to restore hermes.exe" in captured.err
|
||||
assert "move" in captured.err, "must print the literal recovery command"
|
||||
assert captured.out == "", "stdout must stay clean for JSON-RPC"
|
||||
assert orphan.exists(), "the last copy must survive a failed rescue"
|
||||
|
||||
|
||||
def test_cleanup_rescue_is_quiet_when_another_process_wins(tmp_path, capsys):
|
||||
"""Two sweeps, one orphan: the loser must no-op cleanly, not report failure."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
orphan = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
orphan.write_bytes(b"MZ-orphan")
|
||||
original = scripts / "hermes.exe"
|
||||
|
||||
def loses_race(src, dst):
|
||||
# The "winner" lands the shim while our attempt is in flight.
|
||||
original.write_bytes(b"MZ-from-winner")
|
||||
raise PermissionError(32, "being used by another process")
|
||||
|
||||
with _run_cleanup(scripts), patch.object(er.os, "rename", loses_race):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert original.read_bytes() == b"MZ-from-winner"
|
||||
assert captured.err == "", "losing a benign race is not a failure"
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ordering and provenance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cleanup_rescues_newest_by_parsed_stamp_not_lexicographic(tmp_path):
|
||||
"""Mixed-width stamps: ordering must follow the parsed integer.
|
||||
|
||||
``sorted(reverse=True)`` over raw filenames puts ``.old.999`` above a
|
||||
13-digit epoch-ms stamp, which would rescue the wrong bytes onto the live
|
||||
shim name.
|
||||
"""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
(scripts / "hermes.exe.old.999").write_bytes(b"MZ-stray-short-stamp")
|
||||
(scripts / f"hermes.exe.old.{_stamp(60_000)}").write_bytes(b"MZ-genuine")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert (scripts / "hermes.exe").read_bytes() == b"MZ-genuine"
|
||||
|
||||
|
||||
def test_cleanup_ignores_names_it_did_not_create(tmp_path):
|
||||
"""An unparseable suffix is not ours: never rescued, never deleted."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
(scripts / "hermes.exe").write_bytes(b"MZ-live")
|
||||
foreign = scripts / "hermes.exe.old.backup"
|
||||
foreign.write_bytes(b"MZ-someone-elses-file")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert foreign.exists(), "the sweep must not delete files of unknown provenance"
|
||||
assert foreign.read_bytes() == b"MZ-someone-elses-file"
|
||||
assert (scripts / "hermes.exe").read_bytes() == b"MZ-live"
|
||||
|
||||
|
||||
def test_cleanup_does_not_rescue_from_a_foreign_name(tmp_path):
|
||||
"""Missing shim + only a foreign .old: leave it be rather than guess."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
foreign = scripts / "hermes.exe.old.backup"
|
||||
foreign.write_bytes(b"MZ-someone-elses-file")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert not (scripts / "hermes.exe").exists()
|
||||
assert foreign.exists()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# concurrency grace window
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_cleanup_leaves_fresh_quarantine_for_concurrent_update(tmp_path):
|
||||
"""A young .old may belong to an update in flight elsewhere — hands off."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
(scripts / "hermes.exe").write_bytes(b"MZ-live")
|
||||
fresh = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
fresh.write_bytes(b"MZ-inflight")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert fresh.exists(), "a live quarantine must survive another process's sweep"
|
||||
|
||||
|
||||
def test_cleanup_still_sweeps_genuinely_stale_quarantine(tmp_path):
|
||||
"""Past the grace window, with the shim present, it's garbage — sweep it."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
(scripts / "hermes.exe").write_bytes(b"MZ-live")
|
||||
ancient_ms = (cli_main._QUARANTINE_GRACE_SECONDS + 60) * 1000
|
||||
stale = scripts / f"hermes.exe.old.{_stamp(ancient_ms)}"
|
||||
stale.write_bytes(b"MZ-stale")
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert not stale.exists()
|
||||
assert (scripts / "hermes.exe").read_bytes() == b"MZ-live"
|
||||
|
||||
|
||||
def test_cleanup_age_comes_from_filename_not_mtime(tmp_path):
|
||||
"""rename() preserves mtime, so only the name records the quarantine time."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
(scripts / "hermes.exe").write_bytes(b"MZ-live")
|
||||
fresh = scripts / f"hermes.exe.old.{_stamp()}"
|
||||
fresh.write_bytes(b"MZ-inflight")
|
||||
week_ago = time.time() - 7 * 24 * 3600
|
||||
os.utime(fresh, (week_ago, week_ago))
|
||||
|
||||
with _run_cleanup(scripts):
|
||||
cli_main._cleanup_quarantined_exes(scripts)
|
||||
|
||||
assert fresh.exists(), "grace window must key off the .old.<ms> stamp"
|
||||
|
||||
|
||||
def test_quarantine_stamp_ms_parses_and_rejects():
|
||||
assert cli_main._quarantine_stamp_ms(Path("hermes.exe.old.1787020473885")) == 1787020473885
|
||||
assert cli_main._quarantine_stamp_ms(Path("hermes.exe.old.backup")) is None
|
||||
assert cli_main._quarantine_stamp_ms(Path("hermes.exe")) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# the shared restore helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_helper_retries_then_succeeds(tmp_path):
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
quarantined = scripts / "hermes.exe.old.123"
|
||||
quarantined.write_bytes(b"MZ-old-hermes")
|
||||
original = scripts / "hermes.exe"
|
||||
|
||||
real_rename = os.rename
|
||||
calls = {"n": 0}
|
||||
|
||||
def flaky(src, dst):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise PermissionError(32, "being used by another process")
|
||||
return real_rename(src, dst)
|
||||
|
||||
with patch.object(er.os, "rename", flaky):
|
||||
failed = er.restore_quarantined_shims([(original, quarantined)])
|
||||
|
||||
assert failed == []
|
||||
assert original.read_bytes() == b"MZ-old-hermes"
|
||||
assert calls["n"] >= 2
|
||||
|
||||
|
||||
def test_helper_reports_failure_and_returns_the_pair(tmp_path, capsys):
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
quarantined = scripts / "hermes.exe.old.123"
|
||||
quarantined.write_bytes(b"MZ-old-hermes")
|
||||
original = scripts / "hermes.exe"
|
||||
|
||||
def always_locked(src, dst):
|
||||
raise PermissionError(32, "being used by another process")
|
||||
|
||||
with patch.object(er.os, "rename", always_locked):
|
||||
failed = er.restore_quarantined_shims([(original, quarantined)])
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert failed == [(original, quarantined)]
|
||||
assert "FAILED to restore hermes.exe" in captured.err
|
||||
assert "hermes.exe.old.123" in captured.err
|
||||
assert "move" in captured.err
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
def test_helper_is_a_noop_when_installer_wrote_a_fresh_shim(tmp_path, capsys):
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
quarantined = scripts / "hermes.exe.old.123"
|
||||
quarantined.write_bytes(b"MZ-old")
|
||||
original = scripts / "hermes.exe"
|
||||
original.write_bytes(b"MZ-fresh")
|
||||
|
||||
failed = er.restore_quarantined_shims([(original, quarantined)])
|
||||
|
||||
assert failed == []
|
||||
assert original.read_bytes() == b"MZ-fresh", "must not clobber the fresh shim"
|
||||
assert capsys.readouterr().err == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# both call sites route through the helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_main_restore_reports_on_stderr(tmp_path, capsys):
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
quarantined = scripts / "hermes.exe.old.123"
|
||||
quarantined.write_bytes(b"MZ-old-hermes")
|
||||
original = scripts / "hermes.exe"
|
||||
|
||||
def always_locked(src, dst):
|
||||
raise PermissionError(32, "being used by another process")
|
||||
|
||||
with patch.object(er.os, "rename", always_locked):
|
||||
cli_main._restore_quarantined_exes([(original, quarantined)])
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "FAILED to restore hermes.exe" in captured.err
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
def test_repair_restore_reports_on_stderr(tmp_path, capsys):
|
||||
"""The early-recovery path must warn on stderr (acp speaks JSON-RPC on stdout)."""
|
||||
scripts = _make_scripts_dir(tmp_path)
|
||||
quarantined = scripts / "hermes.exe.old.123"
|
||||
quarantined.write_bytes(b"MZ-old-hermes")
|
||||
original = scripts / "hermes.exe"
|
||||
|
||||
def always_locked(src, dst):
|
||||
raise PermissionError(32, "being used by another process")
|
||||
|
||||
with patch.object(er.os, "rename", always_locked):
|
||||
ir._restore_quarantined_exes([(original, quarantined)])
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "FAILED to restore hermes.exe" in captured.err
|
||||
assert captured.out == "", "stdout must stay clean for JSON-RPC"
|
||||
Reference in New Issue
Block a user