From 7eaf13b006d613e82504041f1d0f0b20bf391fcc Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Fri, 30 Oct 2026 12:46:19 -0700 Subject: [PATCH] fix(trash): allow the mount's own trash when the output folder is a mount point (#543) An output root that is exactly a mount root may hold that mount's .Trash-$uid and .Trash/$uid, with the spec's checks intact. Any other trash inside a root stays refused, and so does a root inside the trash. Media routes no longer serve files in trash folders, and a review with nothing to move shows a neutral disabled primary. --- .github/workflows/portability.yml | 39 ++ TESTING.md | 11 +- py/api/images.py | 49 +- py/api/library.py | 4 +- .../browser/workspace-images-delete.spec.mjs | 53 ++ tests/test_trash_freedesktop.py | 12 +- tests/test_trash_mount_root.py | 516 ++++++++++++++++++ utils/diagnostics.py | 9 +- utils/media_folders.py | 9 + utils/trash.py | 22 +- web/js/workspace/workspace.css | 2 +- 11 files changed, 708 insertions(+), 18 deletions(-) create mode 100644 tests/test_trash_mount_root.py diff --git a/.github/workflows/portability.yml b/.github/workflows/portability.yml index b591c0a0..417f1f03 100644 --- a/.github/workflows/portability.yml +++ b/.github/workflows/portability.yml @@ -29,6 +29,9 @@ env: # native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took # about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate # after three green runs. +# native-trash-linux (#543) has no green main runs yet: it installs the dev requirements and runs +# one test module, about 1 minute by the same estimate, so ceil(1.5 x 2) + 3 gives it 6 minutes. +# Recalibrate after three green runs. jobs: test: name: py${{ matrix.python-version }} on ${{ matrix.os }} @@ -104,3 +107,39 @@ jobs: env: PM_NATIVE_TRASH: macos run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v + + # The same tests as the fake mount table, on a real ext4 mount used as the output root (#543). A + # runner can mount with sudo; the tests run as the normal user, so owner and mode checks are real. + native-trash-linux: + name: native trash on a real mount + runs-on: ubuntu-latest + timeout-minutes: 6 + + steps: + - uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: "3.12" + cache: pip + cache-dependency-path: requirements-dev.txt + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements-dev.txt + + - name: Mount an ext4 image + run: | + truncate -s 64M "$RUNNER_TEMP/drive.img" + mkfs.ext4 -q "$RUNNER_TEMP/drive.img" + mkdir "$RUNNER_TEMP/drive" + sudo mount -o loop "$RUNNER_TEMP/drive.img" "$RUNNER_TEMP/drive" + sudo rmdir "$RUNNER_TEMP/drive/lost+found" + sudo chown "$(id -u):$(id -g)" "$RUNNER_TEMP/drive" + findmnt "$RUNNER_TEMP/drive" + + - name: Native trash proof + timeout-minutes: 4 + run: PM_TRASH_MOUNT="$RUNNER_TEMP/drive" python -m unittest tests.test_trash_mount_root -v diff --git a/TESTING.md b/TESTING.md index 22ed40bd..f43d6b0c 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2489,7 +2489,16 @@ files to the system trash. **No code path unlinks or overwrites a media file.** (`_collect_media_files`), the legacy auto-tag listing (`_collect_scan_files`) and the delete review (`trash-folder`), and the image monitor, which tests the folders below the watched folder holding the new file (a source named like a trash folder is still watched). Tests: `tests/test_media_folders.py`, - `tests/test_image_removal_gaps.py`. + `tests/test_image_removal_gaps.py`. Media routes answer not found for a path in such a + folder (serve, inspect, reveal, serving by image id), `/images/link` refuses it and a + prompt's media listing leaves it out (#543). +- **An output folder that is a mount point** (#543): the mount's own `.Trash-$uid` and + `.Trash/$uid` are allowed only for a root that is exactly that mount, with the spec's + checks intact. A root above the mount, a mount nested under the root, a source that is or + sits in the trash folder, and a link in place of the trash folder stay refused. Tests: + `tests/test_trash_mount_root.py`, on a fake mount table in the suite and, with + `PM_TRASH_MOUNT=`, in portability.yml's `native-trash-linux` job (loop-mounted + ext4). Locally, without root: `unshare -Urm` over a tmpfs. - **Trash backends** (`utils/trash.py`, no dependency): Linux and BSD implement the freedesktop Trash specification (home trash on the same mount, else the mount's `.Trash/$uid` or `.Trash-$uid`), reserve the name with `O_EXCL` and move with diff --git a/py/api/images.py b/py/api/images.py index 45daa952..e9cf9f8c 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -66,7 +66,11 @@ try: from ...database.models import normalize_image_path from ...database.run_attribution import recorded_output_lookup from ...utils.card_seed import card_seed, header_params, parsed_positive - from ...utils.media_folders import in_skipped_folder, is_skipped_folder + from ...utils.media_folders import ( + in_skipped_folder, + in_trash_folder, + is_skipped_folder, + ) from ...utils.parallel import map_parallel from ...utils.json_strict import dumps_strict from ...utils.inspection_response import ( @@ -92,7 +96,11 @@ except ImportError: from database.models import normalize_image_path from database.run_attribution import recorded_output_lookup from utils.card_seed import card_seed, header_params, parsed_positive - from utils.media_folders import in_skipped_folder, is_skipped_folder + from utils.media_folders import ( + in_skipped_folder, + in_trash_folder, + is_skipped_folder, + ) from utils.parallel import map_parallel from utils.json_strict import dumps_strict from utils.inspection_response import ( @@ -291,6 +299,21 @@ def _media_access_error(path, allowed_dirs): return None +def _record_in_trash_folder(path, roots) -> bool: + """Whether a recorded *path* lies in a trash folder below every root holding it. + + A source named like a trash folder is still a source, so a path it holds + without a trash folder below it stays servable (#543). + """ + real = os.path.realpath(path) + below = [] + for root in roots: + real_root = os.path.realpath(root) + if real.startswith(os.path.join(real_root, "")): + below.append(os.path.relpath(real, real_root)) + return bool(below) and all(in_trash_folder(relative) for relative in below) + + # Media types the gallery scans and thumbnails (a superset is never served). GALLERY_IMAGE_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".webp", ".gif"}) MEDIA_EXTENSIONS = GALLERY_IMAGE_EXTENSIONS | VIDEO_EXTENSIONS @@ -2640,7 +2663,10 @@ class ImageRoutesMixin: if denied is not None: return denied - if not image_path.is_file(): + if ( + _record_in_trash_folder(image_path, allowed_dirs) + or not image_path.is_file() + ): return web.json_response( {"success": False, "error": "Image file not found"}, status=404 ) @@ -2761,6 +2787,13 @@ class ImageRoutesMixin: if _has_traversal(filepath): return _forbidden() suffix = Path(filepath).suffix.lower() + if in_trash_folder(filepath): + # Thrown away, not an output (#543): as if it were not there. + return web.json_response( + {"success": False, "error": "Image file not found"}, + status=404, + headers={"Cache-Control": "no-store"}, + ) if suffix not in SERVABLE_EXTENSIONS and not ( # A JPEG XL preview is served from thumbnails/ only (#317). suffix in thumbnail_settings.THUMBNAIL_EXTENSIONS @@ -3593,7 +3626,9 @@ class ImageRoutesMixin: if denied is not None: return denied - if not os.path.isfile(image_path): + if _record_in_trash_folder(image_path, allowed_dirs) or not os.path.isfile( + image_path + ): return web.json_response( {"success": False, "error": "Image file not found"}, status=404 ) @@ -3673,6 +3708,8 @@ class ImageRoutesMixin: @staticmethod def _reveal_path(relative: str, root: Path) -> Path | None: """The real path of a regular media file under *root*, else None.""" + if in_trash_folder(relative): + return None resolved = ImageRoutesMixin._resolve_client_media_path(relative, [root]) if resolved is None or _media_access_error(resolved, [root]) is not None: return None @@ -3907,6 +3944,10 @@ class ImageRoutesMixin: if _is_thumbnail_cache_path(relative): # A preview is a placeholder: it carries no metadata (#317). return _forbidden("Previews are not inspected; inspect the original") + if in_trash_folder(relative): + return web.json_response( + {"success": False, "error": "Image file not found"}, status=404 + ) path = Path(roots[index]) / relative denied = _media_access_error(path, [roots[index]]) if denied is None: diff --git a/py/api/library.py b/py/api/library.py index 02ca0088..c19866cd 100644 --- a/py/api/library.py +++ b/py/api/library.py @@ -15,7 +15,7 @@ from pathlib import Path from aiohttp import web -from .images import _canonical, _media_access_error +from .images import _canonical, _media_access_error, _record_in_trash_folder from .pairs import pair_response try: @@ -268,6 +268,8 @@ class LibraryRoutesMixin: denied = _media_access_error(path, roots) if denied is not None or not path.is_file(): continue + if _record_in_trash_folder(path, roots): + continue for index, root in enumerate(roots): canonical_root = _canonical(root) if path.is_relative_to(canonical_root): diff --git a/tests/browser/workspace-images-delete.spec.mjs b/tests/browser/workspace-images-delete.spec.mjs index 7051166b..07d2abbf 100644 --- a/tests/browser/workspace-images-delete.spec.mjs +++ b/tests/browser/workspace-images-delete.spec.mjs @@ -169,6 +169,59 @@ test('without a chosen mode nothing can start a delete', async ({ page }) => { expect(applies).toEqual([]); }); +// #543 (D190): a review where nothing can move shows a primary that clearly cannot act: no danger colors, ignored presses. +test('a trash review with nothing to move has a clearly disabled primary', async ({ page }) => { + const before = await openImages(page); + await check(page, 0); + await page.route('**/images/delete/preview', async route => { + const response = await route.fetch(); + const plan = await response.json(); + const reason = "This drive's trash is inside an output folder."; + await route.fulfill({ response, json: { + ...plan, + trash: { count: 0 }, + trash_refused: { count: 1, counts: { 'trash-in-root': 1 }, rows: [{ name: 'a.png', code: 'trash-in-root', detail: reason }] }, + } }); + }); + const applies = []; + page.on('request', request => { if (request.url().includes('/maintenance/apply')) applies.push(request.url()); }); + await bar(page).getByRole('button', { name: 'Delete…', exact: true }).click(); + const dialog = review(page); + await expect(dialog).toBeVisible(); + await dialog.getByRole('radio', { name: /^Also move the files to the trash/ }).check(); + const primary = dialog.locator('.dialog-actions button.danger'); + await expect(primary).toHaveText('Move 0 images to the trash'); + await expect(primary).toHaveAttribute('aria-disabled', 'true'); + await expect(primary).toHaveCSS('cursor', 'not-allowed'); + const look = element => { const style = getComputedStyle(element); return [style.color, style.borderTopColor]; }; + const cancel = dialog.getByRole('button', { name: 'Cancel', exact: true }); + const [color, border] = await primary.evaluate(look); + const [cancelColor] = await cancel.evaluate(look); + const red = await page.evaluate(() => { + const probe = document.createElement('i'); + probe.style.color = 'var(--red)'; + document.querySelector('.ca-tools').append(probe); + const value = getComputedStyle(probe).color; + probe.remove(); + return value; + }); + expect(color).not.toBe(red); + expect(border).not.toBe(red); + expect(color).not.toBe(cancelColor); + await primary.hover({ force: true }); + await expect(primary).toHaveCSS('background-color', 'rgba(0, 0, 0, 0)'); + await primary.click({ force: true }); + await primary.focus(); + await page.keyboard.press('Enter'); + await expect(dialog).toBeVisible(); + expect(applies).toEqual([]); + await dialog.getByRole('radio', { name: /^Remove from the library only/ }).check(); + await expect(primary).not.toHaveAttribute('aria-disabled', 'true'); + await expect(primary).toHaveCSS('color', red); + await cancel.click(); + await expect(page.locator(CARDS)).toHaveCount(before); +}); + test('a chosen mode starts exactly one delete, with that mode, however often it is pressed', async ({ page }) => { const before = await openImages(page); await check(page, 0); diff --git a/tests/test_trash_freedesktop.py b/tests/test_trash_freedesktop.py index db54baf0..10f88741 100644 --- a/tests/test_trash_freedesktop.py +++ b/tests/test_trash_freedesktop.py @@ -114,11 +114,12 @@ class FreedesktopTrashTests(unittest.TestCase): self.assertEqual(os.listdir(elsewhere), []) def test_trash_inside_an_output_root_is_refused(self): - # The drive's top folder is itself an output root: its .Trash-uid - # would be listed as images. + # A root above the drive's top folder would hold the drive's .Trash-uid. + # (A root that is exactly the top folder is allowed: #543, + # tests/test_trash_mount_root.py.) path = self._file(self.root) with self.assertRaises(trash.TrashRefused) as caught: - self.backend.move(path, [self.drive]) + self.backend.move(path, [self.tmp]) self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT) self.assertTrue(os.path.exists(path)) self.assertFalse( @@ -147,9 +148,8 @@ class FreedesktopTrashTests(unittest.TestCase): def test_available_reports_without_writing(self): path = self._file(self.root) self.assertIsNone(self.backend.available(path, [self.root])) - self.assertEqual( - self.backend.available(path, [self.drive]), trash.TRASH_IN_ROOT - ) + self.assertEqual(self.backend.available(path, [self.tmp]), trash.TRASH_IN_ROOT) + self.assertIsNone(self.backend.available(path, [self.drive])) self.assertEqual(os.listdir(self.drive), ["output"]) # ── names and the no-replace rename ─────────────────────────────── diff --git a/tests/test_trash_mount_root.py b/tests/test_trash_mount_root.py new file mode 100644 index 00000000..f9b0fc78 --- /dev/null +++ b/tests/test_trash_mount_root.py @@ -0,0 +1,516 @@ +"""An output folder that is itself a mount point (#543). + +The freedesktop per-drive trash is then inside the output root. Only the mount's +own ``.Trash-$uid`` and ``.Trash/$uid`` are allowed, only for a root that is +exactly the mount root; every walker and media route ignores them. + +Every class runs on a disposable tree with a fake mount table. With +``PM_TRASH_MOUNT=`` the same tests run on that +real mount (``.github/workflows/portability.yml``, ``native-trash-linux``, or +locally ``unshare -Urm`` over a tmpfs); that folder is the only real place they +touch, and the fake-table-only cases skip. +""" + +import asyncio +import errno +import os +import shutil +import stat +import sys +import tempfile +import unittest +import urllib.parse +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +try: + from tests.disk_waits import DISK_BOUND_SECONDS + from tests.test_api_admin import _write_png + from tests.test_image_delete_routes import ImageDeleteTestCase +except ImportError: # discovered with tests/ as the top-level directory + from disk_waits import DISK_BOUND_SECONDS + from test_api_admin import _write_png + from test_image_delete_routes import ImageDeleteTestCase + +from py.api import images as images_module # noqa: E402 +from py.api.admin import _collect_media_files # noqa: E402 +from py.api.autotag_routes import AutotagRoutesMixin # noqa: E402 +from utils import trash # noqa: E402 +from utils.image_monitor import ImageGenerationHandler # noqa: E402 + +POSIX = os.name == "posix" and sys.platform != "darwin" +REAL_MOUNT = os.environ.get("PM_TRASH_MOUNT") +UID = os.getuid() if hasattr(os, "getuid") else -1 + + +def _empty(folder): + """Remove what a test left inside *folder*, never the folder itself.""" + for name in os.listdir(folder): + path = os.path.join(folder, name) + if os.path.isdir(path) and not os.path.islink(path): + os.chmod(path, 0o700) + shutil.rmtree(path) + else: + os.unlink(path) + + +def restore_as_a_file_manager_does(top, trash_dir, name): + """Put back one trashed file from its ``.trashinfo``, as the spec's readers do.""" + info_path = os.path.join(trash_dir, "info", name + ".trashinfo") + with open(info_path, encoding="utf-8") as handle: + lines = handle.read().splitlines() + recorded = next(line[5:] for line in lines if line.startswith("Path=")) + relative = os.fsdecode(urllib.parse.unquote_to_bytes(recorded)) + # A top-directory trash records a path relative to the mount (spec 1.0). + assert not os.path.isabs(relative), recorded + destination = os.path.join(top, relative) + os.makedirs(os.path.dirname(destination), exist_ok=True) + os.rename(os.path.join(trash_dir, "files", name), destination) + os.unlink(info_path) + return destination + + +@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend") +class MountRootBackendTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="pm-trash-root-") + self.addCleanup(shutil.rmtree, self.tmp, True) + self.data_home = os.path.join(self.tmp, "xdg") + os.makedirs(self.data_home) + if REAL_MOUNT: + self.assertTrue(os.path.ismount(REAL_MOUNT), "PM_TRASH_MOUNT is no mount") + self.drive = os.path.realpath(REAL_MOUNT) + self.addCleanup(_empty, self.drive) + self.backend = trash.FreedesktopBackend(data_home=self.data_home) + else: + self.drive = os.path.join(self.tmp, "drive") + os.makedirs(self.drive) + self.backend = self._fake_table() + self.dash = os.path.join(self.drive, f".Trash-{UID}") + + def _fake_table(self, *extra, uid=None): + mounts = ["/", os.path.realpath(self.drive), *extra] + return trash.FreedesktopBackend( + data_home=self.data_home, mounts=lambda: list(mounts), uid=uid + ) + + def _file(self, relative="a.png", content=b"png-bytes"): + path = os.path.join(self.drive, relative) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as handle: + handle.write(content) + return path + + def _assert_refused(self, path, roots, code): + self.assertEqual(self.backend.available(path, roots), code) + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, roots) + self.assertEqual(caught.exception.code, code) + self.assertTrue(os.path.lexists(path)) + + def _sticky_shared(self): + shared = os.path.join(self.drive, ".Trash") + os.mkdir(shared) + os.chmod(shared, 0o1777) + return shared + + # ── the two allowed folders ─────────────────────────────────────── + + def test_a_mount_root_output_uses_its_dash_uid_trash(self): + path = self._file() + identity = os.lstat(path) + self.assertIsNone(self.backend.available(path, [self.drive])) + self.assertEqual(os.listdir(self.drive), ["a.png"], "available wrote nothing") + label = self.backend.move(path, [self.drive]) + self.assertEqual(label, f".Trash-{UID} on its drive") + self.assertFalse(os.path.lexists(path)) + moved = os.lstat(os.path.join(self.dash, "files", "a.png")) + self.assertEqual( + (moved.st_dev, moved.st_ino), (identity.st_dev, identity.st_ino) + ) + self.assertEqual(stat.S_IMODE(os.lstat(self.dash).st_mode), 0o700) + + def test_a_mount_root_output_uses_a_sticky_shared_trash_first(self): + shared = self._sticky_shared() + path = self._file() + label = self.backend.move(path, [self.drive]) + self.assertEqual(label, f"{os.path.join('.Trash', str(UID))} on its drive") + self.assertTrue( + os.path.exists(os.path.join(shared, str(UID), "files", "a.png")) + ) + self.assertFalse(os.path.lexists(self.dash)) + + def test_a_nested_output_root_beside_the_mount_root_is_fine(self): + path = self._file("sub/a.png") + roots = [self.drive, os.path.join(self.drive, "sub")] + self.assertIsNone(self.backend.available(path, roots)) + self.backend.move(path, roots) + self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png"))) + + # ── the spec's checks stay ──────────────────────────────────────── + + def test_a_shared_trash_without_the_sticky_bit_is_passed_over(self): + shared = os.path.join(self.drive, ".Trash") + os.mkdir(shared, 0o777) + os.chmod(shared, 0o777) + path = self._file() + self.backend.move(path, [self.drive]) + self.assertEqual(os.listdir(shared), []) + self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png"))) + + def test_a_shared_trash_that_is_a_link_is_passed_over(self): + real = os.path.join(self.tmp, "sticky") + os.mkdir(real) + os.chmod(real, 0o1777) + os.symlink(real, os.path.join(self.drive, ".Trash")) + path = self._file() + self.backend.move(path, [self.drive]) + self.assertEqual(os.listdir(real), []) + self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png"))) + + def test_a_dash_uid_link_into_the_root_is_refused(self): + os.mkdir(os.path.join(self.drive, "images")) + os.symlink(os.path.join(self.drive, "images"), self.dash) + path = self._file("images/a.png") + self._assert_refused(path, [self.drive], trash.TRASH_IN_ROOT) + self.assertEqual(os.listdir(os.path.join(self.drive, "images")), ["a.png"]) + + def test_a_dash_uid_link_out_of_the_root_is_refused(self): + elsewhere = os.path.join(self.tmp, "elsewhere") + os.mkdir(elsewhere) + os.symlink(elsewhere, self.dash) + path = self._file() + self._assert_refused(path, [self.drive], trash.NO_TRASH) + self.assertEqual(os.listdir(elsewhere), []) + + def test_a_uid_link_in_the_shared_trash_is_never_followed(self): + shared = self._sticky_shared() + os.mkdir(os.path.join(self.drive, "images")) + os.symlink(os.path.join(self.drive, "images"), os.path.join(shared, str(UID))) + path = self._file("images/a.png") + self._assert_refused(path, [self.drive], trash.TRASH_IN_ROOT) + + def test_a_files_folder_that_is_a_link_is_refused(self): + os.mkdir(self.dash, 0o700) + elsewhere = os.path.join(self.tmp, "elsewhere") + os.mkdir(elsewhere) + os.symlink(elsewhere, os.path.join(self.dash, "files")) + path = self._file() + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, [self.drive]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + self.assertEqual(os.listdir(elsewhere), []) + + @unittest.skipIf(REAL_MOUNT, "needs the fake mount table") + def test_a_trash_folder_owned_by_someone_else_is_refused(self): + other = UID + 1 + os.mkdir(os.path.join(self.drive, f".Trash-{other}"), 0o700) + backend = self._fake_table(uid=other) + path = self._file() + self.assertEqual(backend.available(path, [self.drive]), trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + + def test_a_rename_across_drives_is_refused_and_nothing_is_copied(self): + def cross(*_args): + raise OSError(errno.EXDEV, "cross-device") + + backend = trash.FreedesktopBackend( + data_home=self.data_home, + mounts=lambda: ["/", os.path.realpath(self.drive)], + rename=cross, + ) + path = self._file() + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(path, [self.drive]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + self.assertEqual(os.listdir(os.path.join(self.dash, "files")), []) + self.assertEqual(os.listdir(os.path.join(self.dash, "info")), []) + + # ── any other trash inside a root stays refused ─────────────────── + + def test_a_root_above_the_mount_is_refused(self): + path = self._file() + above = os.path.dirname(self.drive) + self._assert_refused(path, [above], trash.TRASH_IN_ROOT) + self._assert_refused(path, [self.drive, above], trash.TRASH_IN_ROOT) + self.assertFalse(os.path.lexists(self.dash)) + + def test_a_source_named_like_the_trash_folder_is_refused(self): + source = self.dash + path = self._file(f".Trash-{UID}/a.png") + self._assert_refused(path, [source], trash.TRASH_IN_ROOT) + self._assert_refused(path, [self.drive, source], trash.TRASH_IN_ROOT) + + def test_a_source_inside_the_trash_folder_is_refused(self): + source = os.path.join(self.dash, "files") + os.makedirs(source) + path = self._file() + self._assert_refused(path, [self.drive, source], trash.TRASH_IN_ROOT) + self.assertEqual(os.listdir(source), []) + + def test_a_source_that_is_the_shared_trash_is_refused(self): + shared = self._sticky_shared() + path = self._file() + self._assert_refused(path, [self.drive, shared], trash.TRASH_IN_ROOT) + + @unittest.skipIf(REAL_MOUNT, "needs the fake mount table") + def test_a_mount_nested_under_the_root_is_refused(self): + # The file's mount is not the root: its trash is "any other location". + inner = os.path.join(self.drive, "inner") + os.mkdir(inner) + backend = self._fake_table(os.path.realpath(inner)) + path = self._file("inner/a.png") + self.assertEqual(backend.available(path, [self.drive]), trash.TRASH_IN_ROOT) + # ...and the same inner mount as its own root is fine. + self.assertIsNone(backend.available(path, [inner])) + + @unittest.skipIf(REAL_MOUNT, "needs the fake mount table") + def test_the_home_trash_inside_a_mount_root_output_stays_refused(self): + backend = trash.FreedesktopBackend( + data_home=os.path.join(self.drive, "xdg"), + mounts=lambda: ["/", os.path.realpath(self.drive)], + ) + path = self._file() + self.assertEqual(backend.available(path, [self.drive]), trash.TRASH_IN_ROOT) + + # ── restoring ───────────────────────────────────────────────────── + + def test_the_recorded_path_is_relative_to_the_mount(self): + path = self._file("sub dir/b c%é.png") + self.backend.move(path, [self.drive]) + with open( + os.path.join(self.dash, "info", "b c%é.png.trashinfo"), encoding="utf-8" + ) as handle: + info = handle.read() + self.assertIn("Path=sub%20dir/b%20c%25%C3%A9.png\n", info) + + def test_a_restored_file_returns_to_its_place_with_its_bytes(self): + path = self._file("sub/a.png", b"original") + identity = os.lstat(path).st_ino + self.backend.move(path, [self.drive]) + back = restore_as_a_file_manager_does(self.drive, self.dash, "a.png") + self.assertEqual(back, path) + self.assertEqual(os.lstat(path).st_ino, identity) + with open(path, "rb") as handle: + self.assertEqual(handle.read(), b"original") + self.assertEqual(os.listdir(os.path.join(self.dash, "files")), []) + + +@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend") +class MountRootLayoutTests(ImageDeleteTestCase): + """The real backend over the routes: nothing lists the trash, restore relinks.""" + + async def get_application(self): + app = await super().get_application() + self.data_home = Path(self.tmpdir) / "xdg" + self.data_home.mkdir() + if REAL_MOUNT: + self.assertTrue(os.path.ismount(REAL_MOUNT), "PM_TRASH_MOUNT is no mount") + self.output_dir = Path(os.path.realpath(REAL_MOUNT)) + self.addCleanup(_empty, str(self.output_dir)) + self.api._get_all_output_dirs = lambda: [self.output_dir] + mounts = None + else: + drive = os.path.realpath(self.output_dir) + mounts = lambda: ["/", drive] # noqa: E731 + trash.BACKEND = trash.FreedesktopBackend( + data_home=str(self.data_home), mounts=mounts + ) + self.trash_dir = self.output_dir / f".Trash-{UID}" + return app + + async def _trashed_one(self): + """``kept.png`` stays; ``gone.png`` was moved to the drive's trash.""" + _write_png(self.output_dir / "kept.png", prompt_text="a kept prompt") + _write_png(self.output_dir / "gone.png", prompt_text="a thrown prompt") + _plan, job = await self.delete(["gone.png"], "trash") + self.assertEqual(job.detail()["counters"]["trashed"], 1, job.detail()) + self.assertTrue((self.trash_dir / "files" / "gone.png").is_file()) + self.assertFalse((self.output_dir / "gone.png").exists()) + + async def _json(self, method, url, **kwargs): + response = await self.client.request(method, url, **kwargs) + self.assertLess(response.status, 500, await response.text()) + return await response.json() + + async def _scan(self, scope="all"): + answer = await self._json( + "POST", "/prompt_manager/jobs/scan", json={"scope": scope} + ) + task = self.api._scan_job.task + await asyncio.wait_for(asyncio.shield(task), DISK_BOUND_SECONDS) + return answer + + def _linked_paths(self): + return { + os.path.basename(row[0]) + for row in self._rows("SELECT image_path FROM generated_images") + } + + def _rows(self, sql): + import sqlite3 + + conn = sqlite3.connect(self.api.db.model.db_path) + try: + return conn.execute(sql).fetchall() + finally: + conn.close() + + async def test_the_review_can_move_files_on_a_mount_root_output(self): + _write_png(self.output_dir / "a.png", prompt_text="p") + plan = await self.review(["a.png"]) + self.assertEqual(plan["trash"], {"count": 1}) + self.assertEqual(plan["trash_refused"]["count"], 0) + self.assertFalse(self.trash_dir.exists(), "the review wrote nothing") + + async def test_refresh_listing_and_legacy_gallery_ignore_the_trash(self): + await self._trashed_one() + self.assertEqual(await self.listed(), ["kept.png"]) + rescan = await self._json("POST", "/prompt_manager/gallery/rescan") + self.assertNotIn(".Trash", str(rescan)) + legacy = await self._json("GET", "/prompt_manager/images/output") + self.assertNotIn(".Trash", str(legacy)) + folders = await self._json("GET", "/prompt_manager/gallery/subfolders") + self.assertNotIn(".Trash", str(folders)) + + async def test_scan_folder_and_its_walkers_ignore_the_trash(self): + await self._trashed_one() + await self._scan() + self.assertEqual(self._linked_paths(), {"kept.png"}) + collected = _collect_media_files([self.output_dir], [".png"]) + self.assertEqual([Path(p).name for p in collected], ["kept.png"]) + found, _truncated = AutotagRoutesMixin._collect_scan_files(self.output_dir) + self.assertEqual([p.name for p in found], ["kept.png"]) + + async def test_thumbnail_plans_ignore_the_trash(self): + await self._trashed_one() + thumbs = self.output_dir / "thumbnails" + for mode in ("missing", "rebuild"): + plan = self.api._plan_thumbnails(self.output_dir, thumbs, mode=mode) + sources = [src.name for src, _dst, _video in plan["targets"]] + self.assertEqual(sources, ["kept.png"]) + + async def test_the_autotag_listing_ignores_the_trash(self): + await self._trashed_one() + with patch.object( + self.api, "_find_comfyui_output_dir", return_value=str(self.output_dir) + ): + body = await self._json("GET", "/prompt_manager/scan_output_dir") + self.assertEqual([i["filename"] for i in body["images"]], ["kept.png"]) + + async def test_the_monitor_ignores_the_trash_with_nested_roots(self): + await self._trashed_one() + monitor = ImageGenerationHandler.__new__(ImageGenerationHandler) + monitor.supported_extensions = (".png",) + root = str(self.output_dir) + monitor.watched_roots = (root, os.path.join(root, "sub")) + inside = os.path.join(str(self.trash_dir), "files", "gone.png") + self.assertFalse(monitor.is_image_file(inside)) + self.assertTrue(monitor.is_image_file(os.path.join(root, "new.png"))) + self.assertTrue(monitor.is_image_file(os.path.join(root, "sub", "n.png"))) + + async def test_a_source_below_a_trash_named_folder_lists_on_its_own(self): + # The Codex P2 from #502: a source named like a trash folder stays a + # source, while the outer root does not list it. + source = self.output_dir / ".Trash-77" + source.mkdir() + _write_png(source / "mine.png", prompt_text="mine") + _write_png(self.output_dir / "kept.png", prompt_text="kept") + outer = [ + Path(e.path).name for e in images_module._walk_media(self.output_dir, 100) + ] + self.assertEqual(outer, ["kept.png"]) + inner = [Path(e.path).name for e in images_module._walk_media(source, 100)] + self.assertEqual(inner, ["mine.png"]) + monitor = ImageGenerationHandler.__new__(ImageGenerationHandler) + monitor.supported_extensions = (".png",) + monitor.watched_roots = (str(self.output_dir), str(source)) + self.assertTrue(monitor.is_image_file(str(source / "new.png"))) + self.assertFalse( + monitor.is_image_file(str(self.output_dir / ".Trash-1000" / "x.png")) + ) + + async def test_the_media_routes_do_not_serve_a_trashed_file(self): + await self._trashed_one() + relative = f".Trash-{UID}/files/gone.png" + roots = self.api._get_all_output_dirs() + scope = self.api._gallery_root_scope(roots[0], 0) + query = {"root": "0", "root_scope": scope} + served = await self.client.get( + f"/prompt_manager/images/serve/{relative}", params=query + ) + self.assertEqual(served.status, 404) + preview = await self.client.get( + f"/prompt_manager/images/serve/thumbnails/{relative}", params=query + ) + self.assertEqual(preview.status, 404) + self.assertFalse((self.output_dir / "thumbnails" / f".Trash-{UID}").exists()) + inspected = await self.client.get( + f"/prompt_manager/images/inspect/{relative}", params=query + ) + self.assertEqual(inspected.status, 404) + self.assertIsNone(self.api._reveal_path(relative, roots[0])) + self.assertIsNotNone(self.api._reveal_path("kept.png", roots[0])) + + async def test_a_record_pointing_into_the_trash_is_not_served(self): + await self._trashed_one() + inside = self.trash_dir / "files" / "gone.png" + image_id = self.api.db.link_image_to_prompt( + str(self.save("linked by hand")), str(inside) + ) + response = await self.client.get(f"/prompt_manager/images/{image_id}/file") + self.assertEqual(response.status, 404) + + async def test_a_trashed_file_cannot_be_linked_or_listed_under_a_prompt(self): + await self._trashed_one() + inside = self.trash_dir / "files" / "gone.png" + prompt_id = self.save("linked by hand") + refused = await self.client.post( + "/prompt_manager/images/link", + json={"prompt_id": prompt_id, "image_path": str(inside)}, + ) + self.assertEqual(refused.status, 404) + self.assertEqual(self._linked_paths(), set()) + # A record from before this check still lists nothing. + self.api.db.link_image_to_prompt(str(prompt_id), str(inside)) + page = self.api._library_media( + self._rows_as_records(), self.api._get_all_output_dirs() + ) + self.assertEqual(page, []) + + def _rows_as_records(self): + rows = self._rows("SELECT image_path FROM generated_images") + return [{"image_path": row[0]} for row in rows] + + async def test_a_source_named_like_a_trash_folder_is_still_served(self): + source = self.output_dir / ".Trash-77" + source.mkdir() + _write_png(source / "mine.png", prompt_text="mine") + self.api._get_all_output_dirs = lambda: [source] + scope = self.api._gallery_root_scope(source, 0) + served = await self.client.get( + "/prompt_manager/images/serve/mine.png", + params={"root": "0", "root_scope": scope}, + ) + self.assertEqual(served.status, 200) + + async def test_a_restored_file_lists_and_links_as_before(self): + await self._trashed_one() + await self._scan() + self.assertEqual(self._linked_paths(), {"kept.png"}) + back = restore_as_a_file_manager_does( + str(self.output_dir), str(self.trash_dir), "gone.png" + ) + self.assertEqual(back, str(self.output_dir / "gone.png")) + self.assertEqual(sorted(await self.listed()), ["gone.png", "kept.png"]) + await self._scan("new") + self.assertEqual(self._linked_paths(), {"gone.png", "kept.png"}) + prompts = self._rows( + "SELECT p.text FROM generated_images g JOIN prompts p ON p.id = g.prompt_id" + " WHERE g.image_path LIKE '%gone.png'" + ) + self.assertEqual(prompts, [("a thrown prompt",)]) diff --git a/utils/diagnostics.py b/utils/diagnostics.py index 7d1aa769..61172ece 100644 --- a/utils/diagnostics.py +++ b/utils/diagnostics.py @@ -37,6 +37,7 @@ from pathlib import Path from typing import Dict, Any from .logging_config import get_logger +from .media_folders import in_skipped_folder try: from ..database.workflow_documents import DocumentReader, resolve_image_row @@ -298,7 +299,13 @@ class GalleryDiagnostics: try: image_files = [] for ext in [".png", ".jpg", ".jpeg", ".webp"]: - image_files.extend(Path(abs_path).rglob(f"*{ext}")) + image_files.extend( + found + for found in Path(abs_path).rglob(f"*{ext}") + if not in_skipped_folder( + found.relative_to(abs_path).parts[:-1] + ) + ) self.logger.info(f" [IMG] Images found: {len(image_files)}") except Exception as e: self.logger.error(f" [FAIL] Error scanning: {e}") diff --git a/utils/media_folders.py b/utils/media_folders.py index 6052a001..41ec74af 100644 --- a/utils/media_folders.py +++ b/utils/media_folders.py @@ -26,6 +26,15 @@ def is_skipped_folder(name: str) -> bool: return name == PREVIEW_FOLDER or is_trash_folder(name) +def in_trash_folder(relative: str) -> bool: + """Whether a root-relative file path lies in a trash folder (#543). + + Takes either separator. Previews of it (``thumbnails/.Trash-1000/...``) count. + """ + folders = relative.replace("\\", "/").split("/")[:-1] + return any(is_trash_folder(part) for part in folders) + + def in_skipped_folder(parts) -> bool: """Whether any of the folder names *parts* is skipped.""" return any(is_skipped_folder(part) for part in parts) diff --git a/utils/trash.py b/utils/trash.py index 40a5a756..7fcf5121 100644 --- a/utils/trash.py +++ b/utils/trash.py @@ -6,7 +6,10 @@ Two rules hold for every backend: filesystem. Nothing is ever copied, unlinked or overwritten: a file that cannot be moved recoverably stays where it is, and ``TrashRefused`` says why. - **Never into a root.** A trash folder inside an output root would be listed - as images, so it is refused (``trash-in-root``). + as images, so it is refused (``trash-in-root``). The one exception (#543): for + an output root that is exactly a mount root, that mount's own ``.Trash-$uid`` + and ``.Trash/$uid``, which every walker and media route ignores + (``utils.media_folders``). Linux and BSD use :class:`FreedesktopBackend`, the freedesktop.org Trash specification 1.0: the home trash (``$XDG_DATA_HOME/Trash``) for a file on the @@ -184,10 +187,21 @@ def _canonical(path: str) -> str: return os.path.normcase(os.path.realpath(path)) -def _check_roots(trash_dir: str, roots) -> None: +def _check_roots(trash_dir: str, roots, top: str | None = None) -> None: + """Refuse a trash that a root would list, or that holds a root. + + *top* is the mount a per-drive *trash_dir* was built from. A root that is + exactly that mount may hold it, unless a link makes the trash really live + somewhere else (#543). Any other root holding it refuses, also beside one. + """ trash_key = _canonical(trash_dir) + top_key = None if top is None else _canonical(top) + unlinked = trash_key == os.path.normcase(trash_dir) for root in roots: - if _inside(trash_key, _canonical(str(root))): + root_key = _canonical(str(root)) + if _inside(root_key, trash_key): + raise TrashRefused(TRASH_IN_ROOT) + if _inside(trash_key, root_key) and not (unlinked and root_key == top_key): raise TrashRefused(TRASH_IN_ROOT) @@ -267,7 +281,7 @@ class FreedesktopBackend: candidates.append(os.path.join(shared, str(self._uid))) candidates.append(os.path.join(top, f".Trash-{self._uid}")) for trash_dir in candidates: - _check_roots(trash_dir, roots) + _check_roots(trash_dir, roots, top) if self._usable(trash_dir, create=create): label = f"{os.path.relpath(trash_dir, top)} on its drive" return trash_dir, top, label diff --git a/web/js/workspace/workspace.css b/web/js/workspace/workspace.css index 05df5256..8eb8c22c 100644 --- a/web/js/workspace/workspace.css +++ b/web/js/workspace/workspace.css @@ -478,7 +478,7 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc /* D147 (#430): a state that blocks Save shows its reason on its own row above the sticky actions, with Refresh for a changed source; the disabled actions look disabled; a refused capture's empty preview says so. */ .tagging-modal .dialog-actions [aria-disabled="true"], .tagging-modal .tag-model-panel [aria-disabled="true"] { opacity: .55; cursor: not-allowed; /* D147 */ } -#modal .dialog-actions .image-delete-start[aria-disabled="true"] { opacity: .55; cursor: not-allowed; /* #480 D159: no mode chosen yet */ } +#modal .dialog-actions .image-delete-start[aria-disabled="true"], #modal .dialog-actions .image-delete-start[aria-disabled="true"]:hover { color: var(--ca-muted); border-color: var(--ca-border); background: transparent; opacity: .55; cursor: not-allowed; /* #480 D159: no mode chosen yet, #543 D190: or nothing can move: neutral, no danger colors */ } .tagging-modal .tag-action-notice { flex: 1 0 100%; display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 8px; margin: 0; color: var(--ca-text); font-size: 12px; /* D147 */ } .tagging-modal .tag-action-notice > span { flex: 1 1 16em; } .tag-image-preview.tag-preview-empty { display: grid; place-items: center; padding: 8px; text-align: center; font-size: 12px; color: var(--ca-muted); background: var(--ca-field); box-sizing: border-box; /* D147 */ }