fix: block cyclical symlink traversal in internal scanner

src/tagstudio/core/library/scanners.py

@@ -2,12 +2,14 @@
 # SPDX-License-Identifier: MIT
 
 
+import os
+import stat
 import subprocess
 from collections.abc import Iterator
 from pathlib import Path
 
 import structlog
-from wcmatch import pathlib
+import wcmatch.fnmatch as fnmatch
 
 from tagstudio.core.constants import TS_FOLDER_NAME
 from tagstudio.core.library.ignore import PATH_GLOB_FLAGS, ignore_to_glob
@@ -94,17 +96,33 @@ def _scan_with_ripgrep(scan_dir: Path, ignore_patterns: list[str]) -> Iterator[P
 
 
 def _scan_with_internal_scanner(scan_dir: Path, ignore_patterns: list[str]) -> Iterator[Path]:
-    """Scan for files with the internal glob-based scanner (wcmatch)."""
+    """Scan for files with the internal scanner."""
     logger.info("[Scanners] Using internal scanner for scanning", path=scan_dir)
+    matcher = fnmatch.compile(ignore_to_glob(ignore_patterns), PATH_GLOB_FLAGS)
 
-    glob_patterns = ignore_to_glob(ignore_patterns)
-    try:
-        for f in pathlib.Path(str(scan_dir)).glob(
-            "***/*", flags=PATH_GLOB_FLAGS, exclude=glob_patterns
-        ):
-            if f.is_dir():
+    def walk(dir_path: Path, ancestors: frozenset[tuple[int, int]]) -> Iterator[Path]:
+        try:
+            dir_items = list(os.scandir(dir_path))
+        except OSError as e:
+            logger.error("[Scanners] Could not scan directory", path=dir_path, error=e)
+            return
+
+        for item in dir_items:
+            rel = Path(item.path).relative_to(scan_dir)
+            if matcher.match(rel.as_posix()):
                 continue
-            path = Path(f).relative_to(scan_dir)
-            yield path
-    except ValueError:
-        logger.error("[Scanners] ValueError while scanning directory with the internal scanner")
+            try:
+                item_stat = item.stat(follow_symlinks=True)
+            except OSError:
+                continue
+
+            # Check for and handle cyclical symlinks
+            if stat.S_ISDIR(item_stat.st_mode):
+                key = (item_stat.st_dev, item_stat.st_ino)
+                if key not in ancestors:
+                    yield from walk(Path(item.path), ancestors | {key})
+            else:
+                yield rel
+
+    root_stat = scan_dir.stat()
+    yield from walk(scan_dir, frozenset({(root_stat.st_dev, root_stat.st_ino)}))

tests/core/library/test_scanners.py

@@ -0,0 +1,81 @@
+# SPDX-FileCopyrightText: (c) TagStudio Contributors
+# SPDX-License-Identifier: GPL-3.0-only
+
+# pyright: reportPrivateUsage=false
+
+import os
+from pathlib import Path
+
+from tagstudio.core.library.scanners import _scan_with_internal_scanner
+
+
+def test_scan_finds_normal_files(tmp_path: Path):
+    """Normal files are found."""
+    (tmp_path / "a.txt").touch()
+    (tmp_path / "sub").mkdir()
+    (tmp_path / "sub" / "b.txt").touch()
+
+    results = set(_scan_with_internal_scanner(tmp_path, []))
+    assert results == {Path("a.txt"), Path("sub/b.txt")}
+
+
+def test_scan_ignore_directory(tmp_path: Path):
+    """An ignored directory ignores the files within it."""
+    (tmp_path / "keep.txt").touch()
+    (tmp_path / "ignore_dir").mkdir()
+    (tmp_path / "ignore_dir" / "skip.txt").touch()
+
+    results = set(_scan_with_internal_scanner(tmp_path, ["ignore_dir"]))
+    assert results == {Path("keep.txt")}
+
+
+def test_scan_follows_symlinked_directory(tmp_path: Path):
+    """A non-cyclic symlink directory must be followed."""
+    real_target = tmp_path / "original_dir"
+    real_target.mkdir()
+    (real_target / "photo.jpg").touch()
+    os.symlink(real_target, tmp_path / "symlink_dir", target_is_directory=True)
+
+    results = set(_scan_with_internal_scanner(tmp_path, []))
+    assert results == {Path("original_dir/photo.jpg"), Path("symlink_dir/photo.jpg")}
+
+
+def test_scan_stop_infinite_symlink_cycle(tmp_path: Path):
+    """A cyclic symlink shouldn't cause an infinite loop, and be truncated."""
+    (tmp_path / "sub").mkdir()
+    (tmp_path / "sub" / "file.txt").touch()
+    os.symlink(tmp_path, tmp_path / "sub" / "loop", target_is_directory=True)
+
+    results = set(_scan_with_internal_scanner(tmp_path, []))
+    assert results == {Path("sub/file.txt")}
+
+
+def test_scan_handles_mutual_symlink_cycle(tmp_path: Path):
+    """Two directories symlinking into each other must terminate."""
+    # NOTE: Example names are from pnpn packages that triggered this weird case while testing.
+    store = tmp_path / "node_modules" / ".pnpm"
+    a_real = store / "pkgA@1.0.0" / "node_modules" / "pkgA"
+    b_real = store / "pkgB@1.0.0" / "node_modules" / "pkgB"
+    a_real.mkdir(parents=True)
+    b_real.mkdir(parents=True)
+    (a_real / "index.js").touch()
+    (b_real / "index.js").touch()
+    (a_real / "node_modules").mkdir()
+    (b_real / "node_modules").mkdir()
+    os.symlink(a_real, tmp_path / "node_modules" / "pkgA", target_is_directory=True)
+    os.symlink(b_real, tmp_path / "node_modules" / "pkgB", target_is_directory=True)
+    os.symlink(b_real, a_real / "node_modules" / "pkgB", target_is_directory=True)
+    os.symlink(a_real, b_real / "node_modules" / "pkgA", target_is_directory=True)
+
+    results = list(_scan_with_internal_scanner(tmp_path, []))
+    assert len(results) == 8  # Bounded - matches ripgrep's own count on this fixture
+    assert set(results) == {
+        Path("node_modules/pkgA/index.js"),
+        Path("node_modules/pkgA/node_modules/pkgB/index.js"),
+        Path("node_modules/pkgB/index.js"),
+        Path("node_modules/pkgB/node_modules/pkgA/index.js"),
+        Path("node_modules/.pnpm/pkgA@1.0.0/node_modules/pkgA/index.js"),
+        Path("node_modules/.pnpm/pkgA@1.0.0/node_modules/pkgA/node_modules/pkgB/index.js"),
+        Path("node_modules/.pnpm/pkgB@1.0.0/node_modules/pkgB/index.js"),
+        Path("node_modules/.pnpm/pkgB@1.0.0/node_modules/pkgB/node_modules/pkgA/index.js"),
+    }