Guard against malformed or malicious include patterns

Author: nfcampos

libs/langchain_v1/langchain/agents/middleware/anthropic_tools.py

@@ -11,11 +11,16 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any, cast
 
-from langchain_core.messages import AIMessage, ToolMessage
+from langchain_core.messages import ToolMessage
 from langgraph.types import Command
 from typing_extensions import NotRequired, TypedDict
 
-from langchain.agents.middleware.types import AgentMiddleware, AgentState, ModelRequest
+from langchain.agents.middleware.types import (
+    AgentMiddleware,
+    AgentState,
+    ModelRequest,
+    ModelResponse,
+)
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
@@ -182,8 +187,8 @@ def __init__(
     def wrap_model_call(
         self,
         request: ModelRequest,
-        handler: Callable[[ModelRequest], AIMessage],
-    ) -> AIMessage:
+        handler: Callable[[ModelRequest], ModelResponse],
+    ) -> ModelResponse:
         """Inject tool and optional system prompt."""
         # Add tool
         tools = list(request.tools or [])
@@ -610,8 +615,8 @@ def __init__(
     def wrap_model_call(
         self,
         request: ModelRequest,
-        handler: Callable[[ModelRequest], AIMessage],
-    ) -> AIMessage:
+        handler: Callable[[ModelRequest], ModelResponse],
+    ) -> ModelResponse:
         """Inject tool and optional system prompt."""
         # Add tool
         tools = list(request.tools or [])

libs/langchain_v1/langchain/agents/middleware/file_search.py

@@ -21,6 +21,70 @@
 from langchain.agents.middleware.types import AgentMiddleware
 
 
