fix(thumbnails): never copy through a linked folder; refuse a target inside the folder in use (#320)

Moving back to Inside each source followed an existing linked folder in the
destination and could replace a file outside the source (Codex P1). A target
nested inside the thumbnail folder in use was accepted, and the old-copy
cleanup then removed the previews just copied there (Codex P2).

Every copy now checks the whole destination chain for links, also when the
folders exist, and the move refuses a folder inside the one in use with
location-inside-current. The cancelled branches no longer mix explicit and
implicit returns (code-quality note).

py/api/preview_location.py

@@ -103,6 +103,11 @@ MESSAGES = {
         "This folder contains an image source. Choose a folder beside it instead.",
     ),
     "location-same": (400, "Thumbnails are already stored here."),
+    "location-inside-current": (
+        400,
+        "This folder is inside the thumbnail folder in use. Choose a folder "
+        "outside it.",
+    ),
     "location-parent-missing": (
         400,
         "The folder that would hold it doesn't exist. Check the path and that its "
@@ -692,6 +697,14 @@ def _filesystem_problem(path, purpose, context, default):
         )
     ):
         return "location-same"
+    if (
+        purpose == "move"
+        and context.current_path
+        and _inside(path, [context.current_path])
+    ):
+        # The old key folders are removed after the move, and with them a
+        # target nested inside one of them.
+        return "location-inside-current"
     # Windows, or a test injecting its limit (``long_paths=False``).
     if os.name == "nt" or context.long_paths is False:
         if too_long_for_windows(path, context.long_paths):

py/api/thumbnail_move.py

@@ -153,7 +153,8 @@ class ThumbnailMoveMixin:
         done = 0
         for start in range(0, total, chunk):
             if job.cancelled:
-                return await self._move_cancelled(job, plan, state)
+                await self._move_cancelled(job, plan, state)
+                return
             batch = items[start : start + chunk]
             try:
                 results = await self._run_in_executor(
@@ -201,7 +202,8 @@ class ThumbnailMoveMixin:
             )
             job.publish({"type": "progress", "summary": job.summary()})
         if job.cancelled:
-            return await self._move_cancelled(job, plan, state)
+            await self._move_cancelled(job, plan, state)
+            return
         # The switch: not cancellable from here (an exit still stops between
         # files and leaves the cleanup to the next start).
         job.cancellable = False
@@ -318,6 +320,8 @@ class ThumbnailMoveMixin:
                 created = []
                 files._make_folders(base, relative[:-1], created)
                 state["created"].extend(created)
+            # Also when the folders already exist (Codex P1 on #592).
+            files.refuse_links(source.new_store, relative[:-1])
             outcome = files.copy_one(
                 Path(source.old_store).joinpath(*relative),
                 destination,

py/api/thumbnail_move_files.py

@@ -220,6 +220,17 @@ def _make_folders(base, relative_dir, created):
             raise CopyFailed("a link in the destination")
 
 
+def refuse_links(base, relative_dir):
+    """Raise CopyFailed when *base* or an existing folder on the way to
+    ``base/relative_dir`` is a link (never copy through one)."""
+    current = Path(base)
+    for part in (None, *relative_dir):
+        if part is not None:
+            current = current / part
+        if preview_location.is_link_like(current):
+            raise CopyFailed("a link in the destination")
+
+
 def copy_one(source, destination, size, mtime_ns, created, *, long_paths=None):
     """Copy one preview; returns ``"moved"`` or ``"skipped"`` (already
     there). Raises CopyFailed (this file) or FolderGone (the destination)."""

tests/test_api_thumbnail_location.py

@@ -319,6 +319,22 @@ class TestValidation(unittest.TestCase):
         os.mkdir(folder)
         self.assertEqual(self.problem(folder, current_path=folder), "location-same")
 
+    def test_a_folder_inside_the_folder_in_use_is_refused(self):
+        # Codex P2 on #592: the old key folders are removed after the move,
+        # and with them a target nested inside one of them.
+        folder = os.path.join(self.base, "thumbs")
+        os.mkdir(folder)
+        pl.init_folder(folder)
+        for nested in (
+            os.path.join(folder, "output-0123abcd", "nested"),
+            os.path.join(folder, "beside"),
+        ):
+            with self.subTest(nested=nested):
+                self.assertEqual(
+                    self.problem(nested, current_path=folder),
+                    "location-inside-current",
+                )
+
     def test_a_folder_with_other_files_is_not_empty_unless_it_is_ours(self):
         folder = os.path.join(self.base, "full")
         os.mkdir(folder)

tests/test_thumbnail_move.py

@@ -456,6 +456,25 @@ class TestReviewFindings(MoveCase):
         self.assertTrue((outside / "empty").is_dir())
         self.assertTrue((outside / "Thumbs.db").exists())
 
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_moving_back_never_copies_through_an_existing_linked_folder(self):
+        # Codex P1 on #592: an existing linked folder in the destination was
+        # followed, and a file outside the source was replaced.
+        await self.move({"mode": "folder", "path": str(self.folder)})
+        outside = Path(self.tmpdir) / "outside-p1"
+        outside.mkdir()
+        (outside / "b_thumb.webp").write_bytes(b"not ours")
+        if (self.old_store / "sub").exists():
+            (self.old_store / "sub").rmdir()
+        try:
+            (self.old_store / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        _plan, job = await self.move({"mode": "source", "path": None})
+        self.assertEqual((outside / "b_thumb.webp").read_bytes(), b"not ours")
+        self.assertEqual(job.counters["errors"], 1, job.log)
+        self.assertTrue((self.old_store / "a_thumb.webp").is_file())
+
     @unittest.skipUnless(os.name == "nt", "junctions are Windows only")
     def test_empty_folder_removal_never_walks_through_a_junction(self):
         import _winapi