Fix/tool result store read boundary (#12059)

Author: Aarkin7

haystack/hooks/tool_result_offloading/stores.py

@@ -30,6 +30,25 @@ def __init__(self, root: str | Path) -> None:
         """
         self.root = Path(root)
 
+    def _resolve_in_root(self, path_like: str | Path, *, subject: str) -> Path:
+        """
+        Resolve a path-like value and ensure it stays within the configured store root.
+
+        Relative values are interpreted relative to `self.root`; absolute values are used as-is.
+
+        :param path_like: Relative or absolute path-like value to resolve.
+        :param subject: Human-readable label used in the error message.
+        :returns: The resolved absolute path within the store root.
+        :raises ValueError: If the resolved path escapes the store root.
+        """
+        root = self.root.resolve()
+        path = Path(path_like)
+        candidate = path if path.is_absolute() else root / path
+        resolved = candidate.resolve()
+        if not resolved.is_relative_to(root):
+            raise ValueError(f"{subject} '{path_like}' resolves outside the store root '{root}'.")
+        return resolved
+
     def write(self, *, key: str, content: str) -> str:
         """
         Write `content` to `<root>/<key>`, creating parent directories, and return the file path.
@@ -42,10 +61,7 @@ def write(self, *, key: str, content: str) -> str:
         :returns: The absolute path the content was written to, as a string, for use with `read`.
         :raises ValueError: If `key` resolves to a location outside the store root.
         """
-        root = self.root.resolve()
-        path = (root / key).resolve()
-        if not path.is_relative_to(root):
-            raise ValueError(f"Result key '{key}' resolves outside the store root '{root}'.")
+        path = self._resolve_in_root(key, subject="Result key")
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(content, encoding="utf-8")
         return str(path)
@@ -54,10 +70,14 @@ def read(self, reference: str) -> str:
         """
         Read back the content previously written to `reference`.
 
-        :param reference: A path returned by `write`.
+        The resolved reference must stay within the store root: callers must treat it as an opaque
+        store-scoped reference, not as an arbitrary filesystem path.
+
+        :param reference: A store reference returned by `write`.
         :returns: The stored content.
+        :raises ValueError: If `reference` resolves to a location outside the store root.
         """
-        return Path(reference).read_text(encoding="utf-8")
+        return self._resolve_in_root(reference, subject="Result reference").read_text(encoding="utf-8")
 
     def to_dict(self) -> dict[str, Any]:
         """

releasenotes/notes/tool-result-store-read-boundary-2d1720bfa728e738.yaml

@@ -0,0 +1,7 @@
+---
+security:
+  - |
+    Harden ``FileSystemToolResultStore.read()`` so it only reads references that
+    resolve within the configured store root. This closes a boundary gap where
+    callers could previously pass an arbitrary filesystem path to ``read()``
+    instead of a store-scoped reference returned by ``write()``.

test/hooks/tool_result_offloading/test_stores.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import sys
 from pathlib import Path
 
 import pytest
@@ -42,6 +43,35 @@ def test_read_round_trips_written_content(self, tmp_path):
         reference = store.write(key="a.txt", content="round trip")
         assert store.read(reference) == "round trip"
 
+    def test_read_rejects_parent_traversal_reference(self, tmp_path):
+        store = FileSystemToolResultStore(root=tmp_path / "root")
+        outside = tmp_path / "outside.txt"
+        outside.write_text("secret", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="outside the store root"):
+            store.read("../outside.txt")
+
+    def test_read_rejects_absolute_reference_outside_root(self, tmp_path):
+        store = FileSystemToolResultStore(root=tmp_path / "root")
+        outside = tmp_path / "outside.txt"
+        outside.write_text("secret", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="outside the store root"):
+            store.read(str(outside))
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="symlinks require elevated privileges on Windows")
+    def test_read_rejects_symlink_reference_escaping_root(self, tmp_path):
+        root = tmp_path / "root"
+        root.mkdir()
+        outside = tmp_path / "outside.txt"
+        outside.write_text("secret", encoding="utf-8")
+        link = root / "link.txt"
+        link.symlink_to(outside)
+        store = FileSystemToolResultStore(root=root)
+
+        with pytest.raises(ValueError, match="outside the store root"):
+            store.read(str(link))
+
     def test_to_dict_from_dict_roundtrip(self, tmp_path):
         store = FileSystemToolResultStore(root=tmp_path)
         restored = FileSystemToolResultStore.from_dict(store.to_dict())