diff --git a/tests/test_trash_freedesktop.py b/tests/test_trash_freedesktop.py new file mode 100644 index 00000000..a0f3f3ec --- /dev/null +++ b/tests/test_trash_freedesktop.py @@ -0,0 +1,298 @@ +"""The freedesktop trash backend (#480): rename into a trash or refuse, never unlink. + +Every test builds a disposable tree: a fake ``XDG_DATA_HOME`` and, through the +backend's ``mounts`` seam, a fake "drive" mount point, so the real trash of the +user running the tests is never touched. Moves are real renames on one +filesystem. +""" + +import os +import shutil +import stat +import sys +import tempfile +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from utils import trash # noqa: E402 + +POSIX = os.name == "posix" and sys.platform != "darwin" + + +@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend") +class FreedesktopTrashTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="pm-trash-") + self.addCleanup(shutil.rmtree, self.tmp, True) + self.data_home = os.path.join(self.tmp, "xdg") + self.drive = os.path.join(self.tmp, "drive") + self.root = os.path.join(self.drive, "output") + self.home_root = os.path.join(self.tmp, "home-output") + for folder in (self.data_home, self.root, self.home_root): + os.makedirs(folder) + self.mounts = ["/", os.path.realpath(self.drive)] + self.backend = trash.FreedesktopBackend( + data_home=self.data_home, mounts=lambda: list(self.mounts) + ) + + def _file(self, folder, name="a.png", content=b"png-bytes"): + path = os.path.join(folder, name) + with open(path, "wb") as handle: + handle.write(content) + return path + + def _info(self, trash_dir, name): + with open(os.path.join(trash_dir, "info", name + ".trashinfo")) as handle: + return handle.read() + + # ── choosing the trash ──────────────────────────────────────────── + + def test_file_on_the_home_mount_goes_to_the_home_trash(self): + path = self._file(self.home_root) + label = self.backend.move(path, [self.home_root]) + home_trash = os.path.join(self.data_home, "Trash") + self.assertEqual(label, trash.HOME_TRASH_LABEL) + self.assertFalse(os.path.exists(path)) + with open(os.path.join(home_trash, "files", "a.png"), "rb") as handle: + self.assertEqual(handle.read(), b"png-bytes") + info = self._info(home_trash, "a.png") + self.assertIn("[Trash Info]\n", info) + self.assertIn("Path=" + os.path.realpath(path) + "\n", info) + self.assertRegex(info, r"DeletionDate=\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\n") + + def test_file_on_another_mount_uses_its_trash_dash_uid(self): + path = self._file(self.root, "b c%.png") + self.backend.move(path, [self.root]) + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + info_mode = stat.S_IMODE(os.lstat(top).st_mode) + self.assertEqual(info_mode, 0o700) + self.assertTrue(os.path.exists(os.path.join(top, "files", "b c%.png"))) + # A top-directory trash records the path relative to its top folder, + # percent-encoded. + self.assertIn("Path=output/b%20c%25.png\n", self._info(top, "b c%.png")) + + def test_sticky_dot_trash_folder_is_used(self): + shared = os.path.join(self.drive, ".Trash") + os.mkdir(shared) + os.chmod(shared, 0o1777) + path = self._file(self.root) + self.backend.move(path, [self.root]) + self.assertTrue( + os.path.exists(os.path.join(shared, str(os.getuid()), "files", "a.png")) + ) + + def test_dot_trash_without_sticky_bit_is_skipped(self): + shared = os.path.join(self.drive, ".Trash") + os.mkdir(shared, 0o777) + os.chmod(shared, 0o777) + path = self._file(self.root) + self.backend.move(path, [self.root]) + self.assertFalse(os.path.exists(os.path.join(shared, str(os.getuid())))) + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + self.assertTrue(os.path.exists(os.path.join(top, "files", "a.png"))) + + def test_dot_trash_that_is_a_link_is_skipped(self): + elsewhere = os.path.join(self.tmp, "elsewhere") + os.mkdir(elsewhere) + os.chmod(elsewhere, 0o1777) + os.symlink(elsewhere, os.path.join(self.drive, ".Trash")) + path = self._file(self.root) + self.backend.move(path, [self.root]) + self.assertEqual(os.listdir(elsewhere), []) + + def test_trash_dash_uid_that_is_a_link_is_refused(self): + elsewhere = os.path.join(self.tmp, "elsewhere") + os.mkdir(elsewhere) + os.symlink(elsewhere, os.path.join(self.drive, f".Trash-{os.getuid()}")) + path = self._file(self.root) + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, [self.root]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + 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. + path = self._file(self.root) + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, [self.drive]) + self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT) + self.assertTrue(os.path.exists(path)) + self.assertFalse( + os.path.exists(os.path.join(self.drive, f".Trash-{os.getuid()}")) + ) + + def test_home_trash_inside_an_output_root_is_refused(self): + path = self._file(self.home_root) + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, [self.home_root, self.tmp]) + self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT) + self.assertTrue(os.path.exists(path)) + + def test_read_only_top_folder_is_no_trash(self): + path = self._file(self.root) + os.chmod(self.drive, 0o555) + self.addCleanup(os.chmod, self.drive, 0o755) + if os.access(self.drive, os.W_OK): + self.skipTest("running as root: permissions are not enforced") + self.assertEqual(self.backend.available(path, [self.root]), trash.NO_TRASH) + with self.assertRaises(trash.TrashRefused) as caught: + self.backend.move(path, [self.root]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + + 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(os.listdir(self.drive), ["output"]) + + # ── names and the no-replace rename ─────────────────────────────── + + def test_existing_name_in_the_trash_is_never_replaced(self): + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + os.makedirs(os.path.join(top, "files"), mode=0o700) + os.makedirs(os.path.join(top, "info"), mode=0o700) + os.chmod(top, 0o700) + # A stray trashed file with no info file (another tool crashed). + stray = self._file(os.path.join(top, "files"), "a.png", b"older-file") + path = self._file(self.root) + self.backend.move(path, [self.root]) + with open(stray, "rb") as handle: + self.assertEqual(handle.read(), b"older-file") + self.assertTrue(os.path.exists(os.path.join(top, "files", "a.1.png"))) + self.assertTrue(os.path.exists(os.path.join(top, "info", "a.1.png.trashinfo"))) + self.assertFalse(os.path.exists(os.path.join(top, "info", "a.png.trashinfo"))) + + def test_existing_info_name_is_skipped(self): + path = self._file(self.root) + self.backend.move(path, [self.root]) + second = self._file(self.root) + self.backend.move(second, [self.root]) + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + self.assertEqual( + sorted(os.listdir(os.path.join(top, "files"))), ["a.1.png", "a.png"] + ) + + def test_missing_no_replace_rename_is_no_trash(self): + path = self._file(self.root) + for error in (38, 22): # ENOSYS, EINVAL + with self.subTest(error=error): + backend = trash.FreedesktopBackend( + data_home=self.data_home, + mounts=lambda: list(self.mounts), + rename=mock.Mock(side_effect=OSError(error, "no")), + ) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(path, [self.root]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + self.assertEqual(os.listdir(os.path.join(top, "info")), []) + + def test_cross_device_rename_is_no_trash_and_never_copies(self): + path = self._file(self.root) + backend = trash.FreedesktopBackend( + data_home=self.data_home, + mounts=lambda: list(self.mounts), + rename=mock.Mock(side_effect=OSError(18, "EXDEV")), + ) + with mock.patch("shutil.move") as move, mock.patch("shutil.copy2") as copy: + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(path, [self.root]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + move.assert_not_called() + copy.assert_not_called() + self.assertTrue(os.path.exists(path)) + + def test_other_rename_failure_is_failed_and_removes_only_our_info(self): + path = self._file(self.root) + backend = trash.FreedesktopBackend( + data_home=self.data_home, + mounts=lambda: list(self.mounts), + rename=mock.Mock(side_effect=OSError(5, "EIO")), + ) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(path, [self.root]) + self.assertEqual(caught.exception.code, trash.FAILED) + top = os.path.join(self.drive, f".Trash-{os.getuid()}") + self.assertEqual(os.listdir(os.path.join(top, "info")), []) + self.assertTrue(os.path.exists(path)) + + def test_never_removes_or_moves_the_source(self): + path = self._file(self.root) + real_unlink = os.unlink + + def guarded_unlink(target, *args, **kwargs): + # Only our own .trashinfo may ever be removed. + self.assertTrue(str(target).endswith(".trashinfo"), target) + return real_unlink(target, *args, **kwargs) + + backend = trash.FreedesktopBackend( + data_home=self.data_home, + mounts=lambda: list(self.mounts), + rename=mock.Mock(side_effect=OSError(5, "EIO")), + ) + with ( + mock.patch("os.unlink", guarded_unlink), + mock.patch("os.remove", guarded_unlink), + mock.patch("shutil.move") as move, + mock.patch("os.rename") as rename, + ): + with self.assertRaises(trash.TrashRefused): + backend.move(path, [self.root]) + move.assert_not_called() + rename.assert_not_called() + self.assertTrue(os.path.exists(path)) + + def test_mountinfo_parsing_decodes_escapes(self): + text = ( + "36 35 98:0 / / rw - ext4 /dev/root rw\n" + "37 36 98:1 / /mnt/my\\040drive rw - ext4 /dev/sdb1 rw\n" + ) + self.assertEqual(trash.parse_mountinfo(text), ["/", "/mnt/my drive"]) + + def test_mount_of_picks_the_longest_containing_mount(self): + mounts = ["/", "/mnt", "/mnt/data"] + self.assertEqual(trash.mount_of("/mnt/data/x/y.png", mounts), "/mnt/data") + self.assertEqual(trash.mount_of("/mnt/datax/y.png", mounts), "/mnt") + self.assertEqual(trash.mount_of("/home/u/y.png", mounts), "/") + + +class UnavailableBackendTests(unittest.TestCase): + def test_reports_no_trash_and_leaves_the_file(self): + backend = trash.UnavailableBackend() + with tempfile.TemporaryDirectory() as folder: + path = os.path.join(folder, "a.png") + with open(path, "wb") as handle: + handle.write(b"x") + self.assertEqual(backend.available(path, [folder]), trash.NO_TRASH) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(path, [folder]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertTrue(os.path.exists(path)) + + def test_windows_and_macos_have_no_trash_yet(self): + for platform in ("win32", "darwin"): + with self.subTest(platform=platform): + self.assertIsInstance( + trash.default_backend(platform), trash.UnavailableBackend + ) + self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend) + + def test_module_functions_use_the_replaceable_backend(self): + fake = mock.Mock() + fake.available.return_value = "network" + fake.move.return_value = "label" + with mock.patch.object(trash, "BACKEND", fake): + self.assertEqual(trash.trash_available("/x/a.png", ["/x"]), "network") + self.assertEqual(trash.move_to_trash("/x/a.png", ["/x"]), "label") + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/trash.py b/utils/trash.py new file mode 100644 index 00000000..10ce87fb --- /dev/null +++ b/utils/trash.py @@ -0,0 +1,360 @@ +"""Move one media file into the system trash, or refuse (#480). + +Two rules hold for every backend: + +- **Rename or fail.** The only primitive is a no-replace rename within one + 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``). + +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 +same mount, else the file's mount's ``.Trash/$uid`` (only under a real, sticky +``.Trash``) or ``.Trash-$uid``. Names are reserved by creating the +``.trashinfo`` with ``O_EXCL`` and the file is moved with +``renameat2(RENAME_NOREPLACE)``; without that call the move is refused. +Windows and macOS have no backend yet (:class:`UnavailableBackend`): their +own PRs add one with a native proof. + +``BACKEND`` is the one replaceable seam; tests install a fake. +""" + +from __future__ import annotations + +import ctypes +import datetime +import errno +import os +import stat +import sys +import urllib.parse + +NO_TRASH = "no-trash" +NETWORK = "network" +TRASH_IN_ROOT = "trash-in-root" +PATH_TOO_LONG = "path-too-long" +LOCKED = "locked" +FAILED = "failed" +CODES = frozenset({NO_TRASH, NETWORK, TRASH_IN_ROOT, PATH_TOO_LONG, LOCKED, FAILED}) + +HOME_TRASH_LABEL = "home trash" +RENAME_NOREPLACE = 1 +MAX_NAME_ATTEMPTS = 1000 +_DIR_FLAGS = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) +# rename() errors that mean "this trash cannot take it", never a copy. +_NO_TRASH_ERRNOS = { + errno.ENOSYS, + errno.EINVAL, + errno.EXDEV, + errno.EACCES, + errno.EPERM, + errno.EROFS, + getattr(errno, "ENOTSUP", errno.EOPNOTSUPP), + errno.EOPNOTSUPP, +} + + +class TrashRefused(Exception): + """The file was not moved; ``code`` is one of ``CODES``.""" + + def __init__(self, code: str, detail: str = ""): + super().__init__(detail or code) + self.code = code + + +def parse_mountinfo(text: str) -> list[str]: + """Mount points of ``/proc/self/mountinfo`` text (octal escapes decoded).""" + points = [] + for line in text.splitlines(): + fields = line.split(" ") + if len(fields) > 4: + raw = fields[4] + points.append( + bytes(raw, "utf-8") + .decode("unicode_escape") + .encode("latin-1") + .decode("utf-8", "surrogateescape") + ) + return points + + +def _proc_mounts() -> list[str]: + try: + with open( + "/proc/self/mountinfo", encoding="utf-8", errors="surrogateescape" + ) as f: + return parse_mountinfo(f.read()) + except OSError: + return [] + + +def _inside(path: str, folder: str) -> bool: + folder = folder.rstrip("/") or "/" + return path == folder or path.startswith(folder if folder == "/" else folder + "/") + + +def mount_of(path: str, mounts: list[str]) -> str: + """The longest mount point holding *path* (both real paths).""" + best = "/" + for point in mounts: + if _inside(path, point) and len(point) > len(best): + best = point + return best + + +def _ismount_walk(path: str) -> str: + """Fallback without mountinfo (BSD): walk up until ``os.path.ismount``.""" + current = path + while not os.path.ismount(current): + parent = os.path.dirname(current) + if parent == current: + break + current = parent + return current + + +def _libc_renameat2(): + try: + libc = ctypes.CDLL(None, use_errno=True) + function = libc.renameat2 + except (OSError, AttributeError): + return None + function.argtypes = [ + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_uint, + ] + function.restype = ctypes.c_int + return function + + +_RENAMEAT2 = _libc_renameat2() + + +def rename_noreplace(src_dir_fd: int, src: str, dst_dir_fd: int, dst: str) -> None: + """``renameat2(..., RENAME_NOREPLACE)``; raises OSError (ENOSYS when missing).""" + if _RENAMEAT2 is None: + raise OSError(errno.ENOSYS, "renameat2 is not available") + result = _RENAMEAT2( + src_dir_fd, os.fsencode(src), dst_dir_fd, os.fsencode(dst), RENAME_NOREPLACE + ) + if result != 0: + code = ctypes.get_errno() + raise OSError(code, os.strerror(code)) + + +def _canonical(path: str) -> str: + return os.path.normcase(os.path.realpath(path)) + + +def _check_roots(trash_dir: str, roots) -> None: + trash_key = _canonical(trash_dir) + for root in roots: + if _inside(trash_key, _canonical(str(root))): + raise TrashRefused(TRASH_IN_ROOT) + + +def _owned_folder(path: str, uid: int, *, create: bool) -> bool: + """Whether *path* is (or, with *create*, now is) a real folder we own.""" + try: + info = os.lstat(path) + except FileNotFoundError: + if not create: + return False + try: + os.mkdir(path, 0o700) + except OSError: + return False + info = os.lstat(path) + except OSError: + return False + return stat.S_ISDIR(info.st_mode) and info.st_uid == uid + + +class UnavailableBackend: + """No trash on this system: every move is refused, nothing is touched.""" + + def available(self, path: str, roots) -> str | None: + return NO_TRASH + + def move(self, path: str, roots) -> str: + raise TrashRefused(NO_TRASH) + + +class FreedesktopBackend: + """The freedesktop.org Trash specification 1.0 (Linux, BSD).""" + + def __init__(self, data_home=None, mounts=None, rename=None, uid=None, now=None): + self._data_home = data_home + self._mounts = mounts or _proc_mounts + self._rename = rename or rename_noreplace + self._uid = os.getuid() if uid is None else uid + self._now = now or datetime.datetime.now + + def _home_data(self) -> str: + if self._data_home: + return self._data_home + configured = os.environ.get("XDG_DATA_HOME") + if configured and os.path.isabs(configured): + return configured + return os.path.join(os.path.expanduser("~"), ".local", "share") + + def _mount(self, real: str, mounts) -> str: + return mount_of(real, mounts) if mounts else _ismount_walk(real) + + def _plan(self, path: str, roots, *, create: bool): + """``(trash_dir, topdir_or_None, label)`` for *path*; raises TrashRefused.""" + real = os.path.realpath(path) + mounts = [os.path.realpath(point) for point in self._mounts()] + home_data = os.path.realpath(self._home_data()) + if self._mount(real, mounts) == self._mount(home_data, mounts): + trash_dir = os.path.join(home_data, "Trash") + _check_roots(trash_dir, roots) + if create: + os.makedirs(trash_dir, mode=0o700, exist_ok=True) + return trash_dir, None, HOME_TRASH_LABEL + top = self._mount(real, mounts) + shared = os.path.join(top, ".Trash") + try: + info = os.lstat(shared) + sticky = stat.S_ISDIR(info.st_mode) and bool(info.st_mode & stat.S_ISVTX) + except OSError: + sticky = False + candidates = [] + if sticky: + 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) + if self._usable(trash_dir, create=create): + label = f"{os.path.relpath(trash_dir, top)} on its drive" + return trash_dir, top, label + raise TrashRefused(NO_TRASH) + + def _usable(self, trash_dir: str, *, create: bool) -> bool: + if os.path.lexists(trash_dir): + return _owned_folder(trash_dir, self._uid, create=False) + parent = os.path.dirname(trash_dir) + if not create: + return os.access(parent, os.W_OK | os.X_OK) + return _owned_folder(trash_dir, self._uid, create=True) + + def available(self, path: str, roots) -> str | None: + """None when *path* can be trashed, else the refusal code (no writes).""" + try: + self._plan(path, roots, create=False) + except TrashRefused as refused: + return refused.code + except OSError: + return NO_TRASH + return None + + def move(self, path: str, roots) -> str: + """Move *path* into its trash and return the trash's label.""" + try: + trash_dir, top, label = self._plan(path, roots, create=True) + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + real = os.path.realpath(path) + recorded = os.path.relpath(real, top) if top else real + opened = [] + try: + trash_fd = self._open_dir(trash_dir, opened) + files_fd = self._subfolder(trash_fd, "files", opened) + info_fd = self._subfolder(trash_fd, "info", opened) + source_fd = self._open_dir(os.path.dirname(real), opened) + self._move_named( + source_fd, os.path.basename(real), files_fd, info_fd, recorded + ) + finally: + for fd in opened: + os.close(fd) + return label + + def _open_dir(self, path: str, opened: list) -> int: + try: + fd = os.open(path, _DIR_FLAGS) + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + opened.append(fd) + return fd + + def _subfolder(self, parent_fd: int, name: str, opened: list) -> int: + try: + os.mkdir(name, 0o700, dir_fd=parent_fd) + except FileExistsError: + pass + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + try: + info = os.lstat(name, dir_fd=parent_fd) + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + if not stat.S_ISDIR(info.st_mode) or info.st_uid != self._uid: + raise TrashRefused(NO_TRASH, f"{name} is not our folder") + try: + fd = os.open(name, _DIR_FLAGS, dir_fd=parent_fd) + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + opened.append(fd) + return fd + + def _info_text(self, recorded: str) -> bytes: + encoded = urllib.parse.quote(os.fsencode(recorded), safe="/") + stamp = self._now().strftime("%Y-%m-%dT%H:%M:%S") + return f"[Trash Info]\nPath={encoded}\nDeletionDate={stamp}\n".encode("utf-8") + + def _move_named(self, source_fd, name, files_fd, info_fd, recorded) -> None: + stem, extension = os.path.splitext(name) + for attempt in range(MAX_NAME_ATTEMPTS): + candidate = name if attempt == 0 else f"{stem}.{attempt}{extension}" + info_name = candidate + ".trashinfo" + try: + fd = os.open( + info_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), + 0o600, + dir_fd=info_fd, + ) + except FileExistsError: + continue + except OSError as error: + raise TrashRefused(NO_TRASH, str(error)) from error + try: + os.write(fd, self._info_text(recorded)) + finally: + os.close(fd) + try: + self._rename(source_fd, name, files_fd, candidate) + return + except OSError as error: + os.unlink(info_name, dir_fd=info_fd) + if error.errno == errno.EEXIST: + continue # a stray trashed file has that name: never replace it + if error.errno in _NO_TRASH_ERRNOS: + raise TrashRefused(NO_TRASH, str(error)) from error + raise TrashRefused(FAILED, str(error)) from error + raise TrashRefused(FAILED, "no free name in the trash") + + +def default_backend(platform: str = sys.platform): + """The backend for *platform*; Windows and macOS have none yet (#480).""" + if platform in ("win32", "darwin") or platform.startswith("cygwin"): + return UnavailableBackend() + return FreedesktopBackend() + + +BACKEND = default_backend() + + +def trash_available(path: str, roots) -> str | None: + """None when *path* can be trashed, else the refusal code (blocking, no writes).""" + return BACKEND.available(path, roots) + + +def move_to_trash(path: str, roots) -> str: + """Move one regular file into the trash; returns where. Raises TrashRefused.""" + return BACKEND.move(path, roots)