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

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 if not (file_stat.st_mode & stat.S_IWUSR): 

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

58 

59 if stat.S_ISDIR(file_stat.st_mode): 

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

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

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

63 

64 

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

66 """ 

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

68 

69 :param filename: file. 

70 

71 """ 

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

73 

74 

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

76 """ 

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

78 

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

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

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

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

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

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

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

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

87 after the decision. 

88 

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

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

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

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

93 break. 

94 

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

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

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

98 

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

100 

101 """ 

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

103 Path(lock_file).rename(break_path) 

104 try: 

105 st_after = os.lstat(break_path) 

106 except OSError: 

107 return 

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

109 return 

110 Path(break_path).unlink() 

111 

112 

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

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

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

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

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

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

119 os.utime(fd, None) 

120 return 

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

122 

123 

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

125# pathname now names. 

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

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

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

129 

130 

131__all__ = [ 

132 "break_lock_file", 

133 "ensure_directory_exists", 

134 "raise_on_not_writable_file", 

135 "touch", 

136 "write_all", 

137]