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

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

34 """ 

35 attrs = _kernel32.GetFileAttributesW(path) 

36 if attrs == INVALID_FILE_ATTRIBUTES: 

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

38 err = ctypes.get_last_error() 

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

40 return False 

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

42 return False 

43 # Some other error - let caller handle it 

44 return False 

45 return bool(attrs & FILE_ATTRIBUTE_REPARSE_POINT) 

46 

47 class WindowsFileLock(BaseFileLock): 

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

49 

50 def _acquire(self) -> None: 

51 raise_on_not_writable_file(self.lock_file) 

52 ensure_directory_exists(self.lock_file) 

53 

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

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

56 if _is_reparse_point(self.lock_file): 

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

58 raise OSError(msg) 

59 

60 flags = ( 

61 os.O_RDWR # open for read and write 

62 | os.O_CREAT # create file if not exists 

63 | os.O_TRUNC # truncate file if not empty 

64 ) 

65 try: 

66 fd = os.open(self.lock_file, flags, self._context.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 

86 with suppress(OSError): # Probably another instance of the application hat acquired the file lock. 

87 Path(self.lock_file).unlink() 

88 

89else: # pragma: win32 no cover 

90 

91 class WindowsFileLock(BaseFileLock): 

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

93 

94 def _acquire(self) -> None: 

95 raise NotImplementedError 

96 

97 def _release(self) -> None: 

98 raise NotImplementedError 

99 

100 

101__all__ = [ 

102 "WindowsFileLock", 

103]