Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion src/wholocks/_windows.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand Down
5 changes: 5 additions & 0 deletions src/wholocks/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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 "
Expand Down
107 changes: 107 additions & 0 deletions tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
advice_for,
dedupe_holders,
kill_block_reason,
kill_holder,
make_matcher,
split_targets,
)
Expand Down Expand Up @@ -124,3 +125,109 @@ 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)
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"
Loading