diff --git a/design/catalogs.md b/design/catalogs.md index ff8b42cd41..66d4b82e63 100644 --- a/design/catalogs.md +++ b/design/catalogs.md @@ -204,7 +204,10 @@ Current behavior differs by primitive: sources fetch repository catalogs when online and fall back to packaged snapshots only for catalog-unavailable failures; malformed or unsafe content is not masked by the snapshot. Bundle stacks memoize parsed payloads for the - lifetime of one stack instance. + lifetime of one stack instance. When translating an HTTP status error, the + fetcher closes its response body before propagating the catalog error, while + retaining the original exception as its cause and preserving status-based + fallback classification. A future shared fetch boundary may share URL validation, redirect validation, bounded reads, cache records, and error types. It must preserve each diff --git a/design/file-helper.md b/design/file-helper.md index 77ee80fc4a..6fad90eb33 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -67,7 +67,9 @@ the policy. `specify_cli._project.resolve_specify_project_files()` performs the normal existing-project resolution and returns the same handoff for new CLI adapters. Commands using another established root call `project_files` on their actually selected root. Retain a helper only for the current operation; -a later invocation resolves a new helper, so manual edits take effect. +a later invocation resolves a new helper, so manual edits take effect. Use the +returned owner as a context manager, or call `close()` in the owner's cleanup +path, to release retained root handles at that operation boundary. The `specify_cli.integration_state` helpers are an explicit managed-metadata exception to this handoff. Their reads and writes of `.specify/integration.json` @@ -101,12 +103,12 @@ existing lifecycle rollback, including prior policy and unrelated options. ```python from specify_cli.file_helper import FileHelper -files = FileHelper(project_root) # deny symlinks by default -files.mkdir("generated", exist_ok=True) -files.create_text("generated/config.txt", "initial\n") -files.write_text("generated/config.txt", "updated\n") -content = files.read_text("generated/config.txt") -files.delete("generated", recursive=True) +with FileHelper(project_root) as files: # deny symlinks by default + files.mkdir("generated", exist_ok=True) + files.create_text("generated/config.txt", "initial\n") + files.write_text("generated/config.txt", "updated\n") + content = files.read_text("generated/config.txt") + files.delete("generated", recursive=True) ``` ## Boundary and policy @@ -144,6 +146,66 @@ any followed target hierarchy. An unrelated project symlink does not prevent an operation. A helper cannot cross to another root; callers establish separate helpers for independent project/source/managed boundaries. +On supported deletion backends, construction retains a handle to the canonical +root until `close()` or context-manager exit. Keeping the original directory open +prevents its identity from being reused for a different directory while the +helper is alive; stored device/inode numbers alone are not the binding. On POSIX, +the handle uses search-only access where available. On Windows, construction +requests only root metadata access (`FILE_READ_ATTRIBUTES`) and permits delete +sharing; it does not request directory-list access on the canonical ancestors. +The retained reference selects a tagged identity scheme from that handle. +The full 128-bit `FileIdInfo` identity is preferred. If that query reports +`ERROR_INVALID_FUNCTION`, `ERROR_NOT_SUPPORTED`, or `ERROR_INVALID_PARAMETER`, +or returns an all-zero ID, the helper queries the opened volume's filesystem +and uses `GetFileInformationByHandle`'s volume serial plus nonzero 64-bit file +index only for FAT, FAT32, exFAT, or NTFS. The volume query is handle-relative +and works through the filesystem driver rather than a pathname-based volume +lookup, including when a network driver exposes the necessary information. +ReFS and unrecognized filesystems do not fall back to potentially ambiguous +64-bit indexes. Access-denied, sharing, and other unrelated errors still +propagate rather than selecting a weaker identity. +Before accepting it, construction compares Python-native metadata obtained from +the acquired handle with the initially observed root metadata. This comparison +does not reopen the root pathname, so replacement followed by restoration cannot +substitute a different retained object. The metadata query uses a temporary +non-inheritable duplicate and leaves the original reference open. +If that reference or identity query is unavailable, the native capability error +is deferred to `delete()` rather than preventing ordinary read/write operations. +The cached error remains traceback-free; each failed deletion raises an +independent shallow copy retaining the native error type and filesystem details. + +Windows deletion separately pins the canonical ancestry through the drive/share +root without delete sharing, then queries both that root handle and the retained +reference using the already selected scheme. It compares their live identities, +not a cached legacy index: FAT IDs can change after renaming. It neither +renegotiates identity schemes mid-operation nor compares identities from +different schemes. Those operation-time handles require both +`FILE_LIST_DIRECTORY` and `FILE_READ_ATTRIBUTES` on every canonical ancestor, +including those above the helper root, and reject reparse points. An ACL allowing +traversal but not listing can therefore prevent deletion while still allowing +bootstrap reads. Windows deletion also retains non-reparse handles to accessed +parent directories throughout link resolution, before classifying child entries +or reading link targets. Each child is opened relative to its native parent +handle, not by reconstructing an absolute pathname. Metadata and symbolic-link +targets are read from opened leaf handles; directory names are enumerated from +the directory handle. Relative native opens request both +`FILE_SYNCHRONOUS_IO_NONALERT` and `SYNCHRONIZE`, including directories, so +enumeration cannot consume a buffer before a pending directory query completes. +These accessed directories also require +directory-list/read-attribute access. The resolution handles are released after +selection; native mutation parents and root ancestry remain retained during +preflight and removal. Directory sharing locks alone do not prevent in-place +reparse changes: handle-relative operations retain the opened directory object +instead of following its changed pathname. These locks are released at the end of +deletion; the metadata-only reference remains until the helper closes. Automatic +finalization is a compatibility fallback, not a substitute for deterministic +cleanup. Bootstrap configuration and managed integration-state consumers close +their own helpers explicitly. Installed-workflow listing also closes its +`project_files()` owner on every validation, registry-read, and successful exit; +other consumers receiving `project_files()` own its lifetime. +Pinning the root does not make other pathname-based operations +descriptor-relative or change their documented concurrency limitations. + | Operation | Default deny | Explicit allow | |---|---|---| | Read file | Reject accessed parent or leaf links | Follow only existing contained regular-file targets | @@ -158,6 +220,9 @@ helpers for independent project/source/managed boundaries. ## API and failures +- `close()` releases the helper's retained root handles and is idempotent. + `with FileHelper(...) as files:` closes on both success and failure. Operations + on an explicitly closed helper raise `FileHelperError`. - `mkdir(path, parents=False, exist_ok=False)` creates directories. - `read_bytes(path)` / `read_text(path, encoding="utf-8")` read regular files. - `create_bytes(path, content, mode=0o644)` / `create_text(...)` exclusively @@ -224,7 +289,58 @@ helpers for independent project/source/managed boundaries. the attributes of an existing file. - `delete(path, recursive=False)` unlinks files/links or removes directories. Nonrecursive directory deletion requires an empty directory. The established - root cannot be deleted. + root cannot be deleted. Recursive enumeration and each removal use anchored + directories. Each deletion borrows the root handle retained at construction, + checks the currently named root for replacement, and uses that original anchor through + preflight and all removals. POSIX uses this retained root for component + classification, link-target reads and resolution, recursive preflight, and + per-entry revalidation, as well as descriptor-relative enumeration and + `unlink`/`rmdir`. Canonical pathnames identify root-relative entries but are not + reopened for metadata or link resolution. Windows verifies the full canonical + ancestry during root acquisition and then carries native directory handles + through child opens, metadata queries, link-target reads, and enumeration. + Removal opens the leaf itself with `DELETE` access and marks that same handle + with `FileDispositionInfo`; it does not call pathname `unlink` or `rmdir`. + This does not change the trusted treatment of aliases resolved + at root establishment. Allowed contained parent links are resolved within the + retained boundary, and leaf links are still unlinked rather than followed. + Platforms lacking the required directory-descriptor operations fail with + `ENOTSUP`, not pathname-only deletion. + POSIX traversal and removal use search/path-only descriptors (`O_SEARCH` or + `O_PATH`) where available, including Darwin's native `O_SEARCH` value on Python + builds that do not expose it, so known entries in write-and-execute-only parents + can still be removed. Read-capable descriptors are requested only for directories + being enumerated; systems without search/path-only flags retain read-capable + traversal. Recursive enumeration still requires directory read permission, and + removal still requires the operating system's normal directory permissions. + Preflight opens the enumerated child directly as readable, relative to its + anchored parent, without a search-only open or `"."` reopening first. An empty + readable directory therefore does not need its own search permission to be + removed; accessing children in a nonempty directory still requires search + permission. The parent still needs the permissions required to remove its + directory entry. + On systems using the read-capable fallback, root establishment also requires + read permission on the root. + POSIX component resolution and final parent anchoring retain only the current + traversal descriptor, opening its child before releasing the preceding one. + Their live descriptor count is bounded independently of operation-path depth, + in addition to the helper's retained root. Link-target jumps and `..` are + re-anchored through that root rather than following an opened directory's + physical parent after a rename. Windows retains native directory handles for + its child-relative traversal, including during recursive preflight and removal. + POSIX recursive preflight and mutation use explicit worklists and separate + eight-directory handle caches. The selected parent and established root remain + anchored outside those caches. An uncached parent is reacquired component by + component from the nearest retained prefix with no-follow flags, releasing + least-recently-used handles as the cache fills. Removed directories are evicted, + and preflight handles are not reused for mutation. Live descriptors are bounded + independently of recursive depth, including transient opens and enumeration; + worklist and complete-plan memory still scale with the selected tree. + Reacquiring evicted prefixes trades additional opens for that fixed handle + budget. Total opens can grow with the sum of visited path lengths, rather than + being unconditionally linear in the directory count. Windows retains its + native-handle traversal instead of evicting handles needed for child-relative + operations. The complete plan is still preflighted before any removal. - `symlink(target, path)` creates a link exclusively. Relative targets are interpreted relative to the link parent, not the helper root. @@ -283,11 +399,23 @@ 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 -sandbox. For operations that still use validated paths, another process can -exchange parents or entries between checks and syscalls. An I/O failure during -any mutation can leave partial results. Protect against concurrent untrusted -mutation separately. +deletion rechecks each accessed entry. Deletion also anchors directory +enumeration and removal so exchanging an accessed parent for a link cannot +redirect those operations. Reopening the absolute root pathname for each entry is +avoided; the established root handle is never rebound to a replacement, even if +device/inode values are reused. POSIX removal stays attached to +the opened directory even if that directory is renamed; validation continues to +inspect that same directory tree. Windows prevents ancestor renames while handles +are held. Concurrent changes may still cause deletion to fail or leave +partial results, and preflight is not a transaction or an immutable tree snapshot. +POSIX leaf removal acts on the current entry in the anchored parent; Windows +removal acts on its opened leaf handle. No operation is a general race-free +sandbox. For operations that still +use validated paths (reads, directory creation, atomic writes, and symlink +creation), portable path checks are **not** +a race-free sandbox: another process can exchange parents or entries between +checks and syscalls, and an I/O failure during mutation can leave partial +results. Protect against concurrent untrusted mutation separately. Exclusive file creation anchors its resolved parent hierarchy as described above. On POSIX, renaming an opened directory does not redirect creation to its replacement: creation stays with the opened directory, even if it is @@ -317,7 +445,9 @@ rather than falling back to path-based overwrite. Windows path classification and directory-mode reparse rejection have portable regression tests. Native Windows path and junction tests also exist and are skipped on other hosts; running the portable cases on macOS does not validate -Windows filesystem syscalls or junction behavior. Exclusive-creation tests +Windows filesystem syscalls or junction behavior. Deletion replacement tests +exercise swaps at enumeration and removal boundaries on the native host; Windows +handle-sharing behavior requires a native Windows run. Exclusive-creation tests cover parent replacement before anchoring and at the final open, along with native Windows parent-handle pinning and portable handle-lifetime checks; portable checks do not substitute for native Windows execution. diff --git a/src/specify_cli/_file_handles.py b/src/specify_cli/_file_handles.py index 6b38baaffe..cdeddafb27 100644 --- a/src/specify_cli/_file_handles.py +++ b/src/specify_cli/_file_handles.py @@ -6,11 +6,25 @@ import os import secrets import stat +import struct import tempfile from pathlib import Path, PureWindowsPath +from typing import Literal, NamedTuple -def _open_windows_child(parent: int, name: str, *, directory: bool, access: int, sharing: int) -> int: +_IdentityKind = Literal["fileid128", "legacy64"] + + +class _WindowsFileIdentity(NamedTuple): + kind: _IdentityKind + volume: int + file_id: bytes + + +def _open_windows_child( + parent: int, name: str, *, directory: bool | None, access: int, + sharing: int, allow_reparse: bool = False, +) -> 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}") @@ -41,13 +55,15 @@ def _open_windows_child(parent: int, name: str, *, directory: bool, access: int, # 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, + 0x40 | (0 if allow_reparse else 0x1000), None, None, ) handle = wintypes.HANDLE() status = abi._IoStatusBlock() - options = 0x00200000 | (0x1 if directory else 0x40 | 0x20) + options = 0x00200000 | 0x20 # FILE_OPEN_REPARSE_POINT | FILE_SYNCHRONOUS_IO_NONALERT. + if directory is not None: + options |= 0x1 if directory else 0x40 result = create( - ctypes.byref(handle), access | (0 if directory else 0x100000), + ctypes.byref(handle), access | 0x100000, # SYNCHRONIZE is required for synchronous I/O. ctypes.byref(attributes), ctypes.byref(status), None, 0, sharing, 1, options, None, 0, ) @@ -58,6 +74,121 @@ def _open_windows_child(parent: int, name: str, *, directory: bool, access: int, return handle.value +@contextmanager +def _windows_entry(parent: int, name: str, *, deleting: bool = False) -> Iterator[int]: + """Open the leaf itself relative to its directory, including allowed links.""" + handle = _open_windows_child( + parent, name, directory=None, access=0x80 | (0x10000 if deleting else 0), + sharing=0x7, allow_reparse=True, + ) + try: + yield handle + finally: + _windows_api().CloseHandle(handle) + + +def _windows_mode(handle: int) -> int: + import ctypes + + class Attributes(ctypes.Structure): + _fields_ = [("attributes", ctypes.c_uint32), ("tag", ctypes.c_uint32)] + + info = Attributes() + if not _windows_api().GetFileInformationByHandleEx( + handle, 9, ctypes.byref(info), ctypes.sizeof(info) + ): + raise ctypes.WinError(ctypes.get_last_error()) + if info.attributes & 0x400: + if info.tag != 0xA000000C: + raise OSError(errno.ELOOP, "Unsupported Windows reparse point") + return stat.S_IFLNK + return stat.S_IFDIR if info.attributes & 0x10 else stat.S_IFREG + + +def _windows_readlink(handle: int) -> str: + import ctypes + from ctypes import wintypes + + 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, wintypes.LPDWORD, wintypes.LPVOID, + ] + control.restype = wintypes.BOOL + buffer = ctypes.create_string_buffer(16384) + length = wintypes.DWORD() + if not control( + handle, 0x900A8, None, 0, buffer, len(buffer), ctypes.byref(length), None + ): + raise ctypes.WinError(ctypes.get_last_error()) + data = buffer.raw[:length.value] + if len(data) < 20: + raise OSError(errno.EIO, "Incomplete Windows symlink data") + tag, size, _, offset, count, _, _, flags = struct.unpack_from(" len(data) or end > 8 + size + ): + raise OSError(errno.EIO, "Invalid Windows symlink data") + target = data[20 + offset:end].decode("utf-16-le", errors="surrogatepass") + if "\0" in target: + raise OSError(errno.EIO, "Invalid Windows symlink target") + if not flags & 1 and target.startswith("\\??\\"): + target = "\\\\?\\" + target[4:] + return target + + +def _windows_directory_names(handle: int) -> list[str]: + import ctypes + + buffer = ctypes.create_string_buffer(131072) + names = [] + kind = 15 # FileFullDirectoryRestartInfo, then FileFullDirectoryInfo. + while True: + if not _windows_api().GetFileInformationByHandleEx(handle, kind, buffer, len(buffer)): + error = ctypes.get_last_error() + if error == 18: # ERROR_NO_MORE_FILES. + return names + raise ctypes.WinError(error) + kind = 14 + data = buffer.raw + offset = 0 + while True: + if offset + 68 > len(data): + raise OSError(errno.EIO, "Incomplete Windows directory entry") + next_entry = struct.unpack_from(" len(data): + raise OSError(errno.EIO, "Invalid Windows directory entry") + name = data[offset + 68:end].decode("utf-16-le", errors="surrogatepass") + if name not in (".", ".."): + if PureWindowsPath(name).parts != (name,) or PureWindowsPath(name).anchor or "\0" in name: + raise OSError(errno.EIO, "Invalid Windows directory component") + names.append(name) + if not next_entry: + break + if next_entry % 8 or next_entry < 68 + size: + raise OSError(errno.EIO, "Invalid Windows directory entry offset") + offset += next_entry + + +def _windows_delete(handle: int) -> None: + """Mark the opened object for removal, never reopen its pathname.""" + import ctypes + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + dispose = kernel.SetFileInformationByHandle + dispose.argtypes = [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD] + dispose.restype = wintypes.BOOL + delete = ctypes.c_int32(True) + if not dispose(handle, 4, ctypes.byref(delete), ctypes.sizeof(delete)): + raise ctypes.WinError(ctypes.get_last_error()) + + def _windows_api(): import ctypes from ctypes import wintypes @@ -85,6 +216,7 @@ def _windows_api(): def _open_windows_handle( path: Path, *, directory: bool, overwrite: bool = False, create_attributes: int | None = None, parent: int | None = None, traverse: bool = False, + pin: bool = True, ) -> 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") @@ -96,19 +228,24 @@ class AttributeTagInfo(ctypes.Structure): kernel = _windows_api() access = (0x80 | (0x20 if traverse else 0x1)) if directory else 0x80000000 + if not pin: + access = 0x80 if overwrite: access = 0x40000000 | 0x80 + sharing = 0x3 if pin else 0x7 if parent is None: # FILE_TRAVERSE participates in sharing checks without requiring enumeration. handle = kernel.CreateFileW( - str(path), access, 0x3, None, + str(path), access, sharing, 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) + handle = _open_windows_child( + parent, str(path), directory=directory, access=access, sharing=sharing + ) try: attributes = AttributeTagInfo() if not kernel.GetFileInformationByHandleEx( @@ -128,9 +265,14 @@ class AttributeTagInfo(ctypes.Structure): @contextmanager def _windows_handle( path: Path, *, directory: bool, parent: int | None = None, traverse: bool = False, + pin: bool = True, ) -> Iterator[int]: - """Pin a non-reparse path by withholding Windows delete sharing.""" - if parent is None and not traverse: + """Open a non-reparse object, optionally relative to a retained parent.""" + if not pin: + handle = _open_windows_handle( + path, directory=directory, parent=parent, traverse=traverse, pin=False + ) + elif 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) @@ -140,6 +282,132 @@ def _windows_handle( _windows_api().CloseHandle(handle) +def _windows_filesystem(handle: int) -> str: + """Query the opened volume, including network filesystems, without a pathname.""" + import ctypes + from ctypes import wintypes + from . import _windows_abi as abi + + class FilesystemAttributes(ctypes.Structure): + _fields_ = [ + ("attributes", ctypes.c_uint32), + ("maximum_component", ctypes.c_int32), + ("name_length", ctypes.c_uint32), + ("name", ctypes.c_uint16 * 261), + ] + + native = ctypes.WinDLL("ntdll") + query = native.NtQueryVolumeInformationFile + query.argtypes = [ + wintypes.HANDLE, abi._IoStatusBlockPointer, wintypes.LPVOID, + ctypes.c_uint32, ctypes.c_int, + ] + query.restype = ctypes.c_int32 + dos_error = native.RtlNtStatusToDosError + dos_error.argtypes = [ctypes.c_int32] + dos_error.restype = ctypes.c_uint32 + attributes = FilesystemAttributes() + status = abi._IoStatusBlock() + result = query( + handle, ctypes.byref(status), ctypes.byref(attributes), + ctypes.sizeof(attributes), 5, # FileFsAttributeInformation. + ) + if result != 0: + raise ctypes.WinError(dos_error(result)) + length = attributes.name_length + if not length or length % 2 or length > ctypes.sizeof(attributes.name): + raise OSError(errno.EINVAL, "Invalid filesystem name from volume query") + return bytes(attributes.name)[:length].decode("utf-16-le") + + +def _windows_legacy_file_id(handle: int) -> _WindowsFileIdentity: + """Query a live legacy identity only on filesystems with usable 64-bit IDs.""" + import ctypes + from ctypes import wintypes + + filesystem = _windows_filesystem(handle).upper() + # ReFS needs 128 bits; unknown drivers may report non-identifying indexes. + if filesystem not in {"FAT", "FAT32", "EXFAT", "NTFS"}: + raise OSError(errno.ENOTSUP, f"No reliable legacy file identity for {filesystem}") + + class FileInformation(ctypes.Structure): + _fields_ = [ + ("attributes", ctypes.c_uint32), + ("times", ctypes.c_uint32 * 6), + ("volume", ctypes.c_uint32), + ("size_high", ctypes.c_uint32), + ("size_low", ctypes.c_uint32), + ("links", ctypes.c_uint32), + ("index_high", ctypes.c_uint32), + ("index_low", ctypes.c_uint32), + ] + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + query = kernel.GetFileInformationByHandle + query.argtypes = [wintypes.HANDLE, wintypes.LPVOID] + query.restype = wintypes.BOOL + information = FileInformation() + if not query(handle, ctypes.byref(information)): + raise ctypes.WinError(ctypes.get_last_error()) + index = (information.index_high << 32) | information.index_low + if not index: + raise OSError(errno.ENOTSUP, "No usable legacy file identity") + return _WindowsFileIdentity("legacy64", information.volume, index.to_bytes(8, "little")) + + +def _windows_file_id( + handle: int, *, kind: _IdentityKind | None = None +) -> _WindowsFileIdentity: + """Select a supported identity once, or query the already selected scheme.""" + if kind == "legacy64": + return _windows_legacy_file_id(handle) + import ctypes + from ctypes import wintypes + + class FileIdInfo(ctypes.Structure): + _fields_ = [ + ("volume", ctypes.c_ulonglong), + ("file_id", ctypes.c_ubyte * 16), + ] + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + info = kernel.GetFileInformationByHandleEx + info.argtypes = [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD] + info.restype = wintypes.BOOL + identity = FileIdInfo() + if not info(handle, 18, ctypes.byref(identity), ctypes.sizeof(identity)): + error = ctypes.get_last_error() + if kind is None and error in (1, 50, 87): + return _windows_legacy_file_id(handle) + raise ctypes.WinError(error) + value = bytes(identity.file_id) + if not any(value): + if kind is None: + return _windows_legacy_file_id(handle) + raise OSError(errno.ENOTSUP, "No usable extended file identity") + return _WindowsFileIdentity("fileid128", identity.volume, value) + + +def _windows_stat(handle: int) -> os.stat_result: + """Get Python's native metadata without taking ownership of a borrowed handle.""" + 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_NOINHERIT) + except BaseException: + _winapi.CloseHandle(duplicate) + raise + try: + return os.fstat(fd) + finally: + os.close(fd) + + @contextmanager def windows_read_fd(handle: int) -> Iterator[int]: """Read the validated object through an independently owned duplicate.""" diff --git a/src/specify_cli/_file_policy.py b/src/specify_cli/_file_policy.py index e675c9b614..1adae7c28e 100644 --- a/src/specify_cli/_file_policy.py +++ b/src/specify_cli/_file_policy.py @@ -33,12 +33,13 @@ def read_project_options(project_root: Path) -> dict[str, Any]: """Read without following configuration links; only absence defaults.""" try: files = FileHelper(project_root) - except (OSError, RuntimeError) as exc: + except (OSError, FileHelperError, RuntimeError) as exc: raise ProjectPolicyError( f"Cannot establish project root for {INIT_OPTIONS_FILE}: {exc}" ) from exc try: - text = files.read_text(INIT_OPTIONS_FILE) + with files: + text = files.read_text(INIT_OPTIONS_FILE) except FileNotFoundError: return {} except (OSError, UnicodeError, FileHelperError) as exc: @@ -60,7 +61,7 @@ def project_files(project_root: Path) -> FileHelper: options = read_project_options(project_root) try: return FileHelper(project_root, allow_symlinks=options.get("allow_symlinks", False)) - except (OSError, RuntimeError) as exc: + except (OSError, FileHelperError, RuntimeError) as exc: raise ProjectPolicyError(f"Cannot establish project filesystem boundary: {exc}") from exc @@ -74,19 +75,19 @@ def save_project_options(project_root: Path, options: dict[str, Any]) -> None: except (TypeError, ValueError, RecursionError) as exc: raise ProjectPolicyError(f"Cannot encode {INIT_OPTIONS_FILE} as JSON: {exc}") from exc try: - files = FileHelper(project_root) - # Validate the entire destination before creating even .specify. - try: - files.read_text(INIT_OPTIONS_FILE) - except FileNotFoundError: - pass - from .integrations._file_changes import after_file_change, before_file_change + with FileHelper(project_root) as files: + # Validate the entire destination before creating even .specify. + try: + files.read_text(INIT_OPTIONS_FILE) + except FileNotFoundError: + pass + from .integrations._file_changes import after_file_change, before_file_change - destination = project_root / INIT_OPTIONS_FILE - before_file_change(destination) - files.mkdir(".specify", exist_ok=True) - files.overwrite_text(INIT_OPTIONS_FILE, content) - after_file_change(destination) + destination = project_root / INIT_OPTIONS_FILE + before_file_change(destination) + files.mkdir(".specify", exist_ok=True) + files.overwrite_text(INIT_OPTIONS_FILE, content) + after_file_change(destination) except (OSError, UnicodeError, FileHelperError, RuntimeError) as exc: raise ProjectPolicyError( f"Cannot write {INIT_OPTIONS_FILE}: {exc}. " @@ -127,11 +128,11 @@ def preflight_init_root(target: Path) -> Path: if target.exists() or target.is_symlink(): read_project_options(target) else: - files = FileHelper(_existing_init_parent(target)) - try: - files.read_bytes(target) - except FileNotFoundError: - pass + with FileHelper(_existing_init_parent(target)) as files: + try: + files.read_bytes(target) + except FileNotFoundError: + pass return target.resolve() except (OSError, FileHelperError, RuntimeError) as exc: raise ProjectPolicyError( @@ -148,7 +149,9 @@ def prepare_init_root(target: Path) -> Path: target = target.absolute() _validate_init_traversal(target) if target.exists() or target.is_symlink(): - FileHelper(target) + with FileHelper(target): + pass return target - FileHelper(_existing_init_parent(target)).mkdir(target, parents=True, exist_ok=True) + with FileHelper(_existing_init_parent(target)) as files: + files.mkdir(target, parents=True, exist_ok=True) return target diff --git a/src/specify_cli/bundles/adapters.py b/src/specify_cli/bundles/adapters.py index b88969400d..7853254d3e 100644 --- a/src/specify_cli/bundles/adapters.py +++ b/src/specify_cli/bundles/adapters.py @@ -268,13 +268,16 @@ def _validate_redirect(_old_url: str, new_url: str) -> None: # server-side availability responses (408, 429, 5xx) fall back to the # snapshot. Any other 4xx (404, 403, ...) is definitive and must # surface as a hard error. - if exc.code in _TRANSIENT_HTTP_STATUS_CODES or exc.code >= 500: - raise _CatalogUnavailable( + try: + if exc.code in _TRANSIENT_HTTP_STATUS_CODES or exc.code >= 500: + raise _CatalogUnavailable( + f"Failed to fetch catalog from {url}: HTTP {exc.code} {exc.reason}" + ) from exc + raise BundlerError( f"Failed to fetch catalog from {url}: HTTP {exc.code} {exc.reason}" ) from exc - raise BundlerError( - f"Failed to fetch catalog from {url}: HTTP {exc.code} {exc.reason}" - ) from exc + finally: + exc.close() except urllib.error.URLError as exc: # urllib wraps a TLS handshake failure as URLError(reason=); # a certificate-verification failure is a security failure, not a diff --git a/src/specify_cli/file_helper.py b/src/specify_cli/file_helper.py index f293d53b2e..af15246f66 100644 --- a/src/specify_cli/file_helper.py +++ b/src/specify_cli/file_helper.py @@ -6,15 +6,29 @@ import stat import sys import tempfile +from collections import OrderedDict from collections.abc import Iterator from contextlib import ExitStack, contextmanager +from copy import copy from dataclasses import dataclass, field from pathlib import Path, PurePath, PureWindowsPath from typing import TypeVar +from weakref import finalize + +from specify_cli._file_handles import ( + _WindowsFileIdentity, _windows_delete, _windows_directory_names, _windows_entry, + _windows_file_id, _windows_handle, _windows_mode, _windows_overwrite_fd, + _windows_readlink, _windows_stat, windows_read_fd, +) -from specify_cli._file_handles import _windows_handle, _windows_overwrite_fd, windows_read_fd _PathType = TypeVar("_PathType", bound=PurePath) +_DELETE_DIR_FD_SUPPORTED = ( + {os.open, os.stat, os.readlink, os.unlink, os.rmdir} <= os.supports_dir_fd + and os.stat in os.supports_follow_symlinks + and os.scandir in os.supports_fd +) +_DELETE_DIRECTORY_CACHE_SIZE = 8 # Wrapping os.open must not change the underlying platform capability. _POSIX_OPEN = os.open @@ -45,6 +59,122 @@ class UnsupportedPathError(FileHelperError): """An accessed object cannot be used by the requested operation.""" +@dataclass(frozen=True) +class _WindowsDirectory: + path: Path + handle: int + + +@dataclass +class _WalkDirectories: + root: Path + root_handle: int | _WindowsDirectory + flags: int + resources: ExitStack + opened: dict[Path, _WindowsDirectory] = field(default_factory=dict) + current: tuple[Path, int] | None = field(default=None, init=False) + + def __post_init__(self) -> None: + if isinstance(self.root_handle, int): + self.resources.callback(self._release_current) + + def _release_current(self) -> None: + current, self.current = self.current, None + if current is not None: + os.close(current[1]) + + def _posix_directory(self, path: Path, root_fd: int) -> int: + current_path, parent = self.current or (self.root, root_fd) + if path == current_path: + return parent + if path.parent != current_path: + # Re-anchor jumps, including "..", rather than following a moved + # directory's physical parent outside the retained root. + self._release_current() + current_path, parent = self.root, root_fd + for part in path.relative_to(current_path).parts: + child = os.open(part, self.flags, dir_fd=parent) + previous = self.current + current_path = current_path / part + self.current = (current_path, child) + if previous is not None: + os.close(previous[1]) + parent = child + return parent + + def directory(self, path: Path) -> int | _WindowsDirectory: + path.relative_to(self.root) + if isinstance(self.root_handle, int): + return self._posix_directory(path, self.root_handle) + if path == self.root: + return self.root_handle + if path not in self.opened: + parent = self.directory(path.parent) + if not isinstance(parent, _WindowsDirectory): + raise FileHelperError("Expected a native Windows directory") + try: + handle = self.resources.enter_context(_windows_handle( + Path(path.name), directory=True, parent=parent.handle + )) + except OSError as exc: + if exc.errno == errno.EINVAL and getattr(exc, "winerror", None) is None: + raise NotADirectoryError( + errno.ENOTDIR, "Expected a directory", str(path) + ) from exc + raise + self.opened[path] = _WindowsDirectory(path, handle) + return self.opened[path] + + +@dataclass(frozen=True) +class _DeletionEntry: + path: Path + directory: bool + children: tuple["_DeletionEntry", ...] = () + + +@dataclass +class _DeletionDirectories: + root: Path + root_fd: int + flags: int + resources: ExitStack + opened: OrderedDict[Path, int] = field(default_factory=OrderedDict) + + def __post_init__(self) -> None: + self.resources.callback(self.close) + + def close(self) -> None: + while self.opened: + _, fd = self.opened.popitem() + os.close(fd) + + def discard(self, path: Path) -> None: + fd = self.opened.pop(path, None) + if fd is not None: + os.close(fd) + + def directory(self, path: Path) -> int: + path.relative_to(self.root) + ancestor = path + suffix = [] + while ancestor != self.root and ancestor not in self.opened: + suffix.append(ancestor.name) + ancestor = ancestor.parent + parent = self.root_fd if ancestor == self.root else self.opened[ancestor] + if ancestor in self.opened: + self.opened.move_to_end(ancestor) + for name in reversed(suffix): + child = os.open(name, self.flags, dir_fd=parent) + ancestor = ancestor / name + self.opened[ancestor] = child + if len(self.opened) > _DELETE_DIRECTORY_CACHE_SIZE: + _, oldest = self.opened.popitem(last=False) + os.close(oldest) + parent = child + return parent + + @dataclass(frozen=True) class FileHelper: """Own operations beneath one trusted root with an immutable link policy. @@ -56,14 +186,72 @@ class FileHelper: root: Path allow_symlinks: bool = False _canonical_root: Path = field(init=False, repr=False) + _root_stat: os.stat_result = field(init=False, repr=False, compare=False) + _root_anchor: int | Path | None = field(init=False, repr=False, compare=False) + _root_anchor_error: OSError | None = field(init=False, repr=False, compare=False) + _windows_root_id: _WindowsFileIdentity | None = field(init=False, repr=False, compare=False) + _windows_root_handle: int | None = field(init=False, repr=False, compare=False) + _root_cleanup: finalize = field(init=False, repr=False, compare=False) def __post_init__(self) -> None: root = Path(self.root).absolute() canonical = root.resolve(strict=True) - if not canonical.is_dir(): + root_stat = canonical.stat() + if not stat.S_ISDIR(root_stat.st_mode): raise NotADirectoryError(root) object.__setattr__(self, "root", root) object.__setattr__(self, "_canonical_root", canonical) + object.__setattr__(self, "_root_stat", root_stat) + with ExitStack() as resources: + anchor: int | Path | None = None + anchor_error: OSError | None = None + windows_root_id: _WindowsFileIdentity | None = None + windows_root_handle: int | None = None + if os.name == "nt": + with ExitStack() as reference: + try: + handle = reference.enter_context( + _windows_handle(canonical, directory=True, pin=False) + ) + self._check_deletion_root(_windows_stat(handle)) + windows_root_id = _windows_file_id(handle) + except OSError as exc: + # Other operations do not require deletion's handle capability. + exc.__traceback__ = None + anchor_error = exc + else: + self._check_deletion_root(canonical.stat()) + resources.enter_context(reference.pop_all()) + anchor = canonical + windows_root_handle = handle + elif ( + _DELETE_DIR_FD_SUPPORTED + and getattr(os, "O_NOFOLLOW", 0) + and getattr(os, "O_DIRECTORY", 0) + ): + anchor = os.open(canonical, self._directory_flags()) + resources.callback(os.close, anchor) + self._check_deletion_root(os.fstat(anchor)) + object.__setattr__(self, "_root_anchor", anchor) + object.__setattr__(self, "_root_anchor_error", anchor_error) + object.__setattr__(self, "_windows_root_id", windows_root_id) + object.__setattr__(self, "_windows_root_handle", windows_root_handle) + object.__setattr__(self, "_root_cleanup", finalize(self, resources.pop_all().close)) + + def close(self) -> None: + """Release retained root handles; closing more than once is harmless.""" + self._root_cleanup() + + def _require_open(self) -> None: + if not self._root_cleanup.alive: + raise FileHelperError("FileHelper is closed") + + def __enter__(self) -> "FileHelper": + self._require_open() + return self + + def __exit__(self, *args: object) -> None: + self.close() def _parts(self, path: PurePath) -> tuple[str, ...]: original = path @@ -83,8 +271,11 @@ def _parts(self, path: PurePath) -> tuple[str, ...]: raise PathEscapeError(f"Path is outside root {self.root}: {original}") @staticmethod - def _entry_mode(path: Path) -> int: - entry = path.lstat() + def _entry_mode(path: Path, *, dir_fd: int | None = None) -> int: + entry = ( + path.lstat() if dir_fd is None + else os.stat(path.name, dir_fd=dir_fd, follow_symlinks=False) + ) if ( getattr(entry, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT and not stat.S_ISLNK(entry.st_mode) @@ -101,10 +292,13 @@ def _walk( allow_missing: bool = False, follow_leaf: bool = True, links: tuple[Path, ...] = (), + directories: _WalkDirectories | None = None, ) -> Path: current = self._canonical_root for index, part in enumerate(parts): - if not current.is_dir(): + directory = directories.directory(current) if directories is not None else None + directory_fd = directory if isinstance(directory, int) else None + if directories is None and not current.is_dir(): if not (allow_missing and not current.exists()): raise NotADirectoryError(current) if part == "..": @@ -114,7 +308,9 @@ def _walk( continue candidate = current / part try: - mode = self._entry_mode(candidate) + mode = self._deletion_entry_mode(candidate, parent=directory) if ( + isinstance(directory, _WindowsDirectory) + ) else self._entry_mode(candidate, dir_fd=directory_fd) except FileNotFoundError: if not allow_missing: raise @@ -130,12 +326,20 @@ def _walk( return candidate if candidate in links or len(links) >= 40: raise UnsupportedPathError(f"Symlink cycle or excessive chain: {candidate}") - target = Path(os.readlink(candidate)) + if isinstance(directory, _WindowsDirectory): + with _windows_entry(directory.handle, part) as handle: + target = Path(_windows_readlink(handle)) + elif directory_fd is None: + target = Path(os.readlink(candidate)) + else: + target = Path(os.readlink(part, dir_fd=directory_fd)) target_parts = self._parts(target) if not target.is_absolute(): target_parts = current.relative_to(self._canonical_root).parts + target_parts # Resolve the link target strictly, even when creating a new child. - current = self._walk(target_parts, links=links + (candidate,)) + current = self._walk( + target_parts, links=links + (candidate,), directories=directories + ) elif stat.S_ISDIR(mode) or stat.S_ISREG(mode): current = candidate else: @@ -143,15 +347,25 @@ def _walk( return current def _path( - self, path: Path | str, *, allow_missing: bool = False, follow_leaf: bool = True + self, path: Path | str, *, allow_missing: bool = False, follow_leaf: bool = True, + root_handle: int | _WindowsDirectory | None = None, ) -> Path: + self._require_open() original = Path(path) parts = self._parts(original) if ".." in parts: raise PathEscapeError(f"Parent traversal is not an operation path: {original}") - return self._walk( - parts, allow_missing=allow_missing, follow_leaf=follow_leaf - ) + with ExitStack() as resources: + directories = ( + _WalkDirectories( + self._canonical_root, root_handle, + 0 if isinstance(root_handle, _WindowsDirectory) else self._directory_flags(), resources, + ) + if root_handle is not None else None + ) + return self._walk( + parts, allow_missing=allow_missing, follow_leaf=follow_leaf, directories=directories + ) @staticmethod def _require_file(path: Path) -> None: @@ -211,11 +425,7 @@ def _posix_directory(self, directory: Path, *, unsupported_message: str) -> Iter and _POSIX_OPEN in os.supports_dir_fd ): raise OSError(errno.ENOTSUP, unsupported_message) - access = getattr(os, "O_SEARCH", 0) or getattr(os, "O_PATH", 0) - if not access: - # 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) + flags = self._directory_flags() parent = os.open(self._canonical_root, flags) try: for part in parts: @@ -354,37 +564,217 @@ def symlink(self, target: Path | str, path: Path | str) -> None: resolved = self._walk(target_parts) destination.symlink_to(target, target_is_directory=resolved.is_dir()) - def _deletion_plan(self, path: Path, *, recursive: bool) -> list[tuple[Path, bool]]: - mode = self._entry_mode(path) + @staticmethod + def _directory_flags() -> int: + access = getattr(os, "O_PATH", 0) or getattr(os, "O_SEARCH", 0) + if not access: + # Some older Python builds omit Darwin's native search-only flag. + access = (0x40000000 | os.O_DIRECTORY) if sys.platform == "darwin" else os.O_RDONLY + return access | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + + def _check_deletion_root(self, current: os.stat_result) -> None: + if not os.path.samestat(self._root_stat, current): + raise FileHelperError(f"Established root changed during deletion: {self.root}") + + @contextmanager + def _deletion_root(self) -> Iterator[int | _WindowsDirectory]: + """Borrow the root pinned when this helper was established.""" + self._require_open() + if self._root_anchor is None: + if self._root_anchor_error is not None: + error = copy(self._root_anchor_error) + # OSError's copy protocol omits its native Windows error slot. + if hasattr(self._root_anchor_error, "winerror"): + error.winerror = self._root_anchor_error.winerror + raise error + raise OSError(errno.ENOTSUP, "Handle-based deletion is unsupported on this platform") + if isinstance(self._root_anchor, Path): + reference = self._windows_root_handle + identity = self._windows_root_id + if reference is None or identity is None: + raise FileHelperError("Missing retained Windows root identity") + with ExitStack() as resources: + for ancestor in reversed(self._root_anchor.parents): + resources.enter_context(_windows_handle(ancestor, directory=True)) + handle = resources.enter_context( + _windows_handle(self._root_anchor, directory=True) + ) + selected = _windows_file_id(handle, kind=identity.kind) + retained = _windows_file_id(reference, kind=identity.kind) + if selected != retained: + raise FileHelperError(f"Established root changed during deletion: {self.root}") + yield _WindowsDirectory(self._canonical_root, handle) + else: + self._check_deletion_root(self._canonical_root.stat()) + yield self._root_anchor + + @contextmanager + def _child_deletion_directory( + self, parent: int | _WindowsDirectory, name: str, *, readable: bool = False + ) -> Iterator[int | _WindowsDirectory]: + if isinstance(parent, _WindowsDirectory): + with _windows_handle( + Path(name), directory=True, parent=parent.handle + ) as handle: + yield _WindowsDirectory(parent.path / name, handle) + return + flags = ( + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + if readable else self._directory_flags() + ) + child = os.open(name, flags, dir_fd=parent) + try: + yield child + finally: + os.close(child) + + @contextmanager + def _deletion_directory( + self, path: Path, *, root: int | _WindowsDirectory + ) -> Iterator[int | _WindowsDirectory]: + parts = path.relative_to(self._canonical_root).parts + if isinstance(root, int): + with ExitStack() as resources: + directories = _WalkDirectories( + self._canonical_root, root, self._directory_flags(), resources + ) + yield directories.directory(path) + return + with ExitStack() as resources: + parent = root + for part in parts: + parent = resources.enter_context(self._child_deletion_directory(parent, part)) + yield parent + + @staticmethod + def _windows_entry_mode(handle: int, path: Path) -> int: + try: + return _windows_mode(handle) + except OSError as exc: + if exc.errno == errno.ELOOP and getattr(exc, "winerror", None) is None: + raise UnsupportedPathError(f"Unsupported Windows reparse point: {path}") from exc + raise + + def _deletion_entry_mode(self, path: Path, *, parent: int | _WindowsDirectory) -> int: + if isinstance(parent, _WindowsDirectory): + with _windows_entry(parent.handle, path.name) as handle: + return self._windows_entry_mode(handle, path) + return self._entry_mode(path, dir_fd=parent) + + def _deletion_kind(self, path: Path, mode: int) -> bool: if stat.S_ISLNK(mode): if not self.allow_symlinks: raise SymlinkDeniedError(f"Refusing to delete symlink: {path}") - return [(path, False)] + return False if stat.S_ISREG(mode): - return [(path, False)] + return False if not stat.S_ISDIR(mode): raise UnsupportedPathError(f"Unsupported filesystem object: {path}") - plan: list[tuple[Path, bool]] = [] - if recursive: - with os.scandir(path) as entries: - children = sorted((Path(entry.path) for entry in entries)) - for child in children: - plan.extend(self._deletion_plan(child, recursive=True)) - plan.append((path, True)) - return plan + return True + + def _deletion_plan( + self, path: Path, *, recursive: bool, parent: int | _WindowsDirectory + ) -> _DeletionEntry: + if isinstance(parent, int): + return self._posix_deletion_plan(path, recursive=recursive, parent=parent) + directory = self._deletion_kind(path, self._deletion_entry_mode(path, parent=parent)) + children: tuple[_DeletionEntry, ...] = () + if directory and recursive: + with self._child_deletion_directory(parent, path.name, readable=True) as child_directory: + if not isinstance(child_directory, _WindowsDirectory): + raise FileHelperError("Expected a native Windows directory") + names = sorted(_windows_directory_names(child_directory.handle)) + child_plans = [] + for name in names: + child_plans.append( + self._deletion_plan(path / name, recursive=True, parent=child_directory) + ) + children = tuple(child_plans) + return _DeletionEntry(path, directory, children) + + def _posix_deletion_plan( + self, path: Path, *, recursive: bool, parent: int + ) -> _DeletionEntry: + pending: list[tuple[Path, list[str] | None]] = [(path, None)] + completed: dict[Path, _DeletionEntry] = {} + with ExitStack() as resources: + directories = _DeletionDirectories( + path.parent, parent, self._directory_flags(), resources + ) + while pending: + entry, children = pending.pop() + if children is not None: + completed[entry] = _DeletionEntry( + entry, True, tuple(completed.pop(entry / name) for name in children) + ) + continue + entry_parent = directories.directory(entry.parent) + directory = self._deletion_kind( + entry, self._deletion_entry_mode(entry, parent=entry_parent) + ) + if not (directory and recursive): + completed[entry] = _DeletionEntry(entry, directory) + continue + with self._child_deletion_directory( + entry_parent, entry.name, readable=True + ) as readable: + with os.scandir(readable) as entries: + names = sorted(child.name for child in entries) + pending.append((entry, names)) + for name in reversed(names): + pending.append((entry / name, None)) + return completed[path] + + def _delete_plan(self, entry: _DeletionEntry, *, parent: int | _WindowsDirectory) -> None: + if isinstance(parent, int): + self._posix_delete_plan(entry, parent=parent) + return + with _windows_entry(parent.handle, entry.path.name, deleting=True) as handle: + directory = self._deletion_kind( + entry.path, self._windows_entry_mode(handle, entry.path) + ) + if directory != entry.directory: + raise FileHelperError(f"Destination changed during deletion: {entry.path}") + if directory and entry.children: + child_directory = _WindowsDirectory(entry.path, handle) + for child in entry.children: + self._delete_plan(child, parent=child_directory) + _windows_delete(handle) + + def _posix_delete_plan(self, entry: _DeletionEntry, *, parent: int) -> None: + pending = [(entry, False)] + with ExitStack() as resources: + # Do not reuse preflight handles: parent replacements must be checked + # afresh before any removal. + directories = _DeletionDirectories( + entry.path.parent, parent, self._directory_flags(), resources + ) + while pending: + selected, ready = pending.pop() + selected_parent = directories.directory(selected.path.parent) + directory = self._deletion_kind( + selected.path, + self._deletion_entry_mode(selected.path, parent=selected_parent), + ) + if directory != selected.directory: + raise FileHelperError(f"Destination changed during deletion: {selected.path}") + if directory and selected.children and not ready: + pending.append((selected, True)) + for child in reversed(selected.children): + pending.append((child, False)) + continue + if directory: + os.rmdir(selected.path.name, dir_fd=selected_parent) + directories.discard(selected.path) + else: + os.unlink(selected.path.name, dir_fd=selected_parent) def delete(self, path: Path | str, *, recursive: bool = False) -> None: """Delete an entry; recursive deletion never follows descendant links.""" - destination = self._path(path, follow_leaf=False) - if destination == self._canonical_root: - raise FileHelperError(f"Refusing to delete the established root: {self.root}") - plan = self._deletion_plan(destination, recursive=recursive) - for entry, directory in plan: - # Recheck accessed parents/leaf before each mutation, not just preflight. - checked = self._path(entry, follow_leaf=False) - if checked != entry: - raise FileHelperError(f"Destination changed during deletion: {entry}") - if directory: - entry.rmdir() - else: - entry.unlink() + with self._deletion_root() as root: + destination = self._path(path, follow_leaf=False, root_handle=root) + if destination == self._canonical_root: + raise FileHelperError(f"Refusing to delete the established root: {self.root}") + with self._deletion_directory(destination.parent, root=root) as parent: + plan = self._deletion_plan(destination, recursive=recursive, parent=parent) + self._delete_plan(plan, parent=parent) diff --git a/src/specify_cli/integration_state.py b/src/specify_cli/integration_state.py index ae29b833a8..5e9afb8347 100644 --- a/src/specify_cli/integration_state.py +++ b/src/specify_cli/integration_state.py @@ -46,7 +46,7 @@ def _read_integration_json_data( files = FileHelper(project_root) except FileNotFoundError: return None, None - except (OSError, RuntimeError) as exc: + except (OSError, FileHelperError, RuntimeError) as exc: return None, IntegrationReadError( kind="os", detail=f"Cannot establish project root for {INTEGRATION_JSON}: {exc}" ) @@ -57,7 +57,8 @@ def _read_integration_json_data( # other OSErrors (which become loud errors via the IntegrationReadError # path). try: - raw = files.read_text(INTEGRATION_JSON, encoding="utf-8") + with files: + raw = files.read_text(INTEGRATION_JSON, encoding="utf-8") except FileNotFoundError: return None, None except IsADirectoryError as exc: @@ -245,44 +246,44 @@ def write_integration_json( State reads and writes deny links below that root, independently of any lifecycle transaction or operational symlink opt-in. """ - files = FileHelper(project_root) - # Reject the full destination before an observer can snapshot its target. - try: - files.read_bytes(INTEGRATION_JSON) - except FileNotFoundError: - pass - except SymlinkDeniedError as exc: - raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc - dest = project_root / INTEGRATION_JSON - - integration_key = clean_integration_key(integration_key) - installed = dedupe_integration_keys(installed_integrations or []) - if integration_key and integration_key not in installed: - installed.insert(0, integration_key) - if not integration_key and installed: - integration_key = installed[0] - - normalized_settings = normalize_integration_settings(settings or {}) - normalized_settings = { - key: normalized_settings[key] for key in installed if key in normalized_settings - } - - data: dict[str, Any] = { - "version": version, - "integration_state_schema": INTEGRATION_STATE_SCHEMA, - "installed_integrations": installed, - "integration_settings": normalized_settings, - } - if integration_key: - data["integration"] = integration_key - data["default_integration"] = integration_key - - from .integrations._file_changes import after_file_change, before_file_change - - try: - before_file_change(dest) - files.mkdir(".specify", exist_ok=True) - files.overwrite_text(INTEGRATION_JSON, json.dumps(data, indent=2) + "\n", encoding="utf-8") - after_file_change(dest) - except SymlinkDeniedError as exc: - raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc + with FileHelper(project_root) as files: + # Reject the full destination before an observer can snapshot its target. + try: + files.read_bytes(INTEGRATION_JSON) + except FileNotFoundError: + pass + except SymlinkDeniedError as exc: + raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc + dest = project_root / INTEGRATION_JSON + + integration_key = clean_integration_key(integration_key) + installed = dedupe_integration_keys(installed_integrations or []) + if integration_key and integration_key not in installed: + installed.insert(0, integration_key) + if not integration_key and installed: + integration_key = installed[0] + + normalized_settings = normalize_integration_settings(settings or {}) + normalized_settings = { + key: normalized_settings[key] for key in installed if key in normalized_settings + } + + data: dict[str, Any] = { + "version": version, + "integration_state_schema": INTEGRATION_STATE_SCHEMA, + "installed_integrations": installed, + "integration_settings": normalized_settings, + } + if integration_key: + data["integration"] = integration_key + data["default_integration"] = integration_key + + from .integrations._file_changes import after_file_change, before_file_change + + try: + before_file_change(dest) + files.mkdir(".specify", exist_ok=True) + files.overwrite_text(INTEGRATION_JSON, json.dumps(data, indent=2) + "\n", encoding="utf-8") + after_file_change(dest) + except SymlinkDeniedError as exc: + raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc diff --git a/src/specify_cli/workflows/_operation_list.py b/src/specify_cli/workflows/_operation_list.py index 3f0fa902d5..4844207a42 100644 --- a/src/specify_cli/workflows/_operation_list.py +++ b/src/specify_cli/workflows/_operation_list.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass +from contextlib import ExitStack from itertools import islice from pathlib import Path import stat @@ -95,33 +96,34 @@ def list_workflows(request: WorkflowListRequest) -> WorkflowListResult: raise WorkflowListError( "invalid_limit", f"limit must be between 1 and {WORKFLOW_LIST_MAX_LIMIT}" ) - try: - # Unlike is_dir(), stat() preserves permission errors on Python 3.14. + with ExitStack() as owners: try: - root_is_directory = stat.S_ISDIR(root.stat().st_mode) - except (FileNotFoundError, NotADirectoryError, ValueError): - root_is_directory = False - if not root_is_directory: - raise WorkflowListError( - "not_a_spec_kit_project", - "Not a Spec Kit project (no .specify/ directory)", - ) - files = project_files(root) - if not (root / ".specify").is_dir(): - raise WorkflowListError( - "not_a_spec_kit_project", - "Not a Spec Kit project (no .specify/ directory)", - ) - validate_workflow_storage(root) - except (ProjectPolicyError, WorkflowStorageError, OSError) as exc: - raise WorkflowListError("unsafe_workflow_storage", str(exc)) from exc + # Unlike is_dir(), stat() preserves permission errors on Python 3.14. + try: + root_is_directory = stat.S_ISDIR(root.stat().st_mode) + except (FileNotFoundError, NotADirectoryError, ValueError): + root_is_directory = False + if not root_is_directory: + raise WorkflowListError( + "not_a_spec_kit_project", + "Not a Spec Kit project (no .specify/ directory)", + ) + files = owners.enter_context(project_files(root)) + if not (root / ".specify").is_dir(): + raise WorkflowListError( + "not_a_spec_kit_project", + "Not a Spec Kit project (no .specify/ directory)", + ) + validate_workflow_storage(root) + except (ProjectPolicyError, WorkflowStorageError, OSError) as exc: + raise WorkflowListError("unsafe_workflow_storage", str(exc)) from exc - try: - installed = WorkflowRegistry(root, read_files=files).list() - except OSError as exc: - raise WorkflowListError( - "workflow_registry_read_failed", f"Failed to read workflow registry: {exc}" - ) from exc + try: + installed = WorkflowRegistry(root, read_files=files).list() + except OSError as exc: + raise WorkflowListError( + "workflow_registry_read_failed", f"Failed to read workflow registry: {exc}" + ) from exc items: list[WorkflowListEntry | WorkflowListWarning] = [] selected = installed.items() if limit is None else islice(installed.items(), limit) diff --git a/tests/integrations/test_integration_state.py b/tests/integrations/test_integration_state.py index ceaa9cc845..c3f53fcbc8 100644 --- a/tests/integrations/test_integration_state.py +++ b/tests/integrations/test_integration_state.py @@ -5,7 +5,8 @@ import pytest from specify_cli._file_policy import project_files, save_project_options -from specify_cli.file_helper import SymlinkDeniedError +from specify_cli.file_helper import FileHelper, FileHelperError, SymlinkDeniedError +from specify_cli import integration_state from specify_cli.integration_state import ( INTEGRATION_JSON, default_integration_key, @@ -18,6 +19,59 @@ from specify_cli.integrations._file_changes import file_change_observer +@pytest.mark.parametrize("reader", [try_read_integration_json, try_read_integration_json_with_raw]) +def test_state_readers_translate_constructor_contract_failure(tmp_path, monkeypatch, reader): + def failed(*args, **kwargs): + raise FileHelperError("root binding failed") + + monkeypatch.setattr(integration_state, "FileHelper", failed) + *states, error = reader(tmp_path) + assert all(state is None for state in states) + assert error is not None and error.kind == "os" + assert "root binding failed" in error.detail + + +@pytest.mark.parametrize("operation", ["read", "read-with-raw", "write"]) +@pytest.mark.parametrize("failure", [False, True]) +def test_managed_state_consumers_close_retained_helpers(tmp_path, monkeypatch, operation, failure): + (tmp_path / ".specify").mkdir() + (tmp_path / INTEGRATION_JSON).write_text('{"integration": "claude"}') + captured = [] + + def owner(*args, **kwargs): + files = FileHelper(*args, **kwargs) + captured.append(files) + return files + + def denied(*args, **kwargs): + raise PermissionError("managed state access denied") + + monkeypatch.setattr(integration_state, "FileHelper", owner) + if failure: + monkeypatch.setattr(FileHelper, "overwrite_text" if operation == "write" else "read_text", denied) + retained_error = None + if operation == "write": + if failure: + with pytest.raises(PermissionError, match="managed state access denied") as raised: + write_integration_json(tmp_path, version="1", integration_key="claude") + retained_error = raised.value + else: + write_integration_json(tmp_path, version="1", integration_key="claude") + else: + reader = try_read_integration_json if operation == "read" else try_read_integration_json_with_raw + *states, retained_error = reader(tmp_path) + if failure: + assert all(state is None for state in states) + assert retained_error is not None and retained_error.kind == "os" + else: + assert all(state is not None for state in states) + assert retained_error is None + assert len(captured) == 1 + assert not captured[0]._root_cleanup.alive + if failure: + assert retained_error is not None + + def test_normalize_integration_state_strips_default_key_without_duplicates(): state = normalize_integration_state( { diff --git a/tests/specify_cli/bundles/test_adapters.py b/tests/specify_cli/bundles/test_adapters.py index 40a9dd0688..3eac29dad9 100644 --- a/tests/specify_cli/bundles/test_adapters.py +++ b/tests/specify_cli/bundles/test_adapters.py @@ -132,6 +132,32 @@ def fake_open_url(url, timeout=10, extra_headers=None, redirect_validator=None): assert not isinstance(excinfo.value, adapters._CatalogUnavailable) +@pytest.mark.parametrize("status", [403, 404, 408, 429, 500, 503]) +@pytest.mark.parametrize("already_closed", [False, True]) +def test_http_error_response_is_closed_before_catalog_error_translation( + monkeypatch, status, already_closed +): + body = io.BytesIO(b"error response") + error = urllib.error.HTTPError( + "https://example.com/catalog.json", status, "response failed", {}, body, + ) + if already_closed: + error.close() + + def failed(*args, **kwargs): + raise error + + monkeypatch.setattr("specify_cli.authentication.http.open_url", failed) + with pytest.raises(BundlerError, match=f"HTTP {status}") as raised: + adapters._http_get_json("team", error.url) + assert raised.value.__cause__ is error + assert isinstance(raised.value, adapters._CatalogUnavailable) == ( + status in (408, 429) or status >= 500 + ) + assert body.closed + assert error.closed + + def test_http_get_json_preserves_malformed_json(monkeypatch): def fake_open_url(url, timeout=10, extra_headers=None, redirect_validator=None): return _FakeResponse(b"not json", url) diff --git a/tests/specify_cli/test_file_policy.py b/tests/specify_cli/test_file_policy.py index 3f7f356545..24df7c6f13 100644 --- a/tests/specify_cli/test_file_policy.py +++ b/tests/specify_cli/test_file_policy.py @@ -11,7 +11,7 @@ from typer.testing import CliRunner from specify_cli import app -from specify_cli.file_helper import FileHelper +from specify_cli.file_helper import FileHelper, FileHelperError def policy_module(): @@ -20,6 +20,66 @@ def policy_module(): return _file_policy +@pytest.mark.parametrize("operation", ["read_project_options", "project_files"]) +def test_policy_translates_constructor_contract_failure(tmp_path, monkeypatch, operation): + policy = policy_module() + if operation == "project_files": + monkeypatch.setattr(policy, "read_project_options", lambda _: {}) + + def failed(*args, **kwargs): + raise FileHelperError("root binding failed") + + monkeypatch.setattr(policy, "FileHelper", failed) + with pytest.raises(policy.ProjectPolicyError, match="root binding failed") as raised: + getattr(policy, operation)(tmp_path) + assert isinstance(raised.value.__cause__, FileHelperError) + + +@pytest.mark.parametrize( + ("operation", "failing_method"), + [ + ("read_project_options", "read_text"), + ("save_project_options", "overwrite_text"), + ("preflight_init_root", "read_bytes"), + ("prepare_init_root", "mkdir"), + ], +) +@pytest.mark.parametrize("failure", [False, True]) +def test_bootstrap_consumers_close_retained_helpers( + tmp_path, monkeypatch, operation, failing_method, failure +): + policy = policy_module() + (tmp_path / ".specify").mkdir() + (tmp_path / ".specify/init-options.json").write_text("{}") + captured = [] + + def owner(*args, **kwargs): + files = FileHelper(*args, **kwargs) + captured.append(files) + return files + + def denied(*args, **kwargs): + raise PermissionError("bootstrap access denied") + + monkeypatch.setattr(policy, "FileHelper", owner) + if failure: + monkeypatch.setattr(FileHelper, failing_method, denied) + target = tmp_path / "missing/project" if operation.endswith("init_root") else tmp_path + args = (target, {}) if operation == "save_project_options" else (target,) + retained_error = None + if failure: + error_type = PermissionError if operation == "prepare_init_root" else policy.ProjectPolicyError + with pytest.raises(error_type, match="bootstrap access denied") as raised: + getattr(policy, operation)(*args) + retained_error = raised.value + else: + getattr(policy, operation)(*args) + assert len(captured) == 1 + assert not captured[0]._root_cleanup.alive + if failure: + assert retained_error is not None + + @pytest.fixture def symlinks(tmp_path): probe = tmp_path / "probe" diff --git a/tests/specify_cli/workflows/test_operation_list.py b/tests/specify_cli/workflows/test_operation_list.py index c2730dcb23..3379e2cd66 100644 --- a/tests/specify_cli/workflows/test_operation_list.py +++ b/tests/specify_cli/workflows/test_operation_list.py @@ -52,6 +52,47 @@ def test_missing_registry_is_empty_without_creating_storage(tmp_path): assert list((tmp_path / ".specify").iterdir()) == [] +@pytest.mark.parametrize("outcome", ["success", "missing-marker", "storage-error", "registry-error"]) +def test_list_workflows_closes_owner_on_every_exit(tmp_path, monkeypatch, outcome): + if outcome != "missing-marker": + _registry(tmp_path, {}) + captured = [] + original = operation.project_files + + def owner(root): + files = original(root) + captured.append(files) + return files + + def storage_error(*args): + raise operation.WorkflowStorageError("storage unavailable") + + def registry_error(*args): + raise OSError("registry unavailable") + + monkeypatch.setattr(operation, "project_files", owner) + if outcome == "storage-error": + monkeypatch.setattr(operation, "validate_workflow_storage", storage_error) + elif outcome == "registry-error": + monkeypatch.setattr(operation.WorkflowRegistry, "list", registry_error) + error = None + if outcome == "success": + assert operation.list_workflows(operation.WorkflowListRequest(tmp_path)).items == () + else: + with pytest.raises(operation.WorkflowListError) as raised: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + error = raised.value + assert error.code == { + "missing-marker": "not_a_spec_kit_project", + "storage-error": "unsafe_workflow_storage", + "registry-error": "workflow_registry_read_failed", + }[outcome] + assert len(captured) == 1 + assert not captured[0]._root_cleanup.alive + if outcome != "success": + assert error is not None + + @pytest.mark.parametrize("contents", ["{", "[]", '{"workflows": []}', "\udcff"]) def test_corrupt_registry_fails_closed(tmp_path, contents): path = _registry(tmp_path, {}) diff --git a/tests/test_file_handles.py b/tests/test_file_handles.py index a1201e64b6..520edb437a 100644 --- a/tests/test_file_handles.py +++ b/tests/test_file_handles.py @@ -1,25 +1,384 @@ """Ownership and native Windows evidence for shared handle primitives.""" from concurrent.futures import ThreadPoolExecutor -from contextlib import contextmanager +from contextlib import contextmanager, nullcontext import ctypes from ctypes import wintypes import errno import gc import os +import stat from pathlib import Path import sys import subprocess +import struct from types import SimpleNamespace import weakref import pytest from specify_cli import _file_handles, file_helper -from specify_cli._file_handles import _windows_handle +from specify_cli import _file_handles as handles +from specify_cli._file_handles import ( + _WindowsFileIdentity, _windows_file_id, _windows_handle, _windows_stat, +) from specify_cli.file_helper import FileHelper, UnsupportedPathError +@pytest.mark.parametrize("directory", [False, True]) +@pytest.mark.parametrize("pin", [False, True]) +@pytest.mark.parametrize("outcome", ["success", "reparse", "wrong-type", "query-failure", "body-failure", "open-failure"]) +def test_windows_handle_flags_validation_and_cleanup(monkeypatch, directory, pin, outcome): + calls = [] + + def create(*args): + calls.append(("open", args)) + return ctypes.c_void_p(-1).value if outcome == "open-failure" else 123 + + def info(handle, kind, buffer, size): + assert handle == 123 and kind == 9 + attributes = 0x10 if directory else 0 + if outcome == "reparse": + attributes |= 0x400 + if outcome == "wrong-type": + attributes ^= 0x10 + buffer._obj.attributes = attributes + return outcome != "query-failure" + + def close(handle): + calls.append(("close", handle)) + return True + + # Real ctypes functions accept signature attributes; plain Python functions + # do too, allowing this contract to run without loading a Windows DLL. + kernel = SimpleNamespace( + CreateFileW=create, GetFileInformationByHandleEx=info, CloseHandle=close + ) + monkeypatch.setattr(ctypes, "WinDLL", lambda *args, **kwargs: kernel, raising=False) + monkeypatch.setattr(ctypes, "get_last_error", lambda: 5, raising=False) + monkeypatch.setattr( + ctypes, "WinError", lambda code: PermissionError(code, "handle denied"), raising=False + ) + + def use(): + with _windows_handle(Path("selected"), directory=directory, pin=pin) as handle: + assert handle == 123 + assert calls == [calls[0]] + if outcome == "body-failure": + raise RuntimeError("body failed") + + if outcome == "success": + use() + elif outcome == "body-failure": + with pytest.raises(RuntimeError, match="body failed"): + use() + else: + with pytest.raises(OSError) as exc: + use() + if outcome in ("reparse", "wrong-type"): + assert exc.value.errno == (errno.ELOOP if outcome == "reparse" else errno.EINVAL) + args = calls[0][1] + assert args == ( + "selected", (0x81 if directory else 0x80000000) if pin else 0x80, + 0x3 if pin else 0x7, None, 3, 0x00200000 | 0x02000000, None, + ) + assert calls[1:] == ([] if outcome == "open-failure" else [("close", 123)]) + + +@pytest.mark.parametrize("failure", [False, True]) +def test_windows_file_identity_uses_full_volume_and_file_id(monkeypatch, failure): + def info(handle, kind, buffer, size): + assert handle == 0x1FFFFFFFF and kind == 18 and size == 24 + buffer._obj.volume = 0x1FFFFFFFF + buffer._obj.file_id[:] = bytes(range(16)) + return not failure + + monkeypatch.setattr( + ctypes, "WinDLL", + lambda *args, **kwargs: SimpleNamespace(GetFileInformationByHandleEx=info), + raising=False, + ) + monkeypatch.setattr(ctypes, "get_last_error", lambda: 5, raising=False) + monkeypatch.setattr(ctypes, "WinError", lambda code: PermissionError(code, "identity denied"), raising=False) + if failure: + with pytest.raises(PermissionError, match="identity denied"): + _windows_file_id(0x1FFFFFFFF) + else: + assert _windows_file_id(0x1FFFFFFFF) == _WindowsFileIdentity( + "fileid128", 0x1FFFFFFFF, bytes(range(16)) + ) + assert info.argtypes == [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD] + assert info.restype is wintypes.BOOL + + +@pytest.fixture +def identity_kernel(monkeypatch): + state = SimpleNamespace( + extended_error=87, legacy_error=0, last_error=0, filesystem="FAT32", + volume=0xABCDEF12, index=0x123456789ABCDEF, zero_extended=False, + ) + calls = [] + + def extended(handle, kind, output, size): + assert handle == 0x1FFFFFFFF and kind == 18 and size == 24 + calls.append("extended") + if state.extended_error: + state.last_error = state.extended_error + return False + output._obj.volume = state.volume + output._obj.file_id[:] = bytes(16) if state.zero_extended else bytes(range(16)) + return True + + def legacy(handle, output): + assert handle == 0x1FFFFFFFF + assert ctypes.sizeof(output._obj) == 52 + calls.append("legacy") + if state.legacy_error: + state.last_error = state.legacy_error + return False + output._obj.volume = state.volume + output._obj.index_high = state.index >> 32 + output._obj.index_low = state.index & 0xFFFFFFFF + return True + + def native_error(code): + error = OSError(errno.EACCES if code == 5 else errno.ENOTSUP, f"native error {code}") + error.winerror = code + return error + + monkeypatch.setattr(ctypes, "WinDLL", lambda *args, **kwargs: SimpleNamespace( + GetFileInformationByHandleEx=extended, GetFileInformationByHandle=legacy, + ), raising=False) + monkeypatch.setattr(ctypes, "get_last_error", lambda: state.last_error, raising=False) + monkeypatch.setattr(ctypes, "WinError", native_error, raising=False) + monkeypatch.setattr( + handles, "_windows_filesystem", lambda _: state.filesystem, raising=False + ) + return state, calls + + +@pytest.mark.parametrize("unsupported", [1, 50, 87]) +@pytest.mark.parametrize("filesystem", ["FAT", "FAT32", "exFAT", "NTFS"]) +def test_windows_identity_falls_back_only_for_supported_legacy_filesystems( + identity_kernel, unsupported, filesystem +): + state, calls = identity_kernel + state.extended_error = unsupported + state.filesystem = filesystem + identity = _windows_file_id(0x1FFFFFFFF) + assert identity.kind == "legacy64" + assert identity.volume == state.volume + assert identity.file_id == state.index.to_bytes(8, "little") + assert calls == ["extended", "legacy"] + + +@pytest.mark.parametrize("error_code", [5, 6, 32]) +def test_windows_identity_does_not_hide_unrelated_native_errors(identity_kernel, error_code): + state, calls = identity_kernel + state.extended_error = error_code + with pytest.raises(OSError) as raised: + _windows_file_id(0x1FFFFFFFF) + assert raised.value.winerror == error_code + assert calls == ["extended"] + + +@pytest.mark.parametrize("filesystem", ["ReFS", "unknown"]) +def test_windows_identity_rejects_unreliable_legacy_scheme(identity_kernel, filesystem): + state, calls = identity_kernel + state.filesystem = filesystem + with pytest.raises(OSError, match="legacy.*identity"): + _windows_file_id(0x1FFFFFFFF) + assert calls == ["extended"] + + +@pytest.mark.parametrize("scheme", ["legacy64", "fileid128"]) +def test_windows_identity_uses_selected_scheme_without_renegotiating(identity_kernel, scheme): + state, calls = identity_kernel + state.extended_error = 0 + identity = _windows_file_id(0x1FFFFFFFF, kind=scheme) + assert identity.kind == scheme + assert calls == (["legacy"] if scheme == "legacy64" else ["extended"]) + + +def test_windows_identity_selected_extended_scheme_does_not_downgrade(identity_kernel): + state, calls = identity_kernel + with pytest.raises(OSError) as raised: + _windows_file_id(0x1FFFFFFFF, kind="fileid128") + assert raised.value.winerror == state.extended_error + assert calls == ["extended"] + + +@pytest.mark.parametrize("invalid", [False, True]) +def test_windows_filesystem_query_is_handle_relative_and_preserves_abi(monkeypatch, invalid): + from specify_cli import _windows_abi as abi + + def query(handle, status, output, size, kind): + assert handle == 0x1FFFFFFFF and size == 536 and kind == 5 + assert isinstance(status._obj, abi._IoStatusBlock) + name = "exFAT".encode("utf-16-le") + output._obj.name_length = 1 if invalid else len(name) + ctypes.memmove(output._obj.name, name, len(name)) + return 0 + + def dos_error(status): + pytest.fail("A successful native query must not query an error") + + monkeypatch.setattr(ctypes, "WinDLL", lambda *args: SimpleNamespace( + NtQueryVolumeInformationFile=query, RtlNtStatusToDosError=dos_error, + ), raising=False) + if invalid: + with pytest.raises(OSError, match="Invalid filesystem name"): + handles._windows_filesystem(0x1FFFFFFFF) + else: + assert handles._windows_filesystem(0x1FFFFFFFF) == "exFAT" + assert query.argtypes == [ + wintypes.HANDLE, abi._IoStatusBlockPointer, wintypes.LPVOID, + ctypes.c_uint32, ctypes.c_int, + ] + assert query.restype is ctypes.c_int32 + + +def test_windows_filesystem_query_preserves_native_failure(monkeypatch): + def query(*args): + return -1 + + def dos_error(status): + assert status == -1 + return 5 + + def win_error(code): + assert code == 5 + error = PermissionError("volume query denied") + error.winerror = code + return error + + monkeypatch.setattr(ctypes, "WinDLL", lambda *args: SimpleNamespace( + NtQueryVolumeInformationFile=query, RtlNtStatusToDosError=dos_error, + ), raising=False) + monkeypatch.setattr(ctypes, "WinError", win_error, raising=False) + with pytest.raises(PermissionError, match="volume query denied"): + handles._windows_filesystem(123) + + +def test_windows_identity_zero_extended_id_uses_reliable_legacy_identity(identity_kernel): + state, calls = identity_kernel + state.extended_error = 0 + state.zero_extended = True + identity = _windows_file_id(0x1FFFFFFFF) + assert identity.kind == "legacy64" + assert calls == ["extended", "legacy"] + + +def test_windows_identity_zero_legacy_id_is_not_accepted(identity_kernel): + state, calls = identity_kernel + state.index = 0 + with pytest.raises(OSError, match="identity"): + _windows_file_id(0x1FFFFFFFF) + assert calls == ["extended", "legacy"] + + +def test_windows_identity_legacy_query_error_remains_native(identity_kernel): + state, _ = identity_kernel + state.legacy_error = 5 + with pytest.raises(OSError) as raised: + _windows_file_id(0x1FFFFFFFF) + assert raised.value.winerror == 5 + + +@pytest.mark.parametrize("failure", [None, "duplicate", "convert", "stat"]) +def test_windows_stat_preserves_borrowed_handle_and_closes_duplicate(tmp_path, monkeypatch, failure): + expected = tmp_path.stat() + closed = [] + + def duplicate(source_process, handle, target_process, access, inherit, options): + assert source_process == target_process == 0x1FFFFFFFF + assert handle == 0x2FFFFFFFF + assert access == 0 and inherit is False and options == 2 + if failure == "duplicate": + raise OSError("duplicate failed") + return 456 + + def convert(handle, flags): + assert handle == 456 and flags == 128 + if failure == "convert": + raise OSError("convert failed") + return 789 + + def fstat(fd): + assert fd == 789 + if failure == "stat": + raise OSError("stat failed") + return expected + + monkeypatch.setitem(sys.modules, "_winapi", SimpleNamespace( + GetCurrentProcess=lambda: 0x1FFFFFFFF, + DuplicateHandle=duplicate, + DUPLICATE_SAME_ACCESS=2, + CloseHandle=lambda handle: closed.append(("handle", handle)), + )) + monkeypatch.setitem(sys.modules, "msvcrt", SimpleNamespace(open_osfhandle=convert)) + monkeypatch.setattr(handles, "os", SimpleNamespace( + O_RDONLY=0, O_NOINHERIT=128, fstat=fstat, close=lambda fd: closed.append(("fd", fd)), + )) + if failure: + with pytest.raises(OSError, match=f"{failure} failed"): + _windows_stat(0x2FFFFFFFF) + else: + assert _windows_stat(0x2FFFFFFFF) is expected + assert closed == { + None: [("fd", 789)], "duplicate": [], "convert": [("handle", 456)], "stat": [("fd", 789)], + }[failure] + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows directory-handle metadata") +def test_native_windows_stat_matches_path_metadata_without_closing_reference(tmp_path): + with _windows_handle(tmp_path, directory=True, pin=False) as handle: + identity = _windows_file_id(handle) + assert os.path.samestat(tmp_path.stat(), _windows_stat(handle)) + assert _windows_file_id(handle) == identity + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows legacy identity APIs") +@pytest.mark.parametrize("recursive", [False, True]) +@pytest.mark.parametrize("allow", [False, True]) +def test_native_windows_delete_uses_legacy_identity_when_extended_query_is_unsupported( + tmp_path, monkeypatch, recursive, allow +): + parent = tmp_path / "tree/nested" + parent.mkdir(parents=True) + (parent / "victim").write_text("selected") + native = handles._windows_api() + win_dll = ctypes.WinDLL + extended_queries = [] + + def query(handle, kind, output, size): + if kind == 18: + extended_queries.append(handle) + ctypes.set_last_error(87) + return False + return native.GetFileInformationByHandleEx(handle, kind, output, size) + + kernel = SimpleNamespace( + CreateFileW=native.CreateFileW, + CloseHandle=native.CloseHandle, + GetFileInformationByHandleEx=query, + GetFileInformationByHandle=native.GetFileInformationByHandle, + SetFileInformationByHandle=native.SetFileInformationByHandle, + DeviceIoControl=native.DeviceIoControl, + ) + + def load(name, *args, **kwargs): + return kernel if name == "kernel32" else win_dll(name, *args, **kwargs) + + monkeypatch.setattr(ctypes, "WinDLL", load) + with FileHelper(tmp_path, allow_symlinks=allow) as files: + assert files._windows_root_id is not None + assert files._windows_root_id.kind == "legacy64" + files.delete("tree" if recursive else "tree/nested/victim", recursive=recursive) + assert len(extended_queries) == 1 + assert not (parent / "victim").exists() + + @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): @@ -684,9 +1043,9 @@ def create_child(output, access, attributes, status, allocation, file_attributes 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 access == (0x1000A0 if directory else 0x80100000) assert sharing == 3 and disposition == 1 - assert options == (0x200001 if directory else 0x200060) + assert options == (0x200021 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": @@ -737,6 +1096,34 @@ def load(name, **kwargs): 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 synchronous directory I/O") +@pytest.mark.parametrize("traverse", [False, True]) +def test_native_windows_relative_directory_handle_has_synchronous_mode(tmp_path, traverse): + from specify_cli import _windows_abi as abi + + directory = tmp_path / "directory" + directory.mkdir() + (directory / "file").write_text("selected") + native = ctypes.WinDLL("ntdll") + query = native.NtQueryInformationFile + query.argtypes = [ + wintypes.HANDLE, abi._IoStatusBlockPointer, wintypes.LPVOID, + ctypes.c_uint32, ctypes.c_int, + ] + query.restype = ctypes.c_int32 + status = abi._IoStatusBlock() + mode = ctypes.c_uint32() + with _windows_handle(tmp_path, directory=True) as parent: + with _windows_handle( + Path("directory"), directory=True, parent=parent, traverse=traverse, + ) as child: + result = query(child, ctypes.byref(status), ctypes.byref(mode), 4, 16) + assert result == 0 + assert mode.value & 0x20 # FILE_SYNCHRONOUS_IO_NONALERT. + if not traverse: + assert handles._windows_directory_names(child) == ["file"] + + @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): @@ -776,6 +1163,203 @@ def test_windows_child_open_rejects_oversized_unicode_name(): assert error.value.errno == errno.ENAMETOOLONG +@pytest.mark.parametrize("deleting", [False, True]) +@pytest.mark.parametrize("body_failure", [False, True]) +def test_windows_entry_open_uses_native_parent_and_closes_its_handle( + monkeypatch, deleting, body_failure +): + closed = [] + + def create(output, access, attributes, status, allocation, file_attributes, + sharing, disposition, options, ea, ea_length): + assert access == 0x100080 | (0x10000 if deleting else 0) + assert attributes._obj.RootDirectory == 51 + assert attributes._obj.Attributes == 0x40 + assert attributes._obj.ObjectName.contents.Buffer == "entry" + assert sharing == 7 and disposition == 1 and options == 0x200020 + output._obj.value = 77 + return 0 + + def dos_error(status): + pytest.fail("A successful native open must not query an error") + + monkeypatch.setattr(ctypes, "WinDLL", lambda *args: SimpleNamespace( + NtCreateFile=create, RtlNtStatusToDosError=dos_error, + ), raising=False) + monkeypatch.setattr(handles, "_windows_api", lambda: SimpleNamespace(CloseHandle=closed.append)) + expected = pytest.raises(RuntimeError, match="body failed") if body_failure else nullcontext() + with expected: + with handles._windows_entry(51, "entry", deleting=deleting) as handle: + assert handle == 77 and not closed + if body_failure: + raise RuntimeError("body failed") + assert closed == [77] + + +@pytest.mark.parametrize( + ("attributes", "tag", "mode"), + [(0, 0, stat.S_IFREG), (0x10, 0, stat.S_IFDIR), (0x400, 0xA000000C, stat.S_IFLNK)], +) +def test_windows_mode_queries_the_opened_object(monkeypatch, attributes, tag, mode): + def query(handle, kind, output, size): + assert handle == 51 and kind == 9 and size == 8 + output._obj.attributes = attributes + output._obj.tag = tag + return True + + monkeypatch.setattr(handles, "_windows_api", lambda: SimpleNamespace( + GetFileInformationByHandleEx=query, + )) + assert handles._windows_mode(51) == mode + + +def test_windows_mode_rejects_non_symlink_reparse_points(monkeypatch): + def query(handle, kind, output, size): + output._obj.attributes = 0x410 + output._obj.tag = 0xA0000003 + return True + + monkeypatch.setattr(handles, "_windows_api", lambda: SimpleNamespace( + GetFileInformationByHandleEx=query, + )) + with pytest.raises(OSError, match="Unsupported Windows reparse"): + handles._windows_mode(51) + + +@pytest.mark.parametrize("target", ["directory/../caf\u00e9", "name\ud800", "\\??\\C:\\target"]) +def test_windows_readlink_reads_the_opened_reparse_buffer(monkeypatch, target): + encoded = target.encode("utf-16-le", errors="surrogatepass") + relative = not target.startswith("\\??\\") + data = struct.pack("= 2 + for fd in opened: + with pytest.raises(OSError) as exc: + os.fstat(fd) + assert exc.value.errno == errno.EBADF + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX capability check") +def test_delete_unsupported_backend_does_not_remove(root, monkeypatch): + (root / "victim").write_text("keep") + monkeypatch.setattr("specify_cli.file_helper._DELETE_DIR_FD_SUPPORTED", False) + with pytest.raises(OSError) as exc: + FileHelper(root).delete("victim") + assert exc.value.errno == errno.ENOTSUP + assert (root / "victim").read_text() == "keep" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX root descriptor acquisition") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("recursive", [False, True]) +def test_delete_rejects_root_ancestor_exchange_at_open(root, link, monkeypatch, allow, recursive): + ancestor = root / "ancestor" + project = ancestor / "project" + project.mkdir(parents=True) + (project / "tree").mkdir() + (project / "tree/victim").write_text("selected") + outside = root / "outside" + (outside / "project/tree").mkdir(parents=True) + (outside / "project/tree/victim").write_text("keep") + retained = root / "retained" + open_directory = os.open + attempted = False + opened = [] + + def exchange(path, flags, *args, **kwargs): + nonlocal attempted + if not attempted and path == project: + attempted = True + ancestor.rename(retained) + link(ancestor, outside) + fd = open_directory(path, flags, *args, **kwargs) + opened.append(fd) + return fd + + monkeypatch.setattr(os, "open", exchange) + with pytest.raises(FileHelperError, match="root changed"): + with FileHelper(project, allow_symlinks=allow) as files: + files.delete("tree" if recursive else "tree/victim", recursive=recursive) + assert attempted + assert (outside / "project/tree/victim").read_text() == "keep" + assert (retained / "project/tree/victim").read_text() == "selected" + assert len(opened) == 1 + with pytest.raises(OSError) as exc: + os.fstat(opened[0]) + assert exc.value.errno == errno.EBADF + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX root descriptor lifetime") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("recursive", [False, True]) +def test_delete_reuses_root_after_ancestor_exchange(root, link, monkeypatch, allow, recursive): + ancestor = root / "ancestor" + project = ancestor / "project" + project.mkdir(parents=True) + (project / "tree").mkdir() + (project / "tree/victim").write_text("selected") + outside = root / "outside" + (outside / "project/tree").mkdir(parents=True) + (outside / "project/tree/victim").write_text("keep") + files = FileHelper(project, allow_symlinks=allow) + retained = root / "retained" + check = FileHelper._path + attempted = False + + def exchange(self, path, **kwargs): + nonlocal attempted + result = check(self, path, **kwargs) + if not attempted: + attempted = True + ancestor.rename(retained) + link(ancestor, outside) + return result + + monkeypatch.setattr(FileHelper, "_path", exchange) + files.delete("tree" if recursive else "tree/victim", recursive=recursive) + assert attempted + assert (outside / "project/tree/victim").read_text() == "keep" + assert not (retained / "project/tree/victim").exists() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX retained-root path resolution") +@pytest.mark.parametrize("kind", ["file", "directory", "recursive-tree"]) +@pytest.mark.parametrize("target", ["relative", "absolute", "chain", "parent-traversal"]) +def test_delete_resolves_links_in_retained_root(root, link, monkeypatch, kind, target): + ancestor = root / "ancestor" + project = ancestor / "project" + outside = root / "outside/project" + for boundary in (project, outside): + for name in ("selected", "unrelated"): + directory = boundary / name + directory.mkdir(parents=True) + victim = directory / "victim" + if kind == "file": + victim.write_text("keep") + else: + victim.mkdir() + if kind == "recursive-tree": + (victim / "child").write_text("keep") + for boundary, selected in ((project, "selected"), (outside, "unrelated")): + if target == "absolute": + destination = boundary / selected + elif target == "chain": + link(boundary / "first", selected) + destination = "first" + elif target == "parent-traversal": + (boundary / "nested").mkdir() + destination = f"nested/../{selected}" + else: + destination = selected + link(boundary / "alias", destination) + files = FileHelper(project, allow_symlinks=True) + retained = root / "retained" + acquire = FileHelper._deletion_root + + @contextmanager + def exchange(self): + with acquire(self) as handle: + ancestor.rename(retained) + link(ancestor, outside.parent) + yield handle + + monkeypatch.setattr(FileHelper, "_deletion_root", exchange) + files.delete("alias/victim", recursive=kind == "recursive-tree") + assert not (retained / "project/selected/victim").exists() + assert (retained / "project/unrelated/victim").exists() + assert (retained / "project/alias").is_symlink() + assert (outside / "selected/victim").exists() + assert (outside / "unrelated/victim").exists() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX retained-root preflight") +def test_delete_classifies_descendants_in_retained_root(root, link, monkeypatch): + ancestor = root / "ancestor" + project = ancestor / "project" + outside = root / "outside/project" + for boundary in (project, outside): + (boundary / "tree").mkdir(parents=True) + (boundary / "tree/a-file").write_text("keep") + link(project / "tree/z-link", root / "missing") + (outside / "tree/z-link").write_text("replacement") + files = FileHelper(project) + retained = root / "retained" + acquire = FileHelper._deletion_root + + @contextmanager + def exchange(self): + with acquire(self) as handle: + ancestor.rename(retained) + link(ancestor, outside.parent) + yield handle + + monkeypatch.setattr(FileHelper, "_deletion_root", exchange) + with pytest.raises(SymlinkDeniedError): + files.delete("tree", recursive=True) + assert (retained / "project/tree/a-file").read_text() == "keep" + assert (retained / "project/tree/z-link").is_symlink() + assert (outside / "tree/z-link").read_text() == "replacement" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor-relative metadata") +@pytest.mark.parametrize("recursive", [False, True]) +def test_delete_metadata_and_links_use_retained_descriptors(root, link, monkeypatch, recursive): + (root / "selected/tree").mkdir(parents=True) + (root / "selected/tree/victim").write_text("selected") + link(root / "alias", "selected") + files = FileHelper(root, allow_symlinks=True) + acquire = FileHelper._deletion_root + stat_entry = os.stat + readlink = os.readlink + calls = [] + + def relative_stat(path, *args, **kwargs): + assert isinstance(kwargs.get("dir_fd"), int) + assert kwargs.get("follow_symlinks") is False + assert not Path(path).is_absolute() + calls.append("stat") + return stat_entry(path, *args, **kwargs) + + def relative_readlink(path, *args, **kwargs): + assert isinstance(kwargs.get("dir_fd"), int) + assert not Path(path).is_absolute() + calls.append("readlink") + return readlink(path, *args, **kwargs) + + def no_path_metadata(*args, **kwargs): + pytest.fail("Deletion metadata must not reopen an absolute pathname") + + @contextmanager + def observe(self): + with acquire(self) as handle, monkeypatch.context() as scoped: + scoped.setattr(Path, "lstat", no_path_metadata) + scoped.setattr(os, "stat", relative_stat) + scoped.setattr(os, "readlink", relative_readlink) + yield handle + + monkeypatch.setattr(FileHelper, "_deletion_root", observe) + files.delete("alias/tree" if recursive else "alias/tree/victim", recursive=recursive) + assert "stat" in calls and "readlink" in calls + assert not (root / "selected/tree/victim").exists() + assert (root / "alias").is_symlink() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX retained-root revalidation") +@pytest.mark.parametrize("recursive", [False, True]) +def test_delete_revalidates_entries_in_retained_root(root, link, monkeypatch, recursive): + ancestor = root / "ancestor" + project = ancestor / "project" + (project / "selected").mkdir(parents=True) + (project / "selected/victim").write_text("selected") + outside = root / "outside/project" + outside.mkdir(parents=True) + link(outside / "selected", root / "missing") + files = FileHelper(project) + retained = root / "retained" + plan = FileHelper._deletion_plan + selected = project / ("selected" if recursive else "selected/victim") + attempted = False + + def exchange(self, path, **kwargs): + nonlocal attempted + result = plan(self, path, **kwargs) + if path == selected: + attempted = True + ancestor.rename(retained) + link(ancestor, outside.parent) + return result + + monkeypatch.setattr(FileHelper, "_deletion_plan", exchange) + files.delete(selected, recursive=recursive) + assert attempted + assert not (retained / "project/selected/victim").exists() + assert (outside / "selected").is_symlink() + + +@pytest.mark.parametrize( + ("kind", "error"), + [ + ("dangling", FileNotFoundError), + ("cycle", UnsupportedPathError), + ("excessive", UnsupportedPathError), + ("non-directory", NotADirectoryError), + ("escape-and-return", PathEscapeError), + ], +) +def test_delete_parent_link_targets_preserve_failure_contract(root, link, kind, error): + (root / "target").mkdir() + (root / "target/victim").write_text("keep") + if kind == "dangling": + target = "missing" + elif kind == "cycle": + target = "alias" + elif kind == "excessive": + for index in reversed(range(41)): + link(root / f"chain-{index}", "target" if index == 40 else f"chain-{index + 1}") + target = "chain-0" + elif kind == "non-directory": + (root / "file").write_text("keep") + target = "file/../target" + else: + target = f"../{root.name}/target" + link(root / "alias", target) + with pytest.raises(error): + FileHelper(root, allow_symlinks=True).delete("alias/victim") + assert (root / "target/victim").read_text() == "keep" + assert (root / "alias").is_symlink() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions") +@pytest.mark.parametrize("parent_is_root", [False, True]) +@pytest.mark.parametrize("kind", ["file", "empty-directory", "recursive-tree"]) +def test_delete_through_write_execute_only_parent(root, parent_is_root, kind): + if not (sys.platform == "darwin" or getattr(os, "O_SEARCH", 0) or getattr(os, "O_PATH", 0)): + pytest.skip("Search/path-only directory descriptors are unavailable") + if os.geteuid() == 0: + pytest.skip("Root bypasses directory read permissions") + parent = root if parent_is_root else root / "parent" + parent.mkdir(exist_ok=True) + control = parent / "control" + control.write_text("control") + victim = parent / "victim" + if kind == "file": + victim.write_text("selected") + else: + victim.mkdir() + if kind == "recursive-tree": + (victim / "child").write_text("selected") + files = FileHelper(root) + parent.chmod(0o300) + try: + control.unlink() + files.delete(victim, recursive=kind == "recursive-tree") + assert not victim.exists() + finally: + parent.chmod(0o700) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions") +@pytest.mark.parametrize("mode", [0o400, 0o600, 0o444, 0o644]) +@pytest.mark.parametrize("nested", [False, True]) +@pytest.mark.parametrize("allow", [False, True]) +def test_recursive_delete_readable_empty_directory_without_search(root, mode, nested, allow): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory permission checks") + tree = root / "tree" + tree.mkdir() + restricted = tree + if nested: + (tree / "a-first").write_text("selected") + restricted = tree / "z-empty" + restricted.mkdir() + restricted.chmod(mode) + try: + with FileHelper(root, allow_symlinks=allow) as files: + files.delete(tree, recursive=True) + assert not tree.exists() + finally: + if restricted.exists(): + restricted.chmod(0o700) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions") +@pytest.mark.parametrize("mode", [0o400, 0o600]) +@pytest.mark.parametrize("allow", [False, True]) +def test_recursive_delete_nonempty_directory_without_search_fails_preflight(root, mode, allow): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory permission checks") + tree = root / "tree" + restricted = tree / "z-restricted" + restricted.mkdir(parents=True) + (tree / "a-first").write_text("keep") + (restricted / "victim").write_text("keep") + restricted.chmod(mode) + try: + with FileHelper(root, allow_symlinks=allow) as files: + with pytest.raises(PermissionError): + files.delete(tree, recursive=True) + assert (tree / "a-first").read_text() == "keep" + finally: + restricted.chmod(0o700) + assert (restricted / "victim").read_text() == "keep" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions") +@pytest.mark.parametrize("nonempty", [False, True]) +def test_recursive_delete_still_requires_read_permission_for_enumeration(root, nonempty): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory read permissions") + tree = root / "tree" + tree.mkdir() + if nonempty: + (tree / "victim").write_text("keep") + tree.chmod(0o300) + try: + with FileHelper(root) as files: + with pytest.raises(PermissionError): + files.delete(tree, recursive=True) + if nonempty: + assert (tree / "victim").read_text() == "keep" + else: + assert tree.is_dir() + finally: + tree.chmod(0o700) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX directory permissions") +def test_delete_still_requires_write_permission_on_parent(root): + if os.geteuid() == 0: + pytest.skip("Root bypasses directory write permissions") + parent = root / "parent" + parent.mkdir() + (parent / "victim").write_text("keep") + parent.chmod(0o100) + try: + with pytest.raises(PermissionError): + FileHelper(root).delete(parent / "victim") + assert (parent / "victim").read_text() == "keep" + finally: + parent.chmod(0o700) + + +@pytest.mark.parametrize("allow", [False, True]) +def test_delete_rejects_replaced_root_before_validation(root, monkeypatch, allow): + (root / "victim").write_text("selected") + files = FileHelper(root, allow_symlinks=allow) + retained = root.parent / "retained" + root.rename(retained) + root.mkdir() + (root / "victim").write_text("keep") + + def unexpected(*args, **kwargs): + pytest.fail("A replacement root must be rejected before path validation") + + monkeypatch.setattr(FileHelper, "_path", unexpected) + with pytest.raises(FileHelperError, match="root changed"): + files.delete("victim") + assert (root / "victim").read_text() == "keep" + assert (retained / "victim").read_text() == "selected" + + +def test_windows_delete_rejects_replaced_root_during_acquisition_portably( + root, link, monkeypatch +): + import specify_cli.file_helper as module + + ancestor = root / "ancestor" + project = ancestor / "project" + project.mkdir(parents=True) + (project / "victim").write_text("selected") + outside = root / "outside" + (outside / "project").mkdir(parents=True) + (outside / "project/victim").write_text("keep") + retained = root / "retained" + active = [] + identities = {} + metadata = {} + attempted = False + + @contextmanager + def handle(path, *, directory, pin=True): + nonlocal attempted + assert directory + value = len(identities) + 1 + active.append(path) + try: + if pin and path == project: + # Simulate a replacement just before the root handle is acquired. + attempted = True + ancestor.rename(retained) + link(ancestor, outside) + metadata[value] = path.stat() + identities[value] = (1, b"replacement" if attempted else b"original") + yield value + finally: + assert active.pop() == path + + backend = SimpleNamespace(**vars(os)) + backend.name = "nt" + monkeypatch.setattr(module, "os", backend) + monkeypatch.setattr(module, "_windows_handle", handle) + monkeypatch.setattr( + module, "_windows_file_id", + lambda value, **kwargs: _WindowsFileIdentity("fileid128", *identities[value]), + ) + monkeypatch.setattr(module, "_windows_stat", lambda value: metadata[value]) + with FileHelper(project) as files: + assert not attempted + with pytest.raises(FileHelperError, match="root changed"): + files.delete("victim") + assert active == [project] + assert attempted + assert active == [] + assert (outside / "project/victim").read_text() == "keep" + assert (retained / "project/victim").read_text() == "selected" + + +@pytest.mark.parametrize("kind", ["file", "directory", "recursive"]) +def test_windows_link_resolution_pins_accessed_parents_portably( + root, link, monkeypatch, kind, windows_delete_model +): + parent = root / "parent" + parent.mkdir() + outside = root.parent / "outside" + outside.mkdir() + selected = root / "selected" + unrelated = root / "unrelated" + for directory in (selected, unrelated): + directory.mkdir() + if kind == "file": + (directory / "victim").write_text("keep") + else: + (directory / "victim").mkdir() + if kind == "recursive": + (directory / "victim/child").write_text("keep") + link(parent / "alias", selected) + link(outside / "alias", unrelated) + retained = root / "retained" + pinned = [] + attempted = False + blocked = False + lstat = Path.lstat + + @contextmanager + def handle(path, *, directory, pin=True, parent=None): + assert directory + actual = path if parent is None else windows_delete_model.locations[parent] / path + if pin: + pinned.append(actual) + try: + with windows_delete_model.opened(path, parent=parent) as value: + yield value + finally: + if pin: + assert pinned.pop() == actual + + def exchange(path, **kwargs): + nonlocal attempted, blocked + if path == parent / "alias" and not attempted: + attempted = True + blocked = parent in pinned + if not blocked: + parent.rename(retained) + link(parent, outside) + return lstat(path, **kwargs) + + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr( + file_helper, "_windows_file_id", + lambda _, **kwargs: _WindowsFileIdentity("fileid128", 1, b"root"), + ) + monkeypatch.setattr(file_helper, "_windows_stat", lambda _: root.stat()) + monkeypatch.setattr(Path, "lstat", exchange) + with FileHelper(root, allow_symlinks=True) as files: + files.delete("parent/alias/victim", recursive=kind == "recursive") + assert attempted + assert (unrelated / "victim").exists() + assert not (selected / "victim").exists() + assert blocked + assert (parent / "alias").is_symlink() + assert pinned == [] + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows directory-handle semantics") +def test_native_windows_link_resolution_blocks_accessed_parent_exchange(root, link, monkeypatch): + parent = root / "parent" + parent.mkdir() + selected = root / "selected" + selected.mkdir() + (selected / "victim").write_text("selected") + link(parent / "alias", selected) + attempted = False + original_entry = file_helper._windows_entry + + @contextmanager + def exchange(directory, name, *, deleting=False): + nonlocal attempted + with original_entry(directory, name, deleting=deleting) as handle: + if name == "alias" and not attempted: + attempted = True + with pytest.raises(PermissionError): + parent.rename(root / "retained") + yield handle + + monkeypatch.setattr(file_helper, "_windows_entry", exchange) + with FileHelper(root, allow_symlinks=True) as files: + files.delete("parent/alias/victim") + assert attempted + assert not (selected / "victim").exists() + parent.rename(root / "retained") + + +@pytest.mark.parametrize("error_number", [errno.EINVAL, errno.EACCES]) +def test_windows_walk_preserves_parent_type_and_access_errors_portably( + root, monkeypatch, error_number +): + @contextmanager + def denied(*args, **kwargs): + raise OSError(error_number, "directory unavailable") + yield + + monkeypatch.setattr(file_helper, "_windows_handle", denied) + expected = NotADirectoryError if error_number == errno.EINVAL else PermissionError + with file_helper.ExitStack() as resources: + directories = file_helper._WalkDirectories( + root, file_helper._WindowsDirectory(root, 123), 0, resources + ) + with pytest.raises(expected): + directories.directory(root / "child") + + +def test_windows_constructor_rejects_replaced_then_restored_root(root, monkeypatch): + project = root / "project" + replacement = root / "replacement" + saved = root / "saved" + project.mkdir() + replacement.mkdir() + (project / "victim").write_text("selected") + (replacement / "victim").write_text("keep") + acquired = {} + active = [] + + @contextmanager + def handle(path, *, directory, pin=True): + assert path == project and directory and not pin + project.rename(saved) + replacement.rename(project) + acquired[123] = project.stat() + project.rename(replacement) + saved.rename(project) + active.append(123) + try: + yield 123 + finally: + assert active.pop() == 123 + + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr( + file_helper, "_windows_file_id", + lambda _, **kwargs: _WindowsFileIdentity("fileid128", 1, b"replacement"), + ) + monkeypatch.setattr(file_helper, "_windows_stat", lambda value: acquired[value], raising=False) + with pytest.raises(FileHelperError, match="root changed"): + with FileHelper(project): + pytest.fail("A handle to the replacement root must not be accepted") + assert active == [] + assert (project / "victim").read_text() == "selected" + assert (replacement / "victim").read_text() == "keep" + + +@pytest.mark.parametrize("simulate_missing_winerror", [False, True]) +def test_windows_cached_capability_error_has_independent_tracebacks( + root, monkeypatch, simulate_missing_winerror +): + native = PermissionError(errno.EACCES, "reference denied", "source", 5, "destination") + if not hasattr(native, "winerror"): + native.winerror = 5 + if simulate_missing_winerror: + original_copy = file_helper.copy + + def copy_without_winerror(error): + result = original_copy(error) + result.winerror = None + return result - def observe(source, destination): - replacements.append((Path(source), destination)) - assert first.is_symlink() and second.is_symlink() - assert target.read_bytes() == b"old" - replace(source, destination) + monkeypatch.setattr(file_helper, "copy", copy_without_winerror) - monkeypatch.setattr("specify_cli.file_helper.os.replace", observe) - files.write_text(second, "new") - assert len(replacements) == 1 - assert replacements[0][0].parent == target.parent - assert replacements[0][1] == target - assert os.readlink(first) == "target" - assert os.readlink(second) == "first" - assert target.read_bytes() == b"new" - assert sorted(p.name for p in root.iterdir()) == ["first", "second", "target"] + @contextmanager + def denied(*args, **kwargs): + raise native + yield + + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", denied) + with FileHelper(root) as files: + errors = [] + for _ in range(3): + with pytest.raises(PermissionError) as raised: + files.delete("victim") + errors.append(raised.value) + assert files._root_anchor_error is native + assert native.__traceback__ is None + assert len({id(error) for error in errors}) == 3 + for error in errors: + assert error is not native and type(error) is type(native) + assert (error.errno, error.strerror, error.filename, error.filename2, error.winerror) == ( + native.errno, native.strerror, native.filename, native.filename2, native.winerror, + ) + assert error.__traceback__ is not None -def test_allow_operations_through_linked_parent(root, link): - target = root / "target" - target.mkdir() - alias = link(root / "alias", target) - files = FileHelper(root, allow_symlinks=True) - files.mkdir(alias / "created/child", parents=True) - files.create_text(alias / "created/file", "old") - assert files.read_text(alias / "created/file") == "old" - files.write_text(alias / "created/file", "new") - assert (target / "created/file").read_text() == "new" - files.delete(alias / "created", recursive=True) - assert alias.is_symlink() and target.is_dir() - assert list(target.iterdir()) == [] +@pytest.fixture +def windows_victim_read_fd(root, monkeypatch): + @contextmanager + def duplicate(handle): + with (root / "victim").open("rb") as stream: + yield stream.fileno() + monkeypatch.setattr(file_helper, "windows_read_fd", duplicate) -def test_delete_through_parent_link_removes_real_file_not_parent_link(root, link): - target = root / "target" - target.mkdir() - (target / "file").write_text("delete") - alias = link(root / "alias", target) - FileHelper(root, allow_symlinks=True).delete(alias / "file") - assert alias.is_symlink() - assert target.is_dir() - assert not (target / "file").exists() +def test_windows_read_does_not_require_ancestor_listing_portably( + root, monkeypatch, windows_victim_read_fd +): + (root / "victim").write_text("keep") + active = [] -@pytest.mark.parametrize("allow", [False, True]) -def test_empty_directory_nonrecursive_delete(root, allow): - files = FileHelper(root, allow_symlinks=allow) - files.mkdir("empty") - files.delete("empty") - assert not (root / "empty").exists() + @contextmanager + def handle(path, *, directory, pin=True, parent=None, traverse=False): + if pin and directory and not traverse and path != root: + raise PermissionError("ancestor listing denied") + active.append((path, pin)) + try: + yield 123 + finally: + assert active.pop() == (path, pin) + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr( + file_helper, "_windows_file_id", + lambda _, **kwargs: _WindowsFileIdentity("fileid128", 1, b"root"), + ) + monkeypatch.setattr(file_helper, "_windows_stat", lambda _: root.stat()) + with FileHelper(root) as files: + assert files.read_text("victim") == "keep" + with pytest.raises(PermissionError, match="ancestor listing denied"): + files.delete("victim") + assert files.read_text("victim") == "keep" + assert active == [] + + +@pytest.mark.parametrize("failure", ["open", "identity"]) +def test_windows_read_preserves_native_deletion_capability_error( + root, monkeypatch, failure, windows_victim_read_fd +): + (root / "victim").write_text("keep") + active = [] -@pytest.mark.parametrize("operation", ["mkdir", "create_bytes", "write_bytes", "delete"]) -def test_dangling_parent_cannot_be_used_for_mutation(root, link, operation): - alias = link(root / "alias", root / "missing") - files = FileHelper(root, allow_symlinks=True) - path = alias / "child" - args = (path, b"bad") if operation in ("create_bytes", "write_bytes") else (path,) - with pytest.raises(FileNotFoundError): - getattr(files, operation)(*args) - assert alias.is_symlink() - assert not (root / "missing").exists() + @contextmanager + def denied(*args, **kwargs): + if failure == "open" and not kwargs.get("pin", True): + raise PermissionError("root reference denied") + active.append(123) + try: + yield 123 + finally: + active.pop() + def identity(*args): + raise PermissionError("root reference denied") -def test_multi_link_cycle_and_excessive_chain_fail_explicitly(root, link): - link(root / "first", "second") - link(root / "second", "first") - files = FileHelper(root, allow_symlinks=True) - with pytest.raises(UnsupportedPathError, match="cycle"): - files.read_bytes("first") - (root / "target").write_text("ok") - for index in reversed(range(41)): - link(root / f"chain-{index}", "target" if index == 40 else f"chain-{index + 1}") - with pytest.raises(UnsupportedPathError, match="excessive chain"): - files.read_bytes("chain-0") + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", denied) + monkeypatch.setattr(file_helper, "_windows_file_id", identity) + monkeypatch.setattr(file_helper, "_windows_stat", lambda _: root.stat()) + with FileHelper(root) as files: + assert active == [] + assert files.read_text("victim") == "keep" + with pytest.raises(PermissionError, match="root reference denied"): + files.delete("victim") + assert not files._root_cleanup.alive -@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes", "mkdir", "create_bytes", "delete"]) -def test_external_parent_is_never_traversed(root, link, operation): - outside = root.parent / "outside" - outside.mkdir() - (outside / "file").write_bytes(b"old") - alias = link(root / "alias", outside) - files = FileHelper(root, allow_symlinks=True) - path = alias / "file" - args = (path, b"new") if operation in ("write_bytes", "create_bytes") else (path,) - with pytest.raises(PathEscapeError): - getattr(files, operation)(*args) - assert (outside / "file").read_bytes() == b"old" - assert alias.is_symlink() +def test_windows_delete_checks_retained_handle_identity_portably( + root, monkeypatch, windows_victim_read_fd +): + (root / "victim").write_text("keep") + active = [] + identities = {} + @contextmanager + def handle(path, *, directory, pin=True, parent=None, traverse=False): + value = len(identities) + 1 + identities[value] = (1, b"replacement" if pin else b"original") + active.append(value) + try: + yield value + finally: + assert active.pop() == value -@pytest.mark.parametrize("kind", ["external", "dangling", "cycle"]) -@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes"]) -def test_allow_reads_and_updates_reject_invalid_targets(root, link, kind, operation): - if kind == "external": - target = root.parent / "outside" - target.write_bytes(b"old") - error = PathEscapeError - elif kind == "dangling": - target = root / "missing" - error = FileNotFoundError - else: - target = root / "alias" - error = UnsupportedPathError - alias = link(root / "alias", target) - args = (alias, b"new") if operation == "write_bytes" else (alias,) - with pytest.raises(error): - getattr(FileHelper(root, allow_symlinks=True), operation)(*args) - assert alias.is_symlink() - if kind == "external": - assert target.read_bytes() == b"old" - if kind == "dangling": - assert not target.exists() + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr( + file_helper, "_windows_file_id", + lambda value, **kwargs: _WindowsFileIdentity("fileid128", *identities[value]), + ) + monkeypatch.setattr(file_helper, "_windows_stat", lambda _: root.stat()) + with FileHelper(root) as files: + with pytest.raises(FileHelperError, match="root changed"): + files.delete("victim") + assert len(active) == 1 + assert files.read_text("victim") == "keep" + assert active == [] -@pytest.mark.parametrize("kind", ["file", "directory", "external", "dangling", "cycle"]) +@pytest.mark.parametrize("scenario", ["same", "renamed-reference", "replacement-reused-id"]) @pytest.mark.parametrize("recursive", [False, True]) -def test_allow_leaf_delete_unlinks_only_link(root, link, kind, recursive): - target = root / "target" - if kind == "file": - target.write_bytes(b"keep") - elif kind == "directory": - target.mkdir() - (target / "keep").write_bytes(b"keep") - elif kind == "external": - target = root.parent / "outside" - target.write_bytes(b"keep") - elif kind == "cycle": - target = root / "alias" - alias = link(root / "alias", target) - FileHelper(root, allow_symlinks=True).delete(alias, recursive=recursive) - assert not os.path.lexists(alias) - if kind in ("file", "external"): - assert target.read_bytes() == b"keep" - elif kind == "directory": - assert (target / "keep").read_bytes() == b"keep" +def test_windows_legacy_root_identity_is_compared_live_with_the_selected_scheme( + root, monkeypatch, scenario, recursive, windows_delete_model +): + (root / "tree").mkdir() + (root / "tree/victim").write_text("keep") + active = [] + reference = None + queries = [] + changed = False + + @contextmanager + def handle(path, *, directory, pin=True, parent=None): + nonlocal reference + assert directory + with windows_delete_model.opened(path, parent=parent) as value: + active.append(value) + if not pin: + reference = value + try: + yield value + finally: + assert active.pop() == value + + def identity(value, *, kind=None): + queries.append((value, kind)) + assert kind in (None, "legacy64") + index = 1 + if changed and scenario != "same": + index = 2 if value == reference or scenario == "renamed-reference" else 1 + return _WindowsFileIdentity("legacy64", 1, index.to_bytes(8, "little")) + + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr(file_helper, "_windows_file_id", identity) + monkeypatch.setattr(file_helper, "_windows_stat", lambda _: root.stat()) + with FileHelper(root) as files: + assert files._windows_root_id is not None + assert files._windows_root_id.kind == "legacy64" + changed = True + selected = "tree" if recursive else "tree/victim" + if scenario == "replacement-reused-id": + with pytest.raises(FileHelperError, match="root changed"): + files.delete(selected, recursive=recursive) + assert (root / "tree/victim").read_text() == "keep" + else: + files.delete(selected, recursive=recursive) + assert not (root / "tree/victim").exists() + assert active == [reference] + assert active == [] + assert queries[0] == (reference, None) + assert len(queries) == 3 + assert queries[-1] == (reference, "legacy64") + assert all(kind == "legacy64" for _, kind in queries[1:]) + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows directory ACL semantics") +@pytest.mark.parametrize("consumer", ["helper", "policy", "state"]) +def test_windows_readonly_helper_supports_traverse_only_ancestor(root, consumer): + ancestor = root / "ancestor" + project = ancestor / "project" + project.mkdir(parents=True) + (project / "victim").write_text("keep") + (project / ".specify").mkdir() + (project / ".specify/init-options.json").write_text("{}") + (project / ".specify/integration.json").write_text('{"integration": "claude"}') + user = subprocess.check_output(["whoami"], text=True).strip() + subprocess.run( + ["icacls", str(ancestor), "/deny", f"{user}:(RD)"], + check=True, capture_output=True, text=True, + ) + try: + with FileHelper(project) as files: + if consumer == "helper": + assert files.read_text("victim") == "keep" + elif consumer == "policy": + from specify_cli._file_policy import read_project_options + assert read_project_options(project) == {} + else: + from specify_cli.integration_state import try_read_integration_json + state, error = try_read_integration_json(project) + assert error is None and state is not None and state["integration"] == "claude" + with pytest.raises(PermissionError): + files.delete("victim") + assert files.read_text("victim") == "keep" + finally: + subprocess.run( + ["icacls", str(ancestor), "/remove:d", user], + check=True, capture_output=True, text=True, + ) -def test_recursive_deny_preflight_leaves_entire_tree_untouched(root, link): - tree = root / "tree" - tree.mkdir() - (tree / "a-file").write_bytes(b"keep") - (tree / "b-dir").mkdir() - (tree / "b-dir/keep").write_bytes(b"keep") - alias = link(tree / "z-link", root / "missing") - with pytest.raises(SymlinkDeniedError): - FileHelper(root).delete(tree, recursive=True) - assert (tree / "a-file").read_bytes() == b"keep" - assert (tree / "b-dir/keep").read_bytes() == b"keep" +@pytest.mark.parametrize("failure", [False, True]) +@pytest.mark.parametrize("recursive", [False, True]) +def test_windows_delete_holds_ancestors_until_io_completes_portably( + root, link, monkeypatch, failure, recursive, windows_delete_model +): + import specify_cli.file_helper as module + + parent = root / "target/nested" + parent.mkdir(parents=True) + (parent / "victim").write_text("selected") + alias = link(root / "alias", root / "target") + active = [] + references = [] + events = [] + + @contextmanager + def handle(path, *, directory, pin=True, parent=None): + assert directory + actual = path if parent is None else windows_delete_model.locations[parent] / path + selected_handles = active if pin else references + selected_handles.append(actual) + try: + with windows_delete_model.opened(path, parent=parent) as value: + yield value + finally: + assert selected_handles.pop() == actual + + def assert_anchored(handle): + assert handle in windows_delete_model.locations + assert active[0] == Path(root.anchor) + assert root.resolve() in active + assert all(not part.is_symlink() for part in active) + events.append(windows_delete_model.locations[handle]) + + def observe_scan(handle): + assert_anchored(handle) + return windows_delete_model.names(handle) + + def remove(handle): + assert_anchored(handle) + if failure: + raise PermissionError("removal denied") + windows_delete_model.remove(handle) + + backend = SimpleNamespace(**vars(os)) + backend.name = "nt" + monkeypatch.setattr(module, "os", backend) + monkeypatch.setattr(module, "_windows_handle", handle) + monkeypatch.setattr(module, "_windows_directory_names", observe_scan) + monkeypatch.setattr(module, "_windows_delete", remove) + monkeypatch.setattr( + module, "_windows_file_id", + lambda _, **kwargs: _WindowsFileIdentity("fileid128", 1, b"root"), + ) + monkeypatch.setattr(module, "_windows_stat", lambda _: root.stat()) + files = FileHelper(root, allow_symlinks=True) + selected = alias / "nested" if recursive else alias / "nested/victim" + if failure: + with pytest.raises(PermissionError, match="removal denied"): + files.delete(selected, recursive=recursive) + assert (parent / "victim").read_text() == "selected" + else: + files.delete(selected, recursive=recursive) + assert not (parent / "victim").exists() + assert events + assert active == [] + assert references == [root.resolve()] + files.close() + assert references == [] assert alias.is_symlink() -def test_recursive_allow_does_not_descend_through_links(root, link): - tree = root / "tree" - tree.mkdir() - target = root / "retained" - target.mkdir() - (target / "keep").write_bytes(b"keep") +@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor model of native Windows handles") +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("recursive", [False, True]) +def test_windows_deletion_does_not_reopen_changed_parent_path( + root, link, monkeypatch, allow, recursive +): + parent = root / "tree" + parent.mkdir() + victim = parent / "victim" + victim.write_text("selected") outside = root.parent / "outside" outside.mkdir() - (outside / "keep").write_bytes(b"keep") - link(tree / "internal", target) - link(tree / "external", outside) - link(tree / "dangling", root / "missing") - link(tree / "cycle", tree) - (tree / "ordinary").write_bytes(b"delete") - FileHelper(root, allow_symlinks=True).delete(tree, recursive=True) - assert not tree.exists() - assert (target / "keep").read_bytes() == b"keep" - assert (outside / "keep").read_bytes() == b"keep" + (outside / "victim").write_text("keep") + retained = root / "retained" + opened = {} + changed = False + @contextmanager + def handle(path, *, directory, parent=None, **kwargs): + fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent) + try: + yield fd + finally: + os.close(fd) -def test_deletion_rechecks_parents_after_preflight(root, link, monkeypatch): - tree = root / "tree" - (tree / "nested").mkdir(parents=True) - (tree / "nested/file").write_text("keep") - plan = FileHelper._deletion_plan + @contextmanager + def entry(parent, name, *, deleting=False): + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent) + opened[fd] = (parent, name) + try: + yield fd + finally: + del opened[fd] + os.close(fd) - def exchange(self, path, *, recursive): - result = plan(self, path, recursive=recursive) - if path == tree: - (tree / "nested").rename(root / "retained") - link(tree / "nested", root / "retained") - return result + @contextmanager + def anchored_root(self): + with handle(root, directory=True) as fd: + wrapper = getattr(file_helper, "_WindowsDirectory", None) + yield root if wrapper is None else wrapper(root, fd) + + def names(fd): + with os.scandir(fd) as entries: + return [item.name for item in entries] + + def remove(fd): + directory, name = opened[fd] + selected = os.fstat(fd) + current = os.stat(name, dir_fd=directory, follow_symlinks=False) + if not os.path.samestat(selected, current): + raise OSError(errno.ESTALE, "Model cannot remove an object renamed after opening") + if stat.S_ISDIR(selected.st_mode): + os.rmdir(name, dir_fd=directory) + else: + os.unlink(name, dir_fd=directory) - monkeypatch.setattr(FileHelper, "_deletion_plan", exchange) - with pytest.raises(FileHelperError, match="changed during deletion"): - FileHelper(root, allow_symlinks=True).delete(tree, recursive=True) - assert (root / "retained/file").read_text() == "keep" - assert (tree / "nested").is_symlink() + original_delete = FileHelper._delete_plan + + def exchange(self, selected, **kwargs): + nonlocal changed + if selected.path == victim and not changed: + changed = True + parent.rename(retained) + link(parent, outside) + return original_delete(self, selected, **kwargs) + + monkeypatch.setattr(file_helper, "_windows_handle", handle) + monkeypatch.setattr(file_helper, "_windows_entry", entry, raising=False) + monkeypatch.setattr(file_helper, "_windows_mode", lambda fd: os.fstat(fd).st_mode, raising=False) + monkeypatch.setattr(file_helper, "_windows_directory_names", names, raising=False) + monkeypatch.setattr(file_helper, "_windows_delete", remove, raising=False) + monkeypatch.setattr(FileHelper, "_deletion_root", anchored_root) + monkeypatch.setattr(FileHelper, "_delete_plan", exchange) + with FileHelper(root, allow_symlinks=allow) as files: + try: + files.delete(parent if recursive else victim, recursive=recursive) + except OSError: + assert changed + assert changed + assert (outside / "victim").read_text() == "keep" + assert not (retained / "victim").exists() + assert opened == {} + + +@pytest.mark.parametrize("allow", [False, True]) +def test_delete_through_trusted_root_alias(root, link, allow): + alias = link(root.parent / "alias", root) + (root / "tree/nested").mkdir(parents=True) + (root / "tree/nested/file").write_text("selected") + FileHelper(alias, allow_symlinks=allow).delete(alias / "tree", recursive=True) + assert alias.is_symlink() + assert root.is_dir() + assert not (root / "tree").exists() @pytest.mark.parametrize("allow", [False, True]) @@ -1081,6 +2914,7 @@ def test_exclusive_create_rejects_existing_leaf_before_open( target.write_bytes(b"keep") alias = link(root / "alias", root / "alias" if cyclic else target) original_target = os.readlink(alias) + files = FileHelper(root, allow_symlinks=True) opened = [] real_open = os.open @@ -1091,7 +2925,8 @@ def recording_open(path, flags, mode): monkeypatch.setattr("specify_cli.file_helper.os.open", recording_open) content = b"bad" if operation == "create_bytes" else "bad" with pytest.raises(FileExistsError) as error: - getattr(FileHelper(root, allow_symlinks=True), operation)(alias, content) + getattr(files, operation)(alias, content) + files.close() assert error.value.errno == errno.EEXIST assert opened == [] assert alias.is_symlink() @@ -1508,7 +3343,8 @@ def try_exchange(path, flags, *args, **kwargs): monkeypatch.setattr(os, "open", try_exchange) content = b"new" if operation == "create_bytes" else "new" - getattr(FileHelper(root, allow_symlinks=allow), operation)("parent/nested/new", content) + with FileHelper(root, allow_symlinks=allow) as files: + getattr(files, operation)("parent/nested/new", content) assert attempted assert (root / "parent/nested/new").read_bytes() == b"new" parent.rename(retained) @@ -1866,18 +3702,39 @@ def test_directory_reparse_entries_rejected_before_traversal_portably( junction.mkdir() (junction / "keep").write_text("keep") lstat = Path.lstat + stat_entry = os.stat + junction_identity = lstat(junction) + + def as_reparse(result): + return SimpleNamespace( + st_mode=result.st_mode, + st_file_attributes=0x400, + st_reparse_tag=0xA0000003, + ) def directory_reparse(path, **kwargs): result = lstat(path, **kwargs) if path == junction: - return SimpleNamespace( - st_mode=result.st_mode, - st_file_attributes=0x400, - st_reparse_tag=0xA0000003, - ) + return as_reparse(result) + return result + + def directory_reparse_stat(path, *args, **kwargs): + result = stat_entry(path, *args, **kwargs) + if kwargs.get("dir_fd") is not None and Path(path).name == junction.name: + return as_reparse(result) return result monkeypatch.setattr(Path, "lstat", directory_reparse) + monkeypatch.setattr(os, "stat", directory_reparse_stat) + if os.name == "nt": + native_mode = file_helper._windows_mode + + def reparse_mode(handle): + if os.path.samestat(file_helper._windows_stat(handle), junction_identity): + raise OSError(errno.ELOOP, "Unsupported Windows reparse point") + return native_mode(handle) + + monkeypatch.setattr(file_helper, "_windows_mode", reparse_mode) files = FileHelper(root, allow_symlinks=allow) for operation in ( lambda: files.read_bytes(junction / "keep"), diff --git a/tests/test_file_helper_overwrite.py b/tests/test_file_helper_overwrite.py index 6ed2238560..2a785a32d5 100644 --- a/tests/test_file_helper_overwrite.py +++ b/tests/test_file_helper_overwrite.py @@ -370,6 +370,7 @@ def test_unsupported_descriptor_backend_fails_without_path_fallback( path = tmp_path / "file" if existing: path.write_text("old") + files = FileHelper(tmp_path) def unexpected_open(*args, **kwargs): raise AssertionError("An unsupported backend must not open a destination") @@ -382,8 +383,9 @@ def unexpected_open(*args, **kwargs): else: attributes[capability] = value monkeypatch.setattr(file_helper, "os", SimpleNamespace(**attributes)) - with pytest.raises(OSError, match="unsupported") as caught: - FileHelper(tmp_path).overwrite_text("file", "new") + with files: + with pytest.raises(OSError, match="unsupported") as caught: + files.overwrite_text("file", "new") assert caught.value.errno == errno.ENOTSUP if existing: assert path.read_text() == "old" @@ -470,6 +472,7 @@ def test_overwrite_closes_descriptors_on_failure(tmp_path, monkeypatch, failure) parent.mkdir() (parent / "file").write_text("old") opened = [] + opened_leaves = [] original = os.open fstat = os.fstat @@ -478,6 +481,8 @@ def observe(path, flags, *args, **kwargs): raise PermissionError("test failure") fd = original(path, flags, *args, **kwargs) opened.append(fd) + if Path(path).name == "file": + opened_leaves.append(fd) return fd def fail(*args, **kwargs): @@ -497,10 +502,13 @@ def __exit__(self, *args): replacements["fdopen"] = lambda *args, **kwargs: FailedStream() elif failure != "open": replacements[failure] = fail - monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), **replacements})) - with pytest.raises(OSError, match="test failure"): - FileHelper(tmp_path).overwrite_text("parent/file", "new") + with FileHelper(tmp_path) as files: + monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), **replacements})) + with pytest.raises(OSError, match="test failure"): + files.overwrite_text("parent/file", "new") assert opened + if failure != "open": + assert len(opened_leaves) == 1 for fd in opened: with pytest.raises(OSError): fstat(fd) @@ -514,6 +522,7 @@ def test_windows_overwrite_pins_resolved_hierarchy_until_stream_closes( parent.mkdir() leaf = parent / "file" leaf.write_text("old") + files = FileHelper(tmp_path) events = [] active = [] @@ -546,13 +555,14 @@ def open_leaf(path): monkeypatch.setattr(file_helper, "os", SimpleNamespace(**{**vars(os), "name": "nt"})) monkeypatch.setattr(file_helper, "_windows_handle", pin) monkeypatch.setattr(file_helper, "_windows_overwrite_fd", open_leaf) - if failed_parent: - with pytest.raises(PermissionError, match="pin failed"): - FileHelper(tmp_path).overwrite_text("parent/file", "new") - assert leaf.read_text() == "old" - else: - FileHelper(tmp_path).overwrite_text("parent/file", "new") - assert events[-3:] == [("close", leaf), ("unpin", parent), ("unpin", tmp_path)] + with files: + if failed_parent: + with pytest.raises(PermissionError, match="pin failed"): + files.overwrite_text("parent/file", "new") + assert leaf.read_text() == "old" + else: + files.overwrite_text("parent/file", "new") + assert events[-3:] == [("close", leaf), ("unpin", parent), ("unpin", tmp_path)] assert not active