From be1eb1a132d7bf4512a5e3157198787638b194b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BC=98=E9=9B=85=E3=81=AE=E5=92=B8=E9=B1=BC?= <3129538298@qq.com> Date: Wed, 26 Aug 2026 17:51:44 +0800 Subject: [PATCH 1/2] fix: kill_holder honors safety guards and verifies SIGKILL The library API used to skip kill_block_reason and report SIGKILL success even when the process was still alive (D-state / unkillable). --- src/wholocks/core.py | 5 +++++ tests/test_core.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/src/wholocks/core.py b/src/wholocks/core.py index c590219..0308ca2 100644 --- a/src/wholocks/core.py +++ b/src/wholocks/core.py @@ -281,6 +281,9 @@ def kill_block_reason(holder: Holder, force: bool = False) -> Optional[str]: def kill_holder(holder: Holder, force: bool = False, grace: float = 3.0) -> "tuple[bool, str]": """Terminate one holder. Returns (ok, message).""" + reason = kill_block_reason(holder, force=force) + if reason: + return False, "refused: " + reason if sys.platform.startswith("win"): from . import _windows @@ -307,6 +310,8 @@ def kill_holder(holder: Holder, force: bool = False, grace: float = 3.0) -> "tup except PermissionError: return False, "permission denied (try sudo)" time.sleep(0.2) + if _pid_alive(holder.pid): + return False, "still alive after SIGKILL (unkillable or in D state)" return True, "killed (SIGKILL)" return False, ( "did not exit within %.0fs of SIGTERM (or is a zombie awaiting its " diff --git a/tests/test_core.py b/tests/test_core.py index f6e064d..5ed0a02 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -11,6 +11,7 @@ advice_for, dedupe_holders, kill_block_reason, + kill_holder, make_matcher, split_targets, ) @@ -124,3 +125,35 @@ def test_service_requires_force(self): def test_normal_process_allowed(self): h = Holder(pid=44444, name="node.exe", app_type="console") assert kill_block_reason(h) is None + + def test_kill_holder_refuses_self(self): + ok, detail = kill_holder(Holder(pid=os.getpid(), name="python")) + assert ok is False + assert "refused" in detail + assert "itself" in detail + + def test_kill_holder_refuses_pid1_on_posix(self): + if sys.platform.startswith("win"): + pytest.skip("PID 1 guard is POSIX") + ok, detail = kill_holder(Holder(pid=1, name="init")) + assert ok is False + assert "PID 1" in detail + + def test_sigkill_reports_failure_if_still_alive(self, monkeypatch): + import time + + import wholocks.core as core + + ticks = {"n": 0} + + def fake_mono(): + ticks["n"] += 1 + return ticks["n"] * 10 + + monkeypatch.setattr(time, "monotonic", fake_mono) + monkeypatch.setattr(time, "sleep", lambda _s: None) + monkeypatch.setattr(os, "kill", lambda _pid, _sig: None) + monkeypatch.setattr(core, "_pid_alive", lambda _pid: True) + ok, detail = kill_holder(Holder(pid=44444, name="stuck"), force=True, grace=3.0) + assert ok is False + assert "SIGKILL" in detail From e444159c0a56a750343d942ce9f6566dc5f82fd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BC=98=E9=9B=85=E3=81=AE=E5=92=B8=E9=B1=BC?= <3129538298@qq.com> Date: Wed, 26 Aug 2026 18:05:21 +0800 Subject: [PATCH 2/2] fix: Windows kill reports failure when process stays alive TerminateProcess + WaitForSingleObject timeout used to return success ("still shutting down"). After the grace wait, still-alive PIDs are now a failure, matching the POSIX SIGKILL path. The still-alive test only patched os.kill, so Windows CI took _windows.kill(missing pid) and reported already gone. Cover both backends without skipping Windows. --- src/wholocks/_windows.py | 8 ++++- tests/test_core.py | 74 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 1 deletion(-) diff --git a/src/wholocks/_windows.py b/src/wholocks/_windows.py index 7f9f653..cc17939 100644 --- a/src/wholocks/_windows.py +++ b/src/wholocks/_windows.py @@ -36,6 +36,7 @@ SYNCHRONIZE = 0x00100000 STILL_ACTIVE = 259 WAIT_OBJECT_0 = 0 +WAIT_TIMEOUT = 0x00000102 FILE_READ_ATTRIBUTES = 0x0080 FILE_SHARE_ALL = 0x1 | 0x2 | 0x4 @@ -528,7 +529,12 @@ def kill(pid, grace=5.0): waited = _k32.WaitForSingleObject(h, int(grace * 1000)) if waited == WAIT_OBJECT_0: return True, "terminated" - return True, "termination requested (still shutting down)" + # Same honesty as POSIX SIGKILL: a timeout is not a kill. + if _pid_exists(pid): + return False, ( + "still alive after TerminateProcess (protected or unkillable)" + ) + return True, "terminated" finally: _k32.CloseHandle(h) diff --git a/tests/test_core.py b/tests/test_core.py index 5ed0a02..254564d 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -152,8 +152,82 @@ def fake_mono(): monkeypatch.setattr(time, "monotonic", fake_mono) monkeypatch.setattr(time, "sleep", lambda _s: None) + if sys.platform.startswith("win"): + import wholocks._windows as win + + class FakeK32: + def OpenProcess(self, *_a, **_k): + return 1 + + def TerminateProcess(self, *_a, **_k): + return True + + def WaitForSingleObject(self, *_a, **_k): + return win.WAIT_TIMEOUT + + def CloseHandle(self, *_a, **_k): + return True + + monkeypatch.setattr(win, "_load", lambda: None) + monkeypatch.setattr(win, "_k32", FakeK32()) + monkeypatch.setattr(win, "_pid_exists", lambda _pid: True) + ok, detail = kill_holder( + Holder(pid=44444, name="stuck"), force=True, grace=3.0 + ) + assert ok is False + assert "still alive" in detail + assert "TerminateProcess" in detail + return + monkeypatch.setattr(os, "kill", lambda _pid, _sig: None) monkeypatch.setattr(core, "_pid_alive", lambda _pid: True) ok, detail = kill_holder(Holder(pid=44444, name="stuck"), force=True, grace=3.0) assert ok is False assert "SIGKILL" in detail + + def test_windows_kill_reports_failure_if_still_alive(self, monkeypatch): + """Lock the Windows backend on every OS (no rstrtmgr.dll needed).""" + from wholocks import _windows as win + + class FakeK32: + def OpenProcess(self, *_a, **_k): + return 1 + + def TerminateProcess(self, *_a, **_k): + return True + + def WaitForSingleObject(self, *_a, **_k): + return win.WAIT_TIMEOUT + + def CloseHandle(self, *_a, **_k): + return True + + monkeypatch.setattr(win, "_load", lambda: None) + monkeypatch.setattr(win, "_k32", FakeK32()) + monkeypatch.setattr(win, "_pid_exists", lambda _pid: True) + ok, detail = win.kill(44444, grace=3.0) + assert ok is False + assert "still alive" in detail + assert "TerminateProcess" in detail + + def test_windows_kill_ok_when_wait_signals(self, monkeypatch): + from wholocks import _windows as win + + class FakeK32: + def OpenProcess(self, *_a, **_k): + return 1 + + def TerminateProcess(self, *_a, **_k): + return True + + def WaitForSingleObject(self, *_a, **_k): + return win.WAIT_OBJECT_0 + + def CloseHandle(self, *_a, **_k): + return True + + monkeypatch.setattr(win, "_load", lambda: None) + monkeypatch.setattr(win, "_k32", FakeK32()) + ok, detail = win.kill(44444, grace=3.0) + assert ok is True + assert detail == "terminated"