Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_identity.py: 22%

1from __future__ import annotations 

2 

3import os 

4import socket 

5import sys 

6from errno import EPERM, ESRCH 

7from pathlib import Path 

8from typing import Final 

9 

10 

11def host_name() -> str: 

12 """The hostname recorded alongside an owner, so a marker written on another machine is never probed here.""" 

13 return socket.gethostname() 

14 

15 

16def owner_is_stale(pid: int, hostname: str, start_token: int | None) -> bool: 

17 """ 

18 Whether the recorded owner is provably gone, so reclaiming its marker cannot detach a live holder. 

19 

20 Fail closed: return ``True`` only when this process can prove the exact recorded owner is dead. A marker from 

21 another host cannot be probed; a live PID whose start token still matches, or whose token cannot be read, is the 

22 holder or is indistinguishable from it; a live PID whose start token differs is a recycled PID, so the process that 

23 wrote the marker is gone. PostgreSQL, Qt ``QLockFile`` and Mercurial all break a stale lock only on proof of death 

24 and treat an unreadable or foreign owner as still holding. 

25 """ 

26 if hostname != host_name(): 

27 return False 

28 if not process_alive(pid): 

29 return True 

30 if start_token is None: 

31 return False 

32 current = process_start_token(pid) 

33 return current is not None and current != start_token 

34 

35 

36if sys.platform == "win32": # pragma: win32 cover 

37 import ctypes 

38 from ctypes import wintypes 

39 

40 _KERNEL32: Final[ctypes.WinDLL] = ctypes.WinDLL("kernel32", use_last_error=True) 

41 _KERNEL32.CloseHandle.argtypes = [wintypes.HANDLE] 

42 _KERNEL32.CloseHandle.restype = wintypes.BOOL 

43 _KERNEL32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] 

44 _KERNEL32.OpenProcess.restype = wintypes.HANDLE 

45 _KERNEL32.GetProcessTimes.argtypes = [ 

46 wintypes.HANDLE, 

47 ctypes.POINTER(wintypes.FILETIME), 

48 ctypes.POINTER(wintypes.FILETIME), 

49 ctypes.POINTER(wintypes.FILETIME), 

50 ctypes.POINTER(wintypes.FILETIME), 

51 ] 

52 _KERNEL32.GetProcessTimes.restype = wintypes.BOOL 

53 

54 _WIN_SYNCHRONIZE: Final[int] = 0x100000 

55 _WIN_PROCESS_QUERY_LIMITED_INFORMATION: Final[int] = 0x1000 

56 _WIN_ERROR_INVALID_PARAMETER: Final[int] = 87 

57 _WIN_INHERIT_HANDLE: Final[bool] = False 

58 

59 def process_alive(pid: int) -> bool: 

60 """Whether a process with this PID exists, treating an access denial as proof it does.""" 

61 handle = _KERNEL32.OpenProcess(_WIN_SYNCHRONIZE, _WIN_INHERIT_HANDLE, pid) 

62 if handle: 

63 _KERNEL32.CloseHandle(handle) 

64 return True 

65 return ctypes.get_last_error() != _WIN_ERROR_INVALID_PARAMETER 

66 

67 def process_start_token(pid: int) -> int | None: 

68 """The process creation FILETIME as a 100ns tick count, or ``None`` when it cannot be read.""" 

69 handle = _KERNEL32.OpenProcess(_WIN_PROCESS_QUERY_LIMITED_INFORMATION, _WIN_INHERIT_HANDLE, pid) 

70 if not handle: 

71 return None 

72 creation, exit_time, kernel_time, user_time = (wintypes.FILETIME() for _ in range(4)) 

73 try: 

74 if not _KERNEL32.GetProcessTimes( 

75 handle, 

76 ctypes.byref(creation), 

77 ctypes.byref(exit_time), 

78 ctypes.byref(kernel_time), 

79 ctypes.byref(user_time), 

80 ): 

81 return None # pragma: no cover # win32 GetProcessTimes failure path; not reproducible on a live handle 

82 finally: 

83 _KERNEL32.CloseHandle(handle) 

84 return (creation.dwHighDateTime << 32) | creation.dwLowDateTime 

85 

86else: # pragma: win32 no cover 

87 

88 def process_alive(pid: int) -> bool: 

89 """Whether a process with this PID exists, treating an access denial (``EPERM``) as proof it does.""" 

90 try: 

91 os.kill(pid, 0) 

92 except OSError as error: 

93 if error.errno == ESRCH: 

94 return False 

95 if error.errno == EPERM: 

96 return True 

97 raise 

98 return True 

99 

