MCPcopy Create free account
hub / github.com/tox-dev/filelock / WindowsFileLock

Class WindowsFileLock

src/filelock/_windows.py:50–95  ·  view source on GitHub ↗

Uses the :func:`msvcrt.locking` function to hard lock the lock file on Windows systems. Lock file cleanup: Windows attempts to delete the lock file after release, but deletion is not guaranteed in multi-threaded scenarios where another thread holds an open handle. The lock

Source from the content-addressed store, hash-verified

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

Callers

nothing calls this directly

Calls

no outgoing calls

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…