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 :return: True if path is a reparse point, False otherwise 

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

32 """ 

33 attrs = _kernel32.GetFileAttributesW(path) 

34 if attrs == INVALID_FILE_ATTRIBUTES: 

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

36 err = ctypes.get_last_error() 

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

38 return False 

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

40 return False 

41 # Some other error - let caller handle it 

42 return False 

43 return bool(attrs & FILE_ATTRIBUTE_REPARSE_POINT) 

44 

45 class WindowsFileLock(BaseFileLock): 

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

47 

48 def _acquire(self) -> None: 

49 raise_on_not_writable_file(self.lock_file) 

50 ensure_directory_exists(self.lock_file) 

51 

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

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

54 if _is_reparse_point(self.lock_file): 

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

56 raise OSError(msg) 

57 

58 flags = ( 

59 os.O_RDWR # open for read and write 

60 | os.O_CREAT # create file if not exists 

61 ) 

62 try: 

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

64 except OSError as exception: 

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

66 raise 

67 else: 

68 try: 

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

70 except OSError as exception: 

71 os.close(fd) # close file first 

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

73 raise 

74 else: 

75 self._context.lock_file_fd = fd 

76 

77 def _release(self) -> None: 

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

79 self._context.lock_file_fd = None 

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

81 os.close(fd) 

82 

83else: # pragma: win32 no cover 

84 

85 class WindowsFileLock(BaseFileLock): 

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

87 

88 def _acquire(self) -> None: 

89 raise NotImplementedError 

90 

91 def _release(self) -> None: 

92 raise NotImplementedError 

93 

94 

95__all__ = [ 

96 "WindowsFileLock", 

97]