feat(trash): a freedesktop trash backend that renames without replacing or refuses (#480)
This commit is contained in:
Vito Sansevero
@@ -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()
|
||||
+360
@@ -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)
|
||||
Reference in New Issue
Block a user