from __future__ import annotations __lazy_modules__ = ["filelock._util", "msvcrt", "struct"] import os import struct import sys import time from contextlib import contextmanager, suppress from errno import ENOENT, ENOTDIR, EWOULDBLOCK from functools import cache from typing import TYPE_CHECKING, Final, cast from filelock._api import BaseFileLock from filelock._util import raise_on_not_writable_file if TYPE_CHECKING: from collections.abc import Generator 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 _DELETE: Final[int] = 0x00010000 _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: Final[int] = 1 # open the file only if it exists; the NtCreateFile OPEN_EXISTING _FILE_OPEN_IF: Final[int] = 3 # open the file if it exists, create it otherwise; the NtCreateFile OPEN_ALWAYS # Attribute-only access skips share-mode checks, so the probe sees a name's deletion state rather than a reader. _FILE_READ_ATTRIBUTES: Final[int] = 0x00000080 _FILE_SHARE_ALL: Final[int] = 0x00000001 | 0x00000002 | 0x00000004 # read | write | delete _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, return at once instead of waiting for an oplock holder to answer the break, 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_DELETE_ON_CLOSE: Final[int] = 0x00001000 _FILE_COMPLETE_IF_OPLOCKED: Final[int] = 0x00000100 _FILE_OPEN_REPARSE_POINT: Final[int] = 0x00200000 _CREATE_OPTIONS: Final[int] = ( _FILE_SYNCHRONOUS_IO_NONALERT | _FILE_NON_DIRECTORY_FILE | _FILE_COMPLETE_IF_OPLOCKED | _FILE_OPEN_REPARSE_POINT ) # Listing access takes part in share-mode checks, unlike attribute-only access, so a handle that holds it without # FILE_SHARE_DELETE blocks renames and removals of the directory until it closes. _FILE_LIST_DIRECTORY: Final[int] = 0x00000001 _FILE_DIRECTORY_FILE: Final[int] = 0x00000001 _DIRECTORY_OPTIONS: Final[int] = _FILE_SYNCHRONOUS_IO_NONALERT | _FILE_DIRECTORY_FILE | _FILE_OPEN_REPARSE_POINT _OBJ_CASE_INSENSITIVE: Final[int] = 0x00000040 # Win32 name lookups are case-insensitive _OWNER_WRITE: Final[int] = 0o200 _GROUP_AND_OTHERS: Final[int] = 0o077 # The DACL os.mkdir(mode=0o700) applies, less the inheritance flags a file has no use for: protected, so the # directory's inheritable entries stay out, with full access for SYSTEM, Administrators and the owner. _OWNER_ONLY_SDDL: Final[str] = "D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)" _SDDL_REVISION_1: Final[int] = 1 _INVALID_HANDLE_VALUE: Final[int | None] = ctypes.c_void_p(-1).value # 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_ACCESS_DENIED: Final[int] = 5 _ERROR_LOCK_VIOLATION: Final[int] = 33 # another handle holds the byte range _ERROR_NOT_LOCKED: Final[int] = 158 # the byte range is not locked # 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 # A success code that comes with a handle: another handle's oplock is breaking and its holder has not answered yet. _STATUS_OPLOCK_BREAK_IN_PROGRESS: Final[int] = 0x00000108 _STATUS_ACCESS_DENIED: Final[int] = 0xC0000022 _STATUS_OBJECT_NAME_NOT_FOUND: Final[int] = 0xC0000034 _STATUS_OBJECT_PATH_NOT_FOUND: Final[int] = 0xC000003A _STATUS_SHARING_VIOLATION: Final[int] = 0xC0000043 _STATUS_DELETE_PENDING: Final[int] = 0xC0000056 #: Probes, 2 ms apart, for transient STATUS_ACCESS_DENIED while a peer unlinks the name (#604). A count ensures #: every probe runs even when the process stalls; oplock breaks use the caller's wait instead. _ACCESS_DENIED_PROBES: Final[int] = 250 _ACCESS_DENIED_GRACE: Final[float] = 0.5 _ACCESS_DENIED_RETRY: Final[float] = 0.002 _ntdll: Final[ctypes.WinDLL] = ctypes.WinDLL("ntdll") _kernel32: Final[ctypes.WinDLL] = ctypes.WinDLL("kernel32", use_last_error=True) _advapi32: Final[ctypes.WinDLL] = ctypes.WinDLL("advapi32", 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 _SECURITY_ATTRIBUTES(ctypes.Structure): # ruff:ignore[invalid-class-name] # mirrors the Win32 struct name _fields_ = ( ("nLength", wintypes.DWORD), ("lpSecurityDescriptor", ctypes.c_void_p), ("bInheritHandle", wintypes.BOOL), ) 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 _advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(wintypes.ULONG), ] _advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW.restype = wintypes.BOOL _advapi32.GetNamedSecurityInfoW.argtypes = [ wintypes.LPCWSTR, ctypes.c_int, wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(ctypes.c_void_p), ] _advapi32.GetNamedSecurityInfoW.restype = wintypes.DWORD _advapi32.OpenThreadToken.argtypes = [ wintypes.HANDLE, wintypes.DWORD, wintypes.BOOL, ctypes.POINTER(wintypes.HANDLE), ] _advapi32.OpenThreadToken.restype = wintypes.BOOL _advapi32.OpenProcessToken.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)] _advapi32.OpenProcessToken.restype = wintypes.BOOL _advapi32.GetTokenInformation.argtypes = [ wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ] _advapi32.GetTokenInformation.restype = wintypes.BOOL _kernel32.GetCurrentProcess.restype = wintypes.HANDLE _kernel32.GetCurrentThread.restype = wintypes.HANDLE _kernel32.LocalFree.argtypes = [ctypes.c_void_p] _kernel32.LocalFree.restype = ctypes.c_void_p _kernel32.CreateDirectoryW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(_SECURITY_ATTRIBUTES)] _kernel32.CreateDirectoryW.restype = wintypes.BOOL _kernel32.CreateFileW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE, ] _kernel32.CreateFileW.restype = ctypes.c_void_p 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 if ( # pragma: no cover # only an interrupted earlier release leaves our own range unlocked not _kernel32.UnlockFileEx(msvcrt.get_osfhandle(fd), 0, 1, 0, ctypes.byref(overlapped)) and (err := ctypes.get_last_error()) != _ERROR_NOT_LOCKED ): raise ctypes.WinError(err) 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_path) # 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. try: fd = _open_non_reparse_fd(self._lock_path, self._open_mode()) except BlockingIOError: return # an oplock holder has not answered its break; let the retry loop keep its deadline 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: # Closing drops the byte-range lock, so a hold already recorded for this descriptor must go with it. self._mark_descriptor_released() 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): unlink_without_oplock_wait(self._lock_path) def _open_non_reparse_fd(path: str, mode: int, *, oplock_wait: float = _ACCESS_DENIED_GRACE) -> 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. A mode without group and other bits creates it with an owner-only DACL (discretionary access control list, the Windows permission list); any other mode has no Windows counterpart and leaves the file the entries its directory hands down. Both only take effect when the file is created, not when an existing one is opened. :param oplock_wait: seconds to wait for the holder of an oplock on the file to answer its break. :returns: a file descriptor owning the opened handle, or ``None`` on a sharing violation or a delete-pending name, which the caller should treat as contention and retry. :raises BlockingIOError: if an oplock break stays unanswered for *oplock_wait* seconds. :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. An access denial is raised only once it outlasts a run of probes, since a peer unlinking the file as it releases can make the open answer ``STATUS_ACCESS_DENIED`` for a moment. """ # 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) read_only = not mode & _OWNER_WRITE owner_only = not mode & 0o077 handle, status = _nt_open(path, read_only=read_only, owner_only=owner_only) denied_probes = 0 oplock_deadline = time.monotonic() + oplock_wait while True: if status == _STATUS_ACCESS_DENIED: if denied_probes >= _ACCESS_DENIED_PROBES: break denied_probes += 1 elif status == _STATUS_OPLOCK_BREAK_IN_PROGRESS: if time.monotonic() >= oplock_deadline: break else: break time.sleep(_ACCESS_DENIED_RETRY) handle, status = _nt_open(path, read_only=read_only, owner_only=owner_only) if status == _STATUS_OPLOCK_BREAK_IN_PROGRESS: msg = "oplock break in progress" raise BlockingIOError(EWOULDBLOCK, msg, path) if status != _STATUS_SUCCESS: if status in {_STATUS_SHARING_VIOLATION, _STATUS_DELETE_PENDING}: return None raise _status_error(status, path) return _descriptor_for_non_reparse(handle, path, os.O_RDWR) def open_read_without_oplock_wait(path: str) -> int: """ Open *path* read-only and return a descriptor, without waiting for an oplock holder to answer its break. ``os.open`` waits inside the system call until the holder of an oplock on the file answers or closes its handle, so no timeout or ``blocking=False`` ends the wait. A reparse point at *path* is refused rather than followed. :param path: the file to read. :returns: a read-only file descriptor. :raises BlockingIOError: if an oplock break is in progress, which callers treat as contention. :raises FileNotFoundError: if *path* does not exist. :raises OSError: if the open fails for any other reason, or *path* is a reparse point. """ sys.audit("open", path, None, os.O_RDONLY) return _open_existing_without_oplock_wait(path, _GENERIC_READ | _SYNCHRONIZE, os.O_RDONLY) def open_for_delete_without_oplock_wait(path: str) -> int: """ Open *path* with delete access and return a descriptor, without waiting for an oplock holder to answer. The open handle shares read, write and delete, so a rename of *path* can proceed while the caller keeps it. A holder that took an oplock through another handle is asked to answer by this open, which then raises, so the caller skips the rename instead of waiting for it. :param path: the file to rename. :returns: a file descriptor owning the handle; the caller closes it. :raises BlockingIOError: if an oplock break is in progress, which callers treat as contention. :raises FileNotFoundError: if *path* does not exist. :raises OSError: if the open fails for any other reason, or *path* is a reparse point. """ return _open_existing_without_oplock_wait(path, _DELETE | _FILE_READ_ATTRIBUTES | _SYNCHRONIZE, os.O_RDONLY) def unlink_without_oplock_wait(path: str) -> None: """ Delete *path* through a handle that does not wait for an oplock holder to answer its break. ``os.unlink`` waits like ``os.open`` does. Here an unanswered break leaves the file in place, which callers accept since deleting a lock file is best effort. A symlink at *path* is deleted as a link. :param path: the file to delete. :raises FileNotFoundError: if *path* does not exist. :raises OSError: if the open fails for any other reason, such as a sharing violation. """ sys.audit("os.remove", path, None) handle, status = _nt_create( path, _DELETE | _SYNCHRONIZE, _FILE_ATTRIBUTE_NORMAL, _FILE_SHARE_ALL, _FILE_OPEN, options=_CREATE_OPTIONS | _FILE_DELETE_ON_CLOSE, ) if status == _STATUS_SUCCESS: _kernel32.CloseHandle(handle) # deletes the file elif status != _STATUS_OPLOCK_BREAK_IN_PROGRESS: raise _status_error(status, path) def _open_existing_without_oplock_wait(path: str, access: int, flags: int) -> int: handle, status = _nt_create(path, access, _FILE_ATTRIBUTE_NORMAL, _FILE_SHARE_ALL, _FILE_OPEN) if status != _STATUS_SUCCESS: raise _status_error(status, path) return _descriptor_for_non_reparse(handle, path, flags) def _status_error(status: int, path: str) -> OSError: if status == _STATUS_OPLOCK_BREAK_IN_PROGRESS: return BlockingIOError(EWOULDBLOCK, "oplock break in progress", path) if status in {_STATUS_OBJECT_NAME_NOT_FOUND, _STATUS_OBJECT_PATH_NOT_FOUND}: return FileNotFoundError(ENOENT, os.strerror(ENOENT), path) winerror = _ntdll.RtlNtStatusToDosError(status) return OSError(None, ctypes.FormatError(winerror).strip(), path, winerror) def _descriptor_for_non_reparse(handle: int, path: str, flags: int) -> int: 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, flags | 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 _open_deletable_for_reading(path: str) -> int: """ Open *path* read-only without blocking a concurrent unlink and without following a reparse point. ``os.open`` omits ``FILE_SHARE_DELETE``, so while a process has a file open that way, a delete of it fails with a sharing violation. Every contender reads a strict claim and its owner unlinks it, so a reader that keeps the claim open past the owner's retries makes the release fail. """ sys.audit("open", path, None, os.O_RDONLY) handle, status = _nt_create(path, _GENERIC_READ | _SYNCHRONIZE, 0, _FILE_SHARE_ALL, _FILE_OPEN) if status != _STATUS_SUCCESS: winerror = _ntdll.RtlNtStatusToDosError(status) raise OSError(None, ctypes.FormatError(winerror).strip(), path, winerror) try: return msvcrt.open_osfhandle(handle, os.O_RDONLY | 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 _pin_directory(path: str) -> int: """ Open the directory at *path* so that no process can rename or remove it until :func:`_unpin_directory`. Windows has no ``dir_fd``, so an operation resolves every path component by name again. A handle on each directory of a record path, opened without following a reparse point, keeps that name bound to the directory checked here until the operation ends. :raises NotADirectoryError: when *path* is a junction, a symlink or anything other than a directory. """ sys.audit("open", path, None, os.O_RDONLY) # GetFileInformationByHandle needs FILE_READ_ATTRIBUTES; listing access is what takes part in share checks. access = _FILE_LIST_DIRECTORY | _FILE_READ_ATTRIBUTES | _SYNCHRONIZE handle, status = _nt_create(path, access, 0, _FILE_SHARE_READ_WRITE, _FILE_OPEN, _DIRECTORY_OPTIONS) if status != _STATUS_SUCCESS: 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 error = ctypes.get_last_error() _kernel32.CloseHandle(handle) raise ctypes.WinError(error) if info.dwFileAttributes & _FILE_ATTRIBUTE_REPARSE_POINT: _kernel32.CloseHandle(handle) raise NotADirectoryError(ENOTDIR, "is a junction or symlink, not a directory", path) return handle def _unpin_directory(handle: int) -> None: _kernel32.CloseHandle(handle) def _is_unlinked(path: str) -> bool: """ Report whether *path* is gone or pending deletion, which ``os.open`` reports as the same EACCES as a denial. NTFS keeps an unlinked name listed, refusing every open with ``STATUS_DELETE_PENDING``, until the last handle to the file closes. Another process holding such a handle, such as an antivirus scan, can keep it that way for longer than any retry grace, yet the name's owner has already removed it. """ handle, status = _nt_create(path, _FILE_READ_ATTRIBUTES | _SYNCHRONIZE, 0, _FILE_SHARE_ALL, _FILE_OPEN) if handle: _kernel32.CloseHandle(handle) return status in {_STATUS_OBJECT_NAME_NOT_FOUND, _STATUS_DELETE_PENDING} def _nt_open(path: str, *, read_only: bool, owner_only: bool) -> tuple[int, int]: """ Open or create *path* for locking through ``NtCreateFile`` and return ``(handle, status)``. Without *owner_only* a file this call creates gets no DACL of its own and inherits its directory's ACL. """ attributes = _FILE_ATTRIBUTE_READONLY if read_only else _FILE_ATTRIBUTE_NORMAL descriptor = _owner_only_descriptor() if owner_only else None return _nt_create( path, _DESIRED_ACCESS, attributes, _FILE_SHARE_READ_WRITE, _FILE_OPEN_IF, descriptor=descriptor ) @cache def _owner_only_descriptor() -> ctypes.c_void_p: # The DACL os.mkdir(mode=0o700) applies, minus the directory-only inheritance flags: protected (P), so the # directory's inheritable entries stay out, with full access (FA) for SYSTEM, Administrators and the owner. # Converted once and never freed: the open retries every few milliseconds while it waits. descriptor = ctypes.c_void_p() if not _advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW( "D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)", 1, ctypes.byref(descriptor), None ): # pragma: no cover # a constant string converts raise ctypes.WinError(ctypes.get_last_error()) return descriptor def _nt_create( # ruff:ignore[too-many-arguments, too-many-positional-arguments] # NtCreateFile's own knobs path: str, access: int, attributes: int, share: int, disposition: int, options: int = _CREATE_OPTIONS, *, descriptor: ctypes.c_void_p | None = None, ) -> 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``: one that comes with ``STATUS_OPLOCK_BREAK_IN_PROGRESS`` is closed here, since the file is not ours until the oplock holder answers. """ 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: object_attributes = _OBJECT_ATTRIBUTES() object_attributes.Length = ctypes.sizeof(_OBJECT_ATTRIBUTES) object_attributes.ObjectName = ctypes.pointer(nt_name) object_attributes.Attributes = _OBJ_CASE_INSENSITIVE object_attributes.SecurityDescriptor = descriptor handle = wintypes.HANDLE() io_status = _IO_STATUS_BLOCK() status = ( _ntdll.NtCreateFile( ctypes.byref(handle), access, ctypes.byref(object_attributes), ctypes.byref(io_status), None, attributes, share, disposition, options, None, 0, ) & 0xFFFFFFFF ) finally: _ntdll.RtlFreeUnicodeString(ctypes.byref(nt_name)) if status == _STATUS_OPLOCK_BREAK_IN_PROGRESS: _kernel32.CloseHandle(handle) if status != _STATUS_SUCCESS: return 0, status return handle.value or 0, status def _directory_is_private(path: str) -> bool: """ Report whether only the current user, ``SYSTEM`` and ``Administrators`` can create or remove names in *path*. A directory whose owner is anyone else can have its ACL rewritten by that owner. Entries that only a child inherits count too, since the DACL says nothing more precise about who may write. """ owner, dacl, descriptor = ctypes.c_void_p(), ctypes.c_void_p(), ctypes.c_void_p() error = _advapi32.GetNamedSecurityInfoW( path, 1, # SE_FILE_OBJECT 0x1 | 0x4, # OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION ctypes.byref(owner), None, ctypes.byref(dacl), None, ctypes.byref(descriptor), ) if error == _ERROR_ACCESS_DENIED: # C:\Windows\Temp is one. Anything that cannot be proven private is not. return False if error: # pragma: no cover # the directory was opened or listed an instant earlier raise ctypes.WinError(error) try: # SYSTEM, Administrators, CREATOR OWNER and OWNER RIGHTS; the last two stand for the owner. trusted = {_sid_from_text(text) for text in ("S-1-5-18", "S-1-5-32-544", "S-1-3-0", "S-1-3-4")} trusted.add(_user_sid()) # A missing DACL grants everyone full access. return ( bool(dacl.value) and _sid_at(cast("int", owner.value)) in trusted and _name_writers(cast("int", dacl.value)) <= trusted ) finally: _kernel32.LocalFree(descriptor) def _name_writers(dacl: int) -> set[bytes]: (count,) = struct.unpack("<4xH2x", ctypes.string_at(dacl, 8)) writers = set() offset = 8 for _ in range(count): ace_type, _flags, ace_size, mask = struct.unpack(" bytes: token = wintypes.HANDLE() token_query: Final = 0x0008 open_as_self: Final = True # An impersonating thread acts as the account it impersonates, which is the one whose names the lock creates. if not ( _advapi32.OpenThreadToken(_kernel32.GetCurrentThread(), token_query, open_as_self, ctypes.byref(token)) or _advapi32.OpenProcessToken(_kernel32.GetCurrentProcess(), token_query, ctypes.byref(token)) ): # pragma: no cover # every process may query its own token raise ctypes.WinError(ctypes.get_last_error()) try: size = wintypes.DWORD() _advapi32.GetTokenInformation(token, 1, None, 0, ctypes.byref(size)) buffer = ctypes.create_string_buffer(size.value) if not _advapi32.GetTokenInformation( token, 1, buffer, size, ctypes.byref(size) ): # pragma: no cover # the buffer has the size the first call asked for raise ctypes.WinError(ctypes.get_last_error()) return _sid_at(cast("int", ctypes.c_void_p.from_buffer(buffer).value)) finally: _kernel32.CloseHandle(token) def _sid_from_text(text: str) -> bytes: revision, authority, *subauthorities = (int(part) for part in text.removeprefix("S-").split("-")) return ( struct.pack(" bytes: return ctypes.string_at(address, 8 + 4 * ctypes.string_at(address + 1, 1)[0]) def _make_private_directory(path: str) -> None: """Create *path* with the owner-only descriptor, so that nobody else can add a name to it.""" attributes = _SECURITY_ATTRIBUTES( nLength=ctypes.sizeof(_SECURITY_ATTRIBUTES), lpSecurityDescriptor=_owner_only_descriptor(), bInheritHandle=False, ) if not _kernel32.CreateDirectoryW(path, ctypes.byref(attributes)): # pragma: no cover # the name is fresh raise ctypes.WinError(ctypes.get_last_error()) @contextmanager def _pinned(path: str) -> Generator[tuple[int, int]]: """Hold *path* open without delete sharing, which keeps its name from being renamed or removed.""" with _opened(path, _FILE_SHARE_READ_WRITE) as handle: yield _identity_of(handle) def _identity_at(path: str) -> tuple[int, int]: with _opened(path, _FILE_SHARE_ALL) as handle: return _identity_of(handle) def _identity_of_fd(fd: int) -> tuple[int, int]: return _identity_of(msvcrt.get_osfhandle(fd)) @contextmanager def _opened(path: str, share: int) -> Generator[int]: # The final component is opened as itself, never through a reparse point. handle = _kernel32.CreateFileW( path, _FILE_READ_ATTRIBUTES, share, None, 3, # OPEN_EXISTING 0x02000000 | 0x00200000, # FILE_FLAG_BACKUP_SEMANTICS opens a directory; FILE_FLAG_OPEN_REPARSE_POINT None, ) if handle in {None, _INVALID_HANDLE_VALUE}: # pragma: no cover # callers open a name they just made or listed raise ctypes.WinError(ctypes.get_last_error()) try: yield cast("int", handle) finally: _kernel32.CloseHandle(handle) def _identity_of(handle: int) -> tuple[int, int]: info = _BY_HANDLE_FILE_INFORMATION() if not _kernel32.GetFileInformationByHandle(handle, ctypes.byref(info)): # pragma: no cover raise ctypes.WinError(ctypes.get_last_error()) return info.dwVolumeSerialNumber, info.nFileIndexHigh << 32 | info.nFileIndexLow 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", "_directory_is_private", "_identity_at", "_identity_of_fd", "_is_unlinked", "_make_private_directory", "_open_deletable_for_reading", "_open_non_reparse_fd", "_pin_directory", "_pinned", "_unpin_directory", "open_for_delete_without_oplock_wait", "open_read_without_oplock_wait", "unlink_without_oplock_wait", ]