fix(mcp): symlink-safe allow-list for snippet reads + test-lint cleanup

search_code/get_file_neighbors attach a source snippet read from the
indexed File.path. Those paths come from repo content, so a crafted repo
could store/symlink a path resolving outside the indexed worktree and turn
a snippet read into an arbitrary host-file read. _read_snippet now resolves
the real path (following symlinks) and only reads inside an allowed root
(ALLOWED_ANALYSIS_DIR and/or the /<project>/ repo root), preserving the
legacy read only when no root is derivable. project is threaded through
_node_summary, search_code and get_file_neighbors. Adds _is_within +
_snippet_path_allowed helpers and 6 pure-Python guard tests.

Also addresses bot review nits in test_search_cache.py: explain the
best-effort empty-except in the synth_graph teardown and rename the unused
branch unpack to _branch in the 7 tests.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: DvirDukhan

api/mcp/tools/structural.py

@@ -752,11 +752,60 @@ def _raw_name(n: Any) -> Optional[str]:
 """Per-line char cap so a single minified line cannot blow up the payload."""
 
 
+def _is_within(path: str, root: str) -> bool:
+    """True when ``path`` is ``root`` or lives inside it (both absolute)."""
+    try:
+        return os.path.commonpath([path, root]) == root
+    except ValueError:
+        # Different drives / mixed absolute-relative -> not contained.
+        return False
+
+
+def _snippet_path_allowed(abs_path: str, project: Optional[str]) -> bool:
+    """Symlink-safe guard for snippet file reads.
+
+    Snippet paths come from indexed ``File.path`` values, i.e. ultimately from
+    repo content. A malicious repo could store (or symlink) a path resolving
+    outside the indexed worktree, turning a snippet read into an arbitrary
+    host-file read. We resolve the real path (following symlinks) and require it
+    to live inside an allowed root: ``ALLOWED_ANALYSIS_DIR`` when set, and/or the
+    indexed repo root derived from the first ``/<project>/`` marker in the
+    stored path. When neither root is derivable (no allow-list and no project
+    marker in the path) we preserve the legacy read rather than block a
+    legitimate snippet.
+    """
+    try:
+        real = os.path.realpath(abs_path)
+    except OSError:
+        return False
+
+    roots: list[str] = []
+    allowed_env = os.getenv("ALLOWED_ANALYSIS_DIR")
+    if allowed_env:
+        try:
+            roots.append(os.path.realpath(os.path.expanduser(allowed_env)))
+        except OSError:
+            pass
+    if project:
+        marker = f"/{project}/"
+        idx = abs_path.find(marker)
+        if idx != -1:
+            try:
+                roots.append(os.path.realpath(abs_path[: idx + len(marker)]))
+            except OSError:
+                pass
+
+    if not roots:
+        return True
+    return any(_is_within(real, root) for root in roots)
+
+
 def _read_snippet(
     abs_path: Optional[str],
     src_start: Any,
     src_end: Any,
     max_lines: int,
+    project: Optional[str] = None,
 ) -> Optional[str]:
     """Read up to ``max_lines`` leading source lines for an entity from disk.
 
@@ -766,9 +815,15 @@ def _read_snippet(
     info is missing or unreadable. Each line is capped at
     ``_SNIPPET_LINE_CHARS``. Carrying a snippet in the result lets a single
     tool call replace a follow-up ``view`` of a large file.
+
+    ``project`` (the indexed repo identifier) enables a symlink-safe allow-list
+    check via :func:`_snippet_path_allowed` so a snippet read can't escape the
+    indexed worktree.
     """
     if not abs_path or max_lines <= 0:
         return None
+    if not _snippet_path_allowed(abs_path, project):
+        return None
     try:
         start = int(src_start)
     except (TypeError, ValueError):
@@ -852,6 +907,7 @@ def _node_summary(
             props.get("src_start"),
             props.get("src_end"),
             snippet_lines,
+            project=rel_to,
         )
         if snip:
             summary["snippet"] = snip
@@ -1206,6 +1262,7 @@ async def search_code(
             r["src_start"] if r["src_start"] is not None else 1,
             r["src_end"] if r["src_end"] is not None else r["src_start"],
             SEARCH_SNIPPET_LINES,
+            project=project,
         )
         if snip:
             rec["snippet"] = snip
@@ -1366,7 +1423,9 @@ async def get_file_neighbors(
             }
             rep = rep_of.get(ap)
             if rep:
-                snip = _read_snippet(ap, rep[0], rep[1], NEIGHBOR_SNIPPET_LINES)
+                snip = _read_snippet(
+                    ap, rep[0], rep[1], NEIGHBOR_SNIPPET_LINES, project=project
+                )
                 if snip:
                     entry["snippet"] = snip
             neighbors.append(entry)

tests/mcp/test_search_cache.py

@@ -90,6 +90,8 @@ def synth_graph():
         try:
             g.delete()
         except Exception:
