From 47eb8a51c3ce4b29354e086d0d18e6b0dffce373 Mon Sep 17 00:00:00 2001 From: 0xMRMA Date: Tue, 26 May 2026 01:35:28 +0300 Subject: [PATCH 1/2] Enforce directory root boundaries during traversal --- tests/test_scrape_directory_security.py | 63 +++++++++++++++++++++++++ thepipe/scraper.py | 39 +++++++++++++-- 2 files changed, 98 insertions(+), 4 deletions(-) create mode 100644 tests/test_scrape_directory_security.py diff --git a/tests/test_scrape_directory_security.py b/tests/test_scrape_directory_security.py new file mode 100644 index 0000000..4cd73b7 --- /dev/null +++ b/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() diff --git a/thepipe/scraper.py b/thepipe/scraper.py index 942f8d6..a3114b7 100644 --- a/thepipe/scraper.py +++ b/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: From 327dfde5399bd9fea6bc38570868bc32810b4b82 Mon Sep 17 00:00:00 2001 From: 0xMRMA Date: Tue, 26 May 2026 01:45:05 +0300 Subject: [PATCH 2/2] Skip extractor integration test without OpenAI key --- tests/test_extractor.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_extractor.py b/tests/test_extractor.py index 30a1b55..1c3a518 100644 --- a/tests/test_extractor.py +++ b/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()