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

1from __future__ import annotations 

2 

3import os 

4import sys 

5from contextlib import suppress 

6from errno import EACCES 

7from pathlib import Path 

8from typing import cast 

9 

10from ._api import BaseFileLock 

11from ._util import ensure_directory_exists, raise_on_not_writable_file 

12 

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

14 import ctypes 

15 import msvcrt 

16 from ctypes import wintypes 

17 

18 # Windows API constants for reparse point detection 

19 FILE_ATTRIBUTE_REPARSE_POINT = 0x00000400 

20 INVALID_FILE_ATTRIBUTES = 0xFFFFFFFF 

21 

22 # Load kernel32.dll 

23 _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) 

24 _kernel32.GetFileAttributesW.argtypes = [wintypes.LPCWSTR] 

25 _kernel32.GetFileAttributesW.restype = wintypes.DWORD 

26 

27 def _is_reparse_point(path: str) -> bool: 

28 """ 

29 Check if a path is a reparse point (symlink, junction, etc.) on Windows. 

30 

31 :param path: Path to check 

32 

33 :returns: True if path is a reparse point, False otherwise 

34 

35 :raises OSError: If GetFileAttributesW fails for reasons other than file-not-found 

36 

37 """ 

38 attrs = _kernel32.GetFileAttributesW(path) 

39 if attrs == INVALID_FILE_ATTRIBUTES: 

40 # File doesn't exist yet - that's fine, we'll create it 

41 err = ctypes.get_last_error() 

42 if err == 2: # noqa: PLR2004 # ERROR_FILE_NOT_FOUND 

43 return False 

44 if err == 3: # noqa: PLR2004 # ERROR_PATH_NOT_FOUND 

45 return False 

46 # Some other error - let caller handle it 

47 return False 

48 return bool(attrs & FILE_ATTRIBUTE_REPARSE_POINT) 

49 

50 class WindowsFileLock(BaseFileLock): 

51 """ 

52 Uses the :func:`msvcrt.locking` function to hard lock the lock file on Windows systems. 

53 

54 Lock file cleanup: Windows attempts to delete the lock file after release, but deletion is 

55 not guaranteed in multi-threaded scenarios where another thread holds an open handle. The lock 

56 file may persist on disk, which does not affect lock correctness. 

57 """ 

58 

59 def _acquire(self) -> None: 

60 raise_on_not_writable_file(self.lock_file) 

61 ensure_directory_exists(self.lock_file) 

62 

63 # Security check: Refuse to open reparse points (symlinks, junctions) 

64 # This prevents TOCTOU symlink attacks (CVE-TBD) 

65 if _is_reparse_point(self.lock_file): 

66 msg = f"Lock file is a reparse point (symlink/junction): {self.lock_file}" 

67 raise OSError(msg) 

68 

69 flags = ( 

70 os.O_RDWR # open for read and write 

71 | os.O_CREAT # create file if not exists 

72 ) 

73 try: 

74 fd = os.open(self.lock_file, flags, self._open_mode()) 

75 except OSError as exception: 

76 if exception.errno != EACCES: # has no access to this lock 

77 raise 

78 else: 

79 try: 

80 msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) 

81 except OSError as exception: 

82 os.close(fd) # close file first 

83 if exception.errno != EACCES: # file is already locked 

84 raise 

85 else: 

86 self._context.lock_file_fd = fd 

87 

88 def _release(self) -> None: 

89 fd = cast("int", self._context.lock_file_fd) 

90 self._context.lock_file_fd = None 

91 msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) 

92 os.close(fd) 

93 

94 with suppress(OSError): 

95 Path(self.lock_file).unlink() 

96 

97else: # pragma: win32 no cover 

98 

99 class WindowsFileLock(BaseFileLock): 

100 """Uses the :func:`msvcrt.locking` function to hard lock the lock file on Windows systems.""" 

101 

102 def _acquire(self) -> None: 

103 raise NotImplementedError 

104 

105 def _release(self) -> None: 

106 raise NotImplementedError 

107 

108 

109__all__ = [ 

110 "WindowsFileLock", 

111]