100 if sys.platform in {"linux", "android"}: # pragma: linux cover 

101 # Termux/Android reports sys.platform == "android" but runs the Linux kernel, so /proc/<pid>/stat and the boot 

102 # id are the same reliable start-time source; treat it exactly like Linux rather than the tokenless fallback. 

103 # comm (field 2) is wrapped in parentheses and may itself contain spaces or a ')', so the fixed fields start 

104 # after the final ')'. starttime is field 22 overall, the twentieth of those trailing fields (index 19). 

105 _STARTTIME_INDEX: Final[int] = 19 

106 

107 def _read_boot_id() -> int: 

108 # starttime is measured in clock ticks since boot, so on its own it repeats across a reboot. Folding the 

109 # boot id into the high bits makes the Linux token reboot-safe like the absolute clocks macOS and Windows 

110 # expose, while staying a single integer so a 3.29 reader still parses the third marker line. 0 when the 

111 # kernel does not expose a boot id degrades to bare starttime, which stays safe (a reboot collision fails 

112 # closed rather than reclaiming a live marker). 

113 try: 

114 boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii") 

115 return int(boot_id.strip().replace("-", ""), 16) 

116 except (OSError, ValueError): # pragma: no cover # the kernel always exposes boot_id as a UUID on Linux 

117 return 0 

118 

119 _BOOT_ID: Final[int] = _read_boot_id() 

120 

121 def process_start_token(pid: int) -> int | None: 

122 """The ``/proc/<pid>/stat`` ``starttime`` folded with the boot id, or ``None`` when the process is gone.""" 

123 try: 

124 data = Path(f"/proc/{pid}/stat").read_bytes() 

125 except OSError: 

126 return None 

127 # psutil identifies a process by the same (pid, starttime) pair; the boot id extends that across reboots. 

128 fields = data[data.rfind(b")") + 1 :].split() 

129 if len(fields) <= _STARTTIME_INDEX: # pragma: no cover # a truncated /proc read, never seen in practice 

130 return None 

131 try: 

132 starttime = int(fields[_STARTTIME_INDEX]) 

133 except ValueError: # pragma: no cover # /proc always renders starttime as an integer 

134 return None 

135 return (_BOOT_ID << 64) | starttime 

136 

137 elif sys.platform == "darwin": # pragma: darwin cover 

138 import ctypes 

139 import struct 

140 

141 _LIBC: Final[ctypes.CDLL] = ctypes.CDLL(None, use_errno=True) 

142 _CTL_KERN: Final[int] = 1 

143 _KERN_PROC: Final[int] = 14 

144 _KERN_PROC_PID: Final[int] = 1 

145 # kinfo_proc opens with kp_proc.p_starttime (a struct timeval) at offset 0: int64 seconds, int32 microseconds. 

146 # The offset is fixed by the struct chain kinfo_proc -> extern_proc -> p_un, so a read at 0 is not a guess. 

147 _TIMEVAL_AT_ZERO: Final[str] = "<qi" 

148 _TIMEVAL_SIZE: Final[int] = struct.calcsize(_TIMEVAL_AT_ZERO) 

149 

150 def process_start_token(pid: int) -> int | None: 

151 """The process start time in microseconds from ``sysctl(KERN_PROC_PID)``, or ``None`` when it is gone.""" 

152 mib = (ctypes.c_int * 4)(_CTL_KERN, _KERN_PROC, _KERN_PROC_PID, pid) 

153 length = ctypes.c_size_t(0) 

154 # The size probe reports the kinfo_proc size for any PID, so the read below, not the probe, tells a live 

155 # process from a gone one: a gone PID leaves the fetch a zero-length success, so re-check the length after. 

156 if _LIBC.sysctl(mib, 4, None, ctypes.byref(length), None, 0) != 0: 

157 return None 

158 buffer = (ctypes.c_char * length.value)() 

159 if _LIBC.sysctl(mib, 4, buffer, ctypes.byref(length), None, 0) != 0 or length.value < _TIMEVAL_SIZE: 

160 return None 

161 seconds, microseconds = struct.unpack_from(_TIMEVAL_AT_ZERO, buffer.raw, 0) 

162 return seconds * 1_000_000 + microseconds 

163 

164 else: # pragma: no cover # a POSIX platform without a proven start-time source falls back to fail-closed liveness 

165 

166 def process_start_token(pid: int) -> int | None: 

167 """No proven start-time source, so the owner carries no token and liveness rests on the PID alone.""" 

168 del pid 

169 return None 

170 

171 

172__all__ = [ 

173 "host_name", 

174 "owner_is_stale", 

175 "process_alive", 

176 "process_start_token", 

177]