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