diff --git a/py/config.py b/py/config.py index 368b8f69..2e6f679b 100644 --- a/py/config.py +++ b/py/config.py @@ -766,7 +766,23 @@ class Config: return f'/api/lm/previews?path={encoded_path}' def is_preview_path_allowed(self, preview_path: str) -> bool: - """Return ``True`` if ``preview_path`` is within an allowed directory.""" + """Return ``True`` if ``preview_path`` is within an allowed directory. + + If the path is initially rejected, attempts to discover deep symlinks + that were not scanned during initialization. If a symlink is found, + updates the in-memory path mappings and retries the check. + """ + + if self._is_path_in_allowed_roots(preview_path): + return True + + if self._try_discover_deep_symlink(preview_path): + return self._is_path_in_allowed_roots(preview_path) + + return False + + def _is_path_in_allowed_roots(self, preview_path: str) -> bool: + """Check if preview_path is within allowed preview roots without modification.""" if not preview_path: return False @@ -776,29 +792,72 @@ class Config: except Exception: return False - # Use os.path.normcase for case-insensitive comparison on Windows. - # On Windows, Path.relative_to() is case-sensitive for drive letters, - # causing paths like 'a:/folder' to not match 'A:/folder'. candidate_str = os.path.normcase(str(candidate)) for root in self._preview_root_paths: root_str = os.path.normcase(str(root)) - # Check if candidate is equal to or under the root directory if candidate_str == root_str or candidate_str.startswith(root_str + os.sep): return True - if self._preview_root_paths: - logger.debug( - "Preview path rejected: %s (candidate=%s, num_roots=%d, first_root=%s)", - preview_path, - candidate_str, - len(self._preview_root_paths), - os.path.normcase(str(next(iter(self._preview_root_paths)))), - ) - else: - logger.debug( - "Preview path rejected (no roots configured): %s", - preview_path, - ) + logger.debug( + "Path not in allowed roots: %s (candidate=%s, num_roots=%d)", + preview_path, + candidate_str, + len(self._preview_root_paths), + ) + + return False + + def _try_discover_deep_symlink(self, preview_path: str) -> bool: + """Attempt to discover a deep symlink that contains the preview_path. + + Walks up from the preview path to the root directories, checking each + parent directory for symlinks. If a symlink is found, updates the + in-memory path mappings and preview roots. + + Only updates in-memory state (self._path_mappings and self._preview_root_paths), + does not modify the persistent cache file. + + Returns: + True if a symlink was discovered and mappings updated, False otherwise. + """ + if not preview_path: + return False + + try: + candidate = Path(preview_path).expanduser() + except Exception: + return False + + current = candidate + while True: + try: + if self._is_link(str(current)): + try: + target = os.path.realpath(str(current)) + normalized_target = self._normalize_path(target) + normalized_link = self._normalize_path(str(current)) + + self._path_mappings[normalized_target] = normalized_link + self._preview_root_paths.update(self._expand_preview_root(normalized_target)) + self._preview_root_paths.update(self._expand_preview_root(normalized_link)) + + logger.debug( + "Discovered deep symlink: %s -> %s (preview path: %s)", + normalized_link, + normalized_target, + preview_path + ) + + return True + except OSError: + pass + except OSError: + pass + + parent = current.parent + if parent == current: + break + current = parent return False diff --git a/py/routes/handlers/preview_handlers.py b/py/routes/handlers/preview_handlers.py index e3bee61a..7916e716 100644 --- a/py/routes/handlers/preview_handlers.py +++ b/py/routes/handlers/preview_handlers.py @@ -33,6 +33,10 @@ class PreviewHandler: raise web.HTTPBadRequest(text="Invalid preview path encoding") from exc normalized = decoded_path.replace("\\", "/") + + if not self._config.is_preview_path_allowed(normalized): + raise web.HTTPForbidden(text="Preview path is not within an allowed directory") + candidate = Path(normalized) try: resolved = candidate.expanduser().resolve(strict=False) @@ -40,12 +44,8 @@ class PreviewHandler: logger.debug("Failed to resolve preview path %s: %s", normalized, exc) raise web.HTTPBadRequest(text="Unable to resolve preview path") from exc - resolved_str = str(resolved) - if not self._config.is_preview_path_allowed(resolved_str): - raise web.HTTPForbidden(text="Preview path is not within an allowed directory") - if not resolved.is_file(): - logger.debug("Preview file not found at %s", resolved_str) + logger.debug("Preview file not found at %s", str(resolved)) raise web.HTTPNotFound(text="Preview file not found") # aiohttp's FileResponse handles range requests and content headers for us. diff --git a/tests/config/test_symlink_cache.py b/tests/config/test_symlink_cache.py index c0d33567..959e4bd6 100644 --- a/tests/config/test_symlink_cache.py +++ b/tests/config/test_symlink_cache.py @@ -298,6 +298,134 @@ def test_deep_symlink_not_scanned(monkeypatch: pytest.MonkeyPatch, tmp_path): assert normalized_external not in cfg._path_mappings +def test_deep_symlink_discovered_on_preview_access(monkeypatch: pytest.MonkeyPatch, tmp_path): + """Deep symlinks are discovered dynamically when preview is accessed.""" + loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path) + + # Create nested structure with deep symlink at second level + subdir = loras_dir / "anime" + subdir.mkdir() + external_dir = tmp_path / "external" + external_dir.mkdir() + deep_symlink = subdir / "styles" + deep_symlink.symlink_to(external_dir, target_is_directory=True) + + # Create preview file under deep symlink + preview_file = deep_symlink / "model.preview.jpeg" + preview_file.write_bytes(b"preview") + + # Config should not initially detect deep symlinks + cfg = config_module.Config() + normalized_external = _normalize(str(external_dir)) + normalized_deep_link = _normalize(str(deep_symlink)) + assert normalized_external not in cfg._path_mappings + + # First preview access triggers symlink discovery automatically and returns True + is_allowed = cfg.is_preview_path_allowed(str(preview_file)) + + # After discovery, preview should be allowed + assert is_allowed + assert normalized_external in cfg._path_mappings + assert cfg._path_mappings[normalized_external] == normalized_deep_link + + # Verify preview path is now allowed without triggering discovery again + assert cfg.is_preview_path_allowed(str(preview_file)) + + +def test_deep_symlink_at_third_level(monkeypatch: pytest.MonkeyPatch, tmp_path): + """Deep symlinks at third level are also discovered dynamically.""" + loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path) + + # Create nested structure with deep symlink at third level + level1 = loras_dir / "category" + level1.mkdir() + level2 = level1 / "subcategory" + level2.mkdir() + external_dir = tmp_path / "external_deep" + external_dir.mkdir() + deep_symlink = level2 / "deep" + deep_symlink.symlink_to(external_dir, target_is_directory=True) + + # Create preview file under deep symlink + preview_file = deep_symlink / "preview.webp" + preview_file.write_bytes(b"test") + + cfg = config_module.Config() + + # First preview access triggers symlink discovery at third level + is_allowed = cfg.is_preview_path_allowed(str(preview_file)) + + assert is_allowed + normalized_external = _normalize(str(external_dir)) + normalized_deep_link = _normalize(str(deep_symlink)) + assert normalized_external in cfg._path_mappings + assert cfg._path_mappings[normalized_external] == normalized_deep_link + + +def test_deep_symlink_points_outside_roots(monkeypatch: pytest.MonkeyPatch, tmp_path): + """Deep symlinks can point to locations outside configured roots.""" + loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path) + + # Create nested structure with deep symlink pointing outside roots + subdir = loras_dir / "shared" + subdir.mkdir() + outside_root = tmp_path / "storage" + outside_root.mkdir() + deep_symlink = subdir / "models" + deep_symlink.symlink_to(outside_root, target_is_directory=True) + + # Create preview file under deep symlink (outside original roots) + preview_file = deep_symlink / "external.png" + preview_file.write_bytes(b"external") + + cfg = config_module.Config() + + # Preview access triggers symlink discovery + is_allowed = cfg.is_preview_path_allowed(str(preview_file)) + + # After discovery, preview should be allowed even though target is outside roots + assert is_allowed + normalized_outside = _normalize(str(outside_root)) + assert normalized_outside in cfg._path_mappings + + +def test_normal_path_unaffected_by_discovery(monkeypatch: pytest.MonkeyPatch, tmp_path): + """Normal paths (no symlinks) are not affected by symlink discovery logic.""" + loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path) + + # Create normal file structure (no symlinks) + preview_file = loras_dir / "normal.preview.jpeg" + preview_file.write_bytes(b"normal") + + cfg = config_module.Config() + + # Normal paths work without any discovery + assert cfg.is_preview_path_allowed(str(preview_file)) + assert len(cfg._path_mappings) == 0 + + +def test_first_level_symlink_still_works(monkeypatch: pytest.MonkeyPatch, tmp_path): + """First-level symlinks continue to work as before.""" + loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path) + + # Create first-level symlink + external_dir = tmp_path / "first_level_external" + external_dir.mkdir() + first_symlink = loras_dir / "first_level" + first_symlink.symlink_to(external_dir, target_is_directory=True) + + # Create preview file under first-level symlink + preview_file = first_symlink / "model.png" + preview_file.write_bytes(b"first_level") + + cfg = config_module.Config() + + # First-level symlinks are scanned during initialization + normalized_external = _normalize(str(external_dir)) + assert normalized_external in cfg._path_mappings + assert cfg.is_preview_path_allowed(str(preview_file)) + + def test_legacy_symlink_cache_automatic_cleanup(monkeypatch: pytest.MonkeyPatch, tmp_path): """Test that legacy symlink cache is automatically cleaned up after migration.""" settings_dir = tmp_path / "settings" diff --git a/tests/routes/test_preview_routes.py b/tests/routes/test_preview_routes.py index 909f7004..7b9f2853 100644 --- a/tests/routes/test_preview_routes.py +++ b/tests/routes/test_preview_routes.py @@ -188,3 +188,91 @@ def test_is_preview_path_allowed_rejects_prefix_without_separator(tmp_path): # The sibling path should NOT be allowed even though it shares a prefix assert not config.is_preview_path_allowed(str(sibling_file)), \ f"Path in '{sibling_root}' should NOT be allowed when root is '{library_root}'" + + +async def test_preview_handler_serves_from_deep_symlink(tmp_path): + """Test that previews under deep symlinks are served correctly.""" + library_root = tmp_path / "library" + library_root.mkdir() + + # Create nested structure with deep symlink at second level + subdir = library_root / "anime" + subdir.mkdir() + external_dir = tmp_path / "external" + external_dir.mkdir() + deep_symlink = subdir / "styles" + deep_symlink.symlink_to(external_dir, target_is_directory=True) + + # Create preview file under deep symlink + preview_file = deep_symlink / "model.preview.webp" + preview_file.write_bytes(b"preview_content") + + config = Config() + config.apply_library_settings( + { + "folder_paths": { + "loras": [str(library_root)], + "checkpoints": [], + "unet": [], + "embeddings": [], + } + } + ) + + handler = PreviewHandler(config=config) + encoded_path = urllib.parse.quote(str(preview_file), safe="") + request = make_mocked_request("GET", f"/api/lm/previews?path={encoded_path}") + + response = await handler.serve_preview(request) + + assert isinstance(response, web.FileResponse) + assert response.status == 200 + assert Path(response._path) == preview_file.resolve() + + +async def test_deep_symlink_discovered_on_first_access(tmp_path): + """Test that deep symlinks are discovered on first preview access.""" + library_root = tmp_path / "library" + library_root.mkdir() + + # Create nested structure with deep symlink at second level + subdir = library_root / "category" + subdir.mkdir() + external_dir = tmp_path / "storage" + external_dir.mkdir() + deep_symlink = subdir / "models" + deep_symlink.symlink_to(external_dir, target_is_directory=True) + + # Create preview file under deep symlink + preview_file = deep_symlink / "test.png" + preview_file.write_bytes(b"test_image") + + config = Config() + config.apply_library_settings( + { + "folder_paths": { + "loras": [str(library_root)], + "checkpoints": [], + "unet": [], + "embeddings": [], + } + } + ) + + # Deep symlink should not be in mappings initially + normalized_external = os.path.normpath(str(external_dir)).replace(os.sep, '/') + assert normalized_external not in config._path_mappings + + handler = PreviewHandler(config=config) + encoded_path = urllib.parse.quote(str(preview_file), safe="") + request = make_mocked_request("GET", f"/api/lm/previews?path={encoded_path}") + + # First access should trigger symlink discovery and serve the preview + response = await handler.serve_preview(request) + + assert isinstance(response, web.FileResponse) + assert response.status == 200 + assert Path(response._path) == preview_file.resolve() + + # Deep symlink should now be in mappings + assert normalized_external in config._path_mappings