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

1from __future__ import annotations 

2 

3import os 

4import sys 

5from errno import EACCES 

6from typing import cast 

7 

8from ._api import BaseFileLock 

9from ._util import ensure_directory_exists, raise_on_not_writable_file 

10 

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

12 import ctypes 

13 import msvcrt 

14 from ctypes import wintypes 

15 

16 # Windows API constants for reparse point detection 

17 FILE_ATTRIBUTE_REPARSE_POINT = 0x00000400 

18 INVALID_FILE_ATTRIBUTES = 0xFFFFFFFF 

19 

20 # Load kernel32.dll 

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

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

23 _kernel32.GetFileAttributesW.restype = wintypes.DWORD 

24 

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

26 """ 

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

28 

29 :param path: Path to check 

30 

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

32 

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

34 

35 """ 

36 attrs = _kernel32.GetFileAttributesW(path) 

37 if attrs == INVALID_FILE_ATTRIBUTES: 

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

39 err = ctypes.get_last_error() 

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

41 return False 

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

43 return False 

44 # Some other error - let caller handle it 

45 return False 

46 return bool(attrs & FILE_ATTRIBUTE_REPARSE_POINT) 

47 

48 class WindowsFileLock(BaseFileLock): 

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

50 

51 def _acquire(self) -> None: 

52 raise_on_not_writable_file(self.lock_file) 

53 ensure_directory_exists(self.lock_file) 

54 

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

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

57 if _is_reparse_point(self.lock_file): 

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

59 raise OSError(msg) 

60 

61 flags = ( 

62 os.O_RDWR # open for read and write 

63 | os.O_CREAT # create file if not exists 

64 ) 

65 try: 

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

67 except OSError as exception: 

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

69 raise 

70 else: 

71 try: 

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

73 except OSError as exception: 

74 os.close(fd) # close file first 

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

76 raise 

77 else: 

78 self._context.lock_file_fd = fd 

79 

80 def _release(self) -> None: 

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

82 self._context.lock_file_fd = None 

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

84 os.close(fd) 

85 

86else: # pragma: win32 no cover 

87 

88 class WindowsFileLock(BaseFileLock): 

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

90 

91 def _acquire(self) -> None: 

92 raise NotImplementedError 

93 

94 def _release(self) -> None: 

95 raise NotImplementedError 

96 

97 

98__all__ = [ 

99 "WindowsFileLock", 

100]