diff --git a/design/file-helper.md b/design/file-helper.md index cd72104472..77ee80fc4a 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -215,10 +215,10 @@ helpers for independent project/source/managed boundaries. deletion rights. Runtime temporary-storage failures remain explicit errors. A leaf introduced between these calls causes an explicit creation error. The opened leaf remains held through text I/O. Both paths check the opened - file type before truncating that same descriptor. An exchanged parent cannot - redirect creation or truncation through a new link. Allowed links are still - resolved by the normal policy walk; the anchored traversal operates on their - resolved target, not on the original link spelling. + file type before truncating that same descriptor. On POSIX, an exchanged + parent cannot redirect creation or truncation through a new link. Allowed + links are still resolved by the normal policy walk; the anchored traversal + operates on their resolved target, not on the original link spelling. Windows creation also honors the CRT process mask: masking the owner-write bit creates a read-only file, matching native text creation, without changing the attributes of an existing file. @@ -250,6 +250,37 @@ entries, not reject a whole project for excluded/unrelated links. ## Safety limits and migration +POSIX reads, exclusive file creation, and text overwrites share one parent +directory walk for capability checks, search flags, no-follow traversal, and +descriptor cleanup. Leaf open flags, file-type validation, and truncation +remain operation-specific. + +Reads resolve supported contained links under the existing policy, then open +the resolved hierarchy without following substituted links. POSIX reads use +descriptor-relative directory traversal with `O_DIRECTORY` and `O_NOFOLLOW`. +Directory descriptors use `O_SEARCH` or `O_PATH` where available so searchable +directories need not be listable, including Darwin's native `O_SEARCH` value +when a Python build uses an older SDK that omits the constant. Each superseded +parent is closed as soon as its child is open. The leaf uses `O_NOFOLLOW` and +`O_NONBLOCK`. Windows reads open the canonical root with `FILE_TRAVERSE` and +`FILE_READ_ATTRIBUTES`, then open each resolved child relative to its validated +parent handle with `NtCreateFile`. Directory opens do not request enumeration +access. Relative opens prevent an in-place junction change on an already-open +parent from redirecting the next open; write-sharing restrictions alone cannot +prevent attribute-only reparse updates. Each new handle rejects reparse points, +and handles withhold delete sharing. Reads duplicate the validated leaf handle +into an independently owned descriptor, never reopen its pathname. +Both paths validate the opened file with `fstat` before reading its descriptor, +and release all handles on success or failure. Windows child-open ABI +structures and pointer types are initialized lazily in a Windows-only module +and reused through Python's import cache. This avoids type retention in older +supported Python versions without requiring `ctypes` for POSIX startup. +Handle values and path buffers remain operation-local. Platforms without the +required safe-open primitives fail explicitly rather than fall back to +unchecked pathname reads. A concurrent replacement can fail the read or leave +it reading the already-open original file, but cannot redirect it through a +substituted symlink. This does not freeze file contents against in-place edits. + Preflight detects static policy failures before recursive deletion or parent creation. Atomic writes recheck the original path before replacement, and deletion rechecks each accessed entry. No operation is a general race-free @@ -262,8 +293,12 @@ above. On POSIX, renaming an opened directory does not redirect creation to its replacement: creation stays with the opened directory, even if it is renamed out of the root. This is not a guarantee of stable namespace membership under concurrent renames, nor does parent anchoring make every leaf preflight -atomic with the subsequent open. The helper does not constrain subprocesses -or third-party code. +atomic with the subsequent open. +On Windows, exclusive creation and text overwrite still open their leaf by +pathname after pinning the parent hierarchy. Attribute-only reparse updates +can redirect those opens; they do not have the read path's parent-relative +handle protection. +The helper does not constrain subprocesses or third-party code. Existing safe-write mechanisms informed this helper, but existing shared infrastructure warning/skip behavior and development-mode link creation are diff --git a/src/specify_cli/_file_handles.py b/src/specify_cli/_file_handles.py index 74531851ce..6b38baaffe 100644 --- a/src/specify_cli/_file_handles.py +++ b/src/specify_cli/_file_handles.py @@ -1,13 +1,61 @@ """Private platform handles for confined filesystem operations.""" +from collections.abc import Iterator +from contextlib import contextmanager import errno import os import secrets import stat import tempfile -from collections.abc import Iterator -from contextlib import contextmanager -from pathlib import Path +from pathlib import Path, PureWindowsPath + + +def _open_windows_child(parent: int, name: str, *, directory: bool, access: int, sharing: int) -> int: + """Open one component relative to an already validated directory handle.""" + if PureWindowsPath(name).parts != (name,) or PureWindowsPath(name).anchor or name == ".." or "\0" in name: + raise ValueError(f"Expected a single relative Windows path component: {name!r}") + encoded_length = len(name.encode("utf-16-le", errors="surrogatepass")) + if encoded_length > 65532: + raise OSError(errno.ENAMETOOLONG, "Windows path component is too long", name) + + import ctypes + from ctypes import wintypes + from . import _windows_abi as abi + + native = ctypes.WinDLL("ntdll") + create = native.NtCreateFile + create.argtypes = [ + ctypes.POINTER(wintypes.HANDLE), ctypes.c_uint32, + abi._ObjectAttributesPointer, abi._IoStatusBlockPointer, + wintypes.LPVOID, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_uint32, + ctypes.c_uint32, wintypes.LPVOID, ctypes.c_uint32, + ] + create.restype = ctypes.c_int32 + dos_error = native.RtlNtStatusToDosError + dos_error.argtypes = [ctypes.c_int32] + dos_error.restype = ctypes.c_uint32 + buffer = ctypes.create_unicode_buffer(name) + unicode_name = abi._UnicodeString( + encoded_length, encoded_length + 2, ctypes.cast(buffer, wintypes.LPWSTR) + ) + # OBJ_DONT_REPARSE and FILE_OPEN_REPARSE_POINT keep resolution on this handle. + attributes = abi._ObjectAttributes( + ctypes.sizeof(abi._ObjectAttributes), parent, ctypes.pointer(unicode_name), + 0x40 | 0x1000, None, None, + ) + handle = wintypes.HANDLE() + status = abi._IoStatusBlock() + options = 0x00200000 | (0x1 if directory else 0x40 | 0x20) + result = create( + ctypes.byref(handle), access | (0 if directory else 0x100000), + ctypes.byref(attributes), ctypes.byref(status), None, 0, + sharing, 1, options, None, 0, + ) + if result < 0: + raise ctypes.WinError(dos_error(result)) + if not handle.value: + raise OSError(errno.EBADF, "Native child open returned no handle", name) + return handle.value def _windows_api(): @@ -35,8 +83,11 @@ def _windows_api(): def _open_windows_handle( - path: Path, *, directory: bool, overwrite: bool = False, create_attributes: int | None = None + path: Path, *, directory: bool, overwrite: bool = False, + create_attributes: int | None = None, parent: int | None = None, traverse: bool = False, ) -> int: + if parent is not None and (overwrite or create_attributes is not None): + raise ValueError("Relative Windows handle opens do not support creation or overwrite") import ctypes from ctypes import wintypes @@ -44,19 +95,20 @@ class AttributeTagInfo(ctypes.Structure): _fields_ = [("attributes", wintypes.DWORD), ("tag", wintypes.DWORD)] kernel = _windows_api() - # Attribute-only directory handles do not prevent renaming; include - # FILE_LIST_DIRECTORY and withhold FILE_SHARE_DELETE for every ancestor. - handle = kernel.CreateFileW( - str(path), - (0x40000000 | 0x80) if overwrite else ((0x80 | 0x1) if directory else 0x80000000), - 0x3, - None, - 3 if create_attributes is None else 1, # OPEN_EXISTING or CREATE_NEW. - 0x00200000 | 0x02000000 | (create_attributes or 0), - None, - ) - if handle == wintypes.HANDLE(-1).value: - raise ctypes.WinError(ctypes.get_last_error()) + access = (0x80 | (0x20 if traverse else 0x1)) if directory else 0x80000000 + if overwrite: + access = 0x40000000 | 0x80 + if parent is None: + # FILE_TRAVERSE participates in sharing checks without requiring enumeration. + handle = kernel.CreateFileW( + str(path), access, 0x3, None, + 3 if create_attributes is None else 1, # OPEN_EXISTING or CREATE_NEW. + 0x00200000 | 0x02000000 | (create_attributes or 0), None, + ) + if handle == wintypes.HANDLE(-1).value: + raise ctypes.WinError(ctypes.get_last_error()) + else: + handle = _open_windows_child(parent, str(path), directory=directory, access=access, sharing=0x3) try: attributes = AttributeTagInfo() if not kernel.GetFileInformationByHandleEx( @@ -74,15 +126,41 @@ class AttributeTagInfo(ctypes.Structure): @contextmanager -def _windows_handle(path: Path, *, directory: bool) -> Iterator[int]: +def _windows_handle( + path: Path, *, directory: bool, parent: int | None = None, traverse: bool = False, +) -> Iterator[int]: """Pin a non-reparse path by withholding Windows delete sharing.""" - handle = _open_windows_handle(path, directory=directory) + if parent is None and not traverse: + handle = _open_windows_handle(path, directory=directory) + else: + handle = _open_windows_handle(path, directory=directory, parent=parent, traverse=traverse) try: yield handle finally: _windows_api().CloseHandle(handle) +@contextmanager +def windows_read_fd(handle: int) -> Iterator[int]: + """Read the validated object through an independently owned duplicate.""" + import _winapi + import msvcrt + + process = _winapi.GetCurrentProcess() + duplicate = _winapi.DuplicateHandle( + process, handle, process, 0, False, _winapi.DUPLICATE_SAME_ACCESS + ) + try: + fd = msvcrt.open_osfhandle(duplicate, os.O_RDONLY | os.O_BINARY) + except BaseException: + _winapi.CloseHandle(duplicate) + raise + try: + yield fd + finally: + os.close(fd) + + @contextmanager def _windows_overwrite_fd(path: Path) -> Iterator[int]: """Open/create without following a new leaf or changing the CRT mask.""" diff --git a/src/specify_cli/_windows_abi.py b/src/specify_cli/_windows_abi.py new file mode 100644 index 0000000000..36fd74af65 --- /dev/null +++ b/src/specify_cli/_windows_abi.py @@ -0,0 +1,41 @@ +"""Windows ABI metadata, initialized once through lazy module imports.""" + +import ctypes +from ctypes import wintypes + + +# Older Python versions retain POINTER types globally; reuse these ABI classes. +class _UnicodeString(ctypes.Structure): + _fields_ = [ + ("Length", ctypes.c_uint16), + ("MaximumLength", ctypes.c_uint16), + ("Buffer", wintypes.LPWSTR), + ] + + +_UnicodeStringPointer = ctypes.POINTER(_UnicodeString) + + +class _ObjectAttributes(ctypes.Structure): + _fields_ = [ + ("Length", ctypes.c_uint32), + ("RootDirectory", wintypes.HANDLE), + ("ObjectName", _UnicodeStringPointer), + ("Attributes", ctypes.c_uint32), + ("SecurityDescriptor", wintypes.LPVOID), + ("SecurityQualityOfService", wintypes.LPVOID), + ] + + +_ObjectAttributesPointer = ctypes.POINTER(_ObjectAttributes) + + +class _StatusValue(ctypes.Union): + _fields_ = [("Status", ctypes.c_int32), ("Pointer", wintypes.LPVOID)] + + +class _IoStatusBlock(ctypes.Structure): + _fields_ = [("Value", _StatusValue), ("Information", ctypes.c_size_t)] + + +_IoStatusBlockPointer = ctypes.POINTER(_IoStatusBlock) diff --git a/src/specify_cli/file_helper.py b/src/specify_cli/file_helper.py index f820668181..f293d53b2e 100644 --- a/src/specify_cli/file_helper.py +++ b/src/specify_cli/file_helper.py @@ -12,10 +12,11 @@ from pathlib import Path, PurePath, PureWindowsPath from typing import TypeVar -from specify_cli._file_handles import _windows_handle, _windows_overwrite_fd +from specify_cli._file_handles import _windows_handle, _windows_overwrite_fd, windows_read_fd _PathType = TypeVar("_PathType", bound=PurePath) -_OVERWRITE_DIR_FD_SUPPORTED = os.open in os.supports_dir_fd +# Wrapping os.open must not change the underlying platform capability. +_POSIX_OPEN = os.open def _strip_extended_length_prefix(path: _PathType) -> _PathType: @@ -167,35 +168,53 @@ def mkdir( def read_bytes(self, path: Path | str) -> bytes: destination = self._path(path) self._require_file(destination) - return destination.read_bytes() + with ExitStack() as stack: + if os.name == "nt": + parts = destination.relative_to(self._canonical_root).parts + parent = stack.enter_context(_windows_handle( + self._canonical_root, directory=True, traverse=True, + )) + for part in parts[:-1]: + parent = stack.enter_context(_windows_handle( + Path(part), directory=True, parent=parent, traverse=True, + )) + handle = stack.enter_context(_windows_handle( + Path(parts[-1]), directory=False, parent=parent, + )) + fd = stack.enter_context(windows_read_fd(handle)) + else: + unsupported = "Safe scoped reads are unsupported on this platform" + if not getattr(os, "O_NONBLOCK", 0): + raise OSError(errno.ENOTSUP, unsupported) + with self._posix_directory(destination.parent, unsupported_message=unsupported) as parent: + fd = os.open( + destination.name, + os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0), + dir_fd=parent, + ) + stack.callback(os.close, fd) + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise UnsupportedPathError(f"Expected a regular file: {destination}") + with os.fdopen(fd, "rb", closefd=False) as stream: + return stream.read() def read_text(self, path: Path | str, *, encoding: str = "utf-8") -> str: return self.read_bytes(path).decode(encoding) @contextmanager - def _creation_directory(self, directory: Path) -> Iterator[int | Path]: - """Anchor an already policy-resolved parent without following new links.""" + def _posix_directory(self, directory: Path, *, unsupported_message: str) -> Iterator[int]: + """Walk an already policy-resolved parent with bounded descriptor ownership.""" parts = directory.relative_to(self._canonical_root).parts - if os.name == "nt": - with ExitStack() as stack: - current = self._canonical_root - stack.enter_context(_windows_handle(current, directory=True)) - for part in parts: - current = current / part - stack.enter_context(_windows_handle(current, directory=True)) - yield current - return - if not ( getattr(os, "O_NOFOLLOW", 0) and getattr(os, "O_DIRECTORY", 0) - and os.open in os.supports_dir_fd + and _POSIX_OPEN in os.supports_dir_fd ): - raise OSError(errno.ENOTSUP, "Anchored file creation is unsupported on this platform") - access = getattr(os, "O_PATH", 0) or getattr(os, "O_SEARCH", 0) + raise OSError(errno.ENOTSUP, unsupported_message) + access = getattr(os, "O_SEARCH", 0) or getattr(os, "O_PATH", 0) if not access: - # Darwin's O_SEARCH ABI is missing from some older Python builds. - access = (0x40000000 | os.O_DIRECTORY) if sys.platform == "darwin" else os.O_RDONLY + # Older Darwin Python builds omit the kernel's O_EXEC/O_SEARCH bit. + access = 0x40000000 if sys.platform == "darwin" else os.O_RDONLY flags = access | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) parent = os.open(self._canonical_root, flags) try: @@ -207,6 +226,25 @@ def _creation_directory(self, directory: Path) -> Iterator[int | Path]: finally: os.close(parent) + @contextmanager + def _creation_directory(self, directory: Path) -> Iterator[int | Path]: + """Anchor an already policy-resolved parent without following new links.""" + if os.name == "nt": + parts = directory.relative_to(self._canonical_root).parts + with ExitStack() as stack: + current = self._canonical_root + stack.enter_context(_windows_handle(current, directory=True)) + for part in parts: + current = current / part + stack.enter_context(_windows_handle(current, directory=True)) + yield current + return + + with self._posix_directory( + directory, unsupported_message="Anchored file creation is unsupported on this platform", + ) as parent: + yield parent + def create_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> None: """Exclusively create a file; never overwrite an existing entry.""" destination = self._path(path, allow_missing=True, follow_leaf=False) @@ -276,9 +314,9 @@ def overwrite_text( @contextmanager def _overwrite_descriptor(self, destination: Path) -> Iterator[int]: """Anchor the resolved hierarchy before opening a non-truncating leaf.""" - parts = destination.relative_to(self._canonical_root).parts with ExitStack() as stack: if os.name == "nt": + parts = destination.relative_to(self._canonical_root).parts current = self._canonical_root stack.enter_context(_windows_handle(current, directory=True)) for part in parts[:-1]: @@ -287,37 +325,19 @@ def _overwrite_descriptor(self, destination: Path) -> Iterator[int]: yield stack.enter_context(_windows_overwrite_fd(destination)) return - if not ( - _OVERWRITE_DIR_FD_SUPPORTED - and getattr(os, "O_NOFOLLOW", 0) - and getattr(os, "O_DIRECTORY", 0) - and getattr(os, "O_NONBLOCK", 0) - ): - raise OSError(errno.ENOTSUP, "Scoped text overwrite is unsupported on this platform") - directory_access = getattr(os, "O_SEARCH", getattr(os, "O_PATH", None)) - if directory_access is None: - # Older Darwin Python builds omit the kernel's O_EXEC/O_SEARCH bit. - directory_access = 0x40000000 if sys.platform == "darwin" else os.O_RDONLY - directory_flags = ( - directory_access | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) - ) - parent = os.open(self._canonical_root, directory_flags) - try: - for part in parts[:-1]: - child = os.open(part, directory_flags, dir_fd=parent) - os.close(parent) - parent = child + unsupported = "Scoped text overwrite is unsupported on this platform" + if not getattr(os, "O_NONBLOCK", 0): + raise OSError(errno.ENOTSUP, unsupported) + with self._posix_directory(destination.parent, unsupported_message=unsupported) as parent: flags = ( os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0) ) - fd = os.open(parts[-1], flags, 0o666, dir_fd=parent) + fd = os.open(destination.name, flags, 0o666, dir_fd=parent) try: yield fd finally: os.close(fd) - finally: - os.close(parent) def symlink(self, target: Path | str, path: Path | str) -> None: """Create a link to an existing contained regular file or directory. diff --git a/tests/specify_cli/presets/_skill_cleanup_helpers.py b/tests/specify_cli/presets/_skill_cleanup_helpers.py index 6695c9b416..d870e525ab 100644 --- a/tests/specify_cli/presets/_skill_cleanup_helpers.py +++ b/tests/specify_cli/presets/_skill_cleanup_helpers.py @@ -65,7 +65,8 @@ def validate_then_swap(self, directory, **kwargs): monkeypatch.setattr(PresetManager, "_validate_skill_subdir", validate_then_swap) if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) + # Disable descriptor-relative cleanup without disabling safe project reads. + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd - {os.rmdir}) result = runner.invoke(app, command, catch_exceptions=False) assert swapped @@ -97,7 +98,7 @@ def check_cleanup_preserves_user_files( reference.parent.mkdir() reference.write_bytes(original) if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd - {os.rmdir}) result = runner.invoke(app, command, catch_exceptions=False) @@ -133,7 +134,7 @@ def check_cleanup_rollback_preserves_directory_mode( reference.parent.mkdir() reference.write_bytes(b"User reference before cleanup") if fallback: - monkeypatch.setattr(os, "supports_dir_fd", set()) + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd - {os.rmdir}) snapshot = _FileJournal._snapshot snapshotted = [] diff --git a/tests/test_file_handles.py b/tests/test_file_handles.py index e582364cdd..a1201e64b6 100644 --- a/tests/test_file_handles.py +++ b/tests/test_file_handles.py @@ -1,18 +1,278 @@ -"""Portable checks of Windows handle flags, validation, and ownership.""" +"""Ownership and native Windows evidence for shared handle primitives.""" -import ctypes -import os -import sys from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager +import ctypes from ctypes import wintypes +import errno +import gc +import os from pathlib import Path +import sys +import subprocess from types import SimpleNamespace +import weakref import pytest -from specify_cli import _file_handles +from specify_cli import _file_handles, file_helper from specify_cli._file_handles import _windows_handle +from specify_cli.file_helper import FileHelper, UnsupportedPathError + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX startup must not require Windows ctypes support") +@pytest.mark.parametrize("operation", ["help", "files"]) +def test_posix_usage_without_optional_ctypes(tmp_path, operation): + result = subprocess.run( + [sys.executable, "-c", """ +import sys + +class NoCtypes: + def find_spec(self, fullname, path=None, target=None): + if fullname == "_ctypes": + raise ModuleNotFoundError("optional _ctypes unavailable", name="_ctypes") + +for name in tuple(sys.modules): + if name == "_ctypes" or name == "ctypes" or name.startswith("ctypes."): + sys.modules.pop(name) +sys.meta_path.insert(0, NoCtypes()) + +if sys.argv[1] == "help": + from specify_cli import main + sys.argv = ["specify", "--help"] + main() +else: + from pathlib import Path + from specify_cli.file_helper import FileHelper + + root = Path(sys.argv[2]) + for allow in (False, True): + files = FileHelper(root, allow_symlinks=allow) + files.mkdir("nested") + files.create_text("nested/file", "inside") + assert files.read_text("nested/file") == "inside" + files.write_text("nested/file", "updated") + if allow: + files.symlink("file", "nested/alias") + assert files.read_text("nested/alias") == "updated" + files.delete("nested", recursive=True) + assert "specify_cli._windows_abi" not in sys.modules + assert "_ctypes" not in sys.modules +""", operation, str(tmp_path)], + capture_output=True, text=True, check=False, + ) + assert result.returncode == 0, result.stderr + if operation == "help": + assert "Usage" in result.stdout + + +@pytest.mark.parametrize("failure", [None, "duplicate", "convert", "body"]) +def test_windows_read_fd_duplicates_and_closes_only_its_owned_handle(tmp_path, monkeypatch, failure): + selected = tmp_path / "file" + selected.write_bytes(b"inside") + original = os.open(selected, os.O_RDONLY | getattr(os, "O_BINARY", 0)) + duplicates = [] + closed_handles = [] + + def duplicate(source_process, handle, target_process, access, inherit, options): + assert (source_process, handle, target_process, access, inherit, options) == ( + 7, original, 7, 0, False, 2, + ) + if failure == "duplicate": + raise OSError("injected duplicate failure") + fd = os.dup(handle) + duplicates.append(fd) + return fd + + def convert(handle, flags): + assert handle == duplicates[-1] + assert flags == 0x8000 + if failure == "convert": + raise OSError("injected conversion failure") + return handle + + def close(handle): + closed_handles.append(handle) + os.close(handle) + + monkeypatch.setitem(sys.modules, "_winapi", SimpleNamespace( + GetCurrentProcess=lambda: 7, DuplicateHandle=duplicate, + DUPLICATE_SAME_ACCESS=2, CloseHandle=close, + )) + monkeypatch.setitem(sys.modules, "msvcrt", SimpleNamespace(open_osfhandle=convert)) + monkeypatch.setattr(_file_handles, "os", SimpleNamespace( + O_RDONLY=0, O_BINARY=0x8000, close=os.close, + )) + try: + if failure: + with pytest.raises(OSError, match="injected"): + with _file_handles.windows_read_fd(original): + raise OSError("injected body failure") + else: + with _file_handles.windows_read_fd(original) as fd: + assert os.read(fd, 6) == b"inside" + assert not os.get_inheritable(fd) + assert closed_handles == (duplicates if failure == "convert" else []) + assert os.fstat(original).st_size == 6 + for fd in duplicates: + with pytest.raises(OSError): + os.fstat(fd) + finally: + os.close(original) + + +def _set_windows_symlink_reparse( + path: Path, target: Path, *, junction: bool = False, access: int = 0x40000000, +) -> None: + import ctypes + from ctypes import wintypes + import _winapi + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + control = kernel.DeviceIoControl + control.argtypes = [ + wintypes.HANDLE, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, + wintypes.LPVOID, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), wintypes.LPVOID, + ] + control.restype = wintypes.BOOL + template = target.with_name("reparse-template") + if junction: + _winapi.CreateJunction(str(target), str(template)) + else: + template.symlink_to(target) + try: + buffer = ctypes.create_string_buffer(16 * 1024) + returned = wintypes.DWORD() + source = _winapi.CreateFile(str(template), 0, 0x7, 0, 3, 0x02200000, 0) + try: + if not control(source, 0x900A8, None, 0, buffer, len(buffer), ctypes.byref(returned), None): + raise ctypes.WinError(ctypes.get_last_error()) + finally: + _winapi.CloseHandle(source) + handle = _winapi.CreateFile(str(path), access, 0x7, 0, 3, 0x02200000, 0) + try: + if not control(handle, 0x900A4, buffer, returned.value, None, 0, ctypes.byref(returned), None): + raise ctypes.WinError(ctypes.get_last_error()) + finally: + _winapi.CloseHandle(handle) + finally: + if junction: + template.rmdir() + else: + template.unlink() + + +@pytest.mark.parametrize("failure", [None, "get", "set"]) +@pytest.mark.parametrize("junction", [False, True]) +def test_reparse_fixture_copies_native_buffer_and_cleans_up(tmp_path, monkeypatch, failure, junction): + import ctypes + from ctypes import wintypes + + target = tmp_path / "outside" + target.write_bytes(b"outside") + if not junction: + probe = tmp_path / "probe" + try: + probe.symlink_to(target) + except (NotImplementedError, OSError) as exc: + pytest.skip(f"Symlink creation unavailable: {exc}") + probe.unlink() + opened = [] + closed = [] + payload = b"OS-generated reparse buffer" + + def create(path, access, sharing, security, disposition, flags, template): + assert (sharing, security, disposition, flags, template) == (7, 0, 3, 0x02200000, 0) + assert access == (0 if not opened else 0x100 if junction else 0x40000000) + opened.append(len(opened) + 1) + return opened[-1] + + class Control: + def __call__(self, handle, code, source, source_size, destination, size, returned, overlapped): + if code == 0x900A8: + assert handle == 1 and source is None and source_size == 0 + if failure == "get": + return False + ctypes.memmove(destination, payload, len(payload)) + ctypes.cast(returned, ctypes.POINTER(wintypes.DWORD))[0] = len(payload) + else: + assert code == 0x900A4 and handle == 2 + assert ctypes.string_at(source, source_size) == payload + if failure == "set": + return False + return True + + monkeypatch.setitem(sys.modules, "_winapi", SimpleNamespace( + CreateFile=create, CloseHandle=closed.append, + CreateJunction=lambda target, path: Path(path).mkdir(), + )) + monkeypatch.setattr(ctypes, "WinDLL", lambda *args, **kwargs: SimpleNamespace(DeviceIoControl=Control()), raising=False) + monkeypatch.setattr(ctypes, "get_last_error", lambda: 5, raising=False) + monkeypatch.setattr(ctypes, "WinError", lambda code: OSError("injected native failure"), raising=False) + if failure: + with pytest.raises(OSError, match="injected native failure"): + _set_windows_symlink_reparse( + tmp_path / "file", target, junction=junction, access=0x100 if junction else 0x40000000, + ) + else: + _set_windows_symlink_reparse( + tmp_path / "file", target, junction=junction, access=0x100 if junction else 0x40000000, + ) + assert closed == opened + assert not (tmp_path / "reparse-template").is_symlink() + assert not (tmp_path / "reparse-template").exists() + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows reparse changes required") +@pytest.mark.parametrize("allow", [False, True]) +def test_windows_read_never_reopens_leaf_after_in_place_reparse_change(tmp_path, monkeypatch, allow): + root = tmp_path / "project" + root.mkdir() + selected = root / "file" + selected.write_bytes(b"") + outside = tmp_path / "outside" + outside.write_bytes(b"outside") + changed = False + + @contextmanager + def change_after_validation(path, **kwargs): + nonlocal changed + with _file_handles._windows_handle(path, **kwargs) as handle: + if not kwargs["directory"]: + try: + _set_windows_symlink_reparse(selected, outside) + except OSError as exc: + if exc.winerror in (50, 1314): + pytest.skip(f"Symlink reparse updates unavailable: {exc}") + raise + assert selected.read_bytes() == b"outside" + changed = True + yield handle + + monkeypatch.setattr(file_helper, "_windows_handle", change_after_validation) + try: + content = FileHelper(root, allow_symlinks=allow).read_bytes(selected) + except UnsupportedPathError: + pass + else: + assert content == b"" + assert changed + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows sharing semantics required") +def test_windows_traversal_pin_allows_writes_but_prevents_rename(tmp_path): + import _winapi + + def writable_handle(): + return _winapi.CreateFile(str(tmp_path), 0x40000000, 0x7, 0, 3, 0x02200000, 0) + + _winapi.CloseHandle(writable_handle()) + with _file_handles._windows_handle(tmp_path, directory=True, traverse=True): + _winapi.CloseHandle(writable_handle()) + with pytest.raises(OSError) as error: + tmp_path.rename(tmp_path.with_name(tmp_path.name + "-renamed")) + assert error.value.winerror == 32 + _winapi.CloseHandle(writable_handle()) @pytest.fixture @@ -89,6 +349,12 @@ def test_read_handle_retains_existing_access_and_sharing(kernel, directory): assert closed == [123] +@pytest.mark.parametrize("options", [{"overwrite": True}, {"create_attributes": 0}]) +def test_relative_read_handles_reject_mutating_modes(options): + with pytest.raises(ValueError, match="Relative Windows handle opens"): + _file_handles._open_windows_handle(Path("file"), directory=False, parent=51, **options) + + @pytest.mark.parametrize("mask", [0, 0o200, 0o400, 0o600]) @pytest.mark.parametrize("query_ok", [False, True]) def test_existing_windows_overwrite_does_not_query_creation_mask(kernel, monkeypatch, mask, query_ok): @@ -336,8 +602,8 @@ def unlink(path): assert calls[1][4] == 1 -@pytest.mark.parametrize("directory", [False, True]) -def test_windows_handle_declares_complete_createfile_prototype(tmp_path, monkeypatch, directory): +@pytest.mark.parametrize("directory,traverse", [(False, False), (True, False), (True, True)]) +def test_windows_handle_declares_complete_createfile_prototype(tmp_path, monkeypatch, directory, traverse): closed = [] def create(*args): @@ -353,7 +619,7 @@ def create(*args): assert create.restype is wintypes.HANDLE assert args == ( str(tmp_path), - (0x80 | 0x1) if directory else 0x80000000, + (0x80 | (0x20 if traverse else 0x1)) if directory else 0x80000000, 0x3, None, 3, @@ -380,7 +646,211 @@ def load_kernel(name, *, use_last_error): ) monkeypatch.setattr(ctypes, "WinDLL", load_kernel, raising=False) - with _windows_handle(tmp_path, directory=directory) as handle: + with _windows_handle(tmp_path, directory=directory, traverse=traverse) as handle: assert handle == 123 assert closed == [] assert closed == [123] + + +@pytest.mark.parametrize("directory", [False, True]) +@pytest.mark.parametrize("failure", [None, "status", "reparse", "type", "info", "empty"]) +@pytest.mark.parametrize("child_name", ["caf\u00e9", "face \U0001f600", "surrogate\ud800"]) +def test_windows_child_open_is_relative_and_closes_handles(monkeypatch, directory, failure, child_name): + closed = [] + abi_types = {role: set() for role in ( + "name", "attributes", "status_value", "status", + "name_pointer", "attributes_pointer", "status_pointer", + )} + + def create_path(*args): + pytest.fail("A child must not be opened by absolute pathname") + + def create_child(output, access, attributes, status, allocation, file_attributes, + sharing, disposition, options, ea, ea_length): + assert create_child.restype is ctypes.c_int32 + attributes = attributes._obj + declarations = ( + type(attributes.ObjectName.contents), type(attributes), type(status._obj.Value), + type(status._obj), type(attributes.ObjectName), + create_child.argtypes[2], create_child.argtypes[3], + ) + for role, declaration in zip(abi_types, declarations): + abi_types[role].add(weakref.ref(declaration)) + assert attributes.Length == ctypes.sizeof(type(attributes)) + assert attributes.RootDirectory == 51 + assert attributes.Attributes == 0x1040 + assert attributes.SecurityDescriptor is None + assert attributes.SecurityQualityOfService is None + name = attributes.ObjectName.contents + length = len(child_name.encode("utf-16-le", errors="surrogatepass")) + assert (name.Buffer, name.Length, name.MaximumLength) == (child_name, length, length + 2) + assert access == (0xA0 if directory else 0x80100000) + assert sharing == 3 and disposition == 1 + assert options == (0x200001 if directory else 0x200060) + assert allocation is None and file_attributes == 0 and ea is None and ea_length == 0 + assert ctypes.sizeof(status._obj) == 2 * ctypes.sizeof(ctypes.c_void_p) + if failure == "status": + return -1073741790 + if failure != "empty": + output._obj.value = 77 + return 0 + + def info(handle, info_class, buffer, size): + assert handle == 77 and info_class == 9 + attributes = ctypes.cast(buffer, ctypes.POINTER(wintypes.DWORD)) + attributes[0] = 0x10 if directory else 0 + if failure == "reparse": + attributes[0] |= 0x400 + if failure == "type": + attributes[0] ^= 0x10 + return failure != "info" + + def close(handle): + closed.append(handle) + return True + + def error(status): + assert status == -1073741790 + return 5 + + def load(name, **kwargs): + if name == "ntdll": + return SimpleNamespace(NtCreateFile=create_child, RtlNtStatusToDosError=error) + assert name == "kernel32" and kwargs == {"use_last_error": True} + return SimpleNamespace( + CreateFileW=create_path, GetFileInformationByHandleEx=info, CloseHandle=close, + ) + + monkeypatch.setattr(ctypes, "WinDLL", load, raising=False) + monkeypatch.setattr(ctypes, "get_last_error", lambda: 5, raising=False) + monkeypatch.setattr(ctypes, "WinError", lambda code: PermissionError(errno.EACCES, "native denied"), raising=False) + for iteration in range(32): + if failure: + with pytest.raises(OSError): + with _windows_handle(Path(child_name), directory=directory, traverse=directory, parent=51): + pytest.fail("A failing native open must not yield a handle") + else: + with _windows_handle(Path(child_name), directory=directory, traverse=directory, parent=51) as handle: + assert handle == 77 and len(closed) == iteration + assert closed == ([] if failure in ("status", "empty") else [77] * 32) + gc.collect() + assert {role: len(declarations) for role, declarations in abi_types.items()} == dict.fromkeys(abi_types, 1) + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows repeated-read lifetime regression") +@pytest.mark.parametrize("allow", [False, True]) +def test_repeated_windows_reads_do_not_accumulate_abi_types(tmp_path, allow): + root = tmp_path / "project" + selected = root / "nested/file" + selected.parent.mkdir(parents=True) + selected.write_bytes(b"inside") + files = FileHelper(root, allow_symlinks=allow) + abi_modules = {_file_handles.__name__, f"{_file_handles.__package__}._windows_abi"} + + def live_abi_types(): + return { + weakref.ref(obj) for obj in gc.get_objects() + if isinstance(obj, type) and getattr(obj, "__module__", None) in abi_modules + } + + for _ in range(32): + assert files.read_bytes("nested/file") == b"inside" + gc.collect() + baseline = live_abi_types() + assert baseline + for _ in range(256): + assert files.read_bytes("nested/file") == b"inside" + gc.collect() + assert live_abi_types() == baseline + + +@pytest.mark.parametrize("name", ["", ".", "..", "../file", "dir/file", "dir\\file", "C:outside", "C:\\outside", "\0"]) +def test_windows_child_open_rejects_non_component_names(name): + with pytest.raises(ValueError, match="single relative"): + _file_handles._open_windows_child(51, name, directory=False, access=0, sharing=3) + + +def test_windows_child_open_rejects_oversized_unicode_name(): + with pytest.raises(OSError) as error: + _file_handles._open_windows_child(51, "a" * 32767, directory=False, access=0, sharing=3) + assert error.value.errno == errno.ENAMETOOLONG + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows attribute-only junction changes required") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("level", ["root", "parent"]) +def test_windows_read_anchors_children_after_attribute_only_junction_change(tmp_path, monkeypatch, allow, level): + root = tmp_path / "project" + root.mkdir() + root = root.resolve() + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + outside = tmp_path / "outside" + outside_directory = outside / "directory" if level == "root" else outside + outside_directory.mkdir(parents=True) + (outside_directory / "file").write_bytes(b"outside") + locations = {} + changed = False + + @contextmanager + def exchange(path, **kwargs): + nonlocal changed + parent = kwargs.get("parent") + actual = path if parent is None else locations[parent] / path + with _file_handles._windows_handle(path, **kwargs) as handle: + locations[handle] = actual + try: + if kwargs["directory"] and actual == (root if level == "root" else directory): + displaced = directory if level == "root" else selected + displaced.rename(tmp_path / "retained") + _set_windows_symlink_reparse(actual, outside, junction=True, access=0x100) + changed = True + # The former full-path open follows this same converted parent. + with _windows_handle(selected, directory=False) as old_handle: + with _file_handles.windows_read_fd(old_handle) as fd: + assert os.read(fd, 7) == b"outside" + yield handle + finally: + locations.pop(handle) + + monkeypatch.setattr(file_helper, "_windows_handle", exchange) + try: + content = FileHelper(root, allow_symlinks=allow).read_bytes(selected) + except (OSError, ValueError): + pass + else: + assert content == b"inside" + assert changed + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows ACL semantics required") +@pytest.mark.parametrize("restricted", ["root", "parent"]) +@pytest.mark.parametrize("allow", [False, True]) +def test_windows_reads_do_not_require_folder_listing(tmp_path, restricted, allow): + from specify_cli._file_policy import read_project_options + + root = tmp_path / "project" + directory = root / ".specify" + directory.mkdir(parents=True) + selected = directory / "init-options.json" + payload = b'{"allow_symlinks": false}' + selected.write_bytes(payload) + limited = root if restricted == "root" else directory + subprocess.run( + ["icacls", str(limited), "/deny", "*S-1-1-0:(RD)"], check=True, capture_output=True, + ) + try: + assert selected.read_bytes() == payload + with pytest.raises(PermissionError): + list(limited.iterdir()) + with pytest.raises(PermissionError): + with _windows_handle(limited, directory=True): + pytest.fail("The former listing-mode pin must be denied by this ACL") + assert FileHelper(root, allow_symlinks=allow).read_bytes(selected) == payload + assert read_project_options(root)["allow_symlinks"] is False + finally: + subprocess.run( + ["icacls", str(limited), "/remove:d", "*S-1-1-0"], check=True, capture_output=True, + ) diff --git a/tests/test_file_helper.py b/tests/test_file_helper.py index 8bf31e9ba8..d2afd1ac86 100644 --- a/tests/test_file_helper.py +++ b/tests/test_file_helper.py @@ -63,6 +63,534 @@ def test_plain_operations(root, allow): assert list(root.iterdir()) == [] +@pytest.mark.skipif(os.name == "nt", reason="POSIX shared directory traversal") +@pytest.mark.parametrize("operation", ["read_bytes", "create_bytes", "overwrite_text"]) +def test_posix_operations_share_parent_walk_and_preserve_leaf_flags(root, monkeypatch, operation): + parent = root / "parent/nested" + parent.mkdir(parents=True) + selected = parent / "file" + if operation != "create_bytes": + selected.write_bytes(b"inside") + parents = [] + opened = [] + original_walk = FileHelper._posix_directory + original_open = os.open + + @contextmanager + def observe_walk(self, directory, **kwargs): + parents.append(directory) + with original_walk(self, directory, **kwargs) as fd: + yield fd + + def observe_open(path, flags, *args, **kwargs): + opened.append((path, flags, args, kwargs)) + return original_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(FileHelper, "_posix_directory", observe_walk) + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "open": observe_open})) + files = FileHelper(root) + cloexec = getattr(os, "O_CLOEXEC", 0) + if operation == "read_bytes": + assert files.read_bytes(selected) == b"inside" + flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | cloexec + mode = () + elif operation == "create_bytes": + files.create_bytes(selected, b"new") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW + mode = (0o644,) + else: + files.overwrite_text(selected, "new") + flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK | cloexec + mode = (0o666,) + assert parents == [parent.resolve()] + directory_calls = [call for call in opened if call[1] & os.O_DIRECTORY] + assert len(directory_calls) == 3 + assert len({call[1] for call in directory_calls}) == 1 + assert all(call[1] & os.O_NOFOLLOW for call in directory_calls) + leaf_calls = [call for call in opened if not call[1] & os.O_DIRECTORY] + assert len(leaf_calls) == 1 + assert leaf_calls[0][:3] == ("file", flags, mode) + assert leaf_calls[0][3]["dir_fd"] is not None + assert selected.read_bytes() == (b"inside" if operation == "read_bytes" else b"new") + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX shared directory cleanup") +@pytest.mark.parametrize("operation", ["read_bytes", "create_bytes", "overwrite_text"]) +@pytest.mark.parametrize("failure", ["root", "parent", "leaf"]) +def test_posix_parent_walk_closes_descriptors_after_open_failures(root, monkeypatch, operation, failure): + parent = root / "parent/nested" + parent.mkdir(parents=True) + selected = parent / "file" + if operation != "create_bytes": + selected.write_bytes(b"inside") + original_open = os.open + original_close = os.close + live = set() + failed = [] + + def observe_open(path, flags, *args, **kwargs): + fail_here = ( + (failure == "root" and Path(path) == root.resolve()) + or (failure == "parent" and path == "nested") + or (failure == "leaf" and path == "file") + ) + if fail_here: + failed.append(path) + raise PermissionError("injected open failure") + fd = original_open(path, flags, *args, **kwargs) + live.add(fd) + return fd + + def observe_close(fd): + original_close(fd) + live.remove(fd) + + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{ + **vars(os), "open": observe_open, "close": observe_close, + })) + with pytest.raises(PermissionError, match="injected"): + if operation == "read_bytes": + FileHelper(root).read_bytes(selected) + elif operation == "create_bytes": + FileHelper(root).create_bytes(selected, b"new") + else: + FileHelper(root).overwrite_text(selected, "new") + assert len(failed) == 1 + assert live == set() + if operation == "create_bytes": + assert not selected.exists() + else: + assert selected.read_bytes() == b"inside" + + +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("operation", ["read_bytes", "read_text"]) +@pytest.mark.parametrize("kind", ["leaf", "parent", "resolved-link"]) +def test_read_does_not_follow_replacements_after_validation( + root, link, monkeypatch, allow, operation, kind +): + if kind == "resolved-link" and not allow: + pytest.skip("Contained links require allow mode") + directory = root / "directory" + directory.mkdir() + (directory / "file").write_bytes(b"inside") + outside = root.parent / "outside" + outside.mkdir() + (outside / "file").write_bytes(b"outside-secret") + selected = directory / "file" + if kind == "resolved-link": + selected = link(root / "alias", selected) + original_path = FileHelper._path + + def exchange(self, path, **kwargs): + destination = original_path(self, path, **kwargs) + if kind == "parent": + directory.rename(root / "retained") + link(directory, outside) + else: + destination.unlink() + link(destination, outside / "file") + return destination + + monkeypatch.setattr(FileHelper, "_path", exchange) + try: + content = getattr(FileHelper(root, allow_symlinks=allow), operation)(selected) + except (OSError, FileHelperError): + pass + else: + assert content in (b"inside", "inside") + + +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("kind", ["leaf", "parent"]) +def test_read_uses_opened_file_after_replacement(root, link, monkeypatch, allow, kind): + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + outside = root.parent / "outside" + outside.mkdir() + (outside / "file").write_bytes(b"outside-secret") + original_fstat = os.fstat + swapped = False + + def exchange(fd): + nonlocal swapped + info = original_fstat(fd) + if swapped: + return info + if kind == "leaf": + source, replacement = selected, outside / "file" + else: + source, replacement = directory, outside + if os.name == "nt": + with pytest.raises(OSError): + source.rename(root / "retained") + else: + source.rename(root / "retained") + link(source, replacement) + swapped = True + return info + + monkeypatch.setattr("specify_cli.file_helper.os.fstat", exchange) + assert FileHelper(root, allow_symlinks=allow).read_bytes(selected) == b"inside" + assert swapped + if os.name == "nt": + source = selected if kind == "leaf" else directory + source.rename(root / "retained") + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor-relative open regression") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("phase", ["parent", "leaf"]) +def test_read_rejects_replacement_during_open(root, link, monkeypatch, allow, phase): + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + outside = root.parent / "outside" + outside.mkdir() + (outside / "file").write_bytes(b"outside-secret") + original_open = os.open + swapped = False + + def exchange(path, flags, *args, **kwargs): + nonlocal swapped + if path == ("directory" if phase == "parent" else "file"): + source = directory if phase == "parent" else selected + source.rename(root / "retained") + link(source, outside if phase == "parent" else outside / "file") + swapped = True + return original_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {exchange}) + monkeypatch.setattr("specify_cli.file_helper.os.open", exchange) + with pytest.raises(OSError): + FileHelper(root, allow_symlinks=allow).read_bytes(selected) + assert swapped + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor-relative open regression") +def test_read_pins_parent_before_opening_leaf(root, link, monkeypatch): + directory = root / "directory" + directory.mkdir() + (directory / "file").write_bytes(b"inside") + outside = root.parent / "outside" + outside.mkdir() + (outside / "file").write_bytes(b"outside-secret") + original_open = os.open + swapped = False + + def exchange(path, flags, *args, **kwargs): + nonlocal swapped + if path == "file": + directory.rename(root / "retained") + link(directory, outside) + swapped = True + return original_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {exchange}) + monkeypatch.setattr("specify_cli.file_helper.os.open", exchange) + assert FileHelper(root).read_bytes("directory/file") == b"inside" + assert swapped + + +@pytest.mark.parametrize("failure", [None, "parent", "leaf", "duplicate", "fstat"]) +def test_windows_read_pins_entire_resolved_hierarchy(root, monkeypatch, failure): + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + pinned = [] + seen = [] + + @contextmanager + def pin(path, *, directory, parent=None, traverse=False): + if parent is None: + assert directory and path == root.resolve() + actual = path + else: + assert parent == len(pinned) + 10 + assert path.parts == (path.name,) + actual = pinned[-1] / path + if actual.name == {"parent": "directory", "leaf": "file"}.get(failure): + raise OSError("injected handle failure") + if directory: + assert traverse + pinned.append(actual) + seen.append((actual, directory)) + try: + yield len(pinned) + 10 + finally: + assert pinned.pop() == actual + + @contextmanager + def duplicate(handle): + assert handle == 13 + assert pinned == [root.resolve(), directory.resolve(), selected.resolve()] + if failure == "duplicate": + raise OSError("injected duplicate failure") + fd = os.open(selected, os.O_RDONLY | getattr(os, "O_BINARY", 0)) + try: + yield fd + finally: + os.close(fd) + + def no_path_open(*args, **kwargs): + assert pinned == [root.resolve(), directory.resolve(), selected.resolve()] + pytest.fail("Windows reads must duplicate the validated handle, not reopen its path") + + def checked_fstat(fd): + assert pinned == [root.resolve(), directory.resolve(), selected.resolve()] + if failure == "fstat": + raise OSError("injected fstat failure") + return os.fstat(fd) + + windows_os = SimpleNamespace( + name="nt", + O_RDONLY=os.O_RDONLY, + O_BINARY=getattr(os, "O_BINARY", 0), + open=no_path_open, + close=os.close, + fdopen=os.fdopen, + fstat=checked_fstat, + ) + monkeypatch.setattr(file_helper, "os", windows_os) + monkeypatch.setattr(file_helper, "_windows_handle", pin) + monkeypatch.setattr(file_helper, "windows_read_fd", duplicate, raising=False) + files = FileHelper(root) + if failure: + with pytest.raises(OSError, match="injected"): + files.read_bytes(selected) + else: + assert files.read_bytes(selected) == b"inside" + assert seen == [ + (root.resolve(), True), + (directory.resolve(), True), + (selected.resolve(), False), + ] + assert pinned == [] + + +@pytest.mark.skipif(os.name == "nt", reason="Portable replacement model; native reparse coverage is separate") +@pytest.mark.parametrize("allow", [False, True]) +def test_windows_read_keeps_validated_object_when_path_changes(root, link, monkeypatch, allow): + selected = root / "file" + selected.write_bytes(b"inside") + outside = root.parent / "outside" + outside.write_bytes(b"outside") + changed = False + + @contextmanager + def pin(path, *, directory, parent=None, traverse=False): + nonlocal changed + if directory: + assert parent is None and traverse + yield 1 + else: + assert parent == 1 and path == Path("file") + fd = os.open(selected, os.O_RDONLY | getattr(os, "O_BINARY", 0)) + try: + # Model reparse changes after validation without changing the opened object. + selected.unlink() + link(selected, outside) + changed = True + yield fd + finally: + os.close(fd) + + @contextmanager + def duplicate(handle): + fd = os.dup(handle) + try: + yield fd + finally: + os.close(fd) + + monkeypatch.setattr(file_helper, "os", SimpleNamespace( + name="nt", open=os.open, O_RDONLY=os.O_RDONLY, + O_BINARY=getattr(os, "O_BINARY", 0), close=os.close, + fstat=os.fstat, fdopen=os.fdopen, + )) + monkeypatch.setattr(file_helper, "_windows_handle", pin) + monkeypatch.setattr(file_helper, "windows_read_fd", duplicate, raising=False) + assert FileHelper(root, allow_symlinks=allow).read_bytes(selected) == b"inside" + assert changed + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions required") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("restricted", ["root", "parent"]) +@pytest.mark.parametrize("older_darwin_sdk", [ + False, + pytest.param(True, marks=pytest.mark.skipif(sys.platform != "darwin", reason="Darwin SDK regression")), +]) +def test_read_through_search_only_directories(root, monkeypatch, allow, restricted, older_darwin_sdk): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory permission checks") + if older_darwin_sdk: + for flag in ("O_SEARCH", "O_PATH", "O_EXEC"): + monkeypatch.delattr(os, flag, raising=False) + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + limited = root if restricted == "root" else directory + previous_mode = limited.stat().st_mode & 0o777 + limited.chmod(0o111) + try: + assert selected.read_bytes() == b"inside" + with pytest.raises(PermissionError): + list(limited.iterdir()) + assert FileHelper(root, allow_symlinks=allow).read_bytes(selected) == b"inside" + finally: + limited.chmod(previous_mode) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions required") +@pytest.mark.parametrize("restricted", ["root", "parent"]) +@pytest.mark.parametrize("after_validation", [False, True]) +def test_read_rejects_non_searchable_directories(root, monkeypatch, restricted, after_validation): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory permission checks") + directory = root / "directory" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + expected = selected.resolve() + files = FileHelper(root) + limited = root if restricted == "root" else directory + previous_mode = limited.stat().st_mode & 0o777 + checked = [] + opened = [] + original_check = FileHelper._require_file + original_open = os.open + + def revoke(path): + original_check(path) + checked.append(path) + limited.chmod(0) + + def recording_open(*args, **kwargs): + opened.append((args, kwargs)) + return original_open(*args, **kwargs) + + if after_validation: + monkeypatch.setattr(FileHelper, "_require_file", staticmethod(revoke)) + monkeypatch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {recording_open}) + monkeypatch.setattr(os, "open", recording_open) + else: + limited.chmod(0) + try: + with pytest.raises(PermissionError): + files.read_bytes(selected) + if after_validation: + assert checked == [expected] + assert opened + if restricted == "parent": + assert any(kwargs.get("dir_fd") is not None for args, kwargs in opened) + finally: + limited.chmod(previous_mode) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor limits required") +@pytest.mark.parametrize("allow", [False, True]) +def test_read_deep_path_with_small_descriptor_limit(root, allow): + directory = root + for _ in range(96): + directory /= "d" + directory.mkdir() + selected = directory / "file" + selected.write_bytes(b"inside") + result = subprocess.run( + [sys.executable, "-c", """ +import resource +import sys +from pathlib import Path +from specify_cli.file_helper import FileHelper + +root, selected = map(Path, sys.argv[1:3]) +soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) +limit = 64 if soft == resource.RLIM_INFINITY else min(64, soft) +resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard)) +assert selected.read_bytes() == b"inside" +assert FileHelper(root, allow_symlinks=sys.argv[3] == "True").read_bytes(selected) == b"inside" +""", str(root), str(selected), str(allow)], + capture_output=True, text=True, check=False, + ) + assert result.returncode == 0, result.stderr + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor cleanup regression") +@pytest.mark.parametrize("failure", ["parent", "leaf", "fstat", "fdopen", "read"]) +def test_read_closes_descriptors_on_failure(root, monkeypatch, failure): + (root / "directory").mkdir() + selected = root / "directory/file" + selected.write_bytes(b"inside") + opened = [] + original_open = os.open + + def recording_open(path, flags, *args, **kwargs): + if str(path) == {"parent": "directory", "leaf": "file"}.get(failure): + raise PermissionError("injected open failure") + fd = original_open(path, flags, *args, **kwargs) + opened.append(fd) + return fd + + def fail(*args, **kwargs): + raise OSError("injected read failure") + + class FailedRead: + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + read = fail + + with monkeypatch.context() as patch: + patch.setattr(os, "supports_dir_fd", os.supports_dir_fd | {recording_open}) + patch.setattr(os, "open", recording_open) + if failure in ("fstat", "fdopen"): + patch.setattr(os, failure, fail) + elif failure == "read": + patch.setattr(os, "fdopen", lambda *args, **kwargs: FailedRead()) + with pytest.raises(OSError, match="injected"): + FileHelper(root).read_bytes(selected) + assert opened + for fd in opened: + with pytest.raises(OSError) as error: + os.fstat(fd) + assert error.value.errno == errno.EBADF + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX special-file replacement regression") +def test_read_checks_opened_type_without_blocking_on_fifo(root, monkeypatch): + selected = root / "file" + selected.write_bytes(b"inside") + original_check = FileHelper._require_file + + def exchange(path): + original_check(path) + path.unlink() + os.mkfifo(path) + + monkeypatch.setattr(FileHelper, "_require_file", staticmethod(exchange)) + with pytest.raises(UnsupportedPathError, match="regular file"): + FileHelper(root).read_bytes(selected) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX capability refusal regression") +@pytest.mark.parametrize("capability", ["O_NOFOLLOW", "O_DIRECTORY", "O_NONBLOCK", "supports_dir_fd"]) +def test_read_refuses_unsafe_platform_fallback(root, monkeypatch, capability): + (root / "file").write_bytes(b"inside") + monkeypatch.setattr(os, capability, set() if capability == "supports_dir_fd" else 0) + with pytest.raises(OSError) as error: + FileHelper(root).read_bytes("file") + assert error.value.errno == errno.ENOTSUP + + @pytest.mark.parametrize("allow", [False, True]) def test_overwrite_text_preserves_inode_and_native_text_behavior(root, allow): files = FileHelper(root, allow_symlinks=allow) diff --git a/tests/test_file_helper_overwrite.py b/tests/test_file_helper_overwrite.py index f433914d7f..6ed2238560 100644 --- a/tests/test_file_helper_overwrite.py +++ b/tests/test_file_helper_overwrite.py @@ -376,7 +376,7 @@ def unexpected_open(*args, **kwargs): attributes = {**vars(os), "open": unexpected_open} if capability == "dir_fd": - monkeypatch.setattr(file_helper, "_OVERWRITE_DIR_FD_SUPPORTED", False) + attributes["supports_dir_fd"] = set() elif value is None: attributes.pop(capability, None) else: