From f51f349336ed22b909592aeea8a5eff2ae965994 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sun, 1 Nov 2026 03:46:11 -0800 Subject: [PATCH] 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 | 13 +++++++++++++ py/api/thumbnail_move.py | 8 ++++++-- py/api/thumbnail_move_files.py | 11 +++++++++++ tests/test_api_thumbnail_location.py | 16 ++++++++++++++++ tests/test_thumbnail_move.py | 19 +++++++++++++++++++ 5 files changed, 65 insertions(+), 2 deletions(-) diff --git a/py/api/preview_location.py b/py/api/preview_location.py index 7e9f8638..862122ce 100644 --- a/py/api/preview_location.py +++ b/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): diff --git a/py/api/thumbnail_move.py b/py/api/thumbnail_move.py index 9a5a9eda..13bfebce 100644 --- a/py/api/thumbnail_move.py +++ b/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, diff --git a/py/api/thumbnail_move_files.py b/py/api/thumbnail_move_files.py index cd3da88a..031820a4 100644 --- a/py/api/thumbnail_move_files.py +++ b/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).""" diff --git a/tests/test_api_thumbnail_location.py b/tests/test_api_thumbnail_location.py index 67470ddf..fd78be7c 100644 --- a/tests/test_api_thumbnail_location.py +++ b/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) diff --git a/tests/test_thumbnail_move.py b/tests/test_thumbnail_move.py index 0189fc90..790afab6 100644 --- a/tests/test_thumbnail_move.py +++ b/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