File size: 8,415 Bytes
3c8e381 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | from __future__ import annotations
import os
import socket
import sys
from errno import EPERM, ESRCH
from pathlib import Path
from typing import Final
def host_name() -> str:
"""The hostname recorded alongside an owner, so a marker written on another machine is never probed here."""
return socket.gethostname()
def owner_is_stale(pid: int, hostname: str, start_token: int | None) -> bool:
"""
Whether the recorded owner is provably gone, so reclaiming its marker cannot detach a live holder.
Fail closed: return ``True`` only when this process can prove the exact recorded owner is dead. A marker from
another host cannot be probed; a live PID whose start token still matches, or whose token cannot be read, is the
holder or is indistinguishable from it; a live PID whose start token differs is a recycled PID, so the process that
wrote the marker is gone. PostgreSQL, Qt ``QLockFile`` and Mercurial all break a stale lock only on proof of death
and treat an unreadable or foreign owner as still holding.
"""
if hostname != host_name():
return False
if not process_alive(pid):
return True
if start_token is None:
return False
current = process_start_token(pid)
return current is not None and current != start_token
if sys.platform == "win32": # pragma: win32 cover
import ctypes
from ctypes import wintypes
_KERNEL32: Final[ctypes.WinDLL] = ctypes.WinDLL("kernel32", use_last_error=True)
_KERNEL32.CloseHandle.argtypes = [wintypes.HANDLE]
_KERNEL32.CloseHandle.restype = wintypes.BOOL
_KERNEL32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
_KERNEL32.OpenProcess.restype = wintypes.HANDLE
_KERNEL32.GetProcessTimes.argtypes = [
wintypes.HANDLE,
ctypes.POINTER(wintypes.FILETIME),
ctypes.POINTER(wintypes.FILETIME),
ctypes.POINTER(wintypes.FILETIME),
ctypes.POINTER(wintypes.FILETIME),
]
_KERNEL32.GetProcessTimes.restype = wintypes.BOOL
_WIN_SYNCHRONIZE: Final[int] = 0x100000
_WIN_PROCESS_QUERY_LIMITED_INFORMATION: Final[int] = 0x1000
_WIN_ERROR_INVALID_PARAMETER: Final[int] = 87
_WIN_INHERIT_HANDLE: Final[bool] = False
def process_alive(pid: int) -> bool:
"""Whether a process with this PID exists, treating an access denial as proof it does."""
handle = _KERNEL32.OpenProcess(_WIN_SYNCHRONIZE, _WIN_INHERIT_HANDLE, pid)
if handle:
_KERNEL32.CloseHandle(handle)
return True
return ctypes.get_last_error() != _WIN_ERROR_INVALID_PARAMETER
def process_start_token(pid: int) -> int | None:
"""The process creation FILETIME as a 100ns tick count, or ``None`` when it cannot be read."""
handle = _KERNEL32.OpenProcess(_WIN_PROCESS_QUERY_LIMITED_INFORMATION, _WIN_INHERIT_HANDLE, pid)
if not handle:
return None
creation, exit_time, kernel_time, user_time = (wintypes.FILETIME() for _ in range(4))
try:
if not _KERNEL32.GetProcessTimes(
handle,
ctypes.byref(creation),
ctypes.byref(exit_time),
ctypes.byref(kernel_time),
ctypes.byref(user_time),
):
return None # pragma: no cover # win32 GetProcessTimes failure path; not reproducible on a live handle
finally:
_KERNEL32.CloseHandle(handle)
return (creation.dwHighDateTime << 32) | creation.dwLowDateTime
else: # pragma: win32 no cover
def process_alive(pid: int) -> bool:
"""Whether a process with this PID exists, treating an access denial (``EPERM``) as proof it does."""
try:
os.kill(pid, 0)
except OSError as error:
if error.errno == ESRCH:
return False
if error.errno == EPERM:
return True
raise
return True
if sys.platform in {"linux", "android"}: # pragma: linux cover
# Termux/Android reports sys.platform == "android" but runs the Linux kernel, so /proc/<pid>/stat and the boot
# id are the same reliable start-time source; treat it exactly like Linux rather than the tokenless fallback.
# comm (field 2) is wrapped in parentheses and may itself contain spaces or a ')', so the fixed fields start
# after the final ')'. starttime is field 22 overall, the twentieth of those trailing fields (index 19).
_STARTTIME_INDEX: Final[int] = 19
def _read_boot_id() -> int:
# starttime is measured in clock ticks since boot, so on its own it repeats across a reboot. Folding the
# boot id into the high bits makes the Linux token reboot-safe like the absolute clocks macOS and Windows
# expose, while staying a single integer so a 3.29 reader still parses the third marker line. 0 when the
# kernel does not expose a boot id degrades to bare starttime, which stays safe (a reboot collision fails
# closed rather than reclaiming a live marker).
try:
boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii")
return int(boot_id.strip().replace("-", ""), 16)
except (OSError, ValueError): # pragma: no cover # the kernel always exposes boot_id as a UUID on Linux
return 0
_BOOT_ID: Final[int] = _read_boot_id()
def process_start_token(pid: int) -> int | None:
"""The ``/proc/<pid>/stat`` ``starttime`` folded with the boot id, or ``None`` when the process is gone."""
try:
data = Path(f"/proc/{pid}/stat").read_bytes()
except OSError:
return None
# psutil identifies a process by the same (pid, starttime) pair; the boot id extends that across reboots.
fields = data[data.rfind(b")") + 1 :].split()
if len(fields) <= _STARTTIME_INDEX: # pragma: no cover # a truncated /proc read, never seen in practice
return None
try:
starttime = int(fields[_STARTTIME_INDEX])
except ValueError: # pragma: no cover # /proc always renders starttime as an integer
return None
return (_BOOT_ID << 64) | starttime
elif sys.platform == "darwin": # pragma: darwin cover
import ctypes
import struct
_LIBC: Final[ctypes.CDLL] = ctypes.CDLL(None, use_errno=True)
_CTL_KERN: Final[int] = 1
_KERN_PROC: Final[int] = 14
_KERN_PROC_PID: Final[int] = 1
# kinfo_proc opens with kp_proc.p_starttime (a struct timeval) at offset 0: int64 seconds, int32 microseconds.
# The offset is fixed by the struct chain kinfo_proc -> extern_proc -> p_un, so a read at 0 is not a guess.
_TIMEVAL_AT_ZERO: Final[str] = "<qi"
_TIMEVAL_SIZE: Final[int] = struct.calcsize(_TIMEVAL_AT_ZERO)
def process_start_token(pid: int) -> int | None:
"""The process start time in microseconds from ``sysctl(KERN_PROC_PID)``, or ``None`` when it is gone."""
mib = (ctypes.c_int * 4)(_CTL_KERN, _KERN_PROC, _KERN_PROC_PID, pid)
length = ctypes.c_size_t(0)
# The size probe reports the kinfo_proc size for any PID, so the read below, not the probe, tells a live
# process from a gone one: a gone PID leaves the fetch a zero-length success, so re-check the length after.
if _LIBC.sysctl(mib, 4, None, ctypes.byref(length), None, 0) != 0:
return None
buffer = (ctypes.c_char * length.value)()
if _LIBC.sysctl(mib, 4, buffer, ctypes.byref(length), None, 0) != 0 or length.value < _TIMEVAL_SIZE:
return None
seconds, microseconds = struct.unpack_from(_TIMEVAL_AT_ZERO, buffer.raw, 0)
return seconds * 1_000_000 + microseconds
else: # pragma: no cover # a POSIX platform without a proven start-time source falls back to fail-closed liveness
def process_start_token(pid: int) -> int | None:
"""No proven start-time source, so the owner carries no token and liveness rests on the PID alone."""
del pid
return None
__all__ = [
"host_name",
"owner_is_stale",
"process_alive",
"process_start_token",
]
|