From 35167457467dbb8fb036f90f004cbd2c7adcf207 Mon Sep 17 00:00:00 2001 From: Travis Abendshien <46939827+CyanVoxel@users.noreply.github.com> Date: Sun, 13 Sep 2026 02:56:42 -0700 Subject: [PATCH] fix: block cyclical symlink traversal in internal scanner --- src/tagstudio/core/library/scanners.py | 42 +++++++++---- tests/core/library/test_scanners.py | 81 ++++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 12 deletions(-) create mode 100644 tests/core/library/test_scanners.py diff --git a/src/tagstudio/core/library/scanners.py b/src/tagstudio/core/library/scanners.py index bd761829..1aae77b3 100644 --- a/src/tagstudio/core/library/scanners.py +++ b/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)})) diff --git a/tests/core/library/test_scanners.py b/tests/core/library/test_scanners.py new file mode 100644 index 00000000..93289f68 --- /dev/null +++ b/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"), + }