Python: Reject Windows junctions in FileSystemAgentFileStore

python/packages/core/agent_framework/_harness/_file_access.py

@@ -27,6 +27,7 @@
 import logging
 import os
 import re
+import stat
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Mapping, MutableMapping
 from pathlib import Path
@@ -81,6 +82,21 @@
 _ELOOP = errno.ELOOP
 
 
+def _is_link_or_reparse_point(path: Path) -> bool:
+    """Return whether ``path`` is a symbolic link, junction, or other reparse point."""
+    path_stat = path.lstat()
+    if stat.S_ISLNK(path_stat.st_mode):
+        return True
+
+    is_junction = getattr(path, "is_junction", None)
+    if callable(is_junction) and is_junction():
+        return True
+
+    reparse_attribute = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+    file_attributes = getattr(path_stat, "st_file_attributes", 0)
+    return bool(reparse_attribute and file_attributes & reparse_attribute)
+
+
 def _compile_search_regex(pattern: str) -> re.Pattern[str]:
     """Compile a case-insensitive search regex, enforcing the length cap.
 
@@ -856,10 +872,9 @@ def _throw_if_contains_symlink(self, candidate: Path) -> None:
         """Reject any segment between the root and ``candidate`` that is a symlink/reparse point.
 
         Walks each ancestor down from the root on the *unresolved* candidate so
-        ``Path.is_symlink`` observes the on-disk entries instead of their
-        canonical targets. Stops once a segment does not exist on disk so write
-        scenarios remain allowed. ``Path.is_symlink`` detects both POSIX
-        symlinks and Windows reparse points (junctions).
+        ``Path.lstat`` observes the on-disk entries instead of their canonical
+        targets. Stops once a segment does not exist on disk so write scenarios
+        remain allowed.
         """
         try:
             relative_parts = candidate.relative_to(self._root_path).parts
@@ -873,18 +888,19 @@ def _throw_if_contains_symlink(self, candidate: Path) -> None:
         for segment in relative_parts:
             current = current / segment
             try:
-                is_link = current.is_symlink()
+                is_link = _is_link_or_reparse_point(current)
+            except FileNotFoundError:
+                break
             except OSError as exc:
                 # Fail closed: if we cannot verify whether a segment is a
                 # symlink/reparse point we refuse the operation rather than
                 # silently allow access that may escape the root.
+                probed_path = current.relative_to(self._root_path).as_posix()
                 raise ValueError(
-                    f"Invalid path: unable to verify whether '{segment}' is a symbolic link or reparse point."
+                    f"Invalid path: unable to verify whether {probed_path!r} is a symbolic link or reparse point."
                 ) from exc
             if is_link:
                 raise ValueError("Invalid path: the resolved path contains a symbolic link or reparse point.")
-            if not current.exists():
-                break
 
     async def write(self, path: str, content: str, *, overwrite: bool = True) -> None:
         """Write ``content`` to the file at ``path``.
@@ -908,9 +924,9 @@ def _write_file_sync(full_path: Path, content: str, overwrite: bool) -> None:
             flags |= os.O_TRUNC
         else:
             flags |= os.O_EXCL
-        # ``O_NOFOLLOW`` is POSIX-only; on Windows ``Path.is_symlink`` /
-        # reparse-point detection in :meth:`_throw_if_contains_symlink` is the
-        # only line of defence for the leaf segment.
+        # ``O_NOFOLLOW`` is POSIX-only; on Windows the lstat/reparse-point
+        # detection in :meth:`_throw_if_contains_symlink` is the only line of
+        # defence for the leaf segment.
         nofollow = getattr(os, "O_NOFOLLOW", 0)
         flags |= nofollow
         try:
@@ -985,7 +1001,12 @@ def _list_sync(full_dir: Path) -> list[FileStoreEntry]:
         directories: list[FileStoreEntry] = []
         files: list[FileStoreEntry] = []
         for entry in full_dir.iterdir():
-            if entry.is_symlink():
+            try:
+                is_link = _is_link_or_reparse_point(entry)
+            except OSError:
+                # Fail closed when an entry cannot be inspected.
+                continue
+            if is_link:
                 continue
             if entry.is_dir():
                 directories.append(FileStoreEntry(entry.name, FileStoreEntry.DIRECTORY))