+            # Best-effort teardown: never fail a test because cleanup couldn't
+            # reach FalkorDB or the throwaway graph was already gone.
             pass
 
 
@@ -112,7 +114,7 @@ async def test_warm_call_reuses_corpus(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     g = AsyncGraphQuery(name)
@@ -131,7 +133,7 @@ async def test_different_queries_share_one_corpus(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     g = AsyncGraphQuery(name)
@@ -149,7 +151,7 @@ async def test_signature_change_triggers_rebuild(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     g = AsyncGraphQuery(name)
@@ -172,7 +174,7 @@ async def test_reset_corpus_cache_forces_rebuild(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     g = AsyncGraphQuery(name)
@@ -192,7 +194,7 @@ async def test_reset_all_clears_every_graph(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     g = AsyncGraphQuery(name)
@@ -217,7 +219,7 @@ async def test_concurrent_first_calls_build_once(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
     calls = _spy_build(monkeypatch)
 
     async def one():
@@ -238,7 +240,7 @@ async def test_mutation_during_build_is_not_cached(synth_graph, monkeypatch):
     import api.mcp.tools.structural as S
     from api.graph import AsyncGraphQuery
 
-    name, project, branch = synth_graph
+    name, project, _branch = synth_graph
 
     g_probe = AsyncGraphQuery(name)
     orig = S._build_corpus

tests/mcp/test_snippet_guard.py

@@ -0,0 +1,77 @@
+"""Symlink-safe guard for snippet file reads (PR #701 security review).
+
+``search_code`` / ``get_file_neighbors`` attach a short source snippet read
+from the indexed ``File.path``. Those paths ultimately come from repo content,
+so a malicious repo could store (or symlink) a path resolving outside the
+indexed worktree and turn a snippet read into an arbitrary host-file read.
+``_read_snippet`` now resolves the real path and only reads inside an allowed
+root (the ``/<project>/`` repo root and/or ``ALLOWED_ANALYSIS_DIR``).
+
+These tests are pure-Python (real temp files + a symlink); no FalkorDB needed.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from api.mcp.tools.structural import _is_within, _read_snippet, _snippet_path_allowed
+
+
+def _make_repo(tmp_path: Path) -> tuple[Path, Path, Path]:
+    """Create ``<root>/proj/pkg/mod.py`` and an escaping symlink. Returns
+    ``(real_file, escaping_symlink, secret_outside)``."""
+    proj = tmp_path / "root" / "proj"
+    (proj / "pkg").mkdir(parents=True)
+    real_file = proj / "pkg" / "mod.py"
+    real_file.write_text("def alpha():\n    return 1\n\n\nx = 2\n")
+
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.txt"
+    secret.write_text("TOP SECRET LINE 1\nTOP SECRET LINE 2\n")
+
+    evil = proj / "evil.py"
+    evil.symlink_to(secret)
+    return real_file, evil, secret
+
+
+def test_reads_file_inside_repo_root(tmp_path: Path):
+    real_file, _evil, _secret = _make_repo(tmp_path)
+    snip = _read_snippet(str(real_file), 1, 3, 5, project="proj")
+    assert snip is not None
+    assert "def alpha" in snip
+
+
+def test_blocks_symlink_escaping_repo_root(tmp_path: Path):
+    _real_file, evil, _secret = _make_repo(tmp_path)
+    # The path lives under /proj/ but resolves (via symlink) outside the repo.
+    assert _snippet_path_allowed(str(evil), "proj") is False
+    assert _read_snippet(str(evil), 1, 2, 5, project="proj") is None
+
+
+def test_blocks_outside_path_when_allowed_dir_set(tmp_path: Path, monkeypatch):
+    _real_file, _evil, secret = _make_repo(tmp_path)
+    monkeypatch.setenv("ALLOWED_ANALYSIS_DIR", str(tmp_path / "root"))
+    # No /proj/ marker in this path, but ALLOWED_ANALYSIS_DIR excludes it.
+    assert _snippet_path_allowed(str(secret), None) is False
+    assert _read_snippet(str(secret), 1, 2, 5) is None
+
+
+def test_allows_when_no_constraint_derivable(tmp_path: Path):
+    # No project marker in the path and no ALLOWED_ANALYSIS_DIR -> legacy read.
+    f = tmp_path / "loose.py"
+    f.write_text("def beta():\n    return 0\n")
+    assert _snippet_path_allowed(str(f), None) is True
+    assert _read_snippet(str(f), 1, 2, 5) is not None
+
+
+def test_is_within_rejects_sibling_prefix(tmp_path: Path):
+    root = str(tmp_path / "proj")
+    # A sibling dir sharing a string prefix must NOT count as contained.
+    assert _is_within(str(tmp_path / "proj-evil" / "x.py"), root) is False
+    assert _is_within(str(tmp_path / "proj" / "x.py"), root) is True
+
+
+def test_missing_path_returns_none():
+    assert _read_snippet(None, 1, 2, 5, project="proj") is None
+    assert _read_snippet("", 1, 2, 5, project="proj") is None