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 4e37aba5..07263550 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2497,7 +2497,25 @@ 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, resolved first so a link or another spelling cannot pass (#543): + `_media_access_error` is the one gate (by-id, inspect, reveal, link, rescope as + `file-missing`, a prompt's media listing); `/images/serve` and the autotag target ask + `in_trash_below_roots` too, `media_target` refuses by name (annotations, collections, + whose capture store denies links and aliases), and previews are neither made nor served, the fallback past the preview gate's cap + included (`_quick_source`). A History output in one answers `missing`, checked at the + candidate and again at the open (a preview can be a link). The legacy duplicate delete + removes exactly the folder's real path plus the name (a link itself, never its target) + and refuses if either spelling is in a trash folder; `file_problem` refuses too. The delete + and Return flows never use this gate. Windows spellings (trailing dots and spaces, case, + 8.3) run for real on the Windows job in `tests/test_trash_gate_routes.py`. +- **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/admin.py b/py/api/admin.py index 97765b52..64768544 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -36,10 +36,18 @@ except ImportError: try: from ...utils import exit_signal - from ...utils.media_folders import is_skipped_folder + from ...utils.media_folders import ( + in_trash_below_roots, + in_trash_folder, + is_skipped_folder, + ) except ImportError: from utils import exit_signal - from utils.media_folders import is_skipped_folder + from utils.media_folders import ( + in_trash_below_roots, + in_trash_folder, + is_skipped_folder, + ) try: from ...database.image_removal import mark_hides @@ -1089,13 +1097,21 @@ class AdminRoutesMixin: return result def _delete_one_duplicate(self, image_path, output_path): - """Delete one file and its thumbnail; return a failure reason or None.""" + """Delete one file and its thumbnail; return a failure reason or None. + + What is removed is exactly the folder's real path plus the name: a + link is removed itself, never its target, and the preview removed is + that name's. Both the link's place and what it leads to are checked. + """ file_path = Path(image_path) if not file_path.is_absolute(): file_path = output_path / file_path try: - rel_path = file_path.resolve().relative_to(output_path.resolve()) + real_root = output_path.resolve() + rel_path = file_path.resolve().relative_to(real_root) + exact = Path(os.path.realpath(file_path.parent)) / file_path.name + exact_rel = exact.relative_to(real_root) except ValueError: self.logger.warning( f"Attempted to delete file outside output directory: {image_path}" @@ -1104,15 +1120,19 @@ class AdminRoutesMixin: if file_path.suffix.lower() not in DELETABLE_MEDIA_EXTENSIONS: return "not a media file" - if "thumbnails" in rel_path.parts: + if "thumbnails" in rel_path.parts or "thumbnails" in exact_rel.parts: return "thumbnails are managed by the gallery" + if in_trash_below_roots(file_path, [output_path]) or in_trash_folder( + str(exact_rel) + ): + return "it is in a trash folder" # never unlinked from a trash (#543) if not file_path.is_file(): return "file not found" - os.remove(file_path) + os.remove(exact) self.logger.info(f"Deleted duplicate image: {image_path}") - _remove_thumbnail(output_path.resolve(), rel_path, self.logger) + _remove_thumbnail(real_root, exact_rel, self.logger) return None async def get_statistics(self, request): diff --git a/py/api/autotag_routes.py b/py/api/autotag_routes.py index d3ff1629..388e5e3a 100644 --- a/py/api/autotag_routes.py +++ b/py/api/autotag_routes.py @@ -24,9 +24,9 @@ from . import thumbnail_settings from .. import model_gate try: - from ...utils.media_folders import is_skipped_folder + from ...utils.media_folders import in_trash_below_roots, is_skipped_folder except ImportError: - from utils.media_folders import is_skipped_folder + from utils.media_folders import in_trash_below_roots, is_skipped_folder from aiohttp import web @@ -849,9 +849,12 @@ class AutotagRoutesMixin: """Return the real path when it is a media file inside an output directory.""" if Path(str(raw_path)).suffix.lower() not in _MEDIA_EXTENSIONS: return None - for output_dir in self._get_all_output_dirs(): + roots = list(self._get_all_output_dirs()) + for output_dir in roots: if path_is_within(raw_path, output_dir): real = os.path.realpath(raw_path) + if in_trash_below_roots(real, roots): + return None # thrown away, not an output (#543) return real if os.path.isfile(real) else None return None diff --git a/py/api/duplicate_review.py b/py/api/duplicate_review.py index 79f64ba0..d1c1a353 100644 --- a/py/api/duplicate_review.py +++ b/py/api/duplicate_review.py @@ -16,6 +16,11 @@ from pathlib import Path from .admin import DELETABLE_MEDIA_EXTENSIONS from .maintenance import DUPLICATE_FILES +try: + from ...utils.media_folders import in_trash_below_roots, is_trash_folder +except ImportError: + from utils.media_folders import in_trash_below_roots, is_trash_folder + # Keep entries one review takes; with paths of a few hundred characters the # list stays inside the preview body cap (DUPLICATE_PREVIEW_BODY_MAX, 1 MiB). MAX_KEEP = 2000 @@ -197,6 +202,8 @@ def file_problem( parts = Path(os.path.relpath(lexical, base)).parts if "thumbnails" in parts or ".." in parts: return OUTSIDE_SOURCE + if any(is_trash_folder(part) for part in parts[:-1]): + return OUTSIDE_SOURCE # never removed from a trash folder (#543) if os.path.splitext(lexical)[1].lower() not in DELETABLE_MEDIA_EXTENSIONS: return OUTSIDE_SOURCE try: @@ -212,6 +219,8 @@ def file_problem( try: if not _lexically_inside(_canonical_path(path), [_canonical_path(root)]): return OUTSIDE_SOURCE # a linked folder on the way + if in_trash_below_roots(path, [root]): + return OUTSIDE_SOURCE # a linked folder into a trash folder if keeper is not None: same = keeper_problem(path, root, info, keeper, keeper_root) if same: diff --git a/py/api/history_current.py b/py/api/history_current.py index bde14c24..d8c6b7d6 100644 --- a/py/api/history_current.py +++ b/py/api/history_current.py @@ -17,6 +17,7 @@ from aiohttp import web from . import thumbnail_settings from .images import _canonical, _public_path, SERVABLE_EXTENSIONS, VIDEO_EXTENSIONS +from .images import in_trash_below_roots from .images import existing_media_thumbnail, media_thumbnail_rel try: @@ -198,6 +199,9 @@ def resolve_current_candidate( if namespace is None or not namespace.is_dir(): return _unavailable("root_unavailable") target = _canonical(lexical) + if in_trash_below_roots(target, [namespace]): + # Every kind and the outputs listing come through here (#543). + return _unavailable("missing") if not _authorized(target, namespace, scope): return _unavailable("not_authorized") if target.suffix.lower() not in SERVABLE_EXTENSIONS: @@ -248,6 +252,10 @@ def _open_current_target( A *preview* (a file under ``thumbnails/``) may also be JPEG XL (#317).""" if _canonical(target) != target: raise CurrentLocationUnavailable("not_authorized", stale=True) + if in_trash_below_roots(target, [namespace]): + # Also at the open: the cover chooses its preview after the candidate + # check, and a preview can be a link into a trash folder (#543). + raise CurrentLocationUnavailable("missing") if not _authorized(target, namespace, scope): raise CurrentLocationUnavailable("not_authorized") suffix = target.suffix.lower() diff --git a/py/api/images.py b/py/api/images.py index b7786203..2ded1782 100644 --- a/py/api/images.py +++ b/py/api/images.py @@ -66,7 +66,12 @@ 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_below_roots, + 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 +97,12 @@ 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_below_roots, + in_trash_folder, + is_skipped_folder, + ) from utils.parallel import map_parallel from utils.json_strict import dumps_strict from utils.inspection_response import ( @@ -277,10 +287,20 @@ def _forbidden(message="Access denied"): return web.json_response({"success": False, "error": message}, status=403) -def _media_access_error(path, allowed_dirs): - """Return a 403 response when *path* must not be served, else None. +def _file_not_found(): + return web.json_response( + {"success": False, "error": "Image file not found"}, + status=404, + headers={"Cache-Control": "no-store"}, + ) - Fails closed: an empty *allowed_dirs* denies everything. + +def _media_access_error(path, allowed_dirs): + """The response refusing *path*, or None when it may be read. + + Fails closed: an empty *allowed_dirs* denies everything (403). A file in a + trash folder below its root is thrown away, not an output: a 404, as if it + were not there (#543), also through a link or another spelling. """ if Path(path).suffix.lower() not in SERVABLE_EXTENSIONS: return _forbidden("Only media files can be served") @@ -288,6 +308,8 @@ def _media_access_error(path, allowed_dirs): return _forbidden("No allowed image directories configured") if not any(_is_within(path, root) for root in allowed_dirs): return _forbidden() + if in_trash_below_roots(path, allowed_dirs): + return _file_not_found() return None @@ -979,6 +1001,8 @@ def _settled_entry(root, relative): return None if in_skipped_folder(folders) or len(folders) > MAX_SCAN_DEPTH: return None + if in_trash_below_roots(path, [root]): + return None # a linked folder on the way leads into a trash folder folder = os.fspath(root) for part in ("", *folders): folder = os.path.join(folder, part) if part else folder @@ -2761,6 +2785,9 @@ class ImageRoutesMixin: if _has_traversal(filepath): return _forbidden() suffix = Path(filepath).suffix.lower() + if in_trash_folder(filepath): + # Before any file access, so no preview is made for it (#543). + return _file_not_found() if suffix not in SERVABLE_EXTENSIONS and not ( # A JPEG XL preview is served from thumbnails/ only (#317). suffix in thumbnail_settings.THUMBNAIL_EXTENSIONS @@ -2784,6 +2811,8 @@ class ImageRoutesMixin: # a fall-through to the next root. if not _is_within(image_path, output_path): return _forbidden() + if in_trash_below_roots(image_path, output_dirs): + return _file_not_found() if image_path.is_file(): return self._file_response(image_path) @@ -2835,8 +2864,9 @@ class ImageRoutesMixin: if _is_thumbnail_cache_path(filepath): return _forbidden() image_path = root / filepath - if not _is_within(image_path, root): - return _forbidden() + denied = _media_access_error(image_path, [root]) + if denied is not None: + return denied if not image_path.is_file(): return web.json_response( {"success": False, "error": "Image file not found"}, status=404 diff --git a/py/api/media_target.py b/py/api/media_target.py index ee1fcadf..0c16d665 100644 --- a/py/api/media_target.py +++ b/py/api/media_target.py @@ -9,13 +9,15 @@ by the caller. import os import re -from .images import _forbidden, _has_traversal +from .images import _file_not_found, _forbidden, _has_traversal from .prompts import bad_request try: from ...utils.media_capture import CAPTURE_EXTENSIONS + from ...utils.media_folders import in_trash_folder except ImportError: from utils.media_capture import CAPTURE_EXTENSIONS + from utils.media_folders import in_trash_folder MAX_TARGET_PATH = 1024 TARGET_FIELDS = frozenset({"root_index", "root_scope", "path"}) @@ -29,7 +31,9 @@ def media_target(data: object, *, media_message: str = TAGGING_MEDIA_MESSAGE): Only the listing identity is accepted: an int root index, the root's 64-hex scope and a bounded root-relative POSIX media path, never an absolute, drive-qualified, backslash or ``..`` path (a 400 for a malformed - shape, a 403 for an unsafe or non-media path, carrying *media_message*). + shape, a 403 for an unsafe or non-media path, carrying *media_message*, a + 404 for a path in a trash folder, #543). A link or a short-name spelling + is refused later by the capture store's alias check. """ if not isinstance(data, dict) or set(data) != TARGET_FIELDS: return None, bad_request("target must name root_index, root_scope and path") @@ -51,4 +55,6 @@ def media_target(data: object, *, media_message: str = TAGGING_MEDIA_MESSAGE): return None, _forbidden() if os.path.splitext(parts[-1])[1].lower() not in CAPTURE_EXTENSIONS: return None, _forbidden(media_message) + if in_trash_folder(relative): + return None, _file_not_found() return (index, scope, relative), None diff --git a/py/api/source_rescope.py b/py/api/source_rescope.py index caa9bfa9..6f488013 100644 --- a/py/api/source_rescope.py +++ b/py/api/source_rescope.py @@ -41,6 +41,13 @@ def _coded(status: int, code: str, message: str) -> web.Response: ) +def _missing_if_not_found(response: web.Response) -> web.Response: + """A file in a trash folder answers as a missing file does (#543).""" + if response.status == 404: + return _coded(404, "file-missing", "Image file not found") + return response + + def _no_store(response: web.Response) -> web.Response: response.headers["Cache-Control"] = "no-store" return response @@ -71,7 +78,7 @@ class SourceRescopeMixin: path = root / relative denied = _media_access_error(path, [root]) if denied is not None: - return denied + return _missing_if_not_found(denied) try: mtime = path.stat().st_mtime if path.is_file() else None except OSError: @@ -89,7 +96,7 @@ class SourceRescopeMixin: return bad_request("Send exactly one target") target, error = media_target(data["target"]) if error is not None: - return error + return _missing_if_not_found(error) old_index, scope, relative = target if old_index > MAX_ROOT_INDEX: return bad_request("Invalid gallery root") diff --git a/py/api/thumbnail_auto.py b/py/api/thumbnail_auto.py index 3f425dd9..05af0c2d 100644 --- a/py/api/thumbnail_auto.py +++ b/py/api/thumbnail_auto.py @@ -67,6 +67,7 @@ from aiohttp import web from . import thumbnail_settings from .images import ( + in_trash_below_roots, MEDIA_EXTENSIONS, THUMBNAIL_CLEAR_TITLE, VIDEO_EXTENSIONS, @@ -224,6 +225,8 @@ def _quick_source(root, parts): for name in (base + extension, base + extension.upper()): path = folder / name if _regular_file(path) and _is_within(path, root): + if in_trash_below_roots(path, [root]): + return None # also the fallback past the gate's cap (#543) return path, extension in VIDEO_EXTENSIONS return None @@ -248,6 +251,8 @@ def _thumbnail_source(root, parts): found = _scan_folder(root.joinpath(*folder_parts), base, set(wanted)) if found is None or not _is_within(found, root): return None + if in_trash_below_roots(found, [root]): + return None rel = Path(*folder_parts, found.name) dst = root / media_thumbnail_rel(rel) if dst != _current_preview(root, parts): @@ -835,7 +840,7 @@ class AutoThumbnailMixin: if not folded.startswith(base) or not _is_within(path, root): continue rel = Path(os.path.relpath(path, root)) - if "thumbnails" in rel.parts[:-1]: + if "thumbnails" in rel.parts[:-1] or in_trash_below_roots(path, [root]): return None dst = root / media_thumbnail_rel(rel) if not _is_within(root / "thumbnails", root) or not _is_within( diff --git a/tests/browser/workspace-images-delete.spec.mjs b/tests/browser/workspace-images-delete.spec.mjs index c16247fa..9538c84a 100644 --- a/tests/browser/workspace-images-delete.spec.mjs +++ b/tests/browser/workspace-images-delete.spec.mjs @@ -184,6 +184,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_api_history_outputs.py b/tests/test_api_history_outputs.py index b3832030..a6705bb1 100644 --- a/tests/test_api_history_outputs.py +++ b/tests/test_api_history_outputs.py @@ -26,6 +26,7 @@ from tests.test_recorded_saver_anchor import LLM_TEXT, text_generate_run from utils.execution_contexts import ContextBundle from utils.execution_history_events import make_finish, make_start from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots +from py.api.images import media_thumbnail_rel from py.config import GalleryConfig @@ -212,6 +213,57 @@ class HistoryOutputAPITests(ImageAPITestCase): self.assertNotIn("root=", candidate["media"][field]) self.assertNotIn(str(self.comfy_dir), candidate["media"][field]) + async def test_an_output_in_a_trash_folder_is_read_by_no_kind(self): + # #543: inspection, media, thumbnail and the listing share one check. + path = make_png(self.output_dir / "image.png") + start = self.capture([self.ref()]) + media = await self.media(start) + trashed = self.output_dir / ".Trash-1000" / "files" + trashed.mkdir(parents=True) + os.rename(path, trashed / "image.png") + try: + os.symlink(trashed / "image.png", path) + except OSError: + if os.environ.get("GITHUB_ACTIONS"): + raise + self.skipTest("this machine cannot make symlinks") + for field in ("inspection_url", "url", "thumbnail_url"): + with self.subTest(field=field): + response = await self.current(media[field]) + self.assertIn(response.status, (404, 409, 410)) + await response.read() + response, page = await self.outputs(start) + self.assertEqual(response.status, 200, page) + candidate = page["items"][0]["current_candidate"] + self.assertEqual(candidate["status"], "unavailable") + self.assertEqual(candidate["reason"], "missing") + + async def test_a_preview_linked_into_a_trash_folder_does_not_open(self): + # #543: the cover picks its preview after the candidate check. + make_png(self.output_dir / "image.png") + trashed = self.output_dir / ".Trash-1000" / "files" + trashed.mkdir(parents=True) + make_png(trashed / "secret.png", color=(0, 0, 255)) + preview = self.output_dir / media_thumbnail_rel(Path("image.png")) + preview.parent.mkdir(parents=True, exist_ok=True) + try: + os.symlink(trashed / "secret.png", preview) + except OSError: + if os.environ.get("GITHUB_ACTIONS"): + raise + self.skipTest("this machine cannot make symlinks") + media = await self.media(self.capture([self.ref()])) + response = await self.current(media["thumbnail_url"]) + body = await response.read() + # Never the trashed bytes: the link is ignored or refused at the open. + self.assertNotEqual(body, (trashed / "secret.png").read_bytes()) + mod = importlib.import_module("py.api.history_current") + scope = mod.snapshot_current_output_scope(self.api) + with self.assertRaises(mod.CurrentLocationUnavailable): + mod._open_current_target( + mod._canonical(preview), self.output_dir, scope, True + ) + async def test_complete_zero_incomplete_unavailable_and_not_captured_are_distinct( self, ): 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_gate_routes.py b/tests/test_trash_gate_routes.py new file mode 100644 index 00000000..91f69a98 --- /dev/null +++ b/tests/test_trash_gate_routes.py @@ -0,0 +1,372 @@ +"""No route reads a file in a trash folder below its root (#543). + +Each route is tried with the trash folder named outright, through a linked file +and through a linked folder, and with a file it must still read as the control. +The Windows spellings run only on Windows, and they do run there: outside a +developer's machine a missing short name or symlink right is a failure. +""" + +import ctypes +import os +import unittest +import urllib.parse +from pathlib import Path +from unittest.mock import AsyncMock, patch + +try: + from tests.trash_route_audit import ( + FILE_READER_CALLS, + KNOWN_FILE_READERS, + ROUTE_AUDIT, + ) + 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 trash_route_audit import FILE_READER_CALLS, KNOWN_FILE_READERS, ROUTE_AUDIT + from test_api_admin import _write_png + from test_image_delete_routes import ImageDeleteTestCase + +from py.api.duplicate_review import OUTSIDE_SOURCE, file_problem +from py.api.images import media_thumbnail_rel +from py.api.thumbnail_auto import OVER, _quick_source, _thumbnail_source +from utils.media_folders import in_trash_below_roots, in_trash_folder, is_trash_folder + +ON_CI = bool(os.environ.get("GITHUB_ACTIONS")) +NAMED = (".Trash-1000/files/x.png", "$RECYCLE.BIN/x.png", "sub/.Trashes/501/x.png") +LINKED = ("link.png", "linkdir/x.png") + + +def _short_name(path): + """The 8.3 spelling of *path* (Windows only).""" + buffer = ctypes.create_unicode_buffer(1024) + ctypes.windll.kernel32.GetShortPathNameW(str(path), buffer, 1024) + return buffer.value + + +class _GateCase(ImageDeleteTestCase): + async def get_application(self): + app = await super().get_application() + self.root = self.output_dir + _write_png(self.root / "kept.png", prompt_text="kept") + for relative in NAMED: + (self.root / relative).parent.mkdir(parents=True, exist_ok=True) + _write_png(self.root / relative, prompt_text="thrown away") + return app + + def links(self): + """Links into the trash folder, or a skip where they cannot be made.""" + inside = self.root / ".Trash-1000" / "files" + try: + os.symlink(inside / "x.png", self.root / "link.png") + os.symlink(inside, self.root / "linkdir", target_is_directory=True) + except OSError: + if ON_CI: + raise + self.skipTest("this machine cannot make symlinks") + + def query(self): + scope = self.api._gallery_root_scope(self.api._get_all_output_dirs()[0], 0) + return {"root": "0", "root_scope": scope} + + def body(self, relative): + return self.target(relative) + + async def status(self, method, url, **kwargs): + response = await self.client.request(method, url, **kwargs) + await response.read() + return response.status + + async def serve(self, relative): + url = "/prompt_manager/images/serve/" + urllib.parse.quote(relative) + return await self.status("GET", url, params=self.query()) + + async def inspect(self, relative): + url = "/prompt_manager/images/inspect/" + urllib.parse.quote(relative) + return await self.status("GET", url, params=self.query()) + + async def rescope(self, relative): + response = await self.client.post( + "/prompt_manager/images/rescope", json={"target": self.body(relative)} + ) + return response.status, await response.json() + + async def autotag(self, relative): + """The target the legacy single/apply bodies pick, by path.""" + target, error = await self.api._select_autotag_target( + {"root_index": 0, "path": relative} + ) + return target, error + + async def assert_hidden_everywhere(self, relative): + self.assertEqual(await self.serve(relative), 404) + self.assertEqual(await self.inspect(relative), 404) + self.assertIsNone(self.api._reveal_path(relative, self.root)) + status, body = await self.rescope(relative) + self.assertEqual((status, body.get("code")), (404, "file-missing")) + target, error = await self.autotag(relative) + self.assertIsNone(target) + self.assertEqual(error.status, 403) + answer = await self.client.post( + "/prompt_manager/images/link", + json={"prompt_id": self.save("p"), "image_path": str(self.root / relative)}, + ) + self.assertEqual(answer.status, 404) + + +class TestTheGate(_GateCase): + async def test_the_control_file_still_reads_on_every_route(self): + self.assertEqual(await self.serve("kept.png"), 200) + self.assertEqual(await self.inspect("kept.png"), 200) + self.assertIsNotNone(self.api._reveal_path("kept.png", self.root)) + self.assertEqual((await self.rescope("kept.png"))[0], 200) + target, _error = await self.autotag("kept.png") + self.assertIsNotNone(target) + answer = await self.client.post( + "/prompt_manager/images/link", + json={ + "prompt_id": self.save("p"), + "image_path": str(self.root / "kept.png"), + }, + ) + self.assertEqual(answer.status, 200) + observed = await self.client.post( + "/prompt_manager/image-annotations/observe", + json={"target": self.body("kept.png")}, + ) + self.assertEqual(observed.status, 200) + added = await self.client.post( + "/prompt_manager/collections/items", + json={"name": "c", "items": [self.body("kept.png")]}, + ) + self.assertLess(added.status, 300) + + async def test_a_named_trash_folder_is_hidden_on_every_route(self): + for relative in NAMED: + with self.subTest(relative=relative): + await self.assert_hidden_everywhere(relative) + + async def test_a_linked_file_or_folder_is_hidden_on_every_route(self): + self.links() + for relative in LINKED: + with self.subTest(relative=relative): + await self.assert_hidden_everywhere(relative) + + async def test_serving_by_id_autotag_by_id_and_a_prompts_media(self): + self.links() + roots = self.api._get_all_output_dirs() + for relative in (*NAMED, *LINKED): + with self.subTest(relative=relative): + image_id = self.api.db.link_image_to_prompt( + str(self.save(f"linked {relative}")), str(self.root / relative) + ) + url = f"/prompt_manager/images/{image_id}/file" + self.assertEqual(await self.status("GET", url), 404) + target, error = await self.api._select_autotag_target( + {"image_id": image_id} + ) + self.assertIsNone(target) + self.assertEqual(error.status, 403) + records = [{"image_path": str(self.root / relative)}] + self.assertEqual(self.api._library_media(records, roots), []) + kept = [{"image_path": str(self.root / "kept.png")}] + self.assertEqual(len(self.api._library_media(kept, roots)), 1) + + async def test_annotations_refuse_a_trash_path_and_a_link(self): + self.links() + url = "/prompt_manager/image-annotations/observe" + for relative in NAMED: + with self.subTest(relative=relative): + body = {"target": self.body(relative)} + self.assertEqual(await self.status("POST", url, json=body), 404) + for relative in LINKED: + with self.subTest(relative=relative): + body = {"target": self.body(relative)} + status = await self.status("POST", url, json=body) + self.assertGreaterEqual(status, 400) + self.assertLess(status, 500) + + async def test_collections_refuse_a_trash_path(self): + for relative in NAMED: + with self.subTest(relative=relative): + answer = await self.client.post( + "/prompt_manager/collections/items", + json={"name": "c", "items": [self.body(relative)]}, + ) + self.assertEqual(answer.status, 404) + + async def test_the_exact_path_listing_does_not_name_it(self): + self.links() + for relative in (*NAMED, *LINKED): + with self.subTest(relative=relative): + response = await self.client.get( + "/prompt_manager/images/output", + params={"mode": "workspace", "path": relative}, + ) + page = await response.json() + self.assertEqual(page.get("images", []), []) + + async def test_no_preview_is_made_through_a_linked_folder(self): + self.links() + preview = Path(media_thumbnail_rel(Path("linkdir/x.png"))).parts + self.assertIsNone(_thumbnail_source(self.root, preview)) + kept = Path(media_thumbnail_rel(Path("kept.png"))).parts + self.assertIsNotNone(_thumbnail_source(self.root, kept)) + + async def test_the_preview_fallback_past_the_cap_is_gated_too(self): + # The gate's OVER branch looks the source up by name and serves it. + self.links() + gate = self.api._preview_gate() + with patch.object(gate, "run", AsyncMock(return_value=OVER)): + linked = media_thumbnail_rel(Path("linkdir/x.png")) + self.assertEqual(await self.serve(linked), 404) + kept = media_thumbnail_rel(Path("kept.png")) + self.assertEqual(await self.serve(kept), 200) + self.assertIsNone( + _quick_source( + self.root, Path(media_thumbnail_rel(Path("linkdir/x.png"))).parts + ) + ) + + async def test_the_legacy_duplicate_delete_never_unlinks_from_a_trash(self): + self.links() + for relative in (*NAMED, *LINKED): + with self.subTest(relative=relative): + reason = self.api._delete_one_duplicate(relative, self.root) + self.assertEqual(reason, "it is in a trash folder") + self.assertTrue((self.root / relative).exists()) + + async def test_a_link_inside_a_trash_folder_is_never_removed(self): + self.links() + link = self.root / ".Trash-1000" / "files" / "back.png" + os.symlink(self.root / "kept.png", link) + reason = self.api._delete_one_duplicate(".Trash-1000/files/back.png", self.root) + self.assertEqual(reason, "it is in a trash folder") + self.assertTrue(link.is_symlink()) + self.assertTrue((self.root / "kept.png").exists()) + + async def test_the_legacy_delete_removes_the_link_itself_and_its_own_preview(self): + self.links() + os.symlink(self.root / "kept.png", self.root / "alias.png") + thumbs = {} + for name in ("kept.png", "alias.png"): + thumbs[name] = self.root / media_thumbnail_rel(Path(name)) + thumbs[name].parent.mkdir(parents=True, exist_ok=True) + thumbs[name].write_bytes(b"preview") + self.assertIsNone(self.api._delete_one_duplicate("alias.png", self.root)) + self.assertFalse((self.root / "alias.png").is_symlink()) + self.assertTrue((self.root / "kept.png").is_file()) + self.assertFalse(thumbs["alias.png"].exists()) + self.assertTrue(thumbs["kept.png"].exists()) + + async def test_a_linked_folder_out_of_the_root_removes_nothing_there(self): + self.links() + outside = self.outside_dir / "elsewhere" + outside.mkdir(parents=True) + os.symlink(self.root / "kept.png", outside / "back.png") + os.symlink(outside, self.root / "out", target_is_directory=True) + reason = self.api._delete_one_duplicate("out/back.png", self.root) + self.assertEqual(reason, "outside output directory") + self.assertTrue((outside / "back.png").is_symlink()) + + async def test_a_duplicate_removal_review_refuses_a_trash_folder(self): + path = self.root / ".Trash-1000" / "files" / "x.png" + problem = file_problem( + path, self.root, None, path.stat().st_size, hash_check=False + ) + self.assertEqual(problem, OUTSIDE_SOURCE) + + async def test_a_source_named_like_a_trash_folder_still_reads(self): + source = self.root / ".Trash-77" + source.mkdir() + _write_png(source / "mine.png", prompt_text="mine") + self.api._get_all_output_dirs = lambda: [source] + self.assertEqual(await self.serve("mine.png"), 200) + self.assertEqual(await self.inspect("mine.png"), 200) + self.assertEqual((await self.rescope("mine.png"))[0], 200) + self.assertIsNotNone((await self.autotag("mine.png"))[0]) + + async def test_delete_still_moves_a_listed_file(self): + await self.review(["kept.png"]) + _plan, job = await self.delete(["kept.png"], "trash") + self.assertEqual(job.detail()["counters"]["trashed"], 1) + self.assertFalse((self.root / "kept.png").exists()) + + +class TestPredicates(unittest.TestCase): + def test_trailing_dots_and_spaces_are_ignored(self): + for name in ("$RECYCLE.BIN.", "$recycle.bin ", ".Trash-1000. ", ".Trashes..."): + self.assertTrue(is_trash_folder(name), name) + for name in ("trash.", "...", " ", ".Trash-x."): + self.assertFalse(is_trash_folder(name), name) + + def test_either_separator(self): + self.assertTrue(in_trash_folder("a\\.Trash\\b.png")) + self.assertFalse(in_trash_folder(".Trash")) + + def test_a_path_no_root_visibly_holds_fails_closed(self): + self.assertTrue(in_trash_below_roots("/x/.Trash-1/files/a.png", ["/other"])) + self.assertFalse(in_trash_below_roots("/x/a.png", ["/other"])) + + +@unittest.skipUnless(os.name == "nt", "Windows spellings of one folder") +class TestWindowsSpellings(_GateCase): + def spellings(self): + spelled = [ + "$RECYCLE.BIN./x.png", + "$recycle.bin/x.png", + ".TRASH-1000/files/x.png", + ".Trash-1000./files/x.png", + ] + short = _short_name(self.root / ".Trash-1000") + if os.path.basename(short).casefold() == ".trash-1000": + if ON_CI: + self.fail("the job's work disk gave no 8.3 short name") + self.skipTest("this disk makes no 8.3 short names") + spelled.append(os.path.basename(short) + "/files/x.png") + return spelled + + async def test_every_spelling_reaches_the_file_and_is_hidden(self): + for relative in self.spellings(): + with self.subTest(relative=relative): + self.assertTrue((self.root / relative).is_file(), "spelling is real") + await self.assert_hidden_everywhere(relative) + + async def test_a_folder_spelled_with_a_trailing_space_is_hidden_by_name(self): + # Measured on the Windows job (run 41123443497): unlike a trailing dot, + # this spelling does not reach the file through a longer path. The name + # check still answers for it, whatever a volume does. + await self.assert_hidden_everywhere("$RECYCLE.BIN /x.png") + + async def test_a_root_spelled_in_another_case_does_not_fail_open(self): + inside = self.root / ".Trash-1000" / "files" / "x.png" + self.assertTrue(in_trash_below_roots(inside, [str(self.root).upper()])) + kept = self.root / "kept.png" + self.assertFalse(in_trash_below_roots(kept, [str(self.root).upper()])) + + +class TestEveryRouteIsClassified(_GateCase): + async def test_a_new_route_must_say_how_it_treats_media_files(self): + registered = { + f"{route.method} {route.resource.canonical}" + for route in self.app.router.routes() + if route.method != "HEAD" + and route.resource is not None + and route.resource.canonical.startswith("/prompt_manager") + } + audited = {route for routes in ROUTE_AUDIT.values() for route in routes} + self.assertEqual( + registered ^ audited, + set(), + "add the route to tests/trash_route_audit.py with its media check", + ) + + def test_a_file_reader_in_the_api_layer_is_a_known_one(self): + # A new way to open a media file must be looked at for the same gate. + api = Path(__file__).resolve().parent.parent / "py" / "api" + found = set() + for module in api.glob("*.py"): + text = module.read_text(encoding="utf-8") + found.update( + f"{module.name}:{call}" for call in FILE_READER_CALLS if call in text + ) + self.assertEqual(found ^ KNOWN_FILE_READERS, set()) 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/tests/trash_route_audit.py b/tests/trash_route_audit.py new file mode 100644 index 00000000..a18e2387 --- /dev/null +++ b/tests/trash_route_audit.py @@ -0,0 +1,211 @@ +"""Every /prompt_manager route and how it treats a file in a trash folder (#543). + +``tests/test_trash_gate_routes.py`` fails when a route is missing here, so a new +route has to choose a group. ``FILE_READER_CALLS`` is the same idea for code that +opens files in ``py/api``. +""" + +FILE_READER_CALLS = ("FileResponse(", "Image.open(", "os.open(", "read_bytes(") + +KNOWN_FILE_READERS = frozenset( + { + "history_current.py:os.open(", + "images.py:FileResponse(", + "images.py:Image.open(", + "images.py:os.open(", + "__init__.py:Image.open(", + "logging_routes.py:os.open(", + "thumbnail_auto.py:FileResponse(", + "thumbnail_encode.py:Image.open(", + } +) + +ROUTE_AUDIT = { + "gate: _media_access_error, resolved": ( + "GET /prompt_manager/images/{image_id}/file", + "GET /prompt_manager/images/inspect/{image_path}", + "POST /prompt_manager/images/reveal", + "POST /prompt_manager/images/link", + "POST /prompt_manager/images/rescope", + "GET /prompt_manager/library/{positive_id}/images", + "GET /prompt_manager/library/{positive_id}", + "GET /prompt_manager/library", + ), + "in_trash_below_roots, resolved": ( + "GET /prompt_manager/images/serve/{filepath}", + "POST /prompt_manager/autotag/single", + "POST /prompt_manager/autotag/apply", + "GET /prompt_manager/history/runs/{run_id}/outputs", + "GET /prompt_manager/history/runs/{run_id}/output/current/media", + "GET /prompt_manager/history/runs/{run_id}/output/current/thumbnail", + "GET /prompt_manager/history/runs/{run_id}/output/current/inspection", + "POST /prompt_manager/delete_duplicate_images", + ), + "media_target by name, then the capture alias check or listed files": ( + "POST /prompt_manager/image-annotations/observe", + "POST /prompt_manager/image-annotations/suggest", + "PUT /prompt_manager/image-annotations/tags", + "GET /prompt_manager/image-annotations/capture/{token}", + "DELETE /prompt_manager/image-annotations/capture/{token}", + "POST /prompt_manager/jobs/image-tags", + "POST /prompt_manager/collections/items", + "DELETE /prompt_manager/collections/items", + "GET /prompt_manager/collections/membership", + "PUT /prompt_manager/images/category", + "DELETE /prompt_manager/images/category", + ), + "walker: utils/media_folders (listed files only)": ( + "GET /prompt_manager/images/output", + "POST /prompt_manager/gallery/rescan", + "GET /prompt_manager/gallery/subfolders", + "PUT /prompt_manager/images/favorite", + "DELETE /prompt_manager/images/favorite", + "GET /prompt_manager/images/categories", + "POST /prompt_manager/images/delete/preview", + "POST /prompt_manager/images/removed/return", + "POST /prompt_manager/scan", + "POST /prompt_manager/jobs/scan", + "GET /prompt_manager/scan_duplicates", + "POST /prompt_manager/scan_duplicates/stream", + "POST /prompt_manager/jobs/duplicates", + "GET /prompt_manager/maintenance/duplicates/{job_id}", + "GET /prompt_manager/scan_output_dir", + "POST /prompt_manager/autotag/start", + "POST /prompt_manager/autotag/batch/preview", + "POST /prompt_manager/jobs/autotag", + "POST /prompt_manager/jobs/thumbnails", + "GET /prompt_manager/images/thumbnails/plan", + "GET /prompt_manager/images/thumbnails/status", + "POST /prompt_manager/images/generate-thumbnails", + "POST /prompt_manager/images/generate-thumbnails/progress", + "GET /prompt_manager/diagnostics", + ), + "record text only, reads no file; its media is served through the gate": ( + "GET /prompt_manager/images/all", + "GET /prompt_manager/images/recent", + "GET /prompt_manager/images/search", + "GET /prompt_manager/images/prompt/{image_path}", + "GET /prompt_manager/prompts/{prompt_id}/images", + "DELETE /prompt_manager/images/{image_id}", + "GET /prompt_manager/recent", + "GET /prompt_manager/search", + "GET /prompt_manager/export", + "GET /prompt_manager/library/export", + "GET /prompt_manager/history/runs/{run_id}", + "GET /prompt_manager/review/items", + "GET /prompt_manager/review/items/{item_id}", + "POST /prompt_manager/review/items/{item_id}/decision", + "POST /prompt_manager/maintenance/preview", + "POST /prompt_manager/maintenance/apply", + ), + "no media file": ( + "DELETE /prompt_manager/collections/{collection_id}", + "DELETE /prompt_manager/delete/{prompt_id}", + "DELETE /prompt_manager/pairs/{pair_id}", + "DELETE /prompt_manager/prompts/{prompt_id}/tags", + "DELETE /prompt_manager/restore/preview/{token}", + "DELETE /prompt_manager/saved-views/{view_id}", + "DELETE /prompt_manager/tags/{tag_name}", + "GET /prompt_manager/admin", + "GET /prompt_manager/autotag/models", + "GET /prompt_manager/backup", + "GET /prompt_manager/backup/status", + "GET /prompt_manager/categories", + "GET /prompt_manager/collections", + "GET /prompt_manager/gallery", + "GET /prompt_manager/gallery.html", + "GET /prompt_manager/jobs", + "GET /prompt_manager/jobs/{job_id}", + "GET /prompt_manager/jobs/{job_id}/events", + "GET /prompt_manager/jobs/{job_id}/report", + "GET /prompt_manager/js/{filepath}", + "GET /prompt_manager/lib/{filepath}", + "GET /prompt_manager/library/options", + "GET /prompt_manager/library/{positive_id}/history", + "GET /prompt_manager/library/{positive_id}/history/pairs/{pair_id}/newest", + "GET /prompt_manager/logs", + "GET /prompt_manager/logs/config", + "GET /prompt_manager/logs/download/{filename}", + "GET /prompt_manager/logs/files", + "GET /prompt_manager/logs/stats", + "GET /prompt_manager/lora/detect", + "GET /prompt_manager/lora/status", + "GET /prompt_manager/lora/trigger-words", + "GET /prompt_manager/pairs", + "GET /prompt_manager/pairs/export", + "GET /prompt_manager/pairs/{pair_id}", + "GET /prompt_manager/review/events", + "GET /prompt_manager/saved-views", + "GET /prompt_manager/scan/status", + "GET /prompt_manager/scan_duplicates/status", + "GET /prompt_manager/settings", + "GET /prompt_manager/stats", + "GET /prompt_manager/subfolders", + "GET /prompt_manager/system/diagnostics", + "GET /prompt_manager/system/health", + "GET /prompt_manager/system/library", + "GET /prompt_manager/system/overview", + "GET /prompt_manager/system/statistics", + "GET /prompt_manager/tags", + "GET /prompt_manager/tags/filter", + "GET /prompt_manager/tags/stats", + "GET /prompt_manager/tags/suggest", + "GET /prompt_manager/tags/{tag_name}/prompts", + "GET /prompt_manager/test", + "GET /prompt_manager/vocabulary/categories", + "GET /prompt_manager/vocabulary/tags", + "GET /prompt_manager/vocabulary/tags/{tag_id}/impact", + "GET /prompt_manager/web", + "GET /prompt_manager/workspace", + "PATCH /prompt_manager/collections/{collection_id}", + "PATCH /prompt_manager/library/{positive_id}", + "PATCH /prompt_manager/pairs/{pair_id}", + "PATCH /prompt_manager/saved-views/{view_id}", + "PATCH /prompt_manager/vocabulary/tags/{tag_id}", + "POST /prompt_manager/autotag/download/{model_type}", + "POST /prompt_manager/autotag/unload", + "POST /prompt_manager/bulk/category", + "POST /prompt_manager/bulk/delete", + "POST /prompt_manager/bulk/tags", + "POST /prompt_manager/cleanup", + "POST /prompt_manager/diagnostics/test-link", + "POST /prompt_manager/images/clear-thumbnails", + "POST /prompt_manager/jobs/model-download", + "POST /prompt_manager/jobs/{job_id}/cancel", + "POST /prompt_manager/library", + "POST /prompt_manager/library/delete", + "POST /prompt_manager/library/delete/preview", + "POST /prompt_manager/library/restore", + "POST /prompt_manager/library/{positive_id}/variants", + "POST /prompt_manager/logs/config", + "POST /prompt_manager/logs/truncate", + "POST /prompt_manager/lora/enable", + "POST /prompt_manager/lora/import/apply", + "POST /prompt_manager/lora/import/preview", + "POST /prompt_manager/lora/refresh-cache", + "POST /prompt_manager/lora/scan", + "POST /prompt_manager/maintenance", + "POST /prompt_manager/pairs", + "POST /prompt_manager/prompts/tags", + "POST /prompt_manager/prompts/{prompt_id}/tags", + "POST /prompt_manager/restore", + "POST /prompt_manager/restore/apply", + "POST /prompt_manager/restore/preview", + "POST /prompt_manager/save", + "POST /prompt_manager/saved-views", + "POST /prompt_manager/scan/cancel", + "POST /prompt_manager/settings", + "POST /prompt_manager/settings/validate-source", + "POST /prompt_manager/system/health", + "POST /prompt_manager/system/library/keep", + "POST /prompt_manager/system/library/move", + "POST /prompt_manager/tags/merge", + "POST /prompt_manager/vocabulary/categories/apply", + "POST /prompt_manager/vocabulary/categories/preview", + "POST /prompt_manager/vocabulary/tags", + "POST /prompt_manager/vocabulary/tags/{tag_id}/delete", + "PUT /prompt_manager/prompts/{prompt_id}", + "PUT /prompt_manager/prompts/{prompt_id}/rating", + "PUT /prompt_manager/tags/{tag_name}", + ), +} 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..a3f1a0d1 100644 --- a/utils/media_folders.py +++ b/utils/media_folders.py @@ -6,16 +6,22 @@ inside an output root (``D:\\$RECYCLE.BIN`` when ``D:\\`` is a root, the user threw away: they are not outputs, so they are never listed, scanned, linked, tagged or previewed. Every walker asks these predicates instead of testing names itself; only folders *below* a root are tested. + +``in_trash_below_roots`` is the check for a path someone names (#543): it +resolves the path first, so a link or another spelling cannot get past it. """ +import os + TRASH_FOLDER_NAMES = frozenset({".trash", ".trashes", "$recycle.bin"}) PREVIEW_FOLDER = "thumbnails" def is_trash_folder(name: str) -> bool: """A trash folder name, in any case: ``.Trash``, ``.Trash-``, - ``.Trashes`` or ``$RECYCLE.BIN``.""" - folded = name.casefold() + ``.Trashes`` or ``$RECYCLE.BIN``. Windows ignores a name's trailing dots + and spaces, so ``$RECYCLE.BIN.`` is that folder; they are ignored here too.""" + folded = name.rstrip(" .").casefold() if folded in TRASH_FOLDER_NAMES: return True return folded.startswith(".trash-") and folded[7:].isdigit() @@ -26,6 +32,34 @@ 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_trash_below_roots(path, roots) -> bool: + """Whether *path*, resolved, is in a trash folder below every root holding it. + + A source named like a trash folder is still a source, so a file it holds + with no trash folder below it is not in one. Paths compare resolved and + normcased. A path no root visibly holds tests its whole path: a spelling + difference must not fail open. + """ + real = os.path.normcase(os.path.realpath(path)) + below = [] + for root in roots: + real_root = os.path.normcase(os.path.realpath(root)) + if real.startswith(os.path.join(real_root, "")): + below.append(os.path.relpath(real, real_root)) + if not below: + return in_trash_folder(real) + return all(in_trash_folder(relative) for relative in below) + + 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 c5ec0ec9..bfcd4df7 100644 --- a/web/js/workspace/workspace.css +++ b/web/js/workspace/workspace.css @@ -514,7 +514,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 */ }