@@ -1039,7 +1060,12 @@ def _enumerate_search_files(full_dir: Path, recursive: bool) -> list[tuple[str,
         while directories:
             current = directories.pop()
             for entry in current.iterdir():
-                if entry.is_symlink():
+                try:
+                    is_link = _is_link_or_reparse_point(entry)
+                except OSError:
+                    # Fail closed when an entry cannot be inspected.
+                    continue
+                if is_link:
                     continue
                 if entry.is_dir():
                     if recursive:

python/packages/core/tests/core/test_harness_file_access.py

@@ -3,7 +3,11 @@
 from __future__ import annotations
 
 import json
+import os
 import re
+import stat
+import subprocess
+import sys
 import time
 from pathlib import Path
 
@@ -58,6 +62,29 @@ def _text(content: Content) -> str:
     return content.text
 
 
+def _create_junction_or_skip(*, link: Path, target: Path) -> None:
+    if sys.platform != "win32":
+        pytest.skip("Windows directory junctions are only available on Windows")
+
+    result = subprocess.run(
+        [os.environ.get("COMSPEC", "cmd"), "/c", "mklink", "/J", str(link), str(target)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        pytest.skip(f"Could not create Windows directory junction: {result.stderr or result.stdout}")
+
+    is_junction = getattr(link, "is_junction", None)
+    is_reparse_point = bool(
+        getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+        and getattr(link.lstat(), "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
+    )
+    if not (callable(is_junction) and is_junction()) and not is_reparse_point:
+        link.rmdir()
+        pytest.skip("Created junction was not reported as a reparse point")
+
+
 def test_normalize_relative_path_collapses_and_validates() -> None:
     """The path normalizer should accept relative forward/backslash paths and reject unsafe ones."""
     assert _normalize_relative_path("foo/bar.txt") == "foo/bar.txt"
@@ -420,6 +447,31 @@ async def test_filesystem_store_search_and_list_skip_symlinked_directories(tmp_p
     assert {result.file_name for result in results} == {"inside.md"}
 
 
+async def test_filesystem_store_search_and_list_skip_junctioned_directories(tmp_path: Path) -> None:
+    """Recursive search and listing must not follow Windows junctions outside the root."""
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.md").write_text("ERROR outside the root", encoding="utf-8")
+
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "inside.md").write_text("ERROR inside", encoding="utf-8")
+    junction = root / "linked"
+    _create_junction_or_skip(link=junction, target=outside)
+
+    try:
+        store = FileSystemAgentFileStore(root)
+
+        with pytest.raises(ValueError):
+            await store.read("linked/secret.md")
+        assert await _list_dirs(store) == []
+
+        results = await store.search("", "error", recursive=True)
+        assert {result.file_name for result in results} == {"inside.md"}
+    finally:
+        junction.rmdir()
+
+
 async def test_filesystem_store_search_skips_non_utf8_files(tmp_path: Path) -> None:
     """The filesystem store should silently skip non-UTF-8 files instead of aborting the search."""
     store = FileSystemAgentFileStore(tmp_path)
@@ -773,17 +825,22 @@ def slow() -> list[FileSearchResult]:
 async def test_filesystem_store_symlink_probe_fails_closed_on_oserror(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """If ``Path.is_symlink`` raises during the probe, the operation must be refused."""
+    """If ``Path.lstat`` raises during the probe, the operation must be refused."""
     store = FileSystemAgentFileStore(tmp_path)
-    await store.write("ok.txt", "content")
+    await store.write("same/same/ok.txt", "content")
+
+    original_lstat = Path.lstat
+    failing_path = store.root_path / "same" / "same"
 
-    def boom(self: Path) -> bool:
-        raise PermissionError("access denied")
+    def fail_for_target(self: Path) -> os.stat_result:
+        if self == failing_path:
+            raise PermissionError("access denied")
+        return original_lstat(self)
 
-    monkeypatch.setattr(Path, "is_symlink", boom)
+    monkeypatch.setattr(Path, "lstat", fail_for_target)
 
-    with pytest.raises(ValueError, match="symbolic link or reparse point"):
-        await store.read("ok.txt")
+    with pytest.raises(ValueError, match=r"'same/same'"):
+        await store.read("same/same/ok.txt")
 
 
 def test_file_access_harness_classes_are_marked_experimental() -> None: