Isolate Foundry session path handling

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3e5c81ad-75e8-4e92-a883-8bbd676c6c8c
Eduard van Valkenburg (eavanvalkenburg) · Jul 30, 2026 · Jul 24, 2026 · Jul 27, 2026 · Jul 27, 2026 · Jul 27, 2026
commit a22be8969ebb89baa74bc8d7f9e1480385e707e7
@@ -1719,15 +1719,6 @@ def validate_session_id(session_id: str) -> None:
         if len(session_id) > FileSessionStore.MAX_SESSION_ID_LENGTH:
             raise ValueError(f"session_id must be at most {FileSessionStore.MAX_SESSION_ID_LENGTH} characters")
 
-    def get_session_directory(self) -> Path:
-        """Return the directory used for the current file-store operation.
-
-        Subclasses may override this hook to scope operations to a child
-        directory. Filename encoding and containment checks remain owned by the
-        base implementation.
-        """
-        return self._storage_root
-
     @staticmethod
     def _quarantine_corrupt_snapshot(file_path: Path, serialized: bytes) -> Path | None:
         """Move an unchanged corrupt snapshot aside so a retry can recover."""
@@ -1748,17 +1739,18 @@ def _session_file_lock(cls, file_path: Path) -> threading.Lock:
 
     def _session_file_path(self, session_id: str) -> Path:
         """Resolve the contained snapshot path for ``session_id``."""
-        self.validate_session_id(session_id)
-        session_directory = self.get_session_directory().resolve()
-        if not session_directory.is_relative_to(self._storage_root):
-            raise ValueError(f"Session directory escaped storage directory: '{session_directory}'.")
-        session_directory.mkdir(parents=True, exist_ok=True)
-        file_stem = _session_file_stem(session_id, encoded_prefix=self._ENCODED_SESSION_PREFIX)
-        file_path = (session_directory / f"{file_stem}{self._file_extension}").resolve()
-        if not file_path.is_relative_to(session_directory):
+        candidate_path = self._storage_root / self._session_file_name(session_id)
+        file_path = candidate_path.resolve()
+        if file_path != candidate_path or not file_path.is_relative_to(self._storage_root):
             raise ValueError(f"Session path escaped storage directory: {session_id!r}")
         return file_path
 
+    def _session_file_name(self, session_id: str) -> str:
+        """Return the portable snapshot filename for ``session_id``."""
+        self.validate_session_id(session_id)
+        file_stem = _session_file_stem(session_id, encoded_prefix=self._ENCODED_SESSION_PREFIX)
+        return f"{file_stem}{self._file_extension}"
+
 
 class InMemoryHistoryProvider(HistoryProvider):
     """Built-in history provider that stores messages in session.state.

@@ -1223,6 +1223,19 @@ async def test_opaque_session_id_round_trips_through_safe_filename(
         assert session_file.parent == tmp_path
         assert session_file.name.startswith("~session-") is is_encoded
 
+    async def test_rejects_symlinked_session_file_within_storage_root(self, tmp_path: Path) -> None:
+        store = FileSessionStore(tmp_path)
+        target_file = tmp_path / "other-session.json"
+        session_file = tmp_path / "session-1.json"
+        await asyncio.to_thread(target_file.write_text, "{}", encoding="utf-8")
+        try:
+            await asyncio.to_thread(session_file.symlink_to, target_file)
+        except OSError as exc:
+            pytest.skip(f"Symlinks are not available: {exc}")
+
+        with pytest.raises(ValueError, match="escaped storage directory"):
+            await store.get("session-1")
+
     @pytest.mark.parametrize("session_id", ["NUL.txt", "COM¹", "LPT².log"])
     async def test_reserved_windows_filename_is_encoded(self, tmp_path: Path, session_id: str) -> None:
         provider = FileHistoryProvider(tmp_path)

@@ -35,7 +35,21 @@ def __init__(
         """Initialize a Foundry-scoped file store rooted at ``storage_path``."""
         super().__init__(storage_path, serialization_format=serialization_format)
 
-    def get_session_directory(self) -> Path:
-        """Return the active request user's validated session directory."""
+    def _session_file_path(self, session_id: str) -> Path:
+        """Resolve a snapshot path within the active Foundry user's directory."""
         directory_segment = request_user_directory_segment()
-        return self.storage_path / directory_segment if directory_segment else self.storage_path
+        candidate_directory = self._storage_root / directory_segment if directory_segment else self._storage_root
+        session_directory = candidate_directory.resolve()
+        if session_directory != candidate_directory or not session_directory.is_relative_to(self._storage_root):
+            raise ValueError(f"Session directory escaped storage directory: '{session_directory}'.")
+        session_directory.mkdir(parents=True, exist_ok=True)
+
+        candidate_path = session_directory / self._session_file_name(session_id)
+        file_path = candidate_path.resolve()
+        if (
+            file_path != candidate_path
+            or not file_path.is_relative_to(session_directory)
+            or not file_path.is_relative_to(self._storage_root)
+        ):
+            raise ValueError(f"Session path escaped storage directory: {session_id!r}")
+        return file_path
@@ -542,6 +542,23 @@ async def test_scoped_file_store_rejects_symlinked_session_leaf(self, tmp_path:
         with _request_context(user_id="user-A"), pytest.raises(ValueError, match="escaped storage directory"):
             await store.get("conversation-1")
 
+    async def test_scoped_file_store_rejects_symlinked_session_leaf_within_user_directory(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        store = FoundrySessionStore(tmp_path)
+        user_directory = tmp_path / "user-A"
+        user_directory.mkdir()
+        target_file = user_directory / "conversation-2.json"
+        target_file.write_text("outside", encoding="utf-8")
+        try:
+            (user_directory / "conversation-1.json").symlink_to(target_file)
+        except OSError as exc:
+            pytest.skip(f"Symlinks are not available: {exc}")
+
+        with _request_context(user_id="user-A"), pytest.raises(ValueError, match="escaped storage directory"):
+            await store.get("conversation-1")
+
     async def test_scoped_file_store_rejects_symlinked_isolation_directory(self, tmp_path: Path) -> None:
         store = FoundrySessionStore(tmp_path)
         user_directory = "user-A"
@@ -555,6 +572,21 @@ async def test_scoped_file_store_rejects_symlinked_isolation_directory(self, tmp
         with _request_context(user_id="user-A"), pytest.raises(ValueError, match="Session directory escaped"):
             await store.get("conversation-1")
 
+    async def test_scoped_file_store_rejects_symlinked_isolation_directory_within_storage_root(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        store = FoundrySessionStore(tmp_path)
+        other_user_directory = tmp_path / "user-B"
+        other_user_directory.mkdir()
+        try:
+            (tmp_path / "user-A").symlink_to(other_user_directory, target_is_directory=True)
+        except OSError as exc:
+            pytest.skip(f"Symlinks are not available: {exc}")
+
+        with _request_context(user_id="user-A"), pytest.raises(ValueError, match="Session directory escaped"):
+            await store.get("conversation-1")
+
     @pytest.mark.parametrize("user_id", ["../../escape", "/tmp/escape", "user/subdirectory", "user\\subdirectory"])
     async def test_file_store_rejects_unsafe_user_directory(self, tmp_path: Path, user_id: str) -> None:
         store = FoundrySessionStore(tmp_path)