fix(update): reject terminated import probes

This commit is contained in:
fangliquanflq
2026-08-25 16:57:15 +08:00
committed by Teknium
parent 68f9681acb
commit 6c608e2f59
3 changed files with 96 additions and 2 deletions
+15 -2
View File
@@ -744,7 +744,12 @@ def _critical_module_import_failures(
return {}
output = result.stdout or ""
if marker not in output:
return {}
return {
"critical-module probe": (
"terminated before reporting import health "
f"(exit code {result.returncode})"
)
}
try:
import json
@@ -2470,12 +2475,20 @@ def _git_untracked_paths(git_cmd: list[str], cwd: Path) -> set[str]:
errors="surrogateescape",
)
if result.returncode != 0:
print(
" ⚠ Could not enumerate untracked files while validating the "
"restored stash."
)
return set()
return {path for path in result.stdout.split("\0") if path}
def _restored_python_paths(git_cmd: list[str], cwd: Path) -> tuple[str, ...]:
"""Return tracked and untracked Python paths changed from ``HEAD``."""
"""Return restored ``.py`` paths changed from ``HEAD``.
This deliberately validates Python source only; non-Python entry scripts
remain outside the executable import-health check.
"""
changed = subprocess.run(
git_cmd + ["diff", "--name-only", "-z", "HEAD", "--", "*.py"],
cwd=cwd,
+42
View File
@@ -657,6 +657,48 @@ def test_restore_rejects_system_exit_masked_by_preexisting_failure(
assert "gateway was not restarted" in output
def test_restore_rejects_probe_termination(monkeypatch, tmp_path, capsys):
"""A stash cannot bypass import validation by terminating the probe."""
import subprocess
from hermes_cli import update_cmd
def git(*args, check=True):
return subprocess.run(
["git", *args],
cwd=tmp_path,
capture_output=True,
text=True,
check=check,
)
git("init", "-q", "-b", "main")
git("config", "user.email", "t@example.com")
git("config", "user.name", "t")
source = tmp_path / "consumer.py"
source.write_text("VALUE = 1\n", encoding="utf-8")
git("add", "-A")
git("commit", "-qm", "init")
source.write_text("import os\nos._exit(7)\n", encoding="utf-8")
stash_ref = hermes_main._stash_local_changes_if_needed(["git"], tmp_path)
assert stash_ref
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
with pytest.raises(SystemExit) as exc_info:
hermes_main._restore_stashed_changes(
["git"], tmp_path, stash_ref, prompt_user=False
)
assert exc_info.value.code == 1
assert source.read_text(encoding="utf-8") == "VALUE = 1\n"
assert git("status", "--porcelain").stdout == ""
assert git("stash", "list").stdout.strip()
output = capsys.readouterr().out
assert "critical-module probe" in output
assert "exit code 7" in output
assert "gateway was not restarted" in output
def test_gateway_restore_prompt_defaults_to_keep_stash(tmp_path, capsys):
prompts = []
@@ -96,6 +96,45 @@ def test_import_guard_can_report_non_import_errors(monkeypatch, tmp_path):
assert error == "broken config"
def test_import_guard_reports_probe_termination_when_comparing_states(
monkeypatch, tmp_path
):
"""A terminating import is unsafe when validating a restored stash."""
(tmp_path / "consumer.py").write_text("import os\nos._exit(7)\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = hermes_main._validate_critical_modules_import(
tmp_path, report_runtime_errors=True
)
assert ok is False
assert module == "critical-module probe"
assert error == "terminated before reporting import health (exit code 7)"
def test_import_guard_reports_probe_termination_by_default(monkeypatch, tmp_path):
"""A missing health marker must not classify a terminated probe as healthy."""
(tmp_path / "consumer.py").write_text("import os\nos._exit(9)\n")
monkeypatch.setattr(update_cmd, "_UPDATE_CRITICAL_MODULES", ("consumer",))
ok, module, error = hermes_main._validate_critical_modules_import(tmp_path)
assert ok is False
assert module == "critical-module probe"
assert error == "terminated before reporting import health (exit code 9)"
def test_untracked_enumeration_failure_is_visible(monkeypatch, tmp_path, capsys):
class Result:
returncode = 1
stdout = ""
monkeypatch.setattr(update_cmd.subprocess, "run", lambda *_a, **_kw: Result())
assert update_cmd._git_untracked_paths(["git"], tmp_path) == set()
assert "Could not enumerate untracked files" in capsys.readouterr().out
def test_import_guard_is_non_fatal_when_probe_cannot_run(monkeypatch, tmp_path):
"""If we can't spawn the probe, don't block the user's update."""