feat(config): discover deep symlinks dynamically when accessing previews

This commit is contained in:
Will Miao
2026-02-04 00:16:59 +08:00
parent c12aefa82a
commit 7bcf4e4491
4 changed files with 298 additions and 23 deletions

View File

@@ -766,7 +766,23 @@ class Config:
return f'/api/lm/previews?path={encoded_path}' return f'/api/lm/previews?path={encoded_path}'
def is_preview_path_allowed(self, preview_path: str) -> bool: 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: if not preview_path:
return False return False
@@ -776,30 +792,73 @@ class Config:
except Exception: except Exception:
return False 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)) candidate_str = os.path.normcase(str(candidate))
for root in self._preview_root_paths: for root in self._preview_root_paths:
root_str = os.path.normcase(str(root)) 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): if candidate_str == root_str or candidate_str.startswith(root_str + os.sep):
return True return True
if self._preview_root_paths:
logger.debug( logger.debug(
"Preview path rejected: %s (candidate=%s, num_roots=%d, first_root=%s)", "Path not in allowed roots: %s (candidate=%s, num_roots=%d)",
preview_path, preview_path,
candidate_str, candidate_str,
len(self._preview_root_paths), len(self._preview_root_paths),
os.path.normcase(str(next(iter(self._preview_root_paths)))),
) )
else:
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( logger.debug(
"Preview path rejected (no roots configured): %s", "Discovered deep symlink: %s -> %s (preview path: %s)",
preview_path, 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 return False
def apply_library_settings(self, library_config: Mapping[str, object]) -> None: def apply_library_settings(self, library_config: Mapping[str, object]) -> None:

View File

@@ -33,6 +33,10 @@ class PreviewHandler:
raise web.HTTPBadRequest(text="Invalid preview path encoding") from exc raise web.HTTPBadRequest(text="Invalid preview path encoding") from exc
normalized = decoded_path.replace("\\", "/") 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) candidate = Path(normalized)
try: try:
resolved = candidate.expanduser().resolve(strict=False) resolved = candidate.expanduser().resolve(strict=False)
@@ -40,12 +44,8 @@ class PreviewHandler:
logger.debug("Failed to resolve preview path %s: %s", normalized, exc) logger.debug("Failed to resolve preview path %s: %s", normalized, exc)
raise web.HTTPBadRequest(text="Unable to resolve preview path") from 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(): 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") raise web.HTTPNotFound(text="Preview file not found")
# aiohttp's FileResponse handles range requests and content headers for us. # aiohttp's FileResponse handles range requests and content headers for us.

View File

@@ -298,6 +298,134 @@ def test_deep_symlink_not_scanned(monkeypatch: pytest.MonkeyPatch, tmp_path):
assert normalized_external not in cfg._path_mappings 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): def test_legacy_symlink_cache_automatic_cleanup(monkeypatch: pytest.MonkeyPatch, tmp_path):
"""Test that legacy symlink cache is automatically cleaned up after migration.""" """Test that legacy symlink cache is automatically cleaned up after migration."""
settings_dir = tmp_path / "settings" settings_dir = tmp_path / "settings"

View File

@@ -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 # The sibling path should NOT be allowed even though it shares a prefix
assert not config.is_preview_path_allowed(str(sibling_file)), \ assert not config.is_preview_path_allowed(str(sibling_file)), \
f"Path in '{sibling_root}' should NOT be allowed when root is '{library_root}'" 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