fix(delete): the legacy delete removes exactly the checked path (#543)
It tested the resolved path but removed the lexical one, so a link could make the two differ. It now removes the folder's real path plus the name, refuses if either spelling is in a trash folder, and removes that name's preview. History checks a trash folder again at the open, since a preview can be a link.
This commit is contained in:
+4
-1
@@ -2495,7 +2495,10 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
|
||||
`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`. The delete
|
||||
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
|
||||
|
||||
+26
-8
@@ -36,10 +36,18 @@ except ImportError:
|
||||
|
||||
try:
|
||||
from ...utils import exit_signal
|
||||
from ...utils.media_folders import in_trash_below_roots, 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 in_trash_below_roots, 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,17 +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]):
|
||||
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):
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -252,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()
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -237,6 +238,32 @@ class HistoryOutputAPITests(ImageAPITestCase):
|
||||
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,
|
||||
):
|
||||
|
||||
@@ -26,6 +26,7 @@ except ImportError: # discovered with tests/ as the top-level directory
|
||||
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
|
||||
@@ -234,6 +235,46 @@ class TestTheGate(_GateCase):
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user