from __future__ import annotations import os import sys from contextlib import suppress from pathlib import Path from typing import Final, cast from ._api import BaseFileLock from ._util import ensure_directory_exists, raise_on_not_writable_file if sys.platform == "win32": # pragma: win32 cover import ctypes import msvcrt from ctypes import wintypes _GENERIC_READ: Final[int] = 0x80000000 _GENERIC_WRITE: Final[int] = 0x40000000 _SYNCHRONIZE: Final[int] = 0x00100000 _DESIRED_ACCESS: Final[int] = _GENERIC_READ | _GENERIC_WRITE | _SYNCHRONIZE _FILE_SHARE_READ_WRITE: Final[int] = ( 0x00000001 | 0x00000002 ) # read | write; matches os.open (_SH_DENYNO), no delete _FILE_OPEN_IF: Final[int] = 3 # open the file if it exists, create it otherwise; the NtCreateFile OPEN_ALWAYS _FILE_ATTRIBUTE_READONLY: Final[int] = 0x00000001 _FILE_ATTRIBUTE_NORMAL: Final[int] = 0x00000080 _FILE_ATTRIBUTE_REPARSE_POINT: Final[int] = 0x00000400 # CreateOptions: keep the handle synchronous (the CRT and msvcrt.locking rely on a maintained file position), # refuse a directory, and open a reparse point rather than following it so the check below acts on the link itself. _FILE_SYNCHRONOUS_IO_NONALERT: Final[int] = 0x00000020 _FILE_NON_DIRECTORY_FILE: Final[int] = 0x00000040 _FILE_OPEN_REPARSE_POINT: Final[int] = 0x00200000 _CREATE_OPTIONS: Final[int] = _FILE_SYNCHRONOUS_IO_NONALERT | _FILE_NON_DIRECTORY_FILE | _FILE_OPEN_REPARSE_POINT _OBJ_CASE_INSENSITIVE: Final[int] = 0x00000040 # Win32 name lookups are case-insensitive _OWNER_WRITE: Final[int] = 0o200 # LockFileEx locks a byte range at an offset carried in OVERLAPPED, independent of the descriptor's file position. # msvcrt.locking starts at the current position instead, so a metadata write between lock and unlock could shift # the byte a later unlock targets; the explicit offset removes that hazard for both the path lock and #608's # descriptor lock. _LOCKFILE_FAIL_IMMEDIATELY: Final[int] = 0x00000001 _LOCKFILE_EXCLUSIVE_LOCK: Final[int] = 0x00000002 _ERROR_LOCK_VIOLATION: Final[int] = 33 # another handle holds the byte range # NtCreateFile returns the raw NTSTATUS as its value, where CreateFileW collapses several of these into one # ERROR_ACCESS_DENIED. Telling them apart is the point (#604): a name pending deletion or a share conflict is # transient and worth a retry, a real access denial is not. _STATUS_SUCCESS: Final[int] = 0x00000000 _STATUS_ACCESS_DENIED: Final[int] = 0xC0000022 _STATUS_SHARING_VIOLATION: Final[int] = 0xC0000043 _STATUS_DELETE_PENDING: Final[int] = 0xC0000056 _ntdll: Final[ctypes.WinDLL] = ctypes.WinDLL("ntdll") _kernel32: Final[ctypes.WinDLL] = ctypes.WinDLL("kernel32", use_last_error=True) class _UNICODE_STRING(ctypes.Structure): # ruff:ignore[invalid-class-name] # mirrors the Win32 struct name _fields_ = ( ("Length", wintypes.USHORT), # byte length, not character count ("MaximumLength", wintypes.USHORT), ("Buffer", wintypes.LPWSTR), ) class _OBJECT_ATTRIBUTES(ctypes.Structure): # ruff:ignore[invalid-class-name] # mirrors the Win32 struct name _fields_ = ( ("Length", wintypes.ULONG), ("RootDirectory", wintypes.HANDLE), ("ObjectName", ctypes.POINTER(_UNICODE_STRING)), ("Attributes", wintypes.ULONG), ("SecurityDescriptor", ctypes.c_void_p), ("SecurityQualityOfService", ctypes.c_void_p), ) class _IO_STATUS_BLOCK(ctypes.Structure): # ruff:ignore[invalid-class-name] # mirrors the Win32 struct name _fields_ = ( ("Status", ctypes.c_void_p), # a union of NTSTATUS and PVOID, so it is pointer-sized ("Information", ctypes.c_void_p), ) class _OVERLAPPED(ctypes.Structure): # mirrors the Win32 struct name _fields_ = ( ("Internal", ctypes.c_void_p), # ULONG_PTR: pointer-sized, not DWORD, or the x64 layout corrupts Offset ("InternalHigh", ctypes.c_void_p), ("Offset", wintypes.DWORD), # the DUMMYUNIONNAME struct, flattened: low 32 bits of the byte offset ("OffsetHigh", wintypes.DWORD), ("hEvent", wintypes.HANDLE), ) class _BY_HANDLE_FILE_INFORMATION(ctypes.Structure): # ruff:ignore[invalid-class-name] # mirrors the Win32 struct name _fields_ = ( ("dwFileAttributes", wintypes.DWORD), ("ftCreationTime", wintypes.FILETIME), ("ftLastAccessTime", wintypes.FILETIME), ("ftLastWriteTime", wintypes.FILETIME), ("dwVolumeSerialNumber", wintypes.DWORD), ("nFileSizeHigh", wintypes.DWORD), ("nFileSizeLow", wintypes.DWORD), ("nNumberOfLinks", wintypes.DWORD), ("nFileIndexHigh", wintypes.DWORD), ("nFileIndexLow", wintypes.DWORD), ) _ntdll.NtCreateFile.restype = wintypes.LONG # NTSTATUS _ntdll.NtCreateFile.argtypes = [ ctypes.POINTER(wintypes.HANDLE), wintypes.DWORD, ctypes.POINTER(_OBJECT_ATTRIBUTES), ctypes.POINTER(_IO_STATUS_BLOCK), ctypes.POINTER(ctypes.c_longlong), # PLARGE_INTEGER AllocationSize wintypes.ULONG, wintypes.ULONG, wintypes.ULONG, wintypes.ULONG, ctypes.c_void_p, wintypes.ULONG, ] _ntdll.RtlDosPathNameToNtPathName_U_WithStatus.restype = wintypes.LONG # NTSTATUS _ntdll.RtlDosPathNameToNtPathName_U_WithStatus.argtypes = [ wintypes.LPCWSTR, ctypes.POINTER(_UNICODE_STRING), ctypes.c_void_p, ctypes.c_void_p, ] _ntdll.RtlFreeUnicodeString.restype = None _ntdll.RtlFreeUnicodeString.argtypes = [ctypes.POINTER(_UNICODE_STRING)] _ntdll.RtlNtStatusToDosError.restype = wintypes.ULONG _ntdll.RtlNtStatusToDosError.argtypes = [wintypes.LONG] _kernel32.CloseHandle.argtypes = [wintypes.HANDLE] _kernel32.CloseHandle.restype = wintypes.BOOL _kernel32.GetFileInformationByHandle.argtypes = [wintypes.HANDLE, ctypes.POINTER(_BY_HANDLE_FILE_INFORMATION)] _kernel32.GetFileInformationByHandle.restype = wintypes.BOOL _kernel32.LockFileEx.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, ctypes.POINTER(_OVERLAPPED), ] _kernel32.LockFileEx.restype = wintypes.BOOL _kernel32.UnlockFileEx.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, ctypes.POINTER(_OVERLAPPED), ] _kernel32.UnlockFileEx.restype = wintypes.BOOL def _lock_fd_nonblocking(fd: int) -> bool: # One nonblocking exclusive LockFileEx attempt shared by WindowsFileLock and lock_descriptor, over the one-byte # range at offset 0. True on acquisition, False on contention, raise otherwise. The caller owns fd; the handle # from get_osfhandle belongs to the CRT descriptor and must not be closed here. overlapped = _OVERLAPPED() # zero-initialized, so Offset/OffsetHigh/hEvent are 0 flags = _LOCKFILE_EXCLUSIVE_LOCK | _LOCKFILE_FAIL_IMMEDIATELY if _kernel32.LockFileEx(msvcrt.get_osfhandle(fd), flags, 0, 1, 0, ctypes.byref(overlapped)): return True err = ctypes.get_last_error() if err == _ERROR_LOCK_VIOLATION: return False # A non-contention LockFileEx failure is not reproducible in-process. raise ctypes.WinError(err) # pragma: no cover def _unlock_fd(fd: int) -> None: overlapped = _OVERLAPPED() # the same offset 0 and one-byte length the lock used # Unlocking the exact range we hold does not fail. if not _kernel32.UnlockFileEx(msvcrt.get_osfhandle(fd), 0, 1, 0, ctypes.byref(overlapped)): # pragma: no cover raise ctypes.WinError(ctypes.get_last_error()) class WindowsFileLock(BaseFileLock): """ Uses ``LockFileEx`` to hard lock a byte range of 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 file may persist on disk, which does not affect lock correctness. """ def _acquire(self) -> None: raise_on_not_writable_file(self.lock_file) ensure_directory_exists(self.lock_file) # The reparse test is bound to the opened handle, so a symlink or junction swapped in cannot defeat it # through a check-then-open TOCTOU race. fd = _open_non_reparse_fd(self.lock_file, self._open_mode()) if fd is None: return # open contention (share conflict or a name pending deletion); let the retry loop try again try: locked = _lock_fd_nonblocking(fd) if locked: self._mark_descriptor_owned(fd) except BaseException: # pragma: no cover # cleanup only if the lock attempt itself raises os.close(fd) raise if not locked: os.close(fd) # another holder owns the byte-range lock; let the retry loop try again def _release(self) -> None: fd = cast("int", self._context.lock_file_fd) # Retain the descriptor until the OS unlock succeeds: if UnlockFileEx raises, the byte-range lock is still # held, so is_locked must keep reporting held rather than losing the fd. Only after the unlock commits do # close and unlink run as post-unlock cleanup; their failure cannot make the lock held again. _unlock_fd(fd) self._mark_descriptor_released() self._close_released_fd(fd, default_suppresses=False) if not self._preserve_lock_file: # preserve_lock_file keeps a stable file identity for the caller (#605) with suppress(OSError): Path(self.lock_file).unlink() def _open_non_reparse_fd(path: str, mode: int) -> int | None: """ Open *path* for locking while refusing reparse points, bound to the handle actually locked. The file is opened through ``NtCreateFile`` with ``FILE_OPEN_REPARSE_POINT`` so a symlink or junction planted at the path is not followed, and the reparse decision is read from *that* handle via ``GetFileInformationByHandle`` rather than from a prior pathname query. Reading the held handle closes the check-then-open race: an attacker cannot swap the path between validation and use because both act on the same handle. Share mode omits delete so a peer cannot unlink or rename the file out from under a live holder, matching ``os.open``'s ``_SH_DENYNO``. ``NtCreateFile`` is used instead of ``CreateFileW`` because its return value carries the raw ``NTSTATUS``. Windows collapses a transient delete-pending name and a permanent access denial into the same Win32 ``ERROR_ACCESS_DENIED``; the status keeps them apart, so a real denial fails fast instead of spinning until the caller's timeout (#604). The reparse option only guards the final path component; Windows still follows reparse points in intermediate directories. This assumes the lock file sits in a lock directory untrusted users cannot modify. A path with attacker-controlled parent directories would need component-by-component handle validation. :param path: the lock file path. :param mode: the permission mode; as ``os.open`` does on Windows, a cleared owner-write bit creates the file read-only. The attribute only takes effect when the file is created, not when an existing one is opened. :returns: a file descriptor owning the opened handle, or ``None`` on a sharing violation or a delete-pending name the caller should treat as contention and retry. :raises OSError: if the path resolves to a reparse point, or the open fails for any other reason, raised with the Win32 error the status maps to. """ # Emit the audit event os.open would, so consumers watching "open" still see the path-level open and can veto. sys.audit("open", path, None, os.O_RDWR | os.O_CREAT) handle, status = _nt_open(path, read_only=not mode & _OWNER_WRITE) if status != _STATUS_SUCCESS: if status in {_STATUS_SHARING_VIOLATION, _STATUS_DELETE_PENDING}: return None winerror = _ntdll.RtlNtStatusToDosError(status) raise OSError(None, ctypes.FormatError(winerror).strip(), path, winerror) info = _BY_HANDLE_FILE_INFORMATION() # Querying an open handle we just created does not fail. if not _kernel32.GetFileInformationByHandle(handle, ctypes.byref(info)): # pragma: no cover err = ctypes.get_last_error() _kernel32.CloseHandle(handle) raise ctypes.WinError(err) if info.dwFileAttributes & _FILE_ATTRIBUTE_REPARSE_POINT: _kernel32.CloseHandle(handle) msg = f"Lock file is a reparse point (symlink/junction): {path}" raise OSError(msg) try: # O_NOINHERIT mirrors os.open on Windows: the lock fd must not leak into child processes. return msvcrt.open_osfhandle(handle, os.O_RDWR | os.O_NOINHERIT) except BaseException: # pragma: no cover # open_osfhandle audits too; a hook raising must not leak the handle _kernel32.CloseHandle(handle) raise def _nt_open(path: str, *, read_only: bool) -> tuple[int, int]: """ Open *path* through ``NtCreateFile`` and return ``(handle, status)``. ``RtlDosPathNameToNtPathName_U_WithStatus`` translates the Win32 path to the NT namespace, handling relative, drive, UNC and extended-length path forms as Win32 itself would, and allocates a buffer that ``RtlFreeUnicodeString`` releases. The handle is ``0`` unless the status is ``STATUS_SUCCESS``. """ nt_name = _UNICODE_STRING() status = _ntdll.RtlDosPathNameToNtPathName_U_WithStatus(path, ctypes.byref(nt_name), None, None) & 0xFFFFFFFF if status != _STATUS_SUCCESS: return 0, status try: attributes = _OBJECT_ATTRIBUTES() attributes.Length = ctypes.sizeof(_OBJECT_ATTRIBUTES) attributes.ObjectName = ctypes.pointer(nt_name) attributes.Attributes = _OBJ_CASE_INSENSITIVE handle = wintypes.HANDLE() io_status = _IO_STATUS_BLOCK() status = ( _ntdll.NtCreateFile( ctypes.byref(handle), _DESIRED_ACCESS, ctypes.byref(attributes), ctypes.byref(io_status), None, _FILE_ATTRIBUTE_READONLY if read_only else _FILE_ATTRIBUTE_NORMAL, _FILE_SHARE_READ_WRITE, _FILE_OPEN_IF, _CREATE_OPTIONS, None, 0, ) & 0xFFFFFFFF ) finally: _ntdll.RtlFreeUnicodeString(ctypes.byref(nt_name)) if status != _STATUS_SUCCESS: return 0, status return handle.value or 0, status else: # pragma: win32 no cover class WindowsFileLock(BaseFileLock): """Uses ``LockFileEx`` to hard lock a byte range of the lock file on Windows systems.""" def _acquire(self) -> None: raise NotImplementedError def _release(self) -> None: raise NotImplementedError __all__ = [ "WindowsFileLock", ]