fix(thumbnails): never follow a marker or a link out of the thumbnail folder (#320)

Marker keys are data, not paths: only a single <slug>-<hex> folder name
is ever joined onto the thumbnail folder; anything else is dropped and
logged. Every preview write and delete passes one ownership check (below a
key folder, lexically and resolved, no link, junction or reparse point on
the way); make_parent refuses anything else and writes nothing while the
folder lacks this install's marker (an unmounted mount point). Clear never
descends into a junction.

The configured location is in force before any request (unchecked folder
mode is unavailable), and a startup check that fails leaves it invalid,
never the sources. Older listings that run on the event loop no longer
touch the store in folder mode. A leftover under another case of
thumbnails/ on a case-insensitive disk is refused too.

__init__.py

@@ -96,6 +96,10 @@ try:
     # The thumbnail location's startup check (#320), then the temp files of
     # preview writes a crash interrupted (#317), off the import path: the
     # walk may cover a large thumbnails folder.
+    # #320: the configured thumbnail location is in force before any request
+    # (no I/O); the thread below checks it.
+    api.use_preview_location()
+
     def _preview_startup():
         api.start_preview_location()
         api.sweep_thumbnail_temps()

py/api/__init__.py

@@ -855,20 +855,11 @@ class PromptManagerAPI(
                         f"{url_quote(rel_path.as_posix(), safe='/')}"
                     )
 
-                    # Any preview the file owns, the current format first (#317)
-                    from . import thumbnail_settings
-                    from .prompts import thumbnail_url_for
+                    # Any preview the file owns, the current format first
+                    # (#317); no store I/O here in folder mode (#320).
+                    from .images import listing_thumbnail_url
 
-                    existing = thumbnail_settings.existing_thumbnail(
-                        output_path, rel_path
-                    )
-                    thumbnail_url = None
-                    if existing is not None:
-                        thumb_rel = (
-                            thumbnail_settings.thumbnail_stem_rel(rel_path)
-                            + existing.suffix
-                        )
-                        thumbnail_url = thumbnail_url_for(thumb_rel, existing)
+                    thumbnail_url = listing_thumbnail_url(output_path, rel_path)
                     if thumbnail_url:
                         image["thumbnail_url"] = thumbnail_url
                     break  # Found matching root, stop searching

py/api/autotag_routes.py

@@ -19,8 +19,8 @@ from .jobs import (
 )
 
 from .maintenance import disk_plan
-from .prompts import thumbnail_url_for
-from . import preview_store, thumbnail_settings
+from .images import listing_thumbnail_url
+from . import preview_store
 from .. import model_gate
 
 try:
@@ -1371,24 +1371,21 @@ class AutotagRoutesMixin:
             files = await self._run_in_executor(
                 self._without_removed, output_path, files
             )
-            store = preview_store.store_dir(output_path)
-            has_thumbnails = store is not None and store.is_dir()
+            location = preview_store.current()
+            if location.mode == preview_store.FOLDER_MODE:
+                has_thumbnails = location.available  # no store I/O here (#320)
+            else:
+                store = preview_store.store_dir(output_path)
+                has_thumbnails = store is not None and store.is_dir()
 
             images = []
             for image_path in files:
                 rel_path = image_path.relative_to(output_path)
                 thumbnail_url = None
                 if has_thumbnails:
-                    # Any preview the file owns, the current format first (#317)
-                    existing = thumbnail_settings.existing_thumbnail(
-                        output_path, rel_path
-                    )
-                    if existing is not None:
-                        thumb_rel = (
-                            thumbnail_settings.thumbnail_stem_rel(rel_path)
-                            + existing.suffix
-                        )
-                        thumbnail_url = thumbnail_url_for(thumb_rel, existing)
+                    # Any preview the file owns, the current format first
+                    # (#317); in folder mode without store I/O (#320).
+                    thumbnail_url = listing_thumbnail_url(output_path, rel_path)
                 images.append(
                     {
                         "filename": image_path.name,

py/api/images.py

@@ -1096,6 +1096,27 @@ def existing_media_thumbnail(output_path, rel_path):
     )
 
 
+def listing_thumbnail_url(output_path, rel_path):
+    """The thumbnail URL of an older listing (``_enrich_images``, the
+    auto-tag scan), or None. These run on the event loop: in folder mode
+    (#320) the store is never touched there, so the URL names the preview
+    as it is written now and the serve route finds or makes it off the loop
+    (None while the thumbnail folder is unavailable)."""
+    from .prompts import thumbnail_url_for
+
+    location = preview_store.current()
+    if location.mode == preview_store.FOLDER_MODE:
+        if not location.available:
+            return None
+        quoted = urllib.parse.quote(media_thumbnail_rel(rel_path), safe="/")
+        return f"/prompt_manager/images/serve/{quoted}"
+    existing = thumbnail_settings.existing_thumbnail(output_path, rel_path)
+    if existing is None:
+        return None
+    thumb_rel = thumbnail_settings.thumbnail_stem_rel(rel_path) + existing.suffix
+    return thumbnail_url_for(thumb_rel, existing)
+
+
 def media_thumbnail(output_path, rel_path, root_index, *, root_scope=None):
     """``(thumbnail_url, ready)`` for *rel_path* under *output_path* (a few
     blocking stats). The Images listing, the Needs review queue cards and the
@@ -2820,9 +2841,12 @@ class ImageRoutesMixin:
             )
             location = preview_store.current()
             if location.mode == preview_store.FOLDER_MODE:
-                if preview_parts is None and filepath.split("/", 1)[0] == "thumbnails":
+                if preview_parts is None and await self._run_image(
+                    self._in_source_thumbnails, output_dirs, filepath
+                ):
                     # Previews live in the thumbnail folder: anything else
-                    # under a source's thumbnails/ is a leftover (#320).
+                    # under a source's thumbnails/ is a leftover (#320), in
+                    # any case on a case-insensitive disk.
                     return _forbidden()
                 if preview_parts is not None:
                     # Store I/O never runs on the event loop (a share may
@@ -2882,6 +2906,27 @@ class ImageRoutesMixin:
                 {"success": False, "error": _safe_error(e)}, status=500
             )
 
+    @staticmethod
+    def _in_source_thumbnails(output_dirs, filepath):
+        """Whether *filepath* lies in a source's ``thumbnails`` folder: its
+        first part is ``thumbnails`` exactly, or another case of it that is
+        the same folder (a case-insensitive disk). ``Thumbnails/`` media on
+        a case-sensitive disk is another folder and stays served."""
+        first = filepath.split("/", 1)[0]
+        if first == "thumbnails":
+            return True
+        if first.casefold() != "thumbnails":
+            return False
+        for output_path in output_dirs:
+            spelled = output_path / first
+            cache = preview_store.source_folder(output_path)
+            try:
+                if spelled.is_dir() and cache.is_dir() and spelled.samefile(cache):
+                    return True
+            except OSError:
+                continue
+        return False
+
     @staticmethod
     def _find_preview(output_dirs, parts, location):
         """The preview file *parts* names in folder mode, ``"forbidden"``
@@ -3594,7 +3639,11 @@ class ImageRoutesMixin:
             return 0
         for output_path in output_dirs:
             folder = preview_store.store_dir(output_path)
-            if folder is None or folder.is_symlink() or not folder.is_dir():
+            if (
+                folder is None
+                or preview_store.store_is_link(folder)
+                or not folder.is_dir()
+            ):
                 continue
             removed += sweep_stray_temps(folder)[0]
         if removed:
@@ -3619,7 +3668,7 @@ class ImageRoutesMixin:
                 continue
             container = preview_store.store_container(output_path)
             if (
-                thumbnails_dir.is_symlink()
+                preview_store.store_is_link(thumbnails_dir)
                 or not _is_within(thumbnails_dir, container)
                 or _canonical(thumbnails_dir) == _canonical(container)
             ):
@@ -3636,7 +3685,15 @@ class ImageRoutesMixin:
             temps, temp_bytes = sweep_stray_temps(thumbnails_dir, min_age=0)
             cleared_count += temps
             cleared_size += temp_bytes
-            for root, _dirs, files in os.walk(thumbnails_dir):
+            folder_mode = preview_store.current().mode == preview_store.FOLDER_MODE
+            for root, dirs, files in os.walk(thumbnails_dir):
+                if folder_mode:
+                    # Never into a junction or reparse point (#320).
+                    dirs[:] = [
+                        d
+                        for d in dirs
+                        if not preview_store.store_is_link(Path(root) / d)
+                    ]
                 for name in files:
                     path = Path(root) / name
                     if (
@@ -3644,6 +3701,7 @@ class ImageRoutesMixin:
                         or path.suffix.lower()
                         not in thumbnail_settings.THUMBNAIL_EXTENSIONS
                         or path.is_symlink()
+                        or not preview_store.may_write(path)
                     ):
                         continue
                     try:

py/api/preview_location.py

@@ -22,6 +22,7 @@ A source key is a readable slug plus a short hash of the source's identity
 import errno
 import hashlib
 import json
+import logging
 import ntpath
 import os
 import posixpath
@@ -37,10 +38,16 @@ try:
 except ImportError:  # tests import the package top-level
     from utils import library_location
 
+logger = logging.getLogger("prompt_manager.thumbnails")
+
 SOURCE_MODE = "source"
 FOLDER_MODE = "folder"
 MARKER_NAME = ".promptmanager-thumbnails.json"
 MARKER_VERSION = 1
+# A source key: one folder name, a slug and a 8-16 hex hash (``source_key``).
+KEY_PATTERN = re.compile(r"^[a-z0-9_-]{1,24}-[0-9a-f]{8,16}$")
+# A marker is a few keys; anything larger is not ours.
+MAX_MARKER_BYTES = 4 * 2**20
 # OS metadata files a folder may hold and still count as empty.
 OS_METADATA = frozenset({"thumbs.db", "desktop.ini", ".ds_store"})
 KEY_HASH_LENGTHS = (8, 12, 16)
@@ -336,6 +343,7 @@ def file_identity(path):
 def find_key(keys, spelling, identity, dev_ino):
     """The marker key of a source: the same configured spelling, then the
     same identity text, then the same file id; or None."""
+    keys = {k: v for k, v in keys.items() if valid_key(k) and isinstance(v, dict)}
     for key, entry in keys.items():
         if entry.get("path") == spelling:
             return key
@@ -361,18 +369,45 @@ def read_marker(folder):
     ours). OSError other than "not found" propagates (an unreachable share)."""
     try:
         with open(marker_path(folder), "r", encoding="utf-8") as handle:
-            data = json.load(handle)
+            text = handle.read(MAX_MARKER_BYTES + 1)
     except FileNotFoundError:
         return None
     except (ValueError, UnicodeDecodeError):
         return None
+    if len(text) > MAX_MARKER_BYTES:
+        return None
+    try:
+        data = json.loads(text)
+    except ValueError:
+        return None
     if not isinstance(data, dict) or not isinstance(data.get("id"), str):
         return None
-    if not isinstance(data.get("keys"), dict):
-        data["keys"] = {}
+    data["keys"] = valid_keys(data.get("keys"))
     return data
 
 
+def valid_key(key):
+    """Whether *key* can name a source's folder: one folder name of the
+    ``source_key`` form, never a separator, ``..`` or a drive."""
+    return isinstance(key, str) and KEY_PATTERN.fullmatch(key) is not None
+
+
+def valid_keys(keys):
+    """The marker's keys that are safe to join onto the folder; a hand-edited
+    or foreign entry that is not is dropped (and logged), never followed."""
+    if not isinstance(keys, dict):
+        return {}
+    kept = {}
+    for key, entry in keys.items():
+        if valid_key(key) and isinstance(entry, dict):
+            kept[key] = entry
+        else:
+            logger.warning(
+                "Ignoring a thumbnail folder marker entry that is not a source " "key"
+            )
+    return kept
+
+
 def write_marker(folder, mutate):
     """Re-read the marker, apply ``mutate(marker)`` and write it atomically
     (a temp sibling renamed over it). Callers hold ``preview_store.write_lock``.
@@ -400,8 +435,8 @@ def write_marker(folder, mutate):
     finally:
         try:
             os.unlink(temp)
-        except FileNotFoundError:
-            pass
+        except OSError:
+            pass  # gone with the rename; a leftover is swept with the temps
     return marker
 
 

py/api/preview_store.py

@@ -29,7 +29,7 @@ import os
 import threading
 import time
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 
 from . import preview_location, thumbnail_settings
@@ -76,10 +76,16 @@ _clock = time.monotonic
 
 def current():
     """The thumbnail location now."""
-    parsed = _parsed
+    with _state_lock:
+        parsed, available, problem, generation_now = (
+            _parsed,
+            _available,
+            _problem,
+            _generation,
+        )
     if parsed.mode == SOURCE_MODE:
-        return Location(generation=_generation)
-    return Location(FOLDER_MODE, parsed.path, _available, _problem, _generation)
+        return Location(generation=generation_now)
+    return Location(FOLDER_MODE, parsed.path, available, problem, generation_now)
 
 
 def setting():
@@ -168,6 +174,32 @@ def refresh():
     return available
 
 
+def _set_keys(parsed, marker):
+    """Take the keys of a marker just written (availability unchanged)."""
+    with _state_lock:
+        if _parsed is not parsed:
+            return
+        keys = dict((marker or {}).get("keys") or {})
+        if keys != _keys:
+            _keys.clear()
+            _keys.update(keys)
+            _key_cache.clear()
+
+
+def mark_invalid():
+    """The setting cannot be used (a startup check that failed): folder
+    mode stays in force, unavailable, so nothing is written anywhere."""
+    global _parsed, _available, _problem
+    with _state_lock:
+        if _parsed.mode != FOLDER_MODE:
+            _parsed = preview_location.ParsedLocation(
+                mode=FOLDER_MODE, problem="invalid"
+            )
+        else:
+            _parsed = replace(_parsed, problem="invalid")
+        _available, _problem = False, "invalid"
+
+
 def _set_state(parsed, available, problem, marker):
     global _available, _problem
     with _state_lock:
@@ -263,19 +295,22 @@ class _Probe:
         with self._lock:
             if self._thread is not None and self._thread.is_alive():
                 return
-            self._stop.clear()
-            thread = threading.Thread(target=self._run, name=PROBE_THREAD, daemon=True)
+            # Each thread has its own stop: a stopped one never resumes.
+            stop = self._stop = threading.Event()
+            thread = threading.Thread(
+                target=self._run, args=(stop,), name=PROBE_THREAD, daemon=True
+            )
             try:
                 thread.start()
             except RuntimeError:
                 return  # refused at interpreter exit: no probe
             self._thread = thread
 
-    def _run(self):
-        while not self._stop.is_set():
+    def _run(self, stop):
+        while not stop.is_set():
             self._wake.wait(PROBE_INTERVAL)
             self._wake.clear()
-            if self._stop.is_set():
+            if stop.is_set():
                 return
             try:
                 refresh()
@@ -308,6 +343,12 @@ def stop_probe(timeout=2.0):
 # ── stores ───────────────────────────────────────────────────────────
 
 
+def source_folder(root):
+    """``<source>/thumbnails``: the store in source mode, a leftover in
+    folder mode."""
+    return Path(root) / PREVIEW_FOLDER
+
+
 def store_container(root, location=None):
     """The folder a store of *root* must stay inside: the source itself, or
     the thumbnail folder in folder mode."""
@@ -349,7 +390,18 @@ def store_dir(root, location=None):
     if not location.available or not location.path:
         return None
     key = _lookup_key(root)
-    return None if key is None else Path(location.path) / key
+    return None if key is None else _key_folder(location.path, key)
+
+
+def _key_folder(folder, key):
+    """``<folder>/<key>``, only for a key that is one folder name directly
+    below the thumbnail folder (a marker is data, never a path)."""
+    if not preview_location.valid_key(key):
+        return None
+    store = Path(folder) / key
+    if store.parent != Path(folder) or store.name != key:
+        return None
+    return store
 
 
 def ensure_store(root, location=None, *, create=False):
@@ -385,6 +437,8 @@ def _assign_key(root, location):
     assigned = {}
 
     def add(marker):
+        if marker.get("id") != parsed.id:
+            raise ValueError("another folder's marker")  # marker-mismatch
         keys = marker["keys"]
         key = preview_location.find_key(keys, spelling, identity, dev_ino)
         if key is None:
@@ -399,39 +453,86 @@ def _assign_key(root, location):
         assigned["key"] = key
 
     try:
-        with store_io(), write_lock:
+        with write_lock, store_io():
             marker = preview_location.write_marker(location.path, add)
     except (OSError, ValueError):
         return None
-    _set_state(parsed, True, None, marker)
-    return Path(location.path) / assigned["key"]
+    _set_keys(parsed, marker)
+    return _key_folder(location.path, assigned["key"])
+
+
+def may_write(path, location=None):
+    """The one ownership check every preview write and delete passes.
+
+    Source mode: True (the callers' own root rules apply, as before). Folder
+    mode: *path* lies below a source key folder of the thumbnail folder,
+    lexically and resolved, and no link, junction or reparse point sits
+    anywhere from the key folder down to it (the file itself included).
+    """
+    location = location or current()
+    if location.mode == SOURCE_MODE:
+        return True
+    if not location.path:
+        return False
+    folder = Path(os.path.abspath(location.path))
+    path = Path(os.path.abspath(path))
+    try:
+        relative = path.relative_to(folder)
+    except ValueError:
+        return False
+    if len(relative.parts) < 2 or ".." in relative.parts:
+        return False
+    store = _key_folder(folder, relative.parts[0])
+    if store is None or not _within(path, folder):
+        return False
+    try:
+        return _no_link_below(store, path)
+    except OSError:
+        note_failure()
+        return False
 
 
 def make_parent(dst, location=None):
     """Make the folders a preview *dst* is written into. In folder mode only
-    folders below the thumbnail folder are made, never the folder itself (a
-    missing folder raises FileNotFoundError: it may be an unplugged drive)."""
+    folders below the thumbnail folder are made, never the folder itself,
+    and only while it holds this install's marker (an unplugged drive or an
+    unmounted mount point raises FileNotFoundError); a *dst* that does not
+    pass ``may_write`` raises PermissionError."""
     location = location or current()
     dst = Path(dst)
     if location.mode == SOURCE_MODE or not location.path:
+        if location.mode != SOURCE_MODE:
+            raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
         dst.parent.mkdir(parents=True, exist_ok=True)
         return
-    folder = Path(location.path)
-    try:
-        relative = dst.parent.relative_to(folder)
-    except ValueError:
-        dst.parent.mkdir(parents=True, exist_ok=True)  # a store given by a test
-        return
+    folder = Path(os.path.abspath(location.path))
     with store_io():
         if not location.available or not folder.is_dir():
             raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
+        marker = preview_location.read_marker(str(folder))
+        if marker is None or marker.get("id") != _parsed.id:
+            note_failure()
+            raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
+        if not may_write(dst, location):
+            raise PermissionError(errno.EACCES, "outside the thumbnail folder")
         current_folder = folder
-        for part in relative.parts:
+        for part in Path(os.path.abspath(dst.parent)).relative_to(folder).parts:
             current_folder = current_folder / part
             try:
                 os.mkdir(current_folder)
             except FileExistsError:
                 pass
+            if preview_location.is_link_like(current_folder):
+                raise PermissionError(errno.EACCES, "a link in the thumbnail folder")
+
+
+def store_is_link(store, location=None):
+    """Whether a store folder is a link (a junction or reparse point too in
+    folder mode): it is skipped by sweeps and refused by clears."""
+    location = location or current()
+    if location.mode == SOURCE_MODE:
+        return Path(store).is_symlink()
+    return preview_location.is_link_like(store)
 
 
 def given_store(root, store):
@@ -505,11 +606,7 @@ def owns(root, path, location=None):
         return False
     if location.mode == SOURCE_MODE:
         return True
-    try:
-        return _no_link_below(store, path)
-    except OSError:
-        note_failure()
-        return False
+    return may_write(path, location)
 
 
 # ── preview text ─────────────────────────────────────────────────────

py/api/thumbnail_location.py

@@ -98,9 +98,26 @@ class ThumbnailLocationMixin:
             ),
         )
 
+    def use_preview_location(self):
+        """Put the configured location in force at once, unchecked (the
+        package's import, before any request): folder mode is unavailable
+        until ``start_preview_location`` has checked it, so nothing is ever
+        written to the sources meanwhile. No filesystem access."""
+        from ..config import PromptManagerConfig
+
+        try:
+            return preview_store.configure(
+                PromptManagerConfig.THUMBNAIL_LOCATION, check=False
+            )
+        except Exception:
+            self.logger.warning("The thumbnail location is unusable", exc_info=True)
+            preview_store.mark_invalid()
+            return preview_store.current()
+
     def start_preview_location(self):
         """Parse and check the configured location once (startup thread);
-        the probe takes over. Never raises."""
+        the probe takes over. Never raises; a check that fails leaves a
+        folder location unavailable (``invalid``), never the sources."""
         from ..config import PromptManagerConfig
 
         try:
@@ -112,7 +129,14 @@ class ThumbnailLocationMixin:
             location = preview_store.configure(value, context=context)
         except Exception:
             self.logger.warning("The thumbnail location check failed", exc_info=True)
-            return None
+            if (
+                preview_location.parse_location(
+                    PromptManagerConfig.THUMBNAIL_LOCATION
+                ).mode
+                == preview_store.FOLDER_MODE
+            ):
+                preview_store.mark_invalid()
+            return preview_store.current()
         if location.mode == preview_store.FOLDER_MODE and not location.available:
             self.logger.warning(
                 "The thumbnail folder is unavailable "

py/api/thumbnail_settings.py

@@ -224,9 +224,10 @@ def _owned(root, rel_path, preferred):
     found = owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
     if preview_store.current().mode == preview_store.SOURCE_MODE:
         return found
-    # Folder mode: timed, so a stalled share marks the folder unreachable.
+    # Folder mode: timed, so a stalled share marks the folder unreachable,
+    # and only previews no link takes out of the key folder (deletes use it).
     with preview_store.store_io():
-        return iter(list(found))
+        return iter([path for path in found if preview_store.may_write(path)])
 
 
 def owned_thumbnails(root, rel_path, preferred=None):

tests/test_api_thumbnail_location.py

@@ -145,14 +145,18 @@ class TestSourceKeys(unittest.TestCase):
 
     def test_lookup_by_spelling_identity_then_file_id(self):
         keys = {
-            "out-1": {"path": "D:\\Out", "identity": "d:\\out", "dev_ino": [1, 2]},
-            "lib-2": {"path": "E:\\Lib", "identity": "e:\\lib", "dev_ino": None},
+            "out-0123abcd": {
+                "path": "D:\\Out",
+                "identity": "d:\\out",
+                "dev_ino": [1, 2],
+            },
+            "lib-0123abcd": {"path": "E:\\Lib", "identity": "e:\\lib", "dev_ino": None},
         }
-        self.assertEqual(pl.find_key(keys, "D:\\Out", "x", None), "out-1")
-        self.assertEqual(pl.find_key(keys, "Z:\\Out", "d:\\out", None), "out-1")
+        self.assertEqual(pl.find_key(keys, "D:\\Out", "x", None), "out-0123abcd")
+        self.assertEqual(pl.find_key(keys, "Z:\\Out", "d:\\out", None), "out-0123abcd")
         # A letter spelling and its UNC spelling: the same file id.
         self.assertEqual(
-            pl.find_key(keys, "\\\\nas\\out", "\\\\nas\\out", (1, 2)), "out-1"
+            pl.find_key(keys, "\\\\nas\\out", "\\\\nas\\out", (1, 2)), "out-0123abcd"
         )
         self.assertIsNone(pl.find_key(keys, "F:\\New", "f:\\new", (9, 9)))
 
@@ -173,9 +177,15 @@ class TestMarker(unittest.TestCase):
 
     def test_writes_merge_what_another_writer_added(self):
         pl.init_folder(self.folder)
-        pl.write_marker(self.folder, lambda m: m["keys"].update(a={"install": "1"}))
-        pl.write_marker(self.folder, lambda m: m["keys"].update(b={"install": "2"}))
-        self.assertEqual(sorted(pl.read_marker(self.folder)["keys"]), ["a", "b"])
+        pl.write_marker(
+            self.folder, lambda m: m["keys"].update({"a-0123abcd": {"install": "1"}})
+        )
+        pl.write_marker(
+            self.folder, lambda m: m["keys"].update({"b-0123abcd": {"install": "2"}})
+        )
+        self.assertEqual(
+            sorted(pl.read_marker(self.folder)["keys"]), ["a-0123abcd", "b-0123abcd"]
+        )
         # No temp file is left beside it.
         self.assertEqual(os.listdir(self.folder), [pl.MARKER_NAME])
 

tests/test_preview_store_folder.py

@@ -360,3 +360,180 @@ class TestSharedFolders(FolderModeCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestHostileMarker(FolderModeCase):
+    """A hand-edited or foreign marker never takes a write or a delete out of
+    the thumbnail folder (shared and UNC folders can be edited by others)."""
+
+    def plant(self, keys):
+        marker = self.marker()
+        marker["keys"] = keys
+        (self.folder / preview_location.MARKER_NAME).write_text(
+            json.dumps(marker), encoding="utf-8"
+        )
+        self.assertTrue(preview_store.refresh())
+
+    def outside_files(self):
+        return sorted(
+            p.relative_to(self.comfy_dir).as_posix()
+            for p in self.comfy_dir.rglob("*")
+            if p.is_file()
+            and not p.is_relative_to(self.folder)
+            and not p.is_relative_to(self.output_dir)
+            and p.suffix in (".webp", ".jpg", ".png")
+        )
+
+    async def test_keys_that_are_not_one_folder_name_are_dropped(self):
+        entry = {"path": str(self.output_dir), "identity": "x", "install": "i"}
+        self.plant(
+            {
+                "../escape-0123abcd": entry,
+                str(self.comfy_dir / "abs-0123abcd"): entry,
+                "a/b-0123abcd": entry,
+                "a\\b-0123abcd": entry,
+                "Upper-0123abcd": entry,
+                "nohash": entry,
+                "fine-0123abcd": "not a dict",
+            }
+        )
+        self.assertEqual(preview_location.read_marker(str(self.folder))["keys"], {})
+        store = self.key_folder()
+        self.assertEqual(store.parent, self.folder)
+        make_png(self.output_dir / "p.png", size=(640, 480))
+        resp = await self.client.get(SERVE + "thumbnails/p_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertTrue((store / "p_thumb.webp").is_file())
+        self.assertEqual(self.outside_files(), [])
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_key_folder_that_is_a_link_is_never_written_or_cleared(self):
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "q_thumb.webp"
+        victim.write_bytes(b"keep me")
+        store = self.key_folder()
+        store.rmdir()
+        try:
+            store.symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        make_png(self.output_dir / "q.png", size=(640, 480))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("q.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertIn(resp.status, (200, 400))
+        resp = await self.client.get(SERVE + "thumbnails/q_thumb.webp?root=0")
+        self.assertIn(resp.status, (200, 403))
+        self.assertEqual(victim.read_bytes(), b"keep me")
+        self.assertEqual(sorted(p.name for p in outside.iterdir()), ["q_thumb.webp"])
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_linked_folder_below_a_key_is_never_deleted_through(self):
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "r_thumb.webp"
+        victim.write_bytes(b"keep me")
+        try:
+            (self.key_folder() / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        make_png(self.output_dir / "sub" / "r.png", size=(64, 64))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("sub/r.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(victim.read_bytes(), b"keep me")
+
+    @unittest.skipUnless(os.name == "nt", "junctions are Windows only")
+    async def test_a_key_folder_that_is_a_junction_is_never_written_or_cleared(self):
+        import _winapi
+
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "v_thumb.webp"
+        victim.write_bytes(b"keep me")
+        store = self.key_folder()
+        store.rmdir()
+        _winapi.CreateJunction(str(outside), str(store))
+        make_png(self.output_dir / "v.png", size=(640, 480))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("v.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertIn(resp.status, (200, 400))
+        resp = await self.client.get(SERVE + "thumbnails/v_thumb.webp?root=0")
+        self.assertIn(resp.status, (200, 403))
+        self.assertEqual(victim.read_bytes(), b"keep me")
+        self.assertEqual(sorted(p.name for p in outside.iterdir()), ["v_thumb.webp"])
+
+    async def test_a_write_outside_the_folder_is_refused(self):
+        with self.assertRaises(OSError):
+            preview_store.make_parent(self.comfy_dir / "elsewhere" / "a_thumb.webp")
+        self.assertFalse((self.comfy_dir / "elsewhere").exists())
+
+    async def test_an_unmounted_folder_takes_no_write_before_the_next_check(self):
+        """An unmounted mount point is an empty folder: the flag may still
+        say available, but without the marker nothing is made there."""
+        store = self.key_folder()
+        store.rmdir()
+        (self.folder / preview_location.MARKER_NAME).unlink()
+        self.assertTrue(preview_store.current().available)  # not checked yet
+        with self.assertRaises(OSError):
+            preview_store.make_parent(store / "sub" / "s_thumb.webp")
+        self.assertEqual(list(self.folder.iterdir()), [])
+
+
+class TestStartupNeverFallsBackToTheSources(AutoThumbnailCase):
+    async def test_a_folder_setting_is_in_force_before_its_check(self):
+        self.addCleanup(preview_store.configure, None)
+        folder = self.comfy_dir / "thumbs"
+        value = {"mode": "folder", "path": str(folder), "id": "x"}
+        preview_store.configure(value, check=False)
+        location = preview_store.current()
+        self.assertEqual((location.mode, location.available), ("folder", False))
+        make_png(self.output_dir / "t.png", size=(640, 480))
+        resp = await self.client.get(SERVE + "thumbnails/t_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_a_failed_startup_check_leaves_the_folder_invalid(self):
+        from py.config import PromptManagerConfig
+
+        self.addCleanup(preview_store.configure, None)
+        value = {"mode": "folder", "path": str(self.comfy_dir / "thumbs"), "id": "x"}
+        with (
+            patch.object(PromptManagerConfig, "THUMBNAIL_LOCATION", value),
+            patch.object(
+                type(self.api),
+                "_thumbnail_location_context",
+                side_effect=RuntimeError("boom"),
+            ),
+        ):
+            self.api.start_preview_location()
+        location = preview_store.current()
+        self.assertEqual(
+            (location.mode, location.available, location.problem),
+            ("folder", False, "invalid"),
+        )
+
+
+class TestNoStoreIoOnTheEventLoop(FolderModeCase):
+    async def test_legacy_listings_do_not_stat_the_folder(self):
+        make_png(self.output_dir / "u.png", size=(64, 64))
+        self.preview("u.png").write_bytes(b"x")
+        calls = []
+        real = thumbnail_settings.owned_in_folder
+        with patch.object(
+            thumbnail_settings,
+            "owned_in_folder",
+            lambda *a, **k: calls.append(threading.current_thread().name)
+            or real(*a, **k),
+        ):
+            images = self.api._enrich_images(
+                [{"image_path": str(self.output_dir / "u.png")}]
+            )
+        self.assertEqual(calls, [])
+        self.assertEqual(images[0]["thumbnail_url"], SERVE + "thumbnails/u_thumb.webp")