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

1from __future__ import annotations 

2 

3import os 

4import secrets 

5import stat 

6import sys 

7from errno import EACCES, EISDIR 

8from pathlib import Path 

9 

10 

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

12 """ 

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

14 

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

16 

17 :param filename: file to check 

18 

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

20 

21 """ 

22 try: 

23 # lstat, not stat: it settles exists-and-writable in one syscall, and a hostile symlink planted at the lock 

24 # path would otherwise make this inspect the link target, letting an attacker turn a contended acquire into a 

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

26 # O_NOFOLLOW and refuses the symlink anyway. 

27 file_stat = os.lstat(filename) 

28 except OSError: 

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

30 

31 # No mtime guard: the old `if st_mtime != 0` skip existed for NFS/Linux quirks where os.lstat could return an 

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

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

34 if not (file_stat.st_mode & stat.S_IWUSR): 

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

36 

37 if stat.S_ISDIR(file_stat.st_mode): 

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

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

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

41 

42 

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

44 """ 

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

46 

47 :param filename: file. 

48 

49 """ 

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

51 

52 

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

54 """ 

55 Atomically break a stale lock file that was judged stale at modification time *mtime_before*. 

56 

57 The file is renamed to a process-private name before being unlinked, so two processes breaking the same lock 

58 cannot delete each other's work (only one rename of a given inode succeeds; the loser gets ``OSError``). After the 

59 rename the file is re-checked: a newer modification time, or a different inode than *ino_before*, means a peer 

60 recreated the lock between the stale decision and the rename, so we grabbed a live file and must abort, leaving the 

61 renamed file in place rather than rolling back (a rollback rename is itself racy — same trade-off as the soft 

62 read/write marker break). The inode check matters because filesystems with coarse modification-time granularity 

63 (NFS, FAT) can give a same-second recreation the old mtime, so mtime alone would not catch it and a live lock would 

64 be unlinked; the inode is the reliable identity, mirroring the token re-check in the soft read/write marker break. 

65 ``lstat`` is used so a hostile symlink swapped in after the decision is not followed. 

66 

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

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

69 path in the window between the re-verify ``lstat`` above and the ``unlink`` below, so we would delete a live lock 

70 the inode check just approved. A private name means nobody else can target our break path, matching the soft 

71 read/write marker break. 

72 

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

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

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

76 

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

78 

79 """ 

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

81 Path(lock_file).rename(break_path) 

82 try: 

83 st_after = os.lstat(break_path) 

84 except OSError: 

85 return 

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

87 return 

88 Path(break_path).unlink() 

89 

90 

91__all__ = [ 

92 "break_lock_file", 

93 "ensure_directory_exists", 

94 "raise_on_not_writable_file", 

95]