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#: Every marker format in this package caps the hostname at the RFC 1123 limit and reads a longer one as malformed. 

11_HOST_NAME_LIMIT: Final[int] = 253 

12#: The bytes those formats carry verbatim: printable non-space ASCII, less the ``?`` reserved for the escape. 

13_HOST_NAME_VERBATIM: Final[frozenset[int]] = frozenset(range(0x21, 0x7F)) - {ord("?")} 

14 

15 

16def host_name() -> str: 

17 """ 

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

19 

20 Marker formats here hold the field to printable non-space ASCII, while ``socket.gethostname()`` reports whatever 

21 the kernel stores: a space, a newline that forges an extra marker line, or a byte no codec encodes, which Python 

22 hands back as a surrogate. A holder that publishes one raw writes a marker it cannot read back, and loses the lock 

23 to the first peer that ages it out as malformed. ``?`` is illegal in a hostname, so escaping it along with every 

24 out-of-grammar byte as ``?<hex>`` leaves a real name untouched and still keeps two hosts apart, which 

25 :func:`owner_is_stale` relies on to refuse to probe a foreign PID. 

26 """ 

27 name = "" 

28 for byte in socket.gethostname().encode("utf-8", "surrogateescape"): 

29 piece = chr(byte) if byte in _HOST_NAME_VERBATIM else f"?{byte:02x}" 

30 if len(name) + len(piece) > _HOST_NAME_LIMIT: 

31 break 

32 name += piece 

33 # An escape is three characters and a kept byte is never '?', so a bare '?' can only mean an empty hostname. 

34 return name or "?" 

35 

36 

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

38 """ 

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

40 

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

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

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

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

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

46 """ 

47 if hostname != host_name(): 

48 return False 

49 if not process_alive(pid): 

50 return True 

51 if start_token is None: 

52 return False 

53 current = process_start_token(pid) 

54 return current is not None and current != start_token 

55 

56 

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

58 import ctypes 

59 from ctypes import wintypes 

60 

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

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

63 _KERNEL32.CloseHandle.restype = wintypes.BOOL 

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

65 _KERNEL32.OpenProcess.restype = wintypes.HANDLE 

66 _KERNEL32.GetProcessTimes.argtypes = [ 

67 wintypes.HANDLE, 

68 ctypes.POINTER(wintypes.FILETIME), 

69 ctypes.POINTER(wintypes.FILETIME), 

70 ctypes.POINTER(wintypes.FILETIME), 

71 ctypes.POINTER(wintypes.FILETIME), 

72 ] 

73 _KERNEL32.GetProcessTimes.restype = wintypes.BOOL 

74 

75 _WIN_SYNCHRONIZE: Final[int] = 0x100000 

76 _WIN_PROCESS_QUERY_LIMITED_INFORMATION: Final[int] = 0x1000 

77 _WIN_ERROR_INVALID_PARAMETER: Final[int] = 87 

78 _WIN_INHERIT_HANDLE: Final[bool] = False 

79 

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

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

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

83 if handle: 

84 _KERNEL32.CloseHandle(handle) 

85 return True 

86 return ctypes.get_last_error() != _WIN_ERROR_INVALID_PARAMETER 

87 

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

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

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

91 if not handle: 

92 return None 

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

94 try: 

95 if not _KERNEL32.GetProcessTimes( 

96 handle, 

97 ctypes.byref(creation), 

98 ctypes.byref(exit_time), 

99 ctypes.byref(kernel_time), 

100 ctypes.byref(user_time), 

101 ): 

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

103 finally: 

104 _KERNEL32.CloseHandle(handle) 

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

106 

107else: # pragma: win32 no cover 

108 

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

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

111 try: 

112 os.kill(pid, 0) 

113 except OSError as error: 

114 if error.errno == ESRCH: 

115 return False 

116 if error.errno == EPERM: 

117 return True 

118 raise 

119 return True 

120 

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

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

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

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

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

126 _STARTTIME_INDEX: Final[int] = 19 

127 

128 def _read_boot_id() -> int: 

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

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

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

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

133 # closed rather than reclaiming a live marker). 

134 try: 

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

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

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

138 return 0 

139 

140 _BOOT_ID: Final[int] = _read_boot_id() 

141 

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

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

144 try: 

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

146 except OSError: 

147 return None 

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

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

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

151 return None 

152 try: 

153 starttime = int(fields[_STARTTIME_INDEX]) 

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

155 return None 

156 return (_BOOT_ID << 64) | starttime 

157 

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

159 import ctypes 

160 import struct 

161 

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

163 _CTL_KERN: Final[int] = 1 

164 _KERN_PROC: Final[int] = 14 

165 _KERN_PROC_PID: Final[int] = 1 

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

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

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

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

170 

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

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

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

174 length = ctypes.c_size_t(0) 

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

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

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

178 return None 

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

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

181 return None 

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

183 return seconds * 1_000_000 + microseconds 

184 

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

186 

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

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

189 del pid 

190 return None 

191 

192 

193__all__ = [ 

194 "host_name", 

195 "owner_is_stale", 

196 "process_alive", 

197 "process_start_token", 

198]