Contain symlinked extraction inputs

graphify/detect.py

@@ -991,14 +991,11 @@ def _could_contain_included_path(path: Path, root: Path, patterns: list[tuple[Pa
 
 
 def _auto_follow_symlinks(root: Path) -> bool:
-    """Auto-detect: ``True`` if ``root`` has any direct symlinked child.
+    """Return whether ``root`` has any direct symlinked child.
 
-    Allows "fake working dir" patterns (e.g. a folder full of symlinks pointing
-    at scattered source dirs across the user's machine) to work transparently
-    without the caller having to know to pass ``follow_symlinks=True``.
-
-    Override is always possible by passing an explicit ``follow_symlinks=True``
-    or ``follow_symlinks=False`` to :func:`detect` / :func:`detect_incremental`.
+    Kept for callers that import the private helper, but detection no longer
+    enables symlink following automatically. Following symlinks is now an
+    explicit opt-in, and out-of-root symlink targets are never indexed.
     """
     try:
         for p in root.iterdir():
@@ -1009,10 +1006,19 @@ def _auto_follow_symlinks(root: Path) -> bool:
     return False
 
 
+def _resolves_under_root(path: Path, root: Path) -> bool:
+    """True when ``path`` resolves to a target inside ``root``."""
+    try:
+        path.resolve().relative_to(root.resolve())
+    except (OSError, RuntimeError, ValueError):
+        return False
+    return True
+
+
 def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace: bool | None = None, extra_excludes: list[str] | None = None) -> dict:
     root = root.resolve()
     if follow_symlinks is None:
-        follow_symlinks = _auto_follow_symlinks(root)
+        follow_symlinks = False
     google_workspace = google_workspace_enabled() if google_workspace is None else google_workspace
     files: dict[FileType, list[str]] = {
         FileType.CODE: [],
@@ -1072,6 +1078,15 @@ def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace:
                     if not _is_noise_dir(d, dp)
                     and not _is_ignored(dp / d, root, ignore_patterns, _cache=ignore_cache)
                 ]
+                if follow_symlinks:
+                    safe_dirs: list[str] = []
+                    for d in dirnames:
+                        child = dp / d
+                        if child.is_symlink() and not _resolves_under_root(child, root):
+                            skipped_sensitive.append(str(child) + " [symlink target outside scan root]")
+                            continue
+                        safe_dirs.append(d)
+                    dirnames[:] = safe_dirs
             for fname in filenames:
                 if fname in _SKIP_FILES:
                     continue
@@ -1093,6 +1108,9 @@ def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace:
                 continue
         if not in_memory and _is_ignored(p, root, ignore_patterns, _cache=ignore_cache):
             continue
+        if not _resolves_under_root(p, root):
+            skipped_sensitive.append(str(p) + " [symlink target outside scan root]")
+            continue
         if _is_sensitive(p):
             skipped_sensitive.append(str(p))
             continue
@@ -1363,11 +1381,10 @@ def detect_incremental(
     Backwards compatible with legacy manifests storing plain float mtime values
     or {mtime, hash} dicts (treated as ast_hash only; semantic_hash = miss).
 
-    The ``follow_symlinks`` flag is forwarded to :func:`detect` so corpora that
-    rely on symlinked sub-trees (e.g. a ``state_of_truth/`` symlink pointing to a
-    directory outside the scan root) are scanned consistently between full and
-    incremental runs. ``None`` (default) means auto-detect: ``True`` when ``root``
-    contains at least one direct symlinked child, ``False`` otherwise.
+    The ``follow_symlinks`` flag is forwarded to :func:`detect` so in-root
+    symlinked sub-trees are scanned consistently between full and incremental
+    runs. ``None`` (default) does not follow symlinked directories; callers must
+    opt in explicitly, and resolved targets outside the scan root are skipped.
     """
     full = detect(root, follow_symlinks=follow_symlinks, google_workspace=google_workspace, extra_excludes=extra_excludes)
     # Pass ``root`` so a manifest written with relative keys (post-#777) is

graphify/extract.py

@@ -15906,8 +15906,10 @@ def extract(
 
 
 def collect_files(target: Path, *, follow_symlinks: bool = False, root: Path | None = None) -> list[Path]:
+    containment_root = root if root is not None else target
+    from graphify.detect import _resolves_under_root
     if target.is_file():
-        return [target]
+        return [target] if _resolves_under_root(target, containment_root) else []
     _EXTENSIONS = set(_DISPATCH.keys())
     from graphify.detect import _is_ignored, _is_noise_dir, _load_graphifyignore
     ignore_root = root if root is not None else target
@@ -15938,7 +15940,7 @@ def collect_files(target: Path, *, follow_symlinks: bool = False, root: Path | N
             ]
             for fname in filenames:
                 p = dp / fname
-                if p.suffix in _EXTENSIONS and not _ignored(p):
+                if p.suffix in _EXTENSIONS and not _ignored(p) and _resolves_under_root(p, containment_root):
                     results.append(p)
         return sorted(results)
     # Walk with symlink following + cycle detection
@@ -15951,10 +15953,14 @@ def collect_files(target: Path, *, follow_symlinks: bool = False, root: Path | N
                 dirnames.clear()
                 continue
         dp = Path(dirpath)
-        dirnames[:] = [d for d in dirnames if not _is_noise_dir(d)]
+        dirnames[:] = [
+            d for d in dirnames
+            if not _is_noise_dir(d)
+            and (not (dp / d).is_symlink() or _resolves_under_root(dp / d, containment_root))
+        ]
         for fname in filenames:
             p = dp / fname
-            if p.suffix in _EXTENSIONS and not _ignored(p):
+            if p.suffix in _EXTENSIONS and not _ignored(p) and _resolves_under_root(p, containment_root):
                 results.append(p)
     return sorted(results)
 

graphify/llm.py

@@ -449,6 +449,17 @@ def _file_to_text(path: Path) -> str:
     return path.read_text(encoding="utf-8", errors="replace")
 
 
+def _resolve_under_root(path: Path, root: Path) -> Path | None:
+    """Return the resolved path only when it stays inside ``root``."""
+    try:
+        resolved_root = root.resolve()
+        resolved_path = path.resolve()
+        resolved_path.relative_to(resolved_root)
+    except (OSError, RuntimeError, ValueError):
+        return None
+    return resolved_path
+
+
 # Known prompt-injection / chat-template sentinels that a hostile source file
 # might embed to try to break out of the untrusted_source block or impersonate a
 # system/role turn. Neutralised (not deleted — we keep byte offsets stable enough
@@ -505,6 +516,10 @@ def _read_files(units: "list[Path | FileSlice]", root: Path) -> str:
     parts: list[str] = []
     for u in units:
         p = unit_path(u)
+        safe_path = _resolve_under_root(p, root)
+        if safe_path is None:
+            print(f"[graphify] skipping {p}: symlink target outside corpus root", file=sys.stderr)
+            continue
         try:
             rel = str(p.relative_to(root))
         except ValueError:
@@ -513,7 +528,7 @@ def _read_files(units: "list[Path | FileSlice]", root: Path) -> str:
             if isinstance(u, FileSlice):
                 content = read_slice_text(u)
             else:
-                content = _file_to_text(p)
+                content = _file_to_text(safe_path)
         except OSError:
             continue
         # Whole files are still capped (covers non-splittable large files like
@@ -611,6 +626,10 @@ def _build_image_refs(image_files: list[Path], root: Path, *, read_bytes: bool =
     """
     refs: list[_ImageRef] = []
     for p in image_files:
+        abs_path = _resolve_under_root(p, root)
+        if abs_path is None:
+            print(f"[graphify] skipping image {p}: symlink target outside corpus root", file=sys.stderr)
+            continue
         try:
             rel = str(p.relative_to(root))
         except ValueError:
@@ -619,7 +638,7 @@ def _build_image_refs(image_files: list[Path], root: Path, *, read_bytes: bool =
         raw: bytes | None = None
         if read_bytes:
             try:
-                raw = p.read_bytes()
+                raw = abs_path.read_bytes()
             except OSError as exc:
                 print(f"[graphify] could not read image {rel}: {exc}", file=sys.stderr)
                 raw = None
@@ -631,10 +650,6 @@ def _build_image_refs(image_files: list[Path], root: Path, *, read_bytes: bool =
                     file=sys.stderr,
                 )
                 raw = None
-        try:
-            abs_path = p.resolve()
-        except OSError:
-            abs_path = p
         refs.append(_ImageRef(abs_path, rel, media, raw))
     return refs
 

tests/test_detect.py

@@ -236,37 +236,32 @@ def test_detect_handles_circular_symlinks(tmp_path):
     assert any("main.py" in f for f in result["files"]["code"])
 
 
-def test_detect_auto_detects_direct_symlink_child(tmp_path):
-    """When ``root`` has a direct symlinked child, default (None) follows symlinks
-    so the user does not have to know to pass follow_symlinks=True for "fake
-    working dir" patterns (folder of symlinks pointing at scattered sources)."""
+def test_detect_default_does_not_auto_follow_direct_symlink_child(tmp_path):
+    """Symlink directory following is explicit opt-in."""
     real_dir = tmp_path / "real_lib"
     real_dir.mkdir()
     (real_dir / "util.py").write_text("x = 1")
     (tmp_path / "linked_lib").symlink_to(real_dir)
 
-    # Default (no kwarg): auto-detect → follows because of linked_lib symlink
     result = detect(tmp_path)
-    assert any("linked_lib" in f for f in result["files"]["code"])
+    assert any("real_lib" in f for f in result["files"]["code"])
+    assert not any("linked_lib" in f for f in result["files"]["code"])
 
 
 def test_detect_default_does_not_follow_when_no_symlinks(tmp_path):
-    """When ``root`` has no direct symlinks, the auto-detect default stays False
-    (legacy behaviour preserved for ordinary scans)."""
+    """Ordinary scans still walk normal directories by default."""
     (tmp_path / "main.py").write_text("x = 1")
     sub = tmp_path / "sub"
     sub.mkdir()
     (sub / "other.py").write_text("y = 2")
 
-    # Smoke: no symlinks anywhere → auto-detect returns False, scan succeeds
     result = detect(tmp_path)
     assert any("main.py" in f for f in result["files"]["code"])
     assert any("other.py" in f for f in result["files"]["code"])
 
 
 def test_detect_explicit_false_overrides_auto_detect(tmp_path):
-    """An explicit follow_symlinks=False overrides the auto-detect, even when
-    root contains symlinks. Lets callers opt out of the new behaviour."""
+    """An explicit follow_symlinks=False skips symlinked directories."""
     real_dir = tmp_path / "real_lib"
     real_dir.mkdir()
     (real_dir / "util.py").write_text("x = 1")
@@ -277,6 +272,34 @@ def test_detect_explicit_false_overrides_auto_detect(tmp_path):
     assert not any("linked_lib" in f for f in result["files"]["code"])
 
 
+def test_detect_skips_out_of_root_symlinked_directory_even_when_following(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.py").write_text("token = 'outside'")
+    (root / "linked_secret").symlink_to(outside)
+
+    result = detect(root, follow_symlinks=True)
+
+    assert not any("linked_secret" in f for f in result["files"]["code"])
+    assert any("symlink target outside scan root" in item for item in result["skipped_sensitive"])
+
+
+def test_detect_skips_out_of_root_symlinked_file_by_default(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.py").write_text("token = 'outside'")
+    (root / "secret_link.py").symlink_to(outside / "secret.py")
+
+    result = detect(root)
+
+    assert not any("secret_link.py" in f for f in result["files"]["code"])
+    assert any("symlink target outside scan root" in item for item in result["skipped_sensitive"])
+
+
 def test_detect_incremental_propagates_follow_symlinks(tmp_path, monkeypatch):
     """detect_incremental must forward follow_symlinks so symlinked sub-trees
     appear in incremental scans the same way they appear in full scans."""

tests/test_extract.py

@@ -362,6 +362,32 @@ def test_collect_files_follows_symlinked_directory(tmp_path):
     assert [f.name for f in files_yes].count("lib.py") == 2
 
 
+def test_collect_files_skips_out_of_root_symlinked_directory(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.py").write_text("token = 'outside'")
+    (root / "linked_secret").symlink_to(outside)
+
+    files = collect_files(root, follow_symlinks=True)
+
+    assert not any("linked_secret" in str(f) for f in files)
+
+
+def test_collect_files_skips_out_of_root_symlinked_file_by_default(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.py").write_text("token = 'outside'")
+    (root / "secret_link.py").symlink_to(outside / "secret.py")
+
+    files = collect_files(root)
+
+    assert not any(f.name == "secret_link.py" for f in files)
+
+
 def test_collect_files_handles_circular_symlinks(tmp_path):
     sub = tmp_path / "pkg"
     sub.mkdir()

tests/test_image_vision.py

@@ -70,6 +70,22 @@ def test_non_pdf_still_read_as_plain_text(tmp_path):
     assert "# hello" in llm._file_to_text(md)
 
 
+def test_read_files_skips_out_of_root_symlink(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.md"
+    secret.write_text("SECRET SHOULD NOT REACH THE PROMPT")
+    link = root / "secret.md"
+    link.symlink_to(secret)
+
+    out = llm._read_files([link], root)
+
+    assert out == ""
+    assert "SECRET SHOULD NOT REACH THE PROMPT" not in out
+
+
 def test_partition_splits_raster_from_text(tmp_path):
     img, svg, doc = _make_corpus(tmp_path)
     text_files, image_files = llm._partition_semantic_files([doc, img, svg])
@@ -88,6 +104,21 @@ def test_build_image_refs_sets_rel_media_and_bytes(tmp_path):
     assert ref.bedrock_format == "png"
 
 
+def test_build_image_refs_skips_out_of_root_symlink(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    secret = outside / "secret.png"
+    secret.write_bytes(_PNG_BYTES)
+    link = root / "secret.png"
+    link.symlink_to(secret)
+
+    refs = llm._build_image_refs([link], root)
+
+    assert refs == []
+
+
 def test_build_image_refs_drops_oversized(tmp_path, monkeypatch):
     big = tmp_path / "big.jpg"
     big.write_bytes(b"x" * 64)