+def _expand_include_patterns(pattern: str) -> list[str] | None:
+    """Expand brace patterns like ``*.{py,pyi}`` into a list of globs."""
+    if "}" in pattern and "{" not in pattern:
+        return None
+
+    expanded: list[str] = []
+
+    def _expand(current: str) -> None:
+        start = current.find("{")
+        if start == -1:
+            expanded.append(current)
+            return
+
+        end = current.find("}", start)
+        if end == -1:
+            raise ValueError
+
+        prefix = current[:start]
+        suffix = current[end + 1 :]
+        inner = current[start + 1 : end]
+        if not inner:
+            raise ValueError
+
+        for option in inner.split(","):
+            _expand(prefix + option + suffix)
+
+    try:
+        _expand(pattern)
+    except ValueError:
+        return None
+
+    return expanded
+
+
+def _is_valid_include_pattern(pattern: str) -> bool:
+    """Validate glob pattern used for include filters."""
+    if not pattern:
+        return False
+
+    if any(char in pattern for char in ("\x00", "\n", "\r")):
+        return False
+
+    expanded = _expand_include_patterns(pattern)
+    if expanded is None:
+        return False
+
+    try:
+        for candidate in expanded:
+            re.compile(fnmatch.translate(candidate))
+    except re.error:
+        return False
+
+    return True
+
+
+def _match_include_pattern(basename: str, pattern: str) -> bool:
+    """Return True if the basename matches the include pattern."""
+    expanded = _expand_include_patterns(pattern)
+    if not expanded:
+        return False
+
+    return any(fnmatch.fnmatch(basename, candidate) for candidate in expanded)
+
+
 class StateFileSearchMiddleware(AgentMiddleware):
     """Provides Glob and Grep search over state-based files.
 
@@ -159,6 +223,9 @@ def grep_search(  # noqa: D417
             except re.error as e:
                 return f"Invalid regex pattern: {e}"
 
+            if include and not _is_valid_include_pattern(include):
+                return "Invalid include pattern"
+
             # Search files
             files = cast("dict[str, Any]", state.get(self.state_key, {}))
             results: dict[str, list[tuple[int, str]]] = {}
@@ -170,7 +237,7 @@ def grep_search(  # noqa: D417
                 # Check include filter
                 if include:
                     basename = Path(file_path).name
-                    if not self._match_include(basename, include):
+                    if not _match_include_pattern(basename, include):
                         continue
 
                 # Search file content
@@ -190,23 +257,6 @@ def grep_search(  # noqa: D417
         self.grep_search = grep_search
         self.tools = [glob_search, grep_search]
 
-    def _match_include(self, basename: str, pattern: str) -> bool:
-        """Match filename against include pattern."""
-        # Handle brace expansion {a,b,c}
-        if "{" in pattern and "}" in pattern:
-            start = pattern.index("{")
-            end = pattern.index("}")
-            prefix = pattern[:start]
-            suffix = pattern[end + 1 :]
-            alternatives = pattern[start + 1 : end].split(",")
-
-            for alt in alternatives:
-                expanded = prefix + alt + suffix
-                if fnmatch.fnmatch(basename, expanded):
-                    return True
-            return False
-        return fnmatch.fnmatch(basename, pattern)
-
     def _format_grep_results(
         self,
         results: dict[str, list[tuple[int, str]]],
@@ -355,6 +405,9 @@ def grep_search(
             except re.error as e:
                 return f"Invalid regex pattern: {e}"
 
+            if include and not _is_valid_include_pattern(include):
+                return "Invalid include pattern"
+
             # Try ripgrep first if enabled
             results = None
             if self.use_ripgrep:
@@ -416,12 +469,14 @@ def _ripgrep_search(
             return {}
 
         # Build ripgrep command
-        cmd = ["rg", "--json", pattern, str(base_full)]
+        cmd = ["rg", "--json"]
 
         if include:
             # Convert glob pattern to ripgrep glob
             cmd.extend(["--glob", include])
 
+        cmd.extend(["--", pattern, str(base_full)])
+
         try:
             result = subprocess.run(  # noqa: S603
                 cmd,
@@ -475,7 +530,7 @@ def _python_search(
                 continue
 
             # Check include filter
-            if include and not self._match_include(file_path.name, include):
+            if include and not _match_include_pattern(file_path.name, include):
                 continue
 
             # Skip files that are too large
@@ -497,23 +552,6 @@ def _python_search(
 
         return results
 
-    def _match_include(self, basename: str, pattern: str) -> bool:
-        """Match filename against include pattern."""
-        # Handle brace expansion {a,b,c}
-        if "{" in pattern and "}" in pattern:
-            start = pattern.index("{")
-            end = pattern.index("}")
-            prefix = pattern[:start]
-            suffix = pattern[end + 1 :]
-            alternatives = pattern[start + 1 : end].split(",")
-
-            for alt in alternatives:
-                expanded = prefix + alt + suffix
-                if fnmatch.fnmatch(basename, expanded):
-                    return True
-            return False
-        return fnmatch.fnmatch(basename, pattern)
-
     def _format_grep_results(
         self,
         results: dict[str, list[tuple[int, str]]],

libs/langchain_v1/tests/unit_tests/agents/middleware/test_file_search.py

@@ -1,8 +1,14 @@
 """Unit tests for file search middleware."""
 
+from pathlib import Path
+from typing import Any
+
 import pytest
 from langchain.agents.middleware.anthropic_tools import AnthropicToolsState
-from langchain.agents.middleware.file_search import StateFileSearchMiddleware
+from langchain.agents.middleware.file_search import (
+    FilesystemFileSearchMiddleware,
+    StateFileSearchMiddleware,
+)
 from langchain_core.messages import ToolMessage
 
 
@@ -198,7 +204,70 @@ def test_grep_files_with_matches_mode(self) -> None:
         assert "/src/utils.py" in result
         assert "/README.md" not in result
         # Should only have file paths, not line content
-        assert "def foo():" not in result
+
+    def test_grep_invalid_include_pattern(self) -> None:
+        """Return error when include glob is invalid."""
+        middleware = StateFileSearchMiddleware()
+
+        state: AnthropicToolsState = {
+            "messages": [],
+            "text_editor_files": {
+                "/src/main.py": {
+                    "content": ["def foo():"],
+                    "created_at": "2025-01-01T00:00:00",
+                    "modified_at": "2025-01-01T00:00:00",
+                }
+            },
+        }
+
+        result = middleware.grep_search.func(pattern=r"def", include="*.{py", state=state)
+
+        assert result == "Invalid include pattern"
+
+
+class TestFilesystemGrepSearch:
+    """Tests for filesystem-backed grep search."""
+
+    def test_grep_invalid_include_pattern(self, tmp_path: Path) -> None:
+        """Return error when include glob cannot be parsed."""
+
+        (tmp_path / "example.py").write_text("print('hello')\n", encoding="utf-8")
+
+        middleware = FilesystemFileSearchMiddleware(root_path=str(tmp_path), use_ripgrep=False)
+
+        result = middleware.grep_search.func(pattern="print", include="*.{py")
+
+        assert result == "Invalid include pattern"
+
+    def test_ripgrep_command_uses_literal_pattern(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Ensure ripgrep receives pattern after ``--`` to avoid option parsing."""
+
+        (tmp_path / "example.py").write_text("print('hello')\n", encoding="utf-8")
+
+        middleware = FilesystemFileSearchMiddleware(root_path=str(tmp_path), use_ripgrep=True)
+
+        captured: dict[str, list[str]] = {}
+
+        class DummyResult:
+            stdout = ""
+
+        def fake_run(*args: Any, **kwargs: Any) -> DummyResult:
+            cmd = args[0]
+            captured["cmd"] = cmd
+            return DummyResult()
+
+        monkeypatch.setattr("langchain.agents.middleware.file_search.subprocess.run", fake_run)
+
+        middleware._ripgrep_search("--pattern", "/", None)
+
+        assert "cmd" in captured
+        cmd = captured["cmd"]
+        assert cmd[:2] == ["rg", "--json"]
+        assert "--" in cmd
+        separator_index = cmd.index("--")
+        assert cmd[separator_index + 1] == "--pattern"
 
     def test_grep_content_mode(self) -> None:
         """Test grep with content output mode."""