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

1from __future__ import annotations 

2 

3import os 

4import secrets 

5import stat 

6import sys 

7from errno import EACCES, EIO, EISDIR 

8from pathlib import Path 

9from typing import Final 

10 

11 

12def write_all(fd: int, data: bytes) -> None: 

13 """ 

14 Write the whole buffer to *fd*, looping over the short writes ``os.write`` is allowed to make. 

15 

16 A marker written with a bare ``os.write`` can land partially: a peer reading it mid-write parses a truncated record 

17 as malformed or as a foreign holder. Looping until the buffer drains keeps the record atomic in the process and 

18 kernel view. No ``fsync``: filelock needs a complete record, not crash-durable storage. 

19 

20 :param fd: file descriptor open for writing. 

21 :param data: bytes to write in full. 

22 

23 :raises OSError: if a write reports zero progress before the record is complete. 

24 

25 """ 

26 remaining = memoryview(data) 

27 while remaining: 

28 if (written := os.write(fd, remaining)) == 0: 

29 raise OSError(EIO, "os.write wrote 0 bytes before the record was complete") 

30 remaining = remaining[written:] 

31 

32 

33def raise_on_not_writable_file(filename: str) -> None: 

34 """ 

35 Raise an exception if attempting to open the file for writing would fail. 

36 

37 Separates files that can never be written from files that are writable but currently locked. 

38 

39 :param filename: file to check 

40 

41 :raises OSError: as if the file was opened for writing. 

42 

43 """ 

44 try: 

45 # lstat, not stat: settles exists-and-writable in one syscall, and a hostile symlink at the lock path would 

46 # make stat inspect the link target, letting an attacker turn a contended acquire into a misleading 

47 # PermissionError / IsADirectoryError and probe that target's attributes. The real open passes O_NOFOLLOW and 

48 # refuses the symlink anyway. 

49 file_stat = os.lstat(filename) 

50 except OSError: 

51 return # does not exist, or an error the caller cannot act on 

52 

53 # No mtime guard: the old `if st_mtime != 0` skip covered NFS/Linux quirks where os.lstat returned an all-zero 

54 # struct, which it no longer does. Skipping on mtime 0 let a read-only file or a directory at the lock path pass 

55 # as missing, so acquire() blocked forever on an open that cannot succeed. 

56 # Match open() credentials where supported; group permissions and ACLs can grant access without S_IWUSR. 

57 if not (file_stat.st_mode & stat.S_IWUSR) and not os.access( 

58 filename, os.W_OK, effective_ids=os.access in os.supports_effective_ids 

59 ): 

60 try: 

61 os.lstat(filename) 

62 except FileNotFoundError: 

63 # A holder may unlink its marker between lstat() and access(); let the caller attempt creation. 

64 return 

65 raise PermissionError(EACCES, "Permission denied", filename) 

66 

67 if stat.S_ISDIR(file_stat.st_mode): 

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

69 raise PermissionError(EACCES, "Permission denied", filename) 

70 raise IsADirectoryError(EISDIR, "Is a directory", filename) # pragma: win32 no cover 

71 

72 

73def ensure_directory_exists(filename: Path | str) -> None: 

74 """ 

75 Ensure the directory containing the file exists (create it if necessary). 

76 

77 :param filename: file. 

78 

79 """ 

80 Path(filename).parent.mkdir(parents=True, exist_ok=True) 

81 

82 

83def break_lock_file(lock_file: str, mtime_before: float, ino_before: int) -> None: 

84 """ 

85 Atomically break a stale lock file judged stale at modification time *mtime_before*. 

86 

87 Rename the file to a process-private name before unlinking it, so two processes breaking the same lock cannot 

88 delete each other's work: only one rename of a given inode wins, the loser gets ``OSError``. After the rename, 

89 re-check the file. A newer modification time, or a different inode than *ino_before*, means a peer recreated the 

90 lock between the stale decision and the rename, so we grabbed a live file and abort, leaving the renamed file in 

91 place. A rollback rename is itself racy, the same trade-off as the soft read/write marker break. The inode check 

92 matters because filesystems with coarse modification-time granularity (NFS, FAT) can give a same-second recreation 

93 the old mtime, so mtime alone would miss it and unlink a live lock; the inode is the reliable identity, mirroring 

94 the token re-check in the soft read/write marker break. ``lstat`` avoids following a hostile symlink swapped in 

95 after the decision. 

96 

97 The break name carries a random token so it is unguessable and unique per attempt. Without it two breakers in the 

98 same process share ``<lock>.break.<pid>``, and a second break can rename a recreated live lock onto that path in 

99 the window between the re-verify ``lstat`` above and the ``unlink`` below, deleting a live lock the inode check 

100 just approved. A private name keeps anyone else from targeting our break path, matching the soft read/write marker 

101 break. 

102 

103 :param lock_file: path to the lock file to break. 

104 :param mtime_before: modification time observed when the lock was judged stale. 

105 :param ino_before: inode number observed when the lock was judged stale. 

106 

107 :raises OSError: if the rename fails (e.g. the file vanished or is not owned in a sticky directory). 

108 

109 """ 

110 break_path = f"{lock_file}.break.{os.getpid()}.{secrets.token_hex(16)}" 

111 Path(lock_file).rename(break_path) 

112 try: 

113 st_after = os.lstat(break_path) 

114 except OSError: 

115 return 

116 if st_after.st_mtime > mtime_before or st_after.st_ino != ino_before: 

117 return 

118 Path(break_path).unlink() 

119 

120 

121def touch(name: str, *, fd: int | None = None) -> None: 

122 # Prefer the already-open, already-verified fd so a peer that swaps a symlink or a different file in at the 

123 # path after our O_NOFOLLOW read cannot redirect the touch: utime then targets the inode behind the fd. 

124 # Where the platform cannot utime an fd, fall back to a path-based touch that still refuses to follow a 

125 # symlink where supported, matching the O_NOFOLLOW reads used elsewhere here. 

126 if fd is not None and _SUPPORTS_UTIME_FD: # pragma: needs utime-fd 

127 os.utime(fd, None) 

128 return 

129 os.utime(name, None, follow_symlinks=not _SUPPORTS_UTIME_NOFOLLOW) 

130 

131 

132# Retargeting os.utime to an open fd lets a heartbeat refresh the exact inode it verified instead of whatever the 

133# pathname now names. 

134_SUPPORTS_UTIME_FD: Final[bool] = sys.platform != "win32" and os.utime in os.supports_fd 

135# os.utime follows symlinks unless told not to; not every platform can refuse the follow, so probe support. 

136_SUPPORTS_UTIME_NOFOLLOW: Final[bool] = os.utime in os.supports_follow_symlinks 

137 

138 

139__all__ = [ 

140 "break_lock_file", 

141 "ensure_directory_exists", 

142 "raise_on_not_writable_file", 

143 "touch", 

144 "write_all", 

145]