From 8ad49f21b614022306436687d3ca65bdac0efa80 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Tue, 25 Aug 2026 16:06:16 -0400 Subject: [PATCH] feat(client): descriptor-pinned filesystem primitives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third of five slices. Adds `safe_fs.py`, the "write a file under a directory something else may be racing you for" problem solved once. Nothing here knows what a skill is; the materialization layer is its only caller, and it lands next. A path check is only as good as the last path resolution after it. Every `lstat` and containment check validates an inode, but a following `os.replace(tmp, dir / name)` re-resolves `dir` from its name — so anything holding write permission there can move the validated directory aside, leave a symlink in its place, and redirect the write or the unlink somewhere else. Narrowing that window is not a fix; the race is winnable at any width. So the checks hand off to a descriptor and nothing re-resolves a path afterwards. - `open_directory_nofollow` opens with `O_RDONLY | O_DIRECTORY | O_NOFOLLOW` and confirms `S_ISDIR` on the `fstat`, since not every platform defines `O_DIRECTORY`. `open_or_create_directory` adds `os.mkdir` plus an `lstat` on the `FileExistsError` path, because `Path.mkdir(exist_ok=True)` accepts a symlink-to-directory as "already there" and would reopen the hole the caller's check just closed. `pinned_directory` holds either for a block, so a caller states the platform split once and cannot forget the close. - `atomic_write` creates its temp file with `O_CREAT | O_EXCL | O_NOFOLLOW` at that descriptor, `fchmod`s the descriptor rather than a path, writes, fsyncs, renames, and fsyncs the directory so the rename survives a crash. Mode is set explicitly at 0644, never inherited from the umask and never executable. `os.replace` is the single rename call site and `os.rename` must not be substituted for it. - `unlink_file` probes and unlinks descriptor-relative. `unlink` never follows a trailing symlink but it does resolve the directory above it, so the same swap turns a removal into a delete of an arbitrary file. A symlink found where this SDK expects its own file raises `SymlinkRefused` rather than being tidied away — the state on disk is not what the caller believes, and that is the caller's to report. `SUPPORTS_DIR_FD` gates all of it, and the probe deliberately names `os.rename`/`os.stat` rather than the `os.replace`/`os.lstat` this module calls: `os.supports_dir_fd` is populated per underlying syscall, and CPython registers `renameat` under `rename` only and `fstatat` under `stat` only. Probing the names actually called reports "unsupported" on every POSIX platform and silently turns the defense off. Where the family is absent, every operation falls back to the identical full-path sequence. The tests here exercise the module directly, on its own terms. The TOCTOU races these primitives exist to close are proved through the materialization layer, which is what holds a descriptor across a sequence of operations. Testing: `uv run pytest` → 1260 passed, 11 skipped. `ruff check`, `ruff format --check`, and `mypy packages/*/src` all clean. Co-Authored-By: Claude Fable 5 --- packages/client/agents.md | 47 +++ .../src/launchdarkly_ai_server/safe_fs.py | 311 ++++++++++++++++++ packages/client/tests/test_safe_fs.py | 246 ++++++++++++++ 3 files changed, 604 insertions(+) create mode 100644 packages/client/src/launchdarkly_ai_server/safe_fs.py create mode 100644 packages/client/tests/test_safe_fs.py diff --git a/packages/client/agents.md b/packages/client/agents.md index b573356..152f577 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -30,6 +30,7 @@ No other `launchdarkly-ai-*` package may define or duplicate these. They import | `src/launchdarkly_ai_server/types_validation.py` | `parse_ai_config` — validates flag variation shape; `is_valid_skill_key` / `is_valid_skill_version` / `skill_key_rejection_reason` (the canonical key-grammar explanation every layer quotes) | | `src/launchdarkly_ai_server/skills.py` | Agent Skills, retrieval half — `skill_refs`, `get_skill`/`get_skills`/`all_skills`, `InMemorySkillStore`, and the store/telemetry injection points `_set_store` / `_set_emitter_for_testing` | | `src/launchdarkly_ai_server/skills_core.py` | Shared skills internals — the `SkillStore` seam, module state, the telemetry seam and its three recorders, integrity verification, and store resolution. Imported by both `skills.py` and the materialization layer; imports neither | +| `src/launchdarkly_ai_server/safe_fs.py` | Descriptor-pinned filesystem primitives — `atomic_write`, `unlink_file`, `pinned_directory`, `open_directory_nofollow`, `open_or_create_directory`, `SymlinkRefused`, and the `*at()` capability probe. Owns the descriptor-vs-path platform split; knows nothing about skills | | `src/launchdarkly_ai_server/utils.py` | `parse_template`, `parse_json_with_possible_fences`, `create_handler`, `parse_usage`, `make_track_data`, `to_ld_context` | | `src/launchdarkly_ai_server/registry.py` | `Registry`, `global_registry`, `compose`, `resolve_handlers`, `resolve_tools` | | `src/launchdarkly_ai_server/judges.py` | `run_judges`, `build_judge_tasks`, `run_judge` | @@ -334,6 +335,52 @@ The injection path is deliberately narrower than the state's location: `skills.p (`skills._set_store(store)` is the same setter `init_client` uses), and neither should reach into `skills_core` directly. +### Descriptor-pinned filesystem access + +A path check is only as good as the last path resolution after it. Every `lstat`, `realpath` +and containment check validates an *inode*, but a following +`os.replace(tmp, root / key / "SKILL.md")` re-resolves `/` from its *name* — so +anything holding write permission on a managed directory can move the validated directory +aside, leave a symlink in its place, and redirect the write (or an unlink) somewhere else. +Narrowing that window is not a fix; the race is winnable at any width. + +So the checks hand off to a descriptor and nothing re-resolves a path afterwards. The +primitives live in `safe_fs.py`, which knows nothing about skills: + +- `open_directory_nofollow` opens the directory with `O_RDONLY | O_DIRECTORY | O_NOFOLLOW` + and confirms `S_ISDIR` on the `fstat` (the explicit check is what covers platforms with no + `O_DIRECTORY`). `open_or_create_directory` wraps it with `os.mkdir` plus an `lstat` on the + `FileExistsError` path — `Path.mkdir(exist_ok=True)` accepts a symlink-to-directory as + "already there", which would reopen the hole the caller's check just closed. + `pinned_directory` holds either for the duration of a block, so a caller states the + platform split once as `if dir_fd is not None` and cannot forget the `os.close`. +- `atomic_write` creates the temp file with `O_CREAT | O_EXCL | O_NOFOLLOW` **at** that + descriptor (`_mkstemp_at`, since `tempfile` has no `dir_fd` form), `fchmod`s the + descriptor rather than `chmod`ing a path, writes, fsyncs, and renames with + `os.replace(tmp, name, src_dir_fd=fd, dst_dir_fd=fd)`, then fsyncs the directory so the + rename survives a crash. `atomic_write_in` is the same against a directory the caller does + not already hold open. `os.replace` is the single rename call site, reached by attribute + lookup so tests can intercept it, and `os.rename` must not be substituted for it — it is + also the only one with defined overwrite semantics on Windows. +- `unlink_file` probes and unlinks descriptor-relative too. `unlink` never follows a + *trailing* symlink, but it does resolve the directory above it, so the same swap turns a + removal into a delete of an attacker-chosen file. A symlink found where this SDK expects + its own file raises `SymlinkRefused` rather than being tidied away: the state on disk is + not what the caller believes, and that is the caller's to report. + +`safe_fs.SUPPORTS_DIR_FD` gates all of it, and the probe is not the obvious one. +`os.supports_dir_fd` is populated per underlying syscall, and CPython registers `renameat` +under `os.rename` only and `fstatat` under `os.stat` only — even though `os.replace` is the +same `renameat`-backed function and `os.lstat` is `fstatat` with `AT_SYMLINK_NOFOLLOW`. +Probing the names this module actually calls reports "unsupported" on every POSIX platform +and silently turns the defense off, so the probe names the advertised twins +(`{os.rename, os.open, os.unlink, os.stat}`) and a caller's symlink check is spelled +`os.stat(..., follow_symlinks=False)` rather than `os.lstat`. Where the family is absent +(Windows) `open_directory_nofollow` returns `None` after an `lstat` check instead of +attempting the descriptor open — `os.open` cannot open a directory there — and every caller +falls back to the identical full-path sequence, the per-component `lstat` floor. The +residual window on those platforms is documented rather than closed. + --- ## OTel Setup diff --git a/packages/client/src/launchdarkly_ai_server/safe_fs.py b/packages/client/src/launchdarkly_ai_server/safe_fs.py new file mode 100644 index 0000000..3605399 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/safe_fs.py @@ -0,0 +1,311 @@ +""" +Descriptor-pinned filesystem primitives. + +Split out because none of this knows what a skill is: it is the "write a file +under a directory an attacker may be racing you for" problem, solved once. +``skills_fs.py`` is the only caller today. + +The whole point is that a path check is only as good as the last path +resolution after it. Every operation here therefore runs relative to a +descriptor pinned to a directory the caller has already validated, rather than +re-resolving a name — which is what closes the swap window rather than merely +narrowing it. Where the platform has no ``*at()`` syscall family (Windows) the +identical sequence runs against full paths, the per-component ``lstat`` floor. +""" + +from __future__ import annotations + +import errno +import os +import secrets +import stat +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path + +_FILE_MODE = 0o644 +"""Mode set explicitly on every written file — never inherited from the umask, +and never executable.""" + +SUPPORTS_DIR_FD = os.supports_dir_fd.issuperset( + # renameat, openat, unlinkat, fstatat — the four this module needs. + {os.rename, os.open, os.unlink, os.stat} +) +""" +Whether the ``*at()`` syscall family is available, so every operation under the +managed root can be performed relative to a descriptor pinned to a directory +this module has already verified rather than re-resolved from its path. + +That is what closes the swap window rather than merely narrowing it: +a descriptor refers to the inode that was checked, so replacing ``/`` +with a symlink after the check cannot redirect a write or an unlink out of the +root. POSIX has these calls; Windows does not, and there the per-component +``lstat`` floor the spec permits applies instead. + +The probe deliberately names ``os.rename`` and ``os.stat`` rather than the +``os.replace`` and ``os.lstat`` this module actually calls. ``os.supports_dir_fd`` +is populated per underlying syscall, and CPython registers ``renameat`` under +``rename`` only and ``fstatat`` under ``stat`` only — even though ``os.replace`` +is the same ``renameat``-backed function and ``os.lstat`` is ``fstatat`` with +``AT_SYMLINK_NOFOLLOW``, and both accept the descriptor keywords wherever their +advertised twin does (verified on CPython 3.12 and 3.13, macOS). Probing the +names this module calls would report "unsupported" on every POSIX platform and +silently disable the defense. +""" + + +def open_directory_nofollow(directory: Path) -> int | None: + """ + Opens *directory* without following a final symlink, and pins it. + + Everything the caller does afterwards goes through the returned descriptor + instead of the path, which is what turns the "narrow window" into no + window at all: the descriptor names the inode that was checked, so swapping + the path for a symlink between the check and the write cannot redirect the + write out of the managed root. + + On a platform without the ``*at()`` family (Windows) this returns ``None`` + after verifying via ``lstat`` that the path is a real, non-symlink + directory — the per-component floor. It must not attempt the descriptor + open there: ``os.open`` goes through the CRT on Windows, which cannot open + a directory at all, so the descriptor path would fail every operation + rather than fall back. + + Raises ``ValueError`` when the path will not open (or inspect) as a real + directory — the caller reports that as a refusal rather than letting it + escape. + """ + if not SUPPORTS_DIR_FD: + try: + mode = os.lstat(directory).st_mode + except OSError as exc: + raise ValueError(f"the directory could not be inspected: {exc}") from exc + if stat.S_ISLNK(mode): + raise ValueError("the directory is a symlink") + if not stat.S_ISDIR(mode): + raise ValueError("the path is not a directory") + return None + + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + fd = os.open(directory, flags) + except OSError as exc: + raise ValueError( + f"the directory could not be opened without following links: {exc}" + ) from exc + try: + # O_DIRECTORY already guarantees this wherever the platform defines it; + # the explicit check is what covers the platforms that do not. + if not stat.S_ISDIR(os.fstat(fd).st_mode): + raise ValueError("the path is not a directory") + except BaseException: + os.close(fd) + raise + return fd + + +def open_or_create_directory(directory: Path) -> int | None: + """ + Creates *directory* if absent and returns a descriptor pinned to it. + + ``Path.mkdir(exist_ok=True)`` treats an existing symlink-to-directory as + "already there", which would re-open the very hole the caller's ``lstat`` + check just closed. ``os.mkdir`` plus an ``lstat`` on the ``FileExistsError`` + path does not: a link reports as a link, and is refused. + """ + try: + os.mkdir(directory, 0o755) + except FileExistsError: + mode = os.lstat(directory).st_mode + if stat.S_ISLNK(mode): + raise ValueError("the directory is a symlink") from None + if not stat.S_ISDIR(mode): + raise ValueError("the path is not a directory") from None + return open_directory_nofollow(directory) + + +@contextmanager +def pinned_directory(directory: Path, *, create: bool = False) -> Iterator[int | None]: + """ + Holds *directory* pinned for the duration of the block, then releases it. + + Yields what the two openers above return — a descriptor, or ``None`` on the + ``lstat`` floor — so the caller states the platform split once, as + ``if dir_fd is not None``, and cannot forget the ``os.close``. Raises + ``ValueError`` for a directory that will not pin, exactly as they do. + """ + dir_fd = ( + open_or_create_directory(directory) + if create + else open_directory_nofollow(directory) + ) + try: + yield dir_fd + finally: + if dir_fd is not None: + os.close(dir_fd) + + +class SymlinkRefused(OSError): + """ + Raised instead of removing a symlink found where a real file was expected. + + An ``OSError`` subclass so a caller that only cares that the removal failed + keeps its single ``except``; a distinct type so one that must report *this* + refusal specifically does not have to match on a message. + """ + + +def unlink_file(directory: Path, name: str, *, dir_fd: int | None) -> None: + """ + Removes ``/``, refusing to follow a symlink at *name*. + + The mirror of ``atomic_write``, and descriptor-relative for the same reason: + ``unlink`` never follows a *trailing* symlink, but it does resolve the + directory above it, so a ```` swapped for a symlink after the + caller's checks would otherwise turn this into a delete of an + attacker-chosen file. Given a *dir_fd* the probe and the unlink both run + against it; without one the identical sequence runs against full paths. + + Raises ``SymlinkRefused`` when *name* is a symlink. Note that this refuses + rather than removes: ``unlink`` would happily delete the link itself, but a + link where this SDK expects its own file means the state on disk is not what + the manifest describes, and that is the caller's to report rather than to + tidy away. + """ + if dir_fd is None: + # No ``*at()`` family: the trailing-symlink check and the unlink are both + # path-based, the per-component floor. + target = directory / name + if target.is_symlink(): + raise SymlinkRefused(f"{name} is a symlink") + target.unlink() + return + + # os.stat(follow_symlinks=False), not os.lstat: identical result, and it is + # the spelling os.supports_dir_fd actually advertises. + probe = os.stat(name, dir_fd=dir_fd, follow_symlinks=False) + if stat.S_ISLNK(probe.st_mode): + raise SymlinkRefused(f"{name} is a symlink") + os.unlink(name, dir_fd=dir_fd) + + +def _mkstemp_at(dir_fd: int, prefix: str) -> tuple[int, str]: + """ + ``tempfile.mkstemp`` for a directory descriptor. + + ``tempfile`` has no ``dir_fd`` form, so this reproduces the part that + matters: ``O_CREAT | O_EXCL`` against an unpredictable name, retried on + collision, so an existing temp path is never reused and a planted one is + never written through. + """ + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + for _ in range(tempfile.TMP_MAX): + name = f"{prefix}{secrets.token_hex(8)}.tmp" + try: + return os.open(name, flags, 0o600, dir_fd=dir_fd), name + except FileExistsError: + continue + raise OSError(errno.EEXIST, "no usable temporary file name was found") + + +def atomic_write( + directory: Path, name: str, data: bytes, *, dir_fd: int | None = None +) -> None: + """ + Writes *data* to ``/`` so no partial file is ever + observable. + + The temp file is created exclusively in the target's *own* directory — one + anywhere else would make the rename cross-device, and therefore not atomic — + written, fsynced, renamed over the target, and the directory fsynced so the + rename itself survives a crash. Mode is set explicitly rather than left to + the process umask, and the execute bit is never set. + + Given a *dir_fd* on a platform with the ``*at()`` family, every one of those + steps runs relative to that descriptor and both names are bare filenames. + Without one (Windows) the identical sequence runs against full paths, which + is the per-component ``lstat`` floor. + + ``os.replace`` is the one and only rename call site, reached by attribute + lookup on the ``os`` module so tests can intercept it; ``os.rename`` must + not be substituted for it (it is also the only one with defined overwrite + semantics on Windows). + """ + at_fd = dir_fd if dir_fd is not None and SUPPORTS_DIR_FD else None + prefix = f".{name}." + target: str | Path + + if at_fd is not None: + fd, temp = _mkstemp_at(at_fd, prefix) + target = name + else: + # mkstemp opens with O_CREAT|O_EXCL, so an existing temp path is never + # reused. + fd, temp = tempfile.mkstemp(dir=directory, prefix=prefix, suffix=".tmp") + target = directory / name + + try: + try: + # fchmod, not chmod: operating on the descriptor cannot be redirected + # by anything that swaps the temp path underneath us, and it makes the + # mode independent of the process umask (both creation paths open 0600). + os.fchmod(fd, _FILE_MODE) + view = memoryview(data) + while view: + view = view[os.write(fd, view) :] + os.fsync(fd) + finally: + os.close(fd) + if at_fd is not None: + os.replace(temp, target, src_dir_fd=at_fd, dst_dir_fd=at_fd) + else: + os.replace(temp, target) + except BaseException: + try: + if at_fd is not None: + os.unlink(temp, dir_fd=at_fd) + else: + os.unlink(temp) + except OSError: + pass + raise + + if at_fd is not None: + _fsync_directory_fd(at_fd) + else: + _fsync_directory(directory) + + +def atomic_write_in(directory: Path, name: str, data: bytes) -> None: + """ + ``atomic_write`` against a directory this module does not already hold open. + + Used for the skills manifest, whose directory is the managed root. The + descriptor is taken with ``O_NOFOLLOW``, so a root swapped for a symlink after + ``_resolve_root`` validated it fails the write instead of redirecting it — + the caller turns that into a run-level ``error`` action. + """ + with pinned_directory(directory) as dir_fd: + atomic_write(directory, name, data, dir_fd=dir_fd) + + +def _fsync_directory_fd(fd: int) -> None: + """Best effort — not every platform allows fsync on a directory descriptor.""" + try: + os.fsync(fd) + except OSError: + pass + + +def _fsync_directory(directory: Path) -> None: + """Best effort — not every platform lets a directory be opened for fsync.""" + try: + fd = os.open(directory, os.O_RDONLY) + except OSError: + return + try: + _fsync_directory_fd(fd) + finally: + os.close(fd) diff --git a/packages/client/tests/test_safe_fs.py b/packages/client/tests/test_safe_fs.py new file mode 100644 index 0000000..b148ff9 --- /dev/null +++ b/packages/client/tests/test_safe_fs.py @@ -0,0 +1,246 @@ +""" +Tests for the descriptor-pinned filesystem primitives. + +These exercise ``safe_fs`` directly, on its own terms — the module knows nothing +about skills, and its guarantees are worth asserting without a caller in the way. +The TOCTOU races these primitives exist to close are proved through the +materialization layer, which is what actually holds a descriptor across a +sequence of operations. +""" + +from __future__ import annotations + +import os +import stat +from pathlib import Path + +import pytest + +import launchdarkly_ai_server.safe_fs as safe_fs_module +from launchdarkly_ai_server.safe_fs import ( + SymlinkRefused, + atomic_write, + atomic_write_in, + open_directory_nofollow, + open_or_create_directory, + pinned_directory, + unlink_file, +) + + +class TestOpenDirectory: + """Pinning a directory, and refusing anything that is not one.""" + + def test_opens_a_real_directory(self, tmp_path: Path) -> None: + with pinned_directory(tmp_path) as dir_fd: + if dir_fd is None: + pytest.skip("no *at() family on this platform") + assert stat.S_ISDIR(os.fstat(dir_fd).st_mode) + + def test_refuses_a_symlink_to_a_directory(self, tmp_path: Path) -> None: + real = tmp_path / "real" + real.mkdir() + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + with pytest.raises(ValueError): + open_directory_nofollow(link) + + def test_refuses_a_regular_file(self, tmp_path: Path) -> None: + target = tmp_path / "file" + target.write_text("not a directory") + with pytest.raises(ValueError): + open_directory_nofollow(target) + + def test_refuses_an_absent_path(self, tmp_path: Path) -> None: + with pytest.raises(ValueError): + open_directory_nofollow(tmp_path / "nope") + + def test_create_makes_the_directory(self, tmp_path: Path) -> None: + target = tmp_path / "new" + fd = open_or_create_directory(target) + try: + assert target.is_dir() + finally: + if fd is not None: + os.close(fd) + + def test_create_refuses_an_existing_symlink(self, tmp_path: Path) -> None: + """``Path.mkdir(exist_ok=True)`` would accept this and reopen the hole. + + A symlink-to-directory already present reads as "already there" to + ``exist_ok``, so the caller's containment check would be bypassed by + something that was never checked. ``os.mkdir`` plus an ``lstat`` on the + ``FileExistsError`` path refuses it. + """ + real = tmp_path / "real" + real.mkdir() + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + with pytest.raises(ValueError): + open_or_create_directory(link) + + def test_pinned_directory_closes_the_descriptor(self, tmp_path: Path) -> None: + with pinned_directory(tmp_path) as dir_fd: + if dir_fd is None: + pytest.skip("no *at() family on this platform") + held = dir_fd + with pytest.raises(OSError): + os.fstat(held) + + +class TestAtomicWrite: + """Explicit mode, no observable partial file, and one rename call site.""" + + def test_writes_the_bytes(self, tmp_path: Path) -> None: + with pinned_directory(tmp_path) as dir_fd: + atomic_write(tmp_path, "f.txt", b"hello", dir_fd=dir_fd) + assert (tmp_path / "f.txt").read_bytes() == b"hello" + + def test_mode_is_0644_and_never_executable(self, tmp_path: Path) -> None: + """Set explicitly on the descriptor, so the process umask cannot widen or + narrow it and the execute bit is never inherited.""" + previous = os.umask(0o077) + try: + with pinned_directory(tmp_path) as dir_fd: + atomic_write(tmp_path, "f.txt", b"x", dir_fd=dir_fd) + finally: + os.umask(previous) + mode = (tmp_path / "f.txt").stat().st_mode + assert stat.S_IMODE(mode) == 0o644 + assert not mode & stat.S_IXUSR + + def test_overwrites_an_existing_file(self, tmp_path: Path) -> None: + (tmp_path / "f.txt").write_bytes(b"old") + with pinned_directory(tmp_path) as dir_fd: + atomic_write(tmp_path, "f.txt", b"new", dir_fd=dir_fd) + assert (tmp_path / "f.txt").read_bytes() == b"new" + + def test_leaves_no_temp_file_behind(self, tmp_path: Path) -> None: + with pinned_directory(tmp_path) as dir_fd: + atomic_write(tmp_path, "f.txt", b"x", dir_fd=dir_fd) + assert [p.name for p in tmp_path.iterdir()] == ["f.txt"] + + def test_a_failed_rename_removes_the_temp_file( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A crash between write and rename must not leave a partial file, and + must not leave the temp file either.""" + + def _boom(*args: object, **kwargs: object) -> None: + raise OSError("injected rename failure") + + monkeypatch.setattr(os, "replace", _boom) + with pinned_directory(tmp_path) as dir_fd: + with pytest.raises(OSError, match="injected"): + atomic_write(tmp_path, "f.txt", b"x", dir_fd=dir_fd) + assert list(tmp_path.iterdir()) == [] + + def test_rename_goes_through_os_replace( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """``os.replace`` is the single rename call site. + + ``os.rename`` must not be substituted for it: it is the only one with + defined overwrite semantics on Windows, and it is the seam the + materialization tests intercept to prove atomicity. + """ + calls: list[object] = [] + real = os.replace + + def _spy(src: object, dst: object, **kwargs: object) -> None: + calls.append(dst) + real(src, dst, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(os, "replace", _spy) + with pinned_directory(tmp_path) as dir_fd: + atomic_write(tmp_path, "f.txt", b"x", dir_fd=dir_fd) + assert len(calls) == 1 + assert os.path.basename(str(calls[0])) == "f.txt" + + def test_write_without_a_descriptor_uses_the_path_fallback( + self, tmp_path: Path + ) -> None: + """The no-``*at()`` shape must produce an identical result. + + Windows takes this path for every write, so it is not a degenerate case — + the file, its mode, and the absence of a temp file all have to match. + """ + atomic_write(tmp_path, "f.txt", b"fallback", dir_fd=None) + assert (tmp_path / "f.txt").read_bytes() == b"fallback" + assert stat.S_IMODE((tmp_path / "f.txt").stat().st_mode) == 0o644 + assert [p.name for p in tmp_path.iterdir()] == ["f.txt"] + + def test_write_in_pins_the_directory_itself(self, tmp_path: Path) -> None: + atomic_write_in(tmp_path, "f.txt", b"x") + assert (tmp_path / "f.txt").read_bytes() == b"x" + + def test_write_in_refuses_a_symlinked_directory(self, tmp_path: Path) -> None: + real = tmp_path / "real" + real.mkdir() + link = tmp_path / "link" + link.symlink_to(real, target_is_directory=True) + with pytest.raises(ValueError): + atomic_write_in(link, "f.txt", b"x") + + +class TestUnlinkFile: + """Removing only a real file, and refusing a link found in its place.""" + + def test_removes_a_regular_file(self, tmp_path: Path) -> None: + (tmp_path / "f.txt").write_text("x") + with pinned_directory(tmp_path) as dir_fd: + unlink_file(tmp_path, "f.txt", dir_fd=dir_fd) + assert not (tmp_path / "f.txt").exists() + + def test_refuses_a_symlink_rather_than_removing_it(self, tmp_path: Path) -> None: + """It refuses rather than tidies. + + ``unlink`` would happily delete the link itself, but a link where this SDK + expects its own file means the state on disk is not what the manifest + describes — the caller's to report, not this module's to clean up. + """ + outside = tmp_path / "outside.txt" + outside.write_text("do not touch") + link = tmp_path / "f.txt" + link.symlink_to(outside) + + with pinned_directory(tmp_path) as dir_fd: + with pytest.raises(SymlinkRefused): + unlink_file(tmp_path, "f.txt", dir_fd=dir_fd) + + assert link.is_symlink() + assert outside.read_text() == "do not touch" + + def test_refuses_a_symlink_on_the_path_fallback(self, tmp_path: Path) -> None: + outside = tmp_path / "outside.txt" + outside.write_text("do not touch") + (tmp_path / "f.txt").symlink_to(outside) + with pytest.raises(SymlinkRefused): + unlink_file(tmp_path, "f.txt", dir_fd=None) + assert outside.exists() + + def test_symlink_refused_is_an_oserror(self) -> None: + """A caller that only cares the removal failed keeps its single + ``except OSError``; one that must report this refusal specifically does + not have to match on a message.""" + assert issubclass(SymlinkRefused, OSError) + + +class TestDirFdProbe: + """The capability probe names the advertised twins, not the calls made.""" + + def test_probe_names_the_syscalls_python_advertises(self) -> None: + """``os.supports_dir_fd`` is populated per underlying syscall, and CPython + registers ``renameat`` under ``os.rename`` and ``fstatat`` under + ``os.stat``. Probing ``os.replace`` and ``os.lstat`` — the names this + module actually calls — reports "unsupported" on every POSIX platform and + would silently disable the defense. + """ + expected = os.supports_dir_fd.issuperset( + {os.rename, os.open, os.unlink, os.stat} + ) + assert safe_fs_module.SUPPORTS_DIR_FD is expected + + @pytest.mark.skipif(os.name == "nt", reason="POSIX advertises the *at() family") + def test_posix_has_the_family(self) -> None: + assert safe_fs_module.SUPPORTS_DIR_FD is True