Enforce directory root boundaries during traversal

tests/test_extractor.py

@@ -60,6 +60,7 @@ def test_extract_json_from_response(self):
                 result = extract_json_from_response(case["input"])
                 self.assertEqual(result, case["expected"])
 
+    @unittest.skipIf(not os.getenv("OPENAI_API_KEY"), "OpenAI API key required")
     def test_extract(self):
         # provide an explicit client so we cover the new parameter
         client = OpenAI()

tests/test_scrape_directory_security.py

@@ -0,0 +1,63 @@
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+from thepipe.scraper import scrape_directory
+
+
+def create_outside_link(link_path: Path, target_path: Path) -> None:
+    if os.name == "nt":
+        result = subprocess.run(
+            ["cmd", "/c", "mklink", "/J", str(link_path), str(target_path)],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise RuntimeError(result.stderr or result.stdout)
+    else:
+        os.symlink(target_path, link_path, target_is_directory=True)
+
+
+class test_scrape_directory_security(unittest.TestCase):
+    def test_scrape_directory_keeps_normal_files(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "inside.txt").write_text("INSIDE_FILE", encoding="utf-8")
+
+            chunks = scrape_directory(str(root))
+
+            self.assertEqual(len(chunks), 1)
+            self.assertEqual(chunks[0].text, "INSIDE_FILE")
+
+    def test_scrape_directory_stays_within_root(self):
+        canary = "OUTSIDE_ROOT_CANARY"
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            scan_root = root / "scan_root"
+            outside_root = root / "outside_root"
+            linked_dir = scan_root / "linked_outside"
+            scan_root.mkdir()
+            outside_root.mkdir()
+
+            (scan_root / "inside.txt").write_text("INSIDE_FILE", encoding="utf-8")
+            (outside_root / "secret.txt").write_text(canary, encoding="utf-8")
+
+            try:
+                create_outside_link(linked_dir, outside_root)
+            except (OSError, RuntimeError) as exc:
+                self.skipTest(f"unable to create outside-root link: {exc}")
+
+            chunks = scrape_directory(str(scan_root))
+            texts = [chunk.text or "" for chunk in chunks]
+            paths = [Path(chunk.path).resolve() for chunk in chunks if chunk.path]
+
+            self.assertTrue(any("INSIDE_FILE" in text for text in texts))
+            self.assertFalse(any(canary in text for text in texts))
+            self.assertFalse(any(path == (outside_root / "secret.txt") for path in paths))
+
+
+if __name__ == "__main__":
+    unittest.main()

thepipe/scraper.py

@@ -130,6 +130,13 @@ def detect_source_mimetype(source: str) -> str:
     return mimetype
 
 
+def _is_within_directory(root_path: str, candidate_path: str) -> bool:
+    try:
+        return os.path.commonpath([root_path, candidate_path]) == root_path
+    except ValueError:
+        return False
+
+
 def scrape_file(
     filepath: str,
     verbose: bool = False,
@@ -284,6 +291,8 @@ def scrape_directory(
     model: str = DEFAULT_AI_MODEL,
     include_input_images: bool = True,
     include_output_images: bool = True,
+    _root_dir: Optional[str] = None,
+    _visited_dirs: Optional[set[str]] = None,
 ) -> List[Chunk]:
     """
     inclusion_pattern: Optional regex string; only files whose path matches this pattern will be scraped.
@@ -292,10 +301,30 @@ def scrape_directory(
     # compile the include pattern once
     pattern = re.compile(inclusion_pattern) if inclusion_pattern else None
     extraction: List[Chunk] = []
+    canonical_root = os.path.realpath(_root_dir or dir_path)
+    current_dir = os.path.realpath(dir_path)
+    visited_dirs = _visited_dirs if _visited_dirs is not None else set()
+
+    if not _is_within_directory(canonical_root, current_dir):
+        if verbose:
+            print(f"[thepipe] Skipping path outside root: {dir_path}")
+        return extraction
+
+    if current_dir in visited_dirs:
+        if verbose:
+            print(f"[thepipe] Skipping already visited directory: {current_dir}")
+        return extraction
+    visited_dirs.add(current_dir)
 
     try:
         for entry in os.scandir(dir_path):
             path = entry.path
+            resolved_path = os.path.realpath(path)
+
+            if not _is_within_directory(canonical_root, resolved_path):
+                if verbose:
+                    print(f"[thepipe] Skipping path outside root: {path}")
+                continue
 
             # skip ignored directories
             if entry.is_dir() and any(
@@ -321,9 +350,9 @@ def scrape_directory(
                     continue
 
                 if verbose:
-                    print(f"[thepipe] Scraping file: {path}")
+                    print(f"[thepipe] Scraping file: {resolved_path}")
                 extraction += scrape_file(
-                    filepath=path,
+                    filepath=resolved_path,
                     verbose=verbose,
                     openai_client=openai_client,
                     model=model,
@@ -334,15 +363,17 @@ def scrape_directory(
             elif entry.is_dir():
                 # recurse into subdirectory
                 if verbose:
-                    print(f"[thepipe] Entering directory: {path}")
+                    print(f"[thepipe] Entering directory: {resolved_path}")
                 extraction += scrape_directory(
-                    dir_path=path,
+                    dir_path=resolved_path,
                     inclusion_pattern=inclusion_pattern,
                     verbose=verbose,
                     openai_client=openai_client,
                     model=model,
                     include_input_images=include_input_images,
                     include_output_images=include_output_images,
+                    _root_dir=canonical_root,
+                    _visited_dirs=visited_dirs,
                 )
     except PermissionError as e:
         if verbose: