diff --git a/src/borg/archiver/mount_cmds.py b/src/borg/archiver/mount_cmds.py index 020bc9099b..162b053e4f 100644 --- a/src/borg/archiver/mount_cmds.py +++ b/src/borg/archiver/mount_cmds.py @@ -36,27 +36,17 @@ def _do_mount(self, args, repository, manifest): from ..fuse_impl import has_mfusepy if has_mfusepy: - # Use mfusepy implementation - from ..hlfuse import borgfs - - operations = borgfs(manifest, args, repository) - logger.info("Mounting filesystem") - try: - operations.mount(args.mountpoint, args.options, args.foreground, args.show_rc) - except RuntimeError: - # Relevant error message already printed to stderr by FUSE - raise RTError("FUSE mount failed") + from ..hlfuse import borgfs as fuse_operations # high-level FUSE API else: - # Use llfuse/pyfuse3 implementation - from ..fuse import FuseOperations - - operations = FuseOperations(manifest, args, repository) - logger.info("Mounting filesystem") - try: - operations.mount(args.mountpoint, args.options, args.foreground, args.show_rc) - except RuntimeError: - # Relevant error message already printed to stderr by FUSE - raise RTError("FUSE mount failed") + from ..fuse import FuseOperations as fuse_operations # low-level FUSE API + + operations = fuse_operations(manifest, args, repository) + logger.info("Mounting filesystem") + try: + operations.mount(args.mountpoint, args.options, args.foreground, args.show_rc) + except RuntimeError: + # Relevant error message already printed to stderr by FUSE + raise RTError("FUSE mount failed") def do_umount(self, args): """Unmounts the FUSE filesystem.""" diff --git a/src/borg/fuse.py b/src/borg/fuse.py index c61740d2f5..bd185360ad 100644 --- a/src/borg/fuse.py +++ b/src/borg/fuse.py @@ -1,5 +1,9 @@ """ -FUSE filesystem implementation for `borg mount`. +``borg mount`` using llfuse / pyfuse3, the low-level (inode based) FUSE 2 / FUSE 3 API. + +This is a protocol adapter only: what an archive looks like as a file system is +defined in vfs.py, here we just translate between that and the llfuse operations +interface (inode numbers and EntryAttributes in, FUSEErrors out). IMPORTANT ========= @@ -21,20 +25,11 @@ import errno import functools -import io import os -import stat -import struct -import sys -import tempfile import threading -import time -from collections import defaultdict, Counter from signal import SIGINT from typing import TYPE_CHECKING -from .constants import ROBJ_FILE_STREAM, ROBJ_DONTCARE, zeros - if TYPE_CHECKING: # For type checking, assume llfuse is available # This allows mypy to understand llfuse.Operations @@ -64,19 +59,11 @@ def async_wrapper(fn): logger = create_logger() -from .crypto.low_level import blake2b_128 -from .archiver._common import build_matcher, build_filter -from .archive import Archive, get_item_uid_gid -from .hashindex import FuseVersionsIndex -from .helpers import daemonizing, signal_handler, format_file_size, bin_to_hex, Error -from .helpers import HardLinkManager -from .helpers import msgpack +from .helpers import daemonizing, signal_handler from .storelocking import LockRefresher -from .helpers.lrucache import LRUCache -from .item import Item -from .platform import uid2user, gid2group, acl_text_to_xattr -from .platformflags import is_darwin, is_linux -from .repository import Repository +from .vfs import ArchiveVFS, ChunkMissing, parse_mount_options + +BLOCK_SIZE = 512 # Standard filesystem block size for st_blocks and statfs def fuse_main(): @@ -93,525 +80,30 @@ def fuse_main(): return llfuse.main(workers=1) -# on Linux, the kernel exposes POSIX ACLs via these special, binary encoded xattrs. -# maps the xattr name to the borg item attribute holding the ACL text. -# empty on platforms we can not do this for, so the mount just does not offer these xattrs there. -ACL_XATTRS = {b"system.posix_acl_access": "acl_access", b"system.posix_acl_default": "acl_default"} if is_linux else {} - -# size of some LRUCaches (1 element per simultaneously open file) -# note: _inode_cache might have rather large elements - Item.chunks can be large! -# also, simultaneously reading too many files should be avoided anyway. -# thus, do not set FILES to high values. -FILES = 4 - - -class ItemCache: - """ - This is the "meat" of the filesystem's metadata storage. - - This class generates inode numbers that efficiently index items in archives - and retrieves items from these inode numbers. - """ - - # 2 MiB are approximately ~230000 items (depends on the average number of items per metadata chunk). - # - # Since growing a bytearray has to copy it, growing it will converge to O(n^2), however, - # this is not yet relevant due to the swiftness of copying memory. If it becomes an issue, - # use an anonymous mmap and just resize that (or, if on 64 bit, make it so big you never need - # to resize it in the first place; that's free). - GROW_META_BY = 2 * 1024 * 1024 - - indirect_entry_struct = struct.Struct("=cII") - assert indirect_entry_struct.size == 9 - - def __init__(self, repository, repo_objs): - self.repository = repository - self.repo_objs = repo_objs - # self.meta, the "meta-array" is a densely packed array of metadata about where items can be found. - # It is indexed by the inode number minus self.offset. (This is in a way eerily similar to how the first - # unices did this). - # The meta-array contains chunk IDs and item entries (described in iter_archive_items). - # The chunk IDs are referenced by item entries through relative offsets, - # which are bounded by the metadata chunk size. - self.meta = bytearray() - # The current write offset in self.meta - self.write_offset = 0 - - # Offset added to meta-indices, resulting in inodes, - # or subtracted from inodes, resulting in meta-indices. - # XXX: Merge FuseOperations.items and ItemCache to avoid - # this implicit limitation / hack (on the number of synthetic inodes, degenerate - # cases can inflate their number far beyond the number of archives). - self.offset = 1000000 - - # A temporary file that contains direct items, i.e. items directly cached in this layer. - # These are items that span more than one chunk and thus cannot be efficiently cached - # by the object cache (self.chunks), which would require variable-length structures; - # possible but not worth the effort, see iter_archive_items. - self.fd = tempfile.TemporaryFile(prefix="borg-tmp") - - # A small LRU cache for chunks requested by ItemCache.get() from the object cache, - # this significantly speeds up directory traversal and similar operations which - # tend to re-read the same chunks over and over. - # The capacity is kept low because increasing it does not provide any significant advantage, - # but makes LRUCache's square behaviour noticeable and consumes more memory. - self.chunks = LRUCache(capacity=10) - - # Instrumentation - # Count of indirect items, i.e. data is cached in the object cache, not directly in this cache - self.indirect_items = 0 - # Count of direct items, i.e. data is in self.fd - self.direct_items = 0 - - def get(self, inode): - offset = inode - self.offset - if offset < 0: - raise ValueError("ItemCache.get() called with an invalid inode number") - if self.meta[offset] == ord(b"I"): - _, chunk_id_relative_offset, chunk_offset = self.indirect_entry_struct.unpack_from(self.meta, offset) - chunk_id_offset = offset - chunk_id_relative_offset - # bytearray slices are bytearrays as well, explicitly convert to bytes() - chunk_id = bytes(self.meta[chunk_id_offset : chunk_id_offset + 32]) - chunk = self.chunks.get(chunk_id) - if not chunk: - cdata = self.repository.get(chunk_id) - _, chunk = self.repo_objs.parse(chunk_id, cdata, ro_type=ROBJ_DONTCARE) - self.chunks[chunk_id] = chunk - data = memoryview(chunk)[chunk_offset:] - unpacker = msgpack.Unpacker() - unpacker.feed(data) - return Item(internal_dict=next(unpacker)) - elif self.meta[offset] == ord(b"S"): - fd_offset = int.from_bytes(self.meta[offset + 1 : offset + 9], "little") - self.fd.seek(fd_offset, io.SEEK_SET) - return Item(internal_dict=next(msgpack.Unpacker(self.fd, read_size=1024))) - else: - raise ValueError("Invalid entry type in self.meta") - - def iter_archive_items(self, archive_item_ids, filter=None): - unpacker = msgpack.Unpacker() - - # Current offset in the metadata stream, which consists of all metadata chunks glued together - stream_offset = 0 - # Offset of the current chunk in the metadata stream - chunk_begin = 0 - # Length of the chunk preceding the current chunk - last_chunk_length = 0 - msgpacked_bytes = b"" - - write_offset = self.write_offset - meta = self.meta - pack_indirect_into = self.indirect_entry_struct.pack_into - - for key, cdata in zip(archive_item_ids, self.repository.get_many(archive_item_ids)): - _, data = self.repo_objs.parse(key, cdata, ro_type=ROBJ_DONTCARE) - # Store the chunk ID in the meta-array - if write_offset + 32 >= len(meta): - meta.extend(bytes(self.GROW_META_BY)) - meta[write_offset : write_offset + 32] = key - current_id_offset = write_offset - write_offset += 32 - - chunk_begin += last_chunk_length - last_chunk_length = len(data) - - unpacker.feed(data) - while True: - try: - item = unpacker.unpack() - need_more_data = False - except msgpack.OutOfData: - need_more_data = True - - start = stream_offset - chunk_begin - # tell() is not helpful for the need_more_data case, but we know it is the remainder - # of the data in that case. in the other case, tell() works as expected. - length = (len(data) - start) if need_more_data else (unpacker.tell() - stream_offset) - msgpacked_bytes += data[start : start + length] - stream_offset += length - - if need_more_data: - # Need more data, feed the next chunk - break - - item = Item(internal_dict=item) - if filter and not filter(item): - msgpacked_bytes = b"" - continue - - current_item = msgpacked_bytes - current_item_length = len(current_item) - current_spans_chunks = stream_offset - current_item_length < chunk_begin - msgpacked_bytes = b"" - - if write_offset + 9 >= len(meta): - meta.extend(bytes(self.GROW_META_BY)) - - # item entries in the meta-array come in two different flavours, both nine bytes long. - # (1) for items that span chunks: - # - # 'S' + 8 byte offset into the self.fd file, where the msgpacked item starts. - # - # (2) for items that are completely contained in one chunk, which usually is the great majority - # (about 700:1 for system backups) - # - # 'I' + 4 byte offset where the chunk ID is + 4 byte offset in the chunk - # where the msgpacked items starts - # - # The chunk ID offset is the number of bytes _back_ from the start of the entry, i.e.: - # - # |Chunk ID| .... |S1234abcd| - # ^------ offset ----------^ - - if current_spans_chunks: - pos = self.fd.seek(0, io.SEEK_END) - self.fd.write(current_item) - meta[write_offset : write_offset + 9] = b"S" + pos.to_bytes(8, "little") - self.direct_items += 1 - else: - item_offset = stream_offset - current_item_length - chunk_begin - pack_indirect_into(meta, write_offset, b"I", write_offset - current_id_offset, item_offset) - self.indirect_items += 1 - inode = write_offset + self.offset - write_offset += 9 - - yield inode, item - - self.write_offset = write_offset - - -class FuseBackend: - """Virtual filesystem based on archive(s) to provide information to fuse""" +class FuseOperations(llfuse.Operations): + """Export archive contents as a FUSE filesystem""" def __init__(self, manifest, args, repository): - self._args = args - self.numeric_ids = args.numeric_ids + llfuse.Operations.__init__(self) self._manifest = manifest - self.repo_objs = manifest.repo_objs - self.repository_uncached = manifest.repository - # Maps inode numbers to Item instances. This is used for synthetic inodes, i.e. file-system objects that are - # made up and are not contained in the archives. For example archive directories or intermediate directories - # not contained in archives. - self._items = {} - # cache up to Items - self._inode_cache = LRUCache(capacity=FILES) - # _inode_count is the current count of synthetic inodes, i.e. those in self._items - self.inode_count = 0 - # Maps inode numbers to the inode number of the parent - self.parent = {} - # Maps inode numbers to a dictionary mapping byte directory entry names to their inode numbers, - # i.e. this contains all dirents of everything that is mounted. (It becomes really big). - self.contents = defaultdict(dict) - self.default_uid = os.getuid() - self.default_gid = os.getgid() - self.default_dir = None - # Archives to be loaded when first accessed, mapped by their placeholder inode - self.pending_archives = {} - self.cache = ItemCache(repository, self.repo_objs) + self._args = args + self._repository = repository # serializes all repository access (FUSE handlers and the background lock-refresh - # thread), because borgstore connections are not thread-safe. see _lock_refresh. + # thread), because borgstore connections are not thread-safe. self._repo_lock = threading.RLock() - self.allow_damaged_files = False - self.versions = False - self.uid_forced = None - self.gid_forced = None - self.umask = 0 - self.archive_root_dir = {} # archive ID --> directory name - - def _create_filesystem(self): - self._create_dir(parent=1) # first call, create root dir (inode == 1) - self.versions_index = FuseVersionsIndex() - archives = self._manifest.archives.list_considering(self._args) - name_counter = Counter(a.name for a in archives) - duplicate_names = {a.name for a in archives if name_counter[a.name] > 1} - for archive in archives: - name = f"{archive.name}" - if name in duplicate_names: - name += f"-{bin_to_hex(archive.id):.8}" - self.archive_root_dir[archive.id] = name - for archive in archives: - if self.versions: - # process archives immediately - self._process_archive(archive.id) - else: - # lazily load archives, create archive placeholder inode - archive_inode = self._create_dir(parent=1, mtime=int(archive.ts.timestamp() * 1e9)) - name = self.archive_root_dir[archive.id] - self.contents[1][os.fsencode(name)] = archive_inode - self.pending_archives[archive_inode] = archive - - def get_item(self, inode): - item = self._inode_cache.get(inode) - if item is not None: - return item - try: - # this is a cheap get-from-dictionary operation, no need to cache the result. - return self._items[inode] - except KeyError: - # while self.cache does some internal caching, it has still quite some overhead, so we cache the result. - with self._repo_lock: - item = self.cache.get(inode) - self._inode_cache[inode] = item - return item - - def check_pending_archive(self, inode): - # Check if this is an archive we need to load - archive_info = self.pending_archives.pop(inode, None) - if archive_info is not None: - with self._repo_lock: - self._process_archive(archive_info.id, [os.fsencode(self.archive_root_dir[archive_info.id])]) - - def _allocate_inode(self): - self.inode_count += 1 - return self.inode_count - - def _create_dir(self, parent, mtime=None): - """Create directory""" - ino = self._allocate_inode() - if mtime is not None: - self._items[ino] = Item(internal_dict=self.default_dir.as_dict()) - self._items[ino].mtime = mtime - else: - self._items[ino] = self.default_dir - self.parent[ino] = parent - return ino - - def find_inode(self, path, prefix=[]): - segments = prefix + path.split(b"/") - inode = 1 - for segment in segments: - inode = self.contents[inode][segment] - return inode - - def _process_archive(self, archive_id, prefix=[]): - """Build FUSE inode hierarchy from archive metadata""" - self.file_versions = {} # for versions mode: original path -> version - t0 = time.perf_counter() - archive = Archive(self._manifest, archive_id) - strip_components = self._args.strip_components - # omitting args.pattern_roots here, restricting to paths only by cli args.paths: - matcher = build_matcher(self._args.patterns, self._args.paths) - hlm = HardLinkManager(id_type=bytes, info_type=str) # hlid -> path - - filter = build_filter(matcher, strip_components) - for item_inode, item in self.cache.iter_archive_items(archive.metadata.items, filter=filter): - if strip_components: - item.path = os.sep.join(item.path.split(os.sep)[strip_components:]) - path = os.fsencode(item.path) - is_dir = stat.S_ISDIR(item.mode) - if is_dir: - try: - # This can happen if an archive was created with a command line like - # $ borg create ... dir1/file dir1 - # In this case the code below will have created a default_dir inode for dir1 already. - inode = self.find_inode(path, prefix) - except KeyError: - pass - else: - self._items[inode] = item - continue - segments = prefix + path.split(b"/") - parent = 1 - for segment in segments[:-1]: - parent = self._process_inner(segment, parent) - self._process_leaf(segments[-1], item, parent, prefix, is_dir, item_inode, hlm) - duration = time.perf_counter() - t0 - logger.debug("fuse: _process_archive completed in %.1f s for archive %s", duration, archive.name) - - def _process_leaf(self, name, item, parent, prefix, is_dir, item_inode, hlm): - path = item.path - del item.path # save some space - - def file_version(item, path): - if "chunks" in item: - file_id = blake2b_128(path) - current_version, previous_id = self.versions_index.get(file_id, (0, None)) - - contents_id = blake2b_128(b"".join(chunk_id for chunk_id, _ in item.chunks)) - - if contents_id != previous_id: - current_version += 1 - self.versions_index[file_id] = current_version, contents_id - - return current_version - - def make_versioned_name(name, version, add_dir=False): - if add_dir: - # add intermediate directory with same name as filename - path_fname = name.rsplit(b"/", 1) - name += b"/" + path_fname[-1] - # keep original extension at end to avoid confusing tools - name, ext = os.path.splitext(name) - version_enc = os.fsencode(".%05d" % version) - return name + version_enc + ext - - if "hlid" in item: - link_target = hlm.retrieve(id=item.hlid, default=None) - if link_target is not None: - # Hard link was extracted previously, just link - link_target = os.fsencode(link_target) - if self.versions: - # adjust link target name with version - version = self.file_versions[link_target] - link_target = make_versioned_name(link_target, version, add_dir=True) - try: - inode = self.find_inode(link_target, prefix) - except KeyError: - logger.warning("Skipping broken hard link: %s -> %s", path, link_target) - return - item = self.get_item(inode) - item.nlink = item.get("nlink", 1) + 1 - self._items[inode] = item - else: - inode = item_inode - self._items[inode] = item - # remember extracted item path, so that following hard links don't extract twice. - hlm.remember(id=item.hlid, info=path) - else: - inode = item_inode - - if self.versions and not is_dir: - parent = self._process_inner(name, parent) - enc_path = os.fsencode(path) - version = file_version(item, enc_path) - if version is not None: - # regular file, with contents - name = make_versioned_name(name, version) - self.file_versions[enc_path] = version - - self.parent[inode] = parent - if name: - self.contents[parent][name] = inode - - def _process_inner(self, name, parent_inode): - dir = self.contents[parent_inode] - if name in dir: - inode = dir[name] - else: - inode = self._create_dir(parent_inode) - if name: - dir[name] = inode - return inode - - -class FuseOperations(llfuse.Operations, FuseBackend): - """Export archive as a FUSE filesystem""" - - def __init__(self, manifest, args, repository): - llfuse.Operations.__init__(self) - FuseBackend.__init__(self, manifest, args, repository) - data_cache_capacity = int(os.environ.get("BORG_MOUNT_DATA_CACHE_ENTRIES", os.cpu_count() or 1)) - logger.debug("mount data cache capacity: %d chunks", data_cache_capacity) - self.data_cache = LRUCache(capacity=data_cache_capacity) - self._last_pos = LRUCache(capacity=FILES) - - def sig_info_handler(self, sig_no, stack): - logger.debug( - "fuse: %d synth inodes, %d edges (%s)", - self.inode_count, - len(self.parent), - # getsizeof is the size of the dict itself; key and value are two small-ish integers, - # which are shared due to code structure (this has been verified). - format_file_size(sys.getsizeof(self.parent) + len(self.parent) * sys.getsizeof(self.inode_count)), - ) - logger.debug("fuse: %d pending archives", len(self.pending_archives)) - logger.debug( - "fuse: ItemCache %d entries (%d direct, %d indirect), meta-array size %s, direct items size %s", - self.cache.direct_items + self.cache.indirect_items, - self.cache.direct_items, - self.cache.indirect_items, - format_file_size(sys.getsizeof(self.cache.meta)), - format_file_size(os.stat(self.cache.fd.fileno()).st_size), - ) - logger.debug( - "fuse: data cache: %d/%d entries, %s", - len(self.data_cache.items()), - self.data_cache._capacity, - format_file_size(sum(len(chunk) for key, chunk in self.data_cache.items())), - ) + self.vfs = None # created by mount(), once the mount options are known def mount(self, mountpoint, mount_options, foreground=False, show_rc=False): """Mount filesystem on *mountpoint* with *mount_options*.""" - - def pop_option(options, key, present, not_present, wanted_type, int_base=0): - assert isinstance(options, list) # we mutate this - for idx, option in enumerate(options): - if option == key: - options.pop(idx) - return present - if option.startswith(key + "="): - options.pop(idx) - value = option.split("=", 1)[1] - if wanted_type is bool: - v = value.lower() - if v in ("y", "yes", "true", "1"): - return True - if v in ("n", "no", "false", "0"): - return False - raise ValueError("unsupported value in option: %s" % option) - if wanted_type is int: - try: - return int(value, base=int_base) - except ValueError: - raise ValueError("unsupported value in option: %s" % option) from None - try: - return wanted_type(value) - except ValueError: - raise ValueError("unsupported value in option: %s" % option) from None - else: - return not_present - - # default_permissions enables permission checking by the kernel. Without - # this, any umask (or uid/gid) would not have an effect and this could - # cause security issues if used with allow_other mount option. - # When not using allow_other or allow_root, access is limited to the - # mounting user anyway. - options = ["fsname=borgfs", "ro", "default_permissions"] - if mount_options: - options.extend(mount_options.split(",")) - if is_darwin: - # macFUSE supports a volname mount option to give what finder displays on desktop / in directory list. - volname = pop_option(options, "volname", "", "", str) - # if the user did not specify it, we make something up, - # because otherwise it would be "macFUSE Volume 0 (Python)", #7690. - volname = volname or f"{os.path.basename(mountpoint)} (borgfs)" - options.append(f"volname={volname}") - ignore_permissions = pop_option(options, "ignore_permissions", True, False, bool) - if ignore_permissions: - # in case users have a use-case that requires NOT giving "default_permissions", - # this is enabled by the custom "ignore_permissions" mount option which just - # removes "default_permissions" again: - pop_option(options, "default_permissions", True, False, bool) - self.allow_damaged_files = pop_option(options, "allow_damaged_files", True, False, bool) - self.versions = pop_option(options, "versions", True, False, bool) - self.uid_forced = pop_option(options, "uid", None, None, int) - self.gid_forced = pop_option(options, "gid", None, None, int) - self.umask = pop_option(options, "umask", 0, 0, int, int_base=8) # umask is octal, e.g. 222 or 0222 - dir_uid = self.uid_forced if self.uid_forced is not None else self.default_uid - dir_gid = self.gid_forced if self.gid_forced is not None else self.default_gid - dir_user = uid2user(dir_uid) - dir_group = gid2group(dir_gid) - if not isinstance(dir_user, str): - raise Error( - f"uid {dir_uid} can not be resolved to a username. " - f"Please check that the corresponding user exists or do not specify a uid mount option." - ) - if not isinstance(dir_group, str): - raise Error( - f"gid {dir_gid} can not be resolved to a group name. " - f"Please check that the corresponding group exists or do not specify a gid mount option." - ) - dir_mode = 0o40755 & ~self.umask - self.default_dir = Item( - mode=dir_mode, mtime=int(time.time() * 1e9), user=dir_user, group=dir_group, uid=dir_uid, gid=dir_gid - ) - self._create_filesystem() + options, vfs_options = parse_mount_options(self._args, mountpoint, mount_options) + self.vfs = ArchiveVFS(self._manifest, self._args, self._repository, lock=self._repo_lock, options=vfs_options) + self.vfs.create_filesystem() llfuse.init(self, mountpoint, options) if not foreground: with daemonizing(show_rc=show_rc) as (old_id, new_id): # the locking process' PID is changing, migrate it: logger.debug("fuse: mount repo, going to background: migrating lock.") - self.repository_uncached.migrate_lock(old_id, new_id) + self._repository.migrate_lock(old_id, new_id) # If the file system crashes, we do not want to umount because in that # case the mountpoint suddenly appears to become empty. This can have @@ -621,7 +113,7 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0): umount = False # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872). # started here (after a possible daemonizing fork, as threads do not survive fork()). - lock_refreshing_thread = LockRefresher(self.repository_uncached.info, sleep_interval=60, lock=self._repo_lock) + lock_refreshing_thread = LockRefresher(self._repository.info, sleep_interval=60, lock=self._repo_lock) lock_refreshing_thread.start() try: with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler): @@ -632,11 +124,45 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0): lock_refreshing_thread.terminate() llfuse.close(umount) + def sig_info_handler(self, sig_no, stack): + self.vfs.log_stats() + + # -- helpers --------------------------------------------------------------- + + def _getattr(self, inode, ctx=None): + attrs = self.vfs.attrs(inode) + entry = llfuse.EntryAttributes() + entry.st_ino = inode + entry.generation = 0 + entry.entry_timeout = 300 + entry.attr_timeout = 300 + entry.st_mode = attrs.mode + entry.st_nlink = attrs.nlink + entry.st_uid = attrs.uid + entry.st_gid = attrs.gid + entry.st_rdev = attrs.rdev + entry.st_size = attrs.size + entry.st_blksize = BLOCK_SIZE + entry.st_blocks = (entry.st_size + entry.st_blksize - 1) // entry.st_blksize + entry.st_mtime_ns = attrs.mtime_ns + entry.st_atime_ns = attrs.atime_ns + entry.st_ctime_ns = attrs.ctime_ns + entry.st_birthtime_ns = attrs.birthtime_ns + return entry + + def _dir_node(self, inode): + try: + return self.vfs.get_node(inode) + except KeyError: + raise llfuse.FUSEError(errno.ENOTDIR) from None + + # -- filesystem operations ------------------------------------------------- + @async_wrapper def statfs(self, ctx=None): stat_ = llfuse.StatvfsData() - stat_.f_bsize = 512 # Filesystem block size - stat_.f_frsize = 512 # Fragment size + stat_.f_bsize = BLOCK_SIZE # Filesystem block size + stat_.f_frsize = BLOCK_SIZE # Fragment size stat_.f_blocks = 0 # Size of fs in f_frsize units stat_.f_bfree = 0 # Number of free blocks stat_.f_bavail = 0 # Number of free blocks for unprivileged users @@ -646,74 +172,37 @@ def statfs(self, ctx=None): stat_.f_namemax = 255 # == NAME_MAX (depends on archive source OS / FS) return stat_ - def _getattr(self, inode, ctx=None): - item = self.get_item(inode) - entry = llfuse.EntryAttributes() - entry.st_ino = inode - entry.generation = 0 - entry.entry_timeout = 300 - entry.attr_timeout = 300 - entry.st_mode = item.mode & ~self.umask - entry.st_nlink = item.get("nlink", 1) - entry.st_uid, entry.st_gid = get_item_uid_gid( - item, - numeric=self.numeric_ids, - uid_default=self.default_uid, - gid_default=self.default_gid, - uid_forced=self.uid_forced, - gid_forced=self.gid_forced, - ) - entry.st_rdev = item.get("rdev", 0) - entry.st_size = item.get_size() - entry.st_blksize = 512 - entry.st_blocks = (entry.st_size + entry.st_blksize - 1) // entry.st_blksize - # note: older archives only have mtime (not atime nor ctime) - entry.st_mtime_ns = mtime_ns = item.mtime - entry.st_atime_ns = item.get("atime", mtime_ns) - entry.st_ctime_ns = item.get("ctime", mtime_ns) - entry.st_birthtime_ns = item.get("birthtime", mtime_ns) - return entry - @async_wrapper def getattr(self, inode, ctx=None): return self._getattr(inode, ctx=ctx) @async_wrapper def listxattr(self, inode, ctx=None): - item = self.get_item(inode) - names = list(item.get("xattrs", {}).keys()) - # expose the archived POSIX ACLs, so e.g. getfacl or tools copying from the mount can read them. - names.extend(xattr_name for xattr_name, attr in ACL_XATTRS.items() if attr in item) - return names + return self.vfs.listxattr(inode) @async_wrapper def getxattr(self, inode, name, ctx=None): - item = self.get_item(inode) - if name in ACL_XATTRS: - acl = item.get(ACL_XATTRS[name]) - if acl is None: - raise llfuse.FUSEError(ENOATTR) - try: - return acl_text_to_xattr(acl, numeric_ids=self.numeric_ids) - except ValueError: - logger.warning("mount: could not convert ACL of inode %d to the xattr representation", inode) - raise llfuse.FUSEError(errno.EIO) from None try: - return item.get("xattrs", {})[name] or b"" + return self.vfs.getxattr(inode, name) except KeyError: raise llfuse.FUSEError(ENOATTR) from None + except ValueError: + logger.warning("mount: could not convert ACL of inode %d to the xattr representation", inode) + raise llfuse.FUSEError(errno.EIO) from None @async_wrapper def lookup(self, parent_inode, name, ctx=None): - self.check_pending_archive(parent_inode) + node = self._dir_node(parent_inode) if name == b".": inode = parent_inode elif name == b"..": - inode = self.parent[parent_inode] + inode = node.parent.ino if node.parent is not None else parent_inode else: - inode = self.contents[parent_inode].get(name) - if not inode: - raise llfuse.FUSEError(errno.ENOENT) + try: + _, child = self.vfs.lookup(node, os.fsdecode(name)) + except KeyError: + raise llfuse.FUSEError(errno.ENOENT) from None + inode = child.ino return self._getattr(inode) @async_wrapper @@ -722,69 +211,28 @@ def open(self, inode, flags, ctx=None): @async_wrapper def opendir(self, inode, ctx=None): - self.check_pending_archive(inode) + self.vfs.ensure_loaded(self._dir_node(inode)) return inode @async_wrapper def read(self, fh, offset, size): - parts = [] - item = self.get_item(fh) - - # optimize for linear reads: - # we cache the chunk number and the in-file offset of the chunk in _last_pos[fh] - chunk_no, chunk_offset = self._last_pos.get(fh, (0, 0)) - if chunk_offset > offset: - # this is not a linear read, so we lost track and need to start from beginning again... - chunk_no, chunk_offset = (0, 0) - - offset -= chunk_offset - chunks = item.chunks - # note: using index iteration to avoid frequently copying big (sub)lists by slicing - for idx in range(chunk_no, len(chunks)): - id, s = chunks[idx] - if s < offset: - offset -= s - chunk_offset += s - chunk_no += 1 - continue - n = min(size, s - offset) - if id in self.data_cache: - data = self.data_cache[id] - if offset + n == len(data): - # evict fully read chunk from cache - del self.data_cache[id] - else: - try: - with self._repo_lock: - cdata = self.repository_uncached.get(id) - except Repository.ObjectNotFound: - if self.allow_damaged_files: - data = zeros[:s] - assert len(data) == s - else: - raise llfuse.FUSEError(errno.EIO) from None - else: - _, data = self.repo_objs.parse(id, cdata, ro_type=ROBJ_FILE_STREAM) - if offset + n < len(data): - # chunk was only partially read, cache it - self.data_cache[id] = data - parts.append(data[offset : offset + n]) - offset = 0 - size -= n - if not size: - if fh in self._last_pos: - self._last_pos.replace(fh, (chunk_no, chunk_offset)) - else: - self._last_pos[fh] = (chunk_no, chunk_offset) - break - return b"".join(parts) + try: + return self.vfs.read(fh, offset, size, pos_key=fh) + except ChunkMissing: + raise llfuse.FUSEError(errno.EIO) from None + + def _readdir_entries(self, fh): + node = self._dir_node(fh) + parent = node.parent if node.parent is not None else node + entries = [(b".", node.ino), (b"..", parent.ino)] + entries.extend((os.fsencode(name), child.ino) for name, child in self.vfs.children(node)) + return entries # note: we can't have a generator (with yield) and not a generator (async) in the same method if has_pyfuse3: async def readdir(self, fh, off, token): # type: ignore[misc] - entries = [(b".", fh), (b"..", self.parent[fh])] - entries.extend(self.contents[fh].items()) + entries = self._readdir_entries(fh) for i, (name, inode) in enumerate(entries[off:], off): attrs = self._getattr(inode) if not llfuse.readdir_reply(token, name, attrs, i + 1): @@ -793,13 +241,11 @@ async def readdir(self, fh, off, token): # type: ignore[misc] else: def readdir(self, fh, off): # type: ignore[misc] - entries = [(b".", fh), (b"..", self.parent[fh])] - entries.extend(self.contents[fh].items()) + entries = self._readdir_entries(fh) for i, (name, inode) in enumerate(entries[off:], off): attrs = self._getattr(inode) yield name, attrs, i + 1 @async_wrapper def readlink(self, inode, ctx=None): - item = self.get_item(inode) - return os.fsencode(item.target) + return os.fsencode(self.vfs.readlink(inode)) diff --git a/src/borg/hlfuse.py b/src/borg/hlfuse.py index 201dccca9a..242100757e 100644 --- a/src/borg/hlfuse.py +++ b/src/borg/hlfuse.py @@ -1,15 +1,15 @@ -import datetime +""" +``borg mount`` using mfusepy, the high-level (path based) FUSE 2 / FUSE 3 API. + +This is a protocol adapter only: what an archive looks like as a file system is +defined in vfs.py, here we just translate between that and the mfusepy operations +interface (paths and stat dicts in, errnos out). +""" + import errno -import hashlib -import os -import stat import threading -import time -from collections import Counter from typing import TYPE_CHECKING -from .constants import ROBJ_FILE_STREAM, zeros, ROBJ_DONTCARE - if TYPE_CHECKING: # For type checking, assume mfusepy is available # This allows mypy to understand hlfuse.Operations @@ -22,531 +22,46 @@ logger = create_logger() -from .archiver._common import build_matcher, build_filter -from .archive import Archive, get_item_uid_gid -from .hashindex import FuseVersionsIndex -from .helpers import daemonizing, signal_handler, bin_to_hex, Error -from .helpers import HardLinkManager -from .helpers import msgpack +from .helpers import daemonizing, signal_handler from .storelocking import LockRefresher -from .helpers.lrucache import LRUCache -from .item import Item -from .platform import uid2user, gid2group, acl_text_to_xattr -from .platformflags import is_darwin, is_linux -from .repository import Repository +from .vfs import ArchiveVFS, ChunkMissing, parse_mount_options BLOCK_SIZE = 512 # Standard filesystem block size for st_blocks and statfs -# on Linux, the kernel exposes POSIX ACLs via these special, binary encoded xattrs. -# maps the xattr name to the borg item attribute holding the ACL text. -# empty on platforms we can not do this for, so the mount just does not offer these xattrs there. -ACL_XATTRS = {"system.posix_acl_access": "acl_access", "system.posix_acl_default": "acl_default"} if is_linux else {} - -DEBUG_LOG: str | None = None # os.path.join(os.getcwd(), "fuse_debug.log") - - -def debug_log(msg): - """Append debug message to fuse_debug.log""" - if DEBUG_LOG: - timestamp = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3] - with open(DEBUG_LOG, "a") as f: - f.write(f"{timestamp} {msg}\n") - - -class DirEntry: - __slots__ = ("ino", "parent", "children") - def __init__(self, ino, parent=None): - self.ino = ino # inode number - self.parent = parent - self.children = None # name (bytes) -> DirEntry, lazily allocated +class borgfs(hlfuse.Operations): + """Export archive contents as a FUSE filesystem""" - def add_child(self, name, child): - """Add a child entry, lazily allocating the children dict if needed.""" - if self.children is None: - self.children = {} - self.children[name] = child - - def get_child(self, name): - """Get a child entry by name, returns None if not found.""" - if self.children is None: - return None - return self.children.get(name) - - def has_child(self, name): - """Check if a child with the given name exists.""" - if self.children is None: - return False - return name in self.children - - def iter_children(self): - """Iterate over (name, child) pairs.""" - if self.children is None: - return iter([]) - return self.children.items() - - -class FuseBackend: - """Virtual filesystem based on archive(s) to provide information to fuse""" + use_ns = True def __init__(self, manifest, args, repository): - self._args = args - self.numeric_ids = args.numeric_ids + hlfuse.Operations.__init__(self) self._manifest = manifest - self.repo_objs = manifest.repo_objs - self.repository = repository + self._args = args + self._repository = repository # serializes all repository access (FUSE handlers and the background lock-refresh - # thread), because borgstore connections are not thread-safe. see _lock_refresh. + # thread), because borgstore connections are not thread-safe. self._repo_lock = threading.RLock() - - self.default_uid = os.getuid() - self.default_gid = os.getgid() - self.default_dir = None - - self.current_ino = 0 - self.inodes = {} # node.ino -> packed item - self.root = self._create_node() - self.pending_archives = {} # DirEntry -> Archive - - self.allow_damaged_files = False - self.versions = False - self.uid_forced = None - self.gid_forced = None - self.umask = 0 - self.archive_root_dir = {} # archive ID --> directory name - - # Cache for file handles - self.handles = {} + self.vfs = None # created by mount(), once the mount options are known + self.handles = {} # file handle -> node self.handle_count = 0 - # Cache for chunks (moved from ItemCache) - self.chunks_cache = LRUCache(capacity=10) - - def _create_node(self, item=None, parent=None): - self.current_ino += 1 - if item is not None: - self.set_inode(self.current_ino, item) - return DirEntry(self.current_ino, parent) - - def get_inode(self, ino): - packed = self.inodes.get(ino) - if packed is None: - return None - return Item(internal_dict=msgpack.unpackb(packed)) - - def set_inode(self, ino, item): - if item is None: - self.inodes.pop(ino, None) - else: - # Remove path from the item dict before packing to save memory. - # The path is already encoded in the DirEntry tree structure. - item_dict = item.as_dict() - item_dict.pop("path", None) - self.inodes[ino] = msgpack.packb(item_dict) - - def _create_filesystem(self): - self.set_inode(self.root.ino, self.default_dir) - self.versions_index = FuseVersionsIndex() - - if getattr(self._args, "name", None): - archives = [self._manifest.archives.get(self._args.name)] - else: - archives = self._manifest.archives.list_considering(self._args) - - name_counter = Counter(a.name for a in archives) - duplicate_names = {a.name for a in archives if name_counter[a.name] > 1} - - for archive in archives: - name = f"{archive.name}" - if name in duplicate_names: - name += f"-{bin_to_hex(archive.id):.8}" - self.archive_root_dir[archive.id] = name - - for archive in archives: - if self.versions: - self._process_archive(archive.id) - else: - # Create placeholder for archive - name = self.archive_root_dir[archive.id] - name_bytes = os.fsencode(name) - - archive_node = self._create_node(parent=self.root) - # Create a directory item for the archive - item = Item(internal_dict=self.default_dir.as_dict()) - item.mtime = int(archive.ts.timestamp() * 1e9) - self.set_inode(archive_node.ino, item) - - self.root.add_child(name_bytes, archive_node) - self.pending_archives[archive_node] = archive - - def check_pending_archive(self, node): - archive_info = self.pending_archives.pop(node, None) - if archive_info is not None: - with self._repo_lock: - self._process_archive(archive_info.id, node) - - def _iter_archive_items(self, archive_item_ids, filter=None): - unpacker = msgpack.Unpacker() - for id, cdata in zip(archive_item_ids, self.repository.get_many(archive_item_ids)): - _, data = self.repo_objs.parse(id, cdata, ro_type=ROBJ_DONTCARE) - unpacker.feed(data) - for item in unpacker: - item = Item(internal_dict=item) - if filter and not filter(item): - continue - yield item - - def _process_archive(self, archive_id, root_node=None): - if root_node is None: - root_node = self.root - - self.file_versions = {} # for versions mode: original path -> version - - archive = Archive(self._manifest, archive_id) - strip_components = self._args.strip_components - # omitting args.pattern_roots here, restricting to paths only by cli args.paths: - matcher = build_matcher(self._args.patterns, self._args.paths) - hlm = HardLinkManager(id_type=bytes, info_type=str) - - filter = build_filter(matcher, strip_components) - - for item in self._iter_archive_items(archive.metadata.items, filter=filter): - if strip_components: - item.path = os.sep.join(item.path.split(os.sep)[strip_components:]) - - path = os.fsencode(item.path) - segments = path.split(b"/") - is_dir = stat.S_ISDIR(item.mode) - - # For versions mode, handle files differently - if self.versions and not is_dir: - self._process_leaf_versioned(segments, item, root_node, hlm) - else: - # Original non-versions logic - node = root_node - # Traverse/Create directories - for segment in segments[:-1]: - if not node.has_child(segment): - new_node = self._create_node(parent=node) - # We might need a default directory item if it's an implicit directory - self.set_inode(new_node.ino, Item(internal_dict=self.default_dir.as_dict())) - node.add_child(segment, new_node) - node = node.get_child(segment) - - # Leaf (file or explicit directory) - leaf_name = segments[-1] - if node.has_child(leaf_name): - # Already exists (e.g. implicit dir became explicit) - child = node.get_child(leaf_name) - self.set_inode(child.ino, item) # Update item - node = child - else: - new_node = self._create_node(item, parent=node) - node.add_child(leaf_name, new_node) - node = new_node - - # Handle hardlinks (non-versions mode) - if "hlid" in item: - link_target = hlm.retrieve(id=item.hlid, default=None) - if link_target is not None: - target_path = os.fsencode(link_target) - target_node = self._find_node_from_root(root_node, target_path) - if target_node: - # Reuse ino and item from target - node.ino = target_node.ino - # node.item = target_node.item # implicitly shared via ID - item = self.get_inode(node.ino) - if "nlink" not in item: - item.nlink = 1 - item.nlink += 1 - self.set_inode(node.ino, item) - else: - logger.warning("Hardlink target not found: %s", link_target) - else: - hlm.remember(id=item.hlid, info=item.path) - - def _process_leaf_versioned(self, segments, item, root_node, hlm): - """Process a file leaf node in versions mode""" - path = b"/".join(segments) - original_path = item.path - - # Handle hardlinks in versions mode - check if we've seen this hardlink before - is_hardlink = "hlid" in item - link_target = None - if is_hardlink: - link_target = hlm.retrieve(id=item.hlid, default=None) - if link_target is None: - # First occurrence of this hardlink - hlm.remember(id=item.hlid, info=original_path) - - # Calculate version for this file - # If it's a hardlink to a previous file, use that version - if is_hardlink and link_target is not None: - link_target_enc = os.fsencode(link_target) - version = self.file_versions.get(link_target_enc) - else: - version = self._file_version(item, path) - - # Store version for this path - if version is not None: - self.file_versions[path] = version - - # Navigate to parent directory - node = root_node - for segment in segments[:-1]: - if not node.has_child(segment): - new_node = self._create_node(parent=node) - self.set_inode(new_node.ino, Item(internal_dict=self.default_dir.as_dict())) - node.add_child(segment, new_node) - node = node.get_child(segment) - - # Create intermediate directory with the filename - leaf_name = segments[-1] - if not node.has_child(leaf_name): - intermediate_node = self._create_node(parent=node) - self.set_inode(intermediate_node.ino, Item(internal_dict=self.default_dir.as_dict())) - node.add_child(leaf_name, intermediate_node) - else: - intermediate_node = node.get_child(leaf_name) - - # Create versioned filename - if version is not None: - versioned_name = self._make_versioned_name(leaf_name, version) - - # If this is a hardlink to a previous file, reuse that node - if is_hardlink and link_target is not None: - link_target_enc = os.fsencode(link_target) - link_segments = link_target_enc.split(b"/") - link_version = self.file_versions.get(link_target_enc) - if link_version is not None: - # Navigate to the link target - target_node = root_node - for seg in link_segments[:-1]: - if target_node.has_child(seg): - target_node = target_node.get_child(seg) - else: - break - else: - # Get intermediate dir - link_leaf = link_segments[-1] - if target_node.has_child(link_leaf): - target_intermediate = target_node.get_child(link_leaf) - target_versioned = self._make_versioned_name(link_leaf, link_version) - if target_intermediate.has_child(target_versioned): - original_node = target_intermediate.get_child(target_versioned) - # Create new node but reuse the ino and item from original - item = self.get_inode(original_node.ino) - file_node = self._create_node(item, parent=intermediate_node) - file_node.ino = original_node.ino - # Update nlink count - item = self.get_inode(file_node.ino) - if "nlink" not in item: - item.nlink = 1 - item.nlink += 1 - self.set_inode(file_node.ino, item) - intermediate_node.add_child(versioned_name, file_node) - return - - # Not a hardlink or first occurrence - create new node - file_node = self._create_node(item, parent=intermediate_node) - intermediate_node.add_child(versioned_name, file_node) - - def _file_version(self, item, path): - """Calculate version number for a file based on its contents""" - if "chunks" not in item: - return None - - # note: using sha256 here because nowadays it is often hw accelerated. - # shortening the hashes to 16 bytes to save some memory. - file_id = hashlib.sha256(path).digest()[:16] - current_version, previous_id = self.versions_index.get(file_id, (0, None)) - - contents_id = hashlib.sha256(b"".join(chunk_id for chunk_id, _ in item.chunks)).digest()[:16] - - if contents_id != previous_id: - current_version += 1 - self.versions_index[file_id] = current_version, contents_id - - return current_version - - def _make_versioned_name(self, name, version): - """Generate versioned filename like 'file.00001.txt'""" - # keep original extension at end to avoid confusing tools - name_str = name.decode("utf-8", "surrogateescape") if isinstance(name, bytes) else name - name_part, ext = os.path.splitext(name_str) - version_str = ".%05d" % version - versioned = name_part + version_str + ext - return versioned.encode("utf-8", "surrogateescape") if isinstance(name, bytes) else versioned - - def _find_node_from_root(self, root, path): - if path == b"" or path == b".": - return root - segments = path.split(b"/") - node = root - for segment in segments: - child = node.get_child(segment) - if child is not None: - node = child - else: - return None - return node - - def _find_node(self, path): - if isinstance(path, str): - path = os.fsencode(path) - if path == b"/" or path == b"": - return self.root - if path.startswith(b"/"): - path = path[1:] - - segments = path.split(b"/") - node = self.root - for segment in segments: - if node in self.pending_archives: - self.check_pending_archive(node) - child = node.get_child(segment) - if child is not None: - node = child - else: - return None - - if node in self.pending_archives: - self.check_pending_archive(node) - - return node - - def _get_handle(self, node): - self.handle_count += 1 - self.handles[self.handle_count] = node - return self.handle_count - - def _get_node_from_handle(self, fh): - return self.handles.get(fh) - - def _make_stat_dict(self, node): - """Create a stat dictionary from a node.""" - item = self.get_inode(node.ino) - st = {} - st["st_ino"] = node.ino - st["st_mode"] = item.mode & ~self.umask - st["st_nlink"] = item.get("nlink", 1) - if stat.S_ISDIR(st["st_mode"]): - st["st_nlink"] = max(st["st_nlink"], 2) - st["st_uid"], st["st_gid"] = get_item_uid_gid( - item, - numeric=self.numeric_ids, - uid_default=self.default_uid, - gid_default=self.default_gid, - uid_forced=self.uid_forced, - gid_forced=self.gid_forced, - ) - st["st_rdev"] = item.get("rdev", 0) - st["st_size"] = item.get_size() - st["st_blocks"] = (st["st_size"] + BLOCK_SIZE - 1) // BLOCK_SIZE - if getattr(self, "use_ns", False): - st["st_mtime"] = item.mtime - st["st_atime"] = item.get("atime", item.mtime) - st["st_ctime"] = item.get("ctime", item.mtime) - else: - st["st_mtime"] = item.mtime / 1e9 - st["st_atime"] = item.get("atime", item.mtime) / 1e9 - st["st_ctime"] = item.get("ctime", item.mtime) / 1e9 - return st - - -class borgfs(hlfuse.Operations, FuseBackend): - """Export archive as a FUSE filesystem""" - - use_ns = True - - def __init__(self, manifest, args, repository): - hlfuse.Operations.__init__(self) - FuseBackend.__init__(self, manifest, args, repository) - data_cache_capacity = int(os.environ.get("BORG_MOUNT_DATA_CACHE_ENTRIES", os.cpu_count() or 1)) - logger.debug("mount data cache capacity: %d chunks", data_cache_capacity) - self.data_cache = LRUCache(capacity=data_cache_capacity) - self._last_pos = LRUCache(capacity=4) - - def sig_info_handler(self, sig_no, stack): - # Simplified instrumentation - logger.debug("fuse: %d inodes", self.current_ino) - def mount(self, mountpoint, mount_options, foreground=False, show_rc=False): """Mount filesystem on *mountpoint* with *mount_options*.""" - - def pop_option(options, key, present, not_present, wanted_type, int_base=0): - assert isinstance(options, list) # we mutate this - for idx, option in enumerate(options): - if option == key: - options.pop(idx) - return present - if option.startswith(key + "="): - options.pop(idx) - value = option.split("=", 1)[1] - if wanted_type is bool: - v = value.lower() - if v in ("y", "yes", "true", "1"): - return True - if v in ("n", "no", "false", "0"): - return False - raise ValueError("unsupported value in option: %s" % option) - if wanted_type is int: - try: - return int(value, base=int_base) - except ValueError: - raise ValueError("unsupported value in option: %s" % option) from None - try: - return wanted_type(value) - except ValueError: - raise ValueError("unsupported value in option: %s" % option) from None - else: - return not_present - - options = ["fsname=borgfs", "ro", "default_permissions"] - if mount_options: - options.extend(mount_options.split(",")) - if is_darwin: - volname = pop_option(options, "volname", "", "", str) - volname = volname or f"{os.path.basename(mountpoint)} (borgfs)" - options.append(f"volname={volname}") - ignore_permissions = pop_option(options, "ignore_permissions", True, False, bool) - if ignore_permissions: - pop_option(options, "default_permissions", True, False, bool) - self.allow_damaged_files = pop_option(options, "allow_damaged_files", True, False, bool) - self.versions = pop_option(options, "versions", True, False, bool) - self.uid_forced = pop_option(options, "uid", None, None, int) - self.gid_forced = pop_option(options, "gid", None, None, int) - self.umask = pop_option(options, "umask", 0, 0, int, int_base=8) - dir_uid = self.uid_forced if self.uid_forced is not None else self.default_uid - dir_gid = self.gid_forced if self.gid_forced is not None else self.default_gid - dir_user = uid2user(dir_uid) - dir_group = gid2group(dir_gid) - if not isinstance(dir_user, str): - raise Error( - f"uid {dir_uid} can not be resolved to a username. " - f"Please check that the corresponding user exists or do not specify a uid mount option." - ) - if not isinstance(dir_group, str): - raise Error( - f"gid {dir_gid} can not be resolved to a group name. " - f"Please check that the corresponding group exists or do not specify a gid mount option." - ) - dir_mode = 0o40755 & ~self.umask - self.default_dir = Item( - mode=dir_mode, mtime=int(time.time() * 1e9), user=dir_user, group=dir_group, uid=dir_uid, gid=dir_gid - ) - self._create_filesystem() + options, vfs_options = parse_mount_options(self._args, mountpoint, mount_options) + self.vfs = ArchiveVFS(self._manifest, self._args, self._repository, lock=self._repo_lock, options=vfs_options) + self.vfs.create_filesystem() # hlfuse.FUSE will block if foreground=True, otherwise it returns immediately if not foreground: # Background mode: daemonize first, then start FUSE (blocking) with daemonizing(show_rc=show_rc) as (old_id, new_id): logger.debug("fuse: mount repo, going to background: migrating lock.") - self.repository.migrate_lock(old_id, new_id) + self._repository.migrate_lock(old_id, new_id) # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872). # started here (after a possible daemonizing fork, as threads do not survive fork()). - lock_refreshing_thread = LockRefresher(self.repository.info, sleep_interval=60, lock=self._repo_lock) + lock_refreshing_thread = LockRefresher(self._repository.info, sleep_interval=60, lock=self._repo_lock) lock_refreshing_thread.start() try: # Run the FUSE main loop in foreground (we might be daemonized already or not) @@ -555,183 +70,115 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0): finally: lock_refreshing_thread.terminate() - def statfs(self, path): - debug_log(f"statfs(path={path!r})") - stat_ = {} - stat_["f_bsize"] = BLOCK_SIZE - stat_["f_frsize"] = BLOCK_SIZE - stat_["f_blocks"] = 0 - stat_["f_bfree"] = 0 - stat_["f_bavail"] = 0 - stat_["f_files"] = 0 - stat_["f_ffree"] = 0 - stat_["f_favail"] = 0 - stat_["f_namemax"] = 255 - debug_log(f"statfs -> {stat_}") - return stat_ + def sig_info_handler(self, sig_no, stack): + self.vfs.log_stats() - def getattr(self, path, fh=None): - debug_log(f"getattr(path={path!r}, fh={fh})") - if fh is not None: - # use file handle if available to avoid path lookup - node = self._get_node_from_handle(fh) - if node is None: - raise hlfuse.FuseOSError(errno.EBADF) + # -- helpers --------------------------------------------------------------- + + def _find_node(self, path): + """Return the node at *path*; raises ENOENT if there is none.""" + segments = [segment for segment in path.split("/") if segment] + try: + _, node = self.vfs.resolve(segments) + except KeyError: + raise hlfuse.FuseOSError(errno.ENOENT) from None + return node + + def _node_from_handle(self, fh): + node = self.handles.get(fh) + if node is None: + raise hlfuse.FuseOSError(errno.EBADF) + return node + + def _stat(self, node): + """Build the stat dict of *node*.""" + attrs = self.vfs.attrs(node.ino) + st = { + "st_ino": attrs.ino, + "st_mode": attrs.mode, + "st_nlink": attrs.nlink, + "st_uid": attrs.uid, + "st_gid": attrs.gid, + "st_rdev": attrs.rdev, + "st_size": attrs.size, + "st_blocks": (attrs.size + BLOCK_SIZE - 1) // BLOCK_SIZE, + } + if self.use_ns: + st["st_mtime"] = attrs.mtime_ns + st["st_atime"] = attrs.atime_ns + st["st_ctime"] = attrs.ctime_ns else: - node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - st = self._make_stat_dict(node) - debug_log(f"getattr -> {st}") + st["st_mtime"] = attrs.mtime_ns / 1e9 + st["st_atime"] = attrs.atime_ns / 1e9 + st["st_ctime"] = attrs.ctime_ns / 1e9 return st + # -- filesystem operations ------------------------------------------------- + + def statfs(self, path): + return { + "f_bsize": BLOCK_SIZE, + "f_frsize": BLOCK_SIZE, + "f_blocks": 0, + "f_bfree": 0, + "f_bavail": 0, + "f_files": 0, + "f_ffree": 0, + "f_favail": 0, + "f_namemax": 255, # == NAME_MAX (depends on archive source OS / FS) + } + + def getattr(self, path, fh=None): + # use the file handle if we have one, to avoid the path lookup + node = self._node_from_handle(fh) if fh is not None else self._find_node(path) + return self._stat(node) + def listxattr(self, path): - debug_log(f"listxattr(path={path!r})") node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - item = self.get_inode(node.ino) - result = [k.decode("utf-8", "surrogateescape") for k in item.get("xattrs", {}).keys()] - # expose the archived POSIX ACLs, so e.g. getfacl or tools copying from the mount can read them. - result.extend(xattr_name for xattr_name, attr in ACL_XATTRS.items() if attr in item) - debug_log(f"listxattr -> {result}") - return result + return [name.decode("utf-8", "surrogateescape") for name in self.vfs.listxattr(node.ino)] def getxattr(self, path, name, position=0): - debug_log(f"getxattr(path={path!r}, name={name!r}, position={position})") node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - item = self.get_inode(node.ino) - name_str = name if isinstance(name, str) else name.decode("utf-8", "surrogateescape") - if name_str in ACL_XATTRS: - acl = item.get(ACL_XATTRS[name_str]) - if acl is None: - debug_log("getxattr -> ENOATTR") - raise hlfuse.FuseOSError(ENOATTR) - try: - result = acl_text_to_xattr(acl, numeric_ids=self.numeric_ids) - except ValueError: - logger.warning(f"mount: could not convert ACL of {path!r} to the xattr representation") - raise hlfuse.FuseOSError(errno.EIO) from None - debug_log(f"getxattr -> {len(result)} bytes") - return result + if isinstance(name, str): + name = name.encode("utf-8", "surrogateescape") try: - if isinstance(name, str): - name = name.encode("utf-8", "surrogateescape") - result = item.get("xattrs", {})[name] or b"" - debug_log(f"getxattr -> {len(result)} bytes") - return result + return self.vfs.getxattr(node.ino, name) except KeyError: - debug_log("getxattr -> ENOATTR") raise hlfuse.FuseOSError(ENOATTR) from None + except ValueError: + logger.warning(f"mount: could not convert ACL of {path!r} to the xattr representation") + raise hlfuse.FuseOSError(errno.EIO) from None def open(self, path, fi): - debug_log(f"open(path={path!r}, fi={fi})") node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - fh = self._get_handle(node) - fi.fh = fh - debug_log(f"open -> fh={fh}") + self.handle_count += 1 + self.handles[self.handle_count] = node + fi.fh = self.handle_count return 0 def release(self, path, fi): - debug_log(f"release(path={path!r}, fh={fi.fh})") self.handles.pop(fi.fh, None) - self._last_pos.pop(fi.fh, None) + self.vfs.reader.forget(fi.fh) return 0 def create(self, path, mode, fi=None): - debug_log(f"create(path={path!r}, mode={mode}, fi={fi}) -> EROFS") raise hlfuse.FuseOSError(errno.EROFS) def read(self, path, size, offset, fi): - fh = fi.fh - debug_log(f"read(path={path!r}, size={size}, offset={offset}, fh={fh})") - node = self._get_node_from_handle(fh) - if node is None: - raise hlfuse.FuseOSError(errno.EBADF) - - item = self.get_inode(node.ino) - parts = [] - - # optimize for linear reads: - chunk_no, chunk_offset = self._last_pos.get(fh, (0, 0)) - if chunk_offset > offset: - chunk_no, chunk_offset = (0, 0) - - offset -= chunk_offset - chunks = item.chunks - - for idx in range(chunk_no, len(chunks)): - id, s = chunks[idx] - if s < offset: - offset -= s - chunk_offset += s - chunk_no += 1 - continue - n = min(size, s - offset) - if id in self.data_cache: - data = self.data_cache[id] - if offset + n == len(data): - del self.data_cache[id] - else: - try: - # Direct repository access - with self._repo_lock: - cdata = self.repository.get(id) - except Repository.ObjectNotFound: - if self.allow_damaged_files: - data = zeros[:s] - assert len(data) == s - else: - raise hlfuse.FuseOSError(errno.EIO) from None - else: - _, data = self.repo_objs.parse(id, cdata, ro_type=ROBJ_FILE_STREAM) - if offset + n < len(data): - self.data_cache[id] = data - parts.append(data[offset : offset + n]) - offset = 0 - size -= n - if not size: - if fh in self._last_pos: - self._last_pos.replace(fh, (chunk_no, chunk_offset)) - else: - self._last_pos[fh] = (chunk_no, chunk_offset) - break - result = b"".join(parts) - debug_log(f"read -> {len(result)} bytes") - return result + node = self._node_from_handle(fi.fh) + try: + return self.vfs.read(node.ino, offset, size, pos_key=fi.fh) + except ChunkMissing: + raise hlfuse.FuseOSError(errno.EIO) from None def readdir(self, path, fh=None): - debug_log(f"readdir(path={path!r}, fh={fh})") node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - - offset = 0 - offset += 0 # += 1 - debug_log(f"readdir yielding . {offset}") - yield (".", self._make_stat_dict(node), offset) - offset += 0 # += 1 - debug_log(f"readdir yielding .. {offset}") - parent = node.parent if node.parent else node - yield ("..", self._make_stat_dict(parent), offset) - - for name, child_node in node.iter_children(): - name_str = name.decode("utf-8", "surrogateescape") - st = self._make_stat_dict(child_node) - offset += 0 # += 1 - debug_log(f"readdir yielding {name_str} {offset} {st}") - yield (name_str, st, offset) + # offset 0 for all entries: we always return the full directory at once. + yield (".", self._stat(node), 0) + yield ("..", self._stat(node.parent if node.parent else node), 0) + for name, child in self.vfs.children(node): + yield (name, self._stat(child), 0) def readlink(self, path): - debug_log(f"readlink(path={path!r})") node = self._find_node(path) - if node is None: - raise hlfuse.FuseOSError(errno.ENOENT) - item = self.get_inode(node.ino) - result = item.target - debug_log(f"readlink -> {result!r}") - return result + return self.vfs.readlink(node.ino) diff --git a/src/borg/testsuite/archiver/webdav_cmd_test.py b/src/borg/testsuite/archiver/webdav_cmd_test.py index c9fab37834..79d4d04934 100644 --- a/src/borg/testsuite/archiver/webdav_cmd_test.py +++ b/src/borg/testsuite/archiver/webdav_cmd_test.py @@ -11,7 +11,6 @@ import tarfile import threading import time -import unicodedata import urllib.error import urllib.request import xml.etree.ElementTree as ET @@ -26,7 +25,7 @@ from ...manifest import Manifest from ...platform import is_win32 from ...repository import Repository -from ...webdav import make_server, lookup_child, Node +from ...webdav import make_server from .. import are_symlinks_supported, are_hardlinks_supported from . import RK_ENCRYPTION, cmd, create_regular_file, generate_archiver_tests @@ -468,18 +467,23 @@ def test_webdav_data_cache(archivers, request, monkeypatch): with repository: manifest = Manifest.load(repository, Manifest.NO_OPERATION_CHECK) server = make_server(manifest, args, port=0) - assert server.data_cache._capacity == 8 # the env var is honored + data_cache = server.RequestHandlerClass.vfs.reader.data_cache + assert data_cache._capacity == 8 # the env var is honored thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: url = f"http://127.0.0.1:{server.server_address[1]}/test/input/big" - assert len(server.data_cache) == 0 - status, _, body = get(url) # first read populates the cache + assert len(data_cache) == 0 + status, _, body = get(url) # a full download reads each chunk completely ... assert status == 200 and body == big - assert len(server.data_cache) > 0 # the multi-chunk file cached some chunks - # a second (ranged) read hits the cache and must return identical bytes + assert len(data_cache) == 0 # ... so it does not fill (pollute) the cache + # a range request only reads part of a chunk, so that chunk is worth caching status, _, body = get(url, headers={"Range": "bytes=0-99"}) assert status == 206 and body == big[:100] + assert len(data_cache) > 0 + # the next range request in the same chunk is served from the cache + status, _, body = get(url, headers={"Range": "bytes=100-199"}) + assert status == 206 and body == big[100:200] finally: server.shutdown() server.server_close() @@ -522,36 +526,6 @@ def test_webdav_unicode_normalization(archivers, request): assert exc_info.value.code == 404 -def test_webdav_lookup_child(): - # exact matches always win, and names that are ambiguous after normalization are only - # reachable by their exact spelling, so we never serve a different file than requested. - nfc_name = "grüße.txt" # composed - nfd_name = unicodedata.normalize("NFD", nfc_name) # decomposed - assert nfc_name != nfd_name - - only_nfc = Node(0o40755, children={nfc_name: Node(0o100644)}) - # the stored (composed) name is found by both spellings - assert lookup_child(only_nfc, nfc_name)[0] == nfc_name - assert lookup_child(only_nfc, nfd_name)[0] == nfc_name - - only_nfd = Node(0o40755, children={nfd_name: Node(0o100644)}) - assert lookup_child(only_nfd, nfd_name)[0] == nfd_name - assert lookup_child(only_nfd, nfc_name)[0] == nfd_name # the other way round, too - - # an archive may contain both spellings (they are different names on e.g. Linux): - # each one resolves to itself, exactly. - both = Node(0o40755, children={nfc_name: Node(0o100644), nfd_name: Node(0o100644)}) - assert lookup_child(both, nfc_name)[0] == nfc_name - assert lookup_child(both, nfd_name)[0] == nfd_name - - # a name with non-UTF-8 bytes (surrogate escapes) must not break normalization - weird = b"bad\xff.txt".decode("utf-8", "surrogateescape") - assert lookup_child(Node(0o40755, children={weird: Node(0o100644)}), weird)[0] == weird - - with pytest.raises(KeyError): - lookup_child(only_nfc, "no-such-file.txt") - - def test_webdav_file_without_chunks(archivers, request): # An anomalous item - size > 0 but no chunks list (e.g. corrupted metadata) - must not # leave the client hanging after an advertised Content-Length: the server aborts the @@ -566,9 +540,12 @@ def test_webdav_file_without_chunks(archivers, request): manifest = Manifest.load(repository, Manifest.NO_OPERATION_CHECK) server = make_server(manifest, args, port=0) # corrupt the in-memory tree: pretend file1 is non-empty but has no chunks - node, _, _ = server.RequestHandlerClass.vfs.resolve(["test", "input", "file1"]) - node.size = 5 - node.chunks = None + vfs = server.RequestHandlerClass.vfs + _, node = vfs.resolve(["test", "input", "file1"]) + item = vfs.get_item(node.ino) + item.size = 5 + del item.chunks + vfs._set_item(node.ino, item) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: diff --git a/src/borg/testsuite/fuse_test.py b/src/borg/testsuite/fuse_test.py index dda2b24c40..e18440dc24 100644 --- a/src/borg/testsuite/fuse_test.py +++ b/src/borg/testsuite/fuse_test.py @@ -1,38 +1,35 @@ -"""Tests for the FUSE filesystem implementations that do not need an actual mount. +"""Tests for the FUSE adapters that do not need an actual mount. -The ACL xattrs (see ACL_XATTRS) are emulated by the FUSE implementations, so we can -test them by directly calling the relevant methods with a fake item. Unlike the tests -in archiver/mount_cmds_test.py, this also works in environments where the kernel does -not offer ACL xattrs on FUSE mounts (e.g. mounts made inside a user namespace). +The archive VFS (see vfs_test.py) reports a missing xattr as a KeyError and an ACL it +can not convert as a ValueError - both FUSE adapters have to turn these into the right +errno, which is what we check here without mounting anything. """ import errno import pytest -from ..helpers import StableDict -from ..item import Item -from ..platform import acl_text_to_xattr from . import has_llfuse, has_pyfuse3, has_mfusepy, ENOATTR -from .platform.platform_test import skipif_not_linux - -# the ACL xattrs are only emulated on Linux: -pytestmark = skipif_not_linux skipif_no_llfuse_api = pytest.mark.skipif(not (has_llfuse or has_pyfuse3), reason="llfuse/pyfuse3 not available") skipif_no_mfusepy = pytest.mark.skipif(not has_mfusepy, reason="mfusepy not available") -ACCESS_ACL = b"user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-:0\ngroup:root:rw-:0\n" -DEFAULT_ACL = b"user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:r--:0\ngroup:root:r--:0\n" -BROKEN_ACL = b"flubber::rw-\n" # can not be converted to the binary xattr representation +XATTRS = [b"user.foo", b"system.posix_acl_access"] + + +class FakeVFS: + """Minimal stand-in for the archive VFS: answers the xattr calls, however we want.""" + def __init__(self, exception=None): + self.exception = exception -def make_item(*, acls=True, acl_access=ACCESS_ACL): - item = Item(path="file", mode=0o100666, mtime=0, xattrs=StableDict({b"user.foo": b"bar"})) - if acls: - item.acl_access = acl_access - item.acl_default = DEFAULT_ACL - return item + def listxattr(self, ino): + return list(XATTRS) + + def getxattr(self, ino, name): + if self.exception is not None: + raise self.exception + return b"bar" def unwrap(method): @@ -40,21 +37,18 @@ def unwrap(method): return getattr(method, "__wrapped__", method) -def llfuse_ops(item, numeric_ids=False): - """Minimal stand-in for FuseOperations, providing what listxattr/getxattr need.""" +def llfuse_ops(exception=None): from ..fuse import FuseOperations class Operations: pass ops = Operations() - ops.numeric_ids = numeric_ids - ops.get_item = lambda inode: item + ops.vfs = FakeVFS(exception) return ops, unwrap(FuseOperations.listxattr), unwrap(FuseOperations.getxattr) -def mfusepy_ops(item, numeric_ids=False): - """Minimal stand-in for borgfs, providing what listxattr/getxattr need.""" +def mfusepy_ops(exception=None): from ..hlfuse import borgfs class Node: @@ -64,91 +58,71 @@ class Operations: pass ops = Operations() - ops.numeric_ids = numeric_ids + ops.vfs = FakeVFS(exception) ops._find_node = lambda path: Node() - ops.get_inode = lambda ino: item return ops, borgfs.listxattr, borgfs.getxattr @skipif_no_llfuse_api -def test_llfuse_listxattr_acls(): - ops, listxattr, _ = llfuse_ops(make_item()) - assert sorted(listxattr(ops, 1)) == [b"system.posix_acl_access", b"system.posix_acl_default", b"user.foo"] +def test_llfuse_listxattr(): + ops, listxattr, _ = llfuse_ops() + assert listxattr(ops, 1) == XATTRS # the low-level API uses bytes names @skipif_no_llfuse_api -def test_llfuse_listxattr_no_acls(): - ops, listxattr, _ = llfuse_ops(make_item(acls=False)) - assert list(listxattr(ops, 1)) == [b"user.foo"] +def test_llfuse_getxattr(): + ops, _, getxattr = llfuse_ops() + assert getxattr(ops, 1, b"user.foo") == b"bar" @skipif_no_llfuse_api -@pytest.mark.parametrize("numeric_ids", [False, True]) -def test_llfuse_getxattr_acls(numeric_ids): - ops, _, getxattr = llfuse_ops(make_item(), numeric_ids=numeric_ids) - assert getxattr(ops, 1, b"system.posix_acl_access") == acl_text_to_xattr(ACCESS_ACL, numeric_ids=numeric_ids) - assert getxattr(ops, 1, b"system.posix_acl_default") == acl_text_to_xattr(DEFAULT_ACL, numeric_ids=numeric_ids) - assert getxattr(ops, 1, b"user.foo") == b"bar" # "normal" xattrs still work - - -@skipif_no_llfuse_api -def test_llfuse_getxattr_acl_missing(): +def test_llfuse_getxattr_missing(): from ..fuse_impl import llfuse - ops, _, getxattr = llfuse_ops(make_item(acls=False)) + ops, _, getxattr = llfuse_ops(KeyError(b"user.foo")) with pytest.raises(llfuse.FUSEError) as excinfo: - getxattr(ops, 1, b"system.posix_acl_access") + getxattr(ops, 1, b"user.foo") assert excinfo.value.errno == ENOATTR @skipif_no_llfuse_api -def test_llfuse_getxattr_acl_broken(): +def test_llfuse_getxattr_broken_acl(): from ..fuse_impl import llfuse - ops, _, getxattr = llfuse_ops(make_item(acl_access=BROKEN_ACL)) + ops, _, getxattr = llfuse_ops(ValueError("can not convert this ACL")) with pytest.raises(llfuse.FUSEError) as excinfo: getxattr(ops, 1, b"system.posix_acl_access") assert excinfo.value.errno == errno.EIO @skipif_no_mfusepy -def test_mfusepy_listxattr_acls(): - ops, listxattr, _ = mfusepy_ops(make_item()) - assert sorted(listxattr(ops, "/file")) == ["system.posix_acl_access", "system.posix_acl_default", "user.foo"] - - -@skipif_no_mfusepy -def test_mfusepy_listxattr_no_acls(): - ops, listxattr, _ = mfusepy_ops(make_item(acls=False)) - assert list(listxattr(ops, "/file")) == ["user.foo"] +def test_mfusepy_listxattr(): + ops, listxattr, _ = mfusepy_ops() + # the high-level API uses str names + assert listxattr(ops, "/file") == ["user.foo", "system.posix_acl_access"] @skipif_no_mfusepy -@pytest.mark.parametrize("numeric_ids", [False, True]) -def test_mfusepy_getxattr_acls(numeric_ids): - ops, _, getxattr = mfusepy_ops(make_item(), numeric_ids=numeric_ids) - access = getxattr(ops, "/file", "system.posix_acl_access") - assert access == acl_text_to_xattr(ACCESS_ACL, numeric_ids=numeric_ids) - default = getxattr(ops, "/file", "system.posix_acl_default") - assert default == acl_text_to_xattr(DEFAULT_ACL, numeric_ids=numeric_ids) - assert getxattr(ops, "/file", "user.foo") == b"bar" # "normal" xattrs still work +def test_mfusepy_getxattr(): + ops, _, getxattr = mfusepy_ops() + assert getxattr(ops, "/file", "user.foo") == b"bar" @skipif_no_mfusepy -def test_mfusepy_getxattr_acl_missing(): +def test_mfusepy_getxattr_missing(): from ..fuse_impl import hlfuse - ops, _, getxattr = mfusepy_ops(make_item(acls=False)) + ops, _, getxattr = mfusepy_ops(KeyError(b"user.foo")) with pytest.raises(hlfuse.FuseOSError) as excinfo: - getxattr(ops, "/file", "system.posix_acl_default") + getxattr(ops, "/file", "user.foo") assert excinfo.value.errno == ENOATTR @skipif_no_mfusepy -def test_mfusepy_getxattr_acl_broken(): +def test_mfusepy_getxattr_broken_acl(): from ..fuse_impl import hlfuse - ops, _, getxattr = mfusepy_ops(make_item(acl_access=BROKEN_ACL)) + ops, _, getxattr = mfusepy_ops(ValueError("can not convert this ACL")) with pytest.raises(hlfuse.FuseOSError) as excinfo: getxattr(ops, "/file", "system.posix_acl_access") assert excinfo.value.errno == errno.EIO diff --git a/src/borg/testsuite/vfs_test.py b/src/borg/testsuite/vfs_test.py new file mode 100644 index 0000000000..edef69485f --- /dev/null +++ b/src/borg/testsuite/vfs_test.py @@ -0,0 +1,114 @@ +"""Tests for the archive VFS core (vfs.py) that do not need a repository.""" + +import unicodedata + +import pytest + +from ..helpers import StableDict +from ..item import Item +from ..platform import acl_text_to_xattr +from ..vfs import VFSNode, item_getxattr, item_listxattr, lookup_child, versioned_name +from .platform.platform_test import skipif_not_linux + +ACCESS_ACL = b"user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:rw-:0\ngroup:root:rw-:0\n" +DEFAULT_ACL = b"user::rw-\ngroup::r--\nmask::rw-\nother::---\nuser:root:r--:0\ngroup:root:r--:0\n" +BROKEN_ACL = b"flubber::rw-\n" # can not be converted to the binary xattr representation + + +def make_item(*, acls=True, acl_access=ACCESS_ACL): + item = Item(path="file", mode=0o100666, mtime=0, xattrs=StableDict({b"user.foo": b"bar"})) + if acls: + item.acl_access = acl_access + item.acl_default = DEFAULT_ACL + return item + + +def make_dir(names): + """Build a directory node with a leaf child for each name.""" + node = VFSNode(1, is_dir=True) + for ino, name in enumerate(names, start=2): + node.children[name] = VFSNode(ino, node) + return node + + +def test_versioned_name(): + assert versioned_name("file.txt", 1) == "file.00001.txt" # the extension stays at the end + assert versioned_name("file", 42) == "file.00042" + + +def test_lookup_child(): + # exact matches always win, and names that are ambiguous after normalization are only + # reachable by their exact spelling, so we never serve a different file than requested. + nfc_name = "grüße.txt" # composed + nfd_name = unicodedata.normalize("NFD", nfc_name) # decomposed + assert nfc_name != nfd_name + + only_nfc = make_dir([nfc_name]) + # the stored (composed) name is found by both spellings + assert lookup_child(only_nfc, nfc_name)[0] == nfc_name + assert lookup_child(only_nfc, nfd_name)[0] == nfc_name + + only_nfd = make_dir([nfd_name]) + assert lookup_child(only_nfd, nfd_name)[0] == nfd_name + assert lookup_child(only_nfd, nfc_name)[0] == nfd_name # the other way round, too + + # an archive may contain both spellings (they are different names on e.g. Linux): + # each one resolves to itself, exactly. + both = make_dir([nfc_name, nfd_name]) + assert lookup_child(both, nfc_name)[0] == nfc_name + assert lookup_child(both, nfd_name)[0] == nfd_name + + # a name with non-UTF-8 bytes (surrogate escapes) must not break normalization + weird = b"bad\xff.txt".decode("utf-8", "surrogateescape") + assert lookup_child(make_dir([weird]), weird)[0] == weird + + with pytest.raises(KeyError): + lookup_child(only_nfc, "no-such-file.txt") + + with pytest.raises(KeyError): + lookup_child(VFSNode(1), "anything") # not a directory + + +# The ACL xattrs (see vfs.ACL_XATTRS) are emulated by the VFS, so we can test them by +# calling the xattr methods with a fake item. Unlike the tests in archiver/mount_cmds_test.py, +# this also works in environments where the kernel does not offer ACL xattrs on FUSE mounts +# (e.g. mounts made inside a user namespace). They are only emulated on Linux: + + +@skipif_not_linux +def test_listxattr_acls(): + names = item_listxattr(make_item()) + assert sorted(names) == [b"system.posix_acl_access", b"system.posix_acl_default", b"user.foo"] + + +@skipif_not_linux +def test_listxattr_no_acls(): + assert item_listxattr(make_item(acls=False)) == [b"user.foo"] + + +@skipif_not_linux +@pytest.mark.parametrize("numeric_ids", [False, True]) +def test_getxattr_acls(numeric_ids): + item = make_item() + access = item_getxattr(item, b"system.posix_acl_access", numeric_ids=numeric_ids) + assert access == acl_text_to_xattr(ACCESS_ACL, numeric_ids=numeric_ids) + default = item_getxattr(item, b"system.posix_acl_default", numeric_ids=numeric_ids) + assert default == acl_text_to_xattr(DEFAULT_ACL, numeric_ids=numeric_ids) + assert item_getxattr(item, b"user.foo") == b"bar" # "normal" xattrs still work + + +@skipif_not_linux +def test_getxattr_acl_missing(): + with pytest.raises(KeyError): + item_getxattr(make_item(acls=False), b"system.posix_acl_access") + + +@skipif_not_linux +def test_getxattr_acl_broken(): + with pytest.raises(ValueError): + item_getxattr(make_item(acl_access=BROKEN_ACL), b"system.posix_acl_access") + + +def test_getxattr_missing(): + with pytest.raises(KeyError): + item_getxattr(make_item(), b"user.nope") diff --git a/src/borg/vfs.py b/src/borg/vfs.py new file mode 100644 index 0000000000..e15a091b72 --- /dev/null +++ b/src/borg/vfs.py @@ -0,0 +1,723 @@ +""" +The read-only "archive as a file system" implementation, shared by all its users. + +borg presents archive contents as a browsable directory tree in several places: + +- ``borg mount`` via mfusepy (high-level FUSE 2/3), see hlfuse.py +- ``borg mount`` via llfuse / pyfuse3 (low-level FUSE 2/3), see fuse.py +- ``borg webdav`` (WebDAV / HTTP server), see webdav.py + +They all need the same thing: build a directory tree from the items of the selected +archives, map borg item metadata to file attributes and read file content out of the +chunk lists. That is what this module does - the modules above are thin protocol +adapters over it, so "what an archive looks like as a file system" is defined here +and only here. + +Threading: an ArchiveVFS is not thread-safe by itself. Users that access it from +multiple threads (webdav) pass a lock; all repository access (building an archive's +tree, fetching chunks) happens while holding it, because borgstore connections are +not thread-safe. The lock is also used to keep the repository lock refresher off the +repository while a request is being served, see storelocking.LockRefresher. +""" + +import hashlib +import os +import stat +import threading +import time +import unicodedata +from collections import Counter +from typing import NamedTuple + +from .archive import Archive, DownloadPipeline, get_item_uid_gid +from .constants import ROBJ_FILE_STREAM +from .hashindex import FuseVersionsIndex +from .helpers import Error, bin_to_hex, format_file_size, remove_surrogates +from .helpers import msgpack +from .helpers.lrucache import LRUCache +from .item import ChunkListEntry, Item +from .logger import create_logger +from .platform import acl_text_to_xattr, uid2user, gid2group +from .platformflags import is_darwin, is_linux, is_win32 + +logger = create_logger() + +# on Linux, the kernel exposes POSIX ACLs via these special, binary encoded xattrs. +# maps the xattr name to the borg item attribute holding the ACL text. +# empty on platforms we can not do this for, so the mount just does not offer these xattrs there. +ACL_XATTRS = {b"system.posix_acl_access": "acl_access", b"system.posix_acl_default": "acl_default"} if is_linux else {} + +DEFAULT_DIR_MODE = 0o40755 + +# size of the cache of unpacked items (see ArchiveVFS.get_item). +# note: such an item can be rather large - Item.chunks can be a long list. +ITEM_CACHE_SIZE = 8 + + +class ChunkMissing(Exception): + """A chunk of a file's content is missing in the repository.""" + + +class VFSOptions: + """How the archive contents shall be presented; the defaults suit a plain, read-only view. + + *versions* merges all selected archives into one tree, showing each file as a + directory that contains all its versions (see ArchiveVFS._load_archive). + *numeric_ids*, *uid_forced*, *gid_forced* and *umask* control the ownership and + permissions mapping, *strip_components* and *item_filter* which items are shown, + *allow_damaged_files* whether reads of files with missing chunks return zeros + instead of failing, and *dir_item* is the item used for synthesized directories. + """ + + def __init__( + self, + *, + versions=False, + numeric_ids=False, + uid_forced=None, + gid_forced=None, + umask=0, + allow_damaged_files=False, + strip_components=0, + item_filter=None, + dir_item=None, + ): + self.versions = versions + self.numeric_ids = numeric_ids + self.uid_forced = uid_forced + self.gid_forced = gid_forced + self.umask = umask + self.allow_damaged_files = allow_damaged_files + self.strip_components = strip_components + self.item_filter = item_filter + self.dir_item = dir_item + + +class Attrs(NamedTuple): + """The file attributes of a node, as far as an archived item defines them.""" + + ino: int + mode: int + nlink: int + uid: int + gid: int + rdev: int + size: int + mtime_ns: int + atime_ns: int + ctime_ns: int + birthtime_ns: int + + +class VFSNode: + """A node of the file system tree: a directory (children is a dict) or a leaf. + + A node only knows its place in the tree - the item metadata lives in ArchiveVFS, + keyed by the inode number. Hard links to the same archived file are separate nodes + sharing one inode number (and thus one item). + """ + + __slots__ = ("ino", "parent", "children", "nfc_names") + + def __init__(self, ino, parent=None, *, is_dir=False): + self.ino = ino + self.parent = parent + self.children = {} if is_dir else None # name (str) -> VFSNode + self.nfc_names = None # lazily built by ArchiveVFS.lookup(), see there + + @property + def is_dir(self): + return self.children is not None + + +def nfc(name): + """NFC-normalize *name*; names with surrogates (non-UTF-8 bytes) pass through unchanged.""" + return unicodedata.normalize("NFC", name) + + +def item_listxattr(item): + """Return the xattr names (bytes) of *item*.""" + names = list(item.get("xattrs", {}).keys()) + # expose the archived POSIX ACLs, so e.g. getfacl or tools copying from the mount can read them. + names.extend(xattr_name for xattr_name, attr in ACL_XATTRS.items() if attr in item) + return names + + +def item_getxattr(item, name, *, numeric_ids=False): + """Return the value of xattr *name* (bytes) of *item*. + + Raises KeyError if the item does not have that xattr, ValueError if an archived + ACL can not be converted to the binary xattr representation. + """ + if name in ACL_XATTRS: + acl = item.get(ACL_XATTRS[name]) + if acl is None: + raise KeyError(name) + return acl_text_to_xattr(acl, numeric_ids=numeric_ids) # raises ValueError + return item.get("xattrs", {})[name] or b"" + + +def lookup_child(node, name): + """Return (child_name, child_node) for *name* in directory *node*. Raises KeyError. + + An exact match always wins. If there is none, we retry comparing Unicode NFC forms: + file systems disagree about normalization (macOS decomposes, so its clients ask for + "gru..." where the archive stores "grü..."), and an exact-only + lookup would answer "not found" for a file the client just saw in a listing. + + The NFC index is built lazily (only for directories that actually get such a request) + and skips names that are ambiguous, i.e. where several different names share one NFC + form - those keep requiring an exact match, so we never silently serve the wrong file. + """ + children = node.children + if children is None: + raise KeyError(name) # not a directory + try: + return name, children[name] + except KeyError: + pass + if node.nfc_names is None: + index = {} + for child_name in children: + key = nfc(child_name) + index[key] = None if key in index else child_name # None marks an ambiguous key + node.nfc_names = index + child_name = node.nfc_names.get(nfc(name)) + if child_name is None: + raise KeyError(name) + return child_name, children[child_name] + + +def versioned_name(name, version): + """Build the name a file has in the versions view, e.g. "file.00001.txt".""" + # keep the original extension at the end, to avoid confusing tools + name, ext = os.path.splitext(name) + return f"{name}.{version:05d}{ext}" + + +class ArchiveVFS: + """A read-only file system view of the archives selected by *args*. + + The root directory has one directory per (deduplicated) archive name; each of + these archive trees is built when it is first accessed - building it means + reading the whole item metadata stream of that archive, which takes a while for + a big archive. In the versions view, the contents of all selected archives are + merged into one tree instead, which is built completely upfront. + + Directories are addressable by inode number (get_node); everything else is + reached by looking up names in a directory node (lookup / resolve). + """ + + def __init__(self, manifest, args, repository=None, *, lock=None, options=None): + self.manifest = manifest + self.repository = repository if repository is not None else manifest.repository + self.args = args + self.options = options if options is not None else VFSOptions() + self.lock = lock if lock is not None else threading.RLock() + self.pipeline = DownloadPipeline(self.repository, manifest.repo_objs) + self.reader = DataReader(self.pipeline, lock=self.lock, allow_damaged_files=self.options.allow_damaged_files) + # the ids of the mounting user are what items without a usable uid/gid fall back to. + # note: borg webdav also runs on Windows, which has no uid/gid (nor os.getuid). + self.default_uid = 0 if is_win32 else os.getuid() + self.default_gid = 0 if is_win32 else os.getgid() + self.archives = {} # display name -> ArchiveInfo + self.root_mtime = int(time.time() * 1e9) # replaced by create_filesystem() + self._ino_count = 0 + self._items = {} # inode number -> msgpacked item (None -> options.dir_item) + self._item_cache = LRUCache(capacity=ITEM_CACHE_SIZE) # inode number -> Item + self._nodes = {} # inode number -> VFSNode, for directories only + self._nlinks = {} # inode number -> link count, for hard linked items only + self._pending = {} # VFSNode -> ArchiveInfo, archives whose tree is not built yet + self._versions_index = None + self.root = self._new_node(None, is_dir=True) + assert self.root.ino == 1 # the root inode number is 1 by convention + + # -- setting up --------------------------------------------------------------- + + def create_filesystem(self): + """Populate the root directory (this also selects the archives to be shown).""" + archives = self.manifest.archives.list_considering(self.args) + # archives of a series all have the same name, so make the display names unique + name_counter = Counter(archive.name for archive in archives) + for archive in archives: + name = archive.name + if name_counter[name] > 1: + name += f"-{bin_to_hex(archive.id):.8}" + self.archives[name] = archive + timestamps = [archive.ts for archive in archives] + self.root_mtime = int(max(timestamps).timestamp() * 1e9) if timestamps else int(time.time() * 1e9) + self._set_item(self.root.ino, self._dir_item(self.root_mtime)) + if self.options.versions: + # the versions view merges all archives into one tree, so there is nothing + # to defer: build it now. + self._versions_index = FuseVersionsIndex() + for archive in self.archives.values(): + self._load_archive(archive, self.root) + else: + for name, archive in self.archives.items(): + node = self._new_node(self.root, is_dir=True) + self._set_item(node.ino, self._dir_item(int(archive.ts.timestamp() * 1e9))) + self.root.children[name] = node + self._pending[node] = archive + + def _dir_item(self, mtime): + """Return the item used for a synthesized directory, with mtime *mtime*.""" + item = Item(internal_dict=self._default_dir().as_dict()) + item.mtime = mtime + return item + + def _default_dir(self): + """The item used for all synthesized directories that keep the default mtime.""" + if self.options.dir_item is None: + uid = self.options.uid_forced if self.options.uid_forced is not None else self.default_uid + gid = self.options.gid_forced if self.options.gid_forced is not None else self.default_gid + self.options.dir_item = Item( + mode=DEFAULT_DIR_MODE & ~self.options.umask, mtime=int(time.time() * 1e9), uid=uid, gid=gid + ) + return self.options.dir_item + + # -- building the tree -------------------------------------------------------- + + def ensure_loaded(self, node): + """Build the tree of an archive directory, if that did not happen yet.""" + if node not in self._pending: + return # not an archive directory, or its tree is complete + with self.lock: + archive = self._pending.get(node) + if archive is None: + return # another thread built it while we waited for the lock + self._load_archive(archive, node) + # only now (with the tree complete) let other threads past the check above. + del self._pending[node] + + def _load_archive(self, archive_info, root): + """Add the items of one archive to the tree below *root*.""" + t0 = time.perf_counter() + archive = Archive(self.manifest, archive_info.id) + archive_mtime = int(archive_info.ts.timestamp() * 1e9) + strip_components = self.options.strip_components + versions = self.options.versions + hardlinks = {} # hlid -> node of the first item with that hlid + for item in archive.iter_items(self.options.item_filter): + if strip_components: + item.path = os.sep.join(item.path.split(os.sep)[strip_components:]) + segments = [segment for segment in item.path.split("/") if segment] + if not segments: + continue + parent = self._make_dirs(root, segments[:-1], mtime=archive_mtime) + name = segments[-1] + if stat.S_ISDIR(item.mode): + # the directory may have been synthesized already (as the parent of an + # item that came first): then just replace its item metadata. + node = parent.children.get(name) + if node is None or not node.is_dir: + node = self._new_node(parent, is_dir=True) + parent.children[name] = node + self._set_item(node.ino, item) + continue + if versions: + # the file name becomes a directory that contains all versions of that file + parent = self._make_dirs(parent, [name]) + version = self._file_version(item, item.path) + if version is not None: # a regular file, with contents + name = versioned_name(name, version) + hlid = item.get("hlid") + if hlid is not None: + first = hardlinks.get(hlid) + if first is not None: + # another link to a file we already have: share its inode (and item), + # so the mount shows the hard link as such. + self._nlinks[first.ino] = self._nlinks.get(first.ino, 1) + 1 + parent.children[name] = VFSNode(first.ino, parent) + continue + node = self._new_node(parent, is_dir=False) + self._set_item(node.ino, item) + parent.children[name] = node + if hlid is not None: + hardlinks[hlid] = node + duration = time.perf_counter() - t0 + logger.debug("vfs: built the tree of archive %s in %.1f s", remove_surrogates(archive_info.name), duration) + + def _make_dirs(self, node, segments, mtime=None): + """Return the directory at *segments* below *node*, synthesizing what is missing. + + A synthesized directory gets an item of its own if *mtime* is given (so it can show + the time of the archive it belongs to) - otherwise it shares the default directory + item, which is what the many intermediate directories of the versions view do. + """ + for segment in segments: + child = node.children.get(segment) + if child is None or not child.is_dir: + # a directory we have not seen an item for (yet): synthesize it. + child = self._new_node(node, is_dir=True) + if mtime is not None: + self._set_item(child.ino, self._dir_item(mtime)) + node.children[segment] = child + node = child + return node + + def _new_node(self, parent, *, is_dir): + self._ino_count += 1 + node = VFSNode(self._ino_count, parent, is_dir=is_dir) + if is_dir: + self._nodes[node.ino] = node + return node + + def _file_version(self, item, path): + """Return the version number of *item* in the versions view (None if it has no contents). + + Files are versioned by their content: a file keeps its version number as long as + its chunk list does not change from one archive to the next. + """ + if "chunks" not in item: + return None + # note: using sha256 here because nowadays it is often hw accelerated. + # shortening the hashes to 16 bytes to save some memory. + file_id = hashlib.sha256(path.encode("utf-8", "surrogateescape")).digest()[:16] + current_version, previous_id = self._versions_index.get(file_id, (0, None)) + contents_id = hashlib.sha256(b"".join(chunk_id for chunk_id, _ in item.chunks)).digest()[:16] + if contents_id != previous_id: + current_version += 1 + self._versions_index[file_id] = current_version, contents_id + return current_version + + # -- looking things up -------------------------------------------------------- + + def get_node(self, ino): + """Return the directory node with inode number *ino*. Raises KeyError.""" + return self._nodes[ino] + + def lookup(self, node, name): + """Return (canonical_name, child_node) for *name* in directory *node*. Raises KeyError.""" + self.ensure_loaded(node) + return lookup_child(node, name) + + def resolve(self, segments, node=None): + """Resolve path *segments* to (canonical_segments, node). Raises KeyError. + + *canonical_segments* are the names as they are stored in the archive, which may + differ from what was asked for, see lookup(). + """ + node = self.root if node is None else node + canonical = [] + for segment in segments: + name, node = self.lookup(node, segment) + canonical.append(name) + return canonical, node + + def children(self, node): + """Return the [(name, child_node), ...] of directory *node*.""" + self.ensure_loaded(node) + return list(node.children.items()) + + def get_item(self, ino): + """Return the item of the node with inode number *ino*.""" + packed = self._items.get(ino) + if packed is None: + return self._default_dir() # a synthesized directory + # note: two threads can miss on the same inode and both unpack and store the item. + # that is a bit of duplicated work, but both get a valid item. + item = self._item_cache.get(ino) + if item is None: + item = Item(internal_dict=msgpack.unpackb(packed)) + if "chunks" in item: # msgpack does not know about our namedtuple + item.chunks = [ChunkListEntry(*chunk) for chunk in item.chunks] + self._item_cache[ino] = item + return item + + def _set_item(self, ino, item): + item_dict = item.as_dict() + # the path is already encoded in the tree structure, so drop it to save memory. + item_dict.pop("path", None) + self._items[ino] = msgpack.packb(item_dict) + self._item_cache.pop(ino, None) + + # -- item metadata ------------------------------------------------------------ + + def attrs(self, ino): + """Return the file attributes of the node with inode number *ino*.""" + item = self.get_item(ino) + options = self.options + mode = item.mode & ~options.umask + nlink = self._nlinks.get(ino, 1) + if stat.S_ISDIR(mode): + nlink = max(nlink, 2) # every directory has at least "." and its entry in the parent + uid, gid = get_item_uid_gid( + item, + numeric=options.numeric_ids, + uid_default=self.default_uid, + gid_default=self.default_gid, + uid_forced=options.uid_forced, + gid_forced=options.gid_forced, + ) + # note: old archives only have mtime (not atime nor ctime nor birthtime) + mtime_ns = item.mtime + return Attrs( + ino=ino, + mode=mode, + nlink=nlink, + uid=uid, + gid=gid, + rdev=item.get("rdev", 0), + size=item.get_size(), + mtime_ns=mtime_ns, + atime_ns=item.get("atime", mtime_ns), + ctime_ns=item.get("ctime", mtime_ns), + birthtime_ns=item.get("birthtime", mtime_ns), + ) + + def listxattr(self, ino): + """Return the xattr names (bytes) of the node with inode number *ino*.""" + return item_listxattr(self.get_item(ino)) + + def getxattr(self, ino, name): + """Return the value of xattr *name* (bytes) of the node with inode number *ino*. + + Raises KeyError if the item does not have that xattr, ValueError if an archived + ACL can not be converted to the binary xattr representation. + """ + return item_getxattr(self.get_item(ino), name, numeric_ids=self.options.numeric_ids) + + def readlink(self, ino): + """Return the target of the symlink with inode number *ino*.""" + return self.get_item(ino).target + + def log_stats(self): + """Log some statistics about this file system (used by the SIGUSR1/SIGINFO handler).""" + logger.debug( + "vfs: %d inodes, %d items (%s), %d archives not loaded yet", + self._ino_count, + len(self._items), + format_file_size(sum(len(packed) for packed in self._items.values())), + len(self._pending), + ) + data_cache = self.reader.data_cache + logger.debug( + "vfs: data cache: %d/%d entries, %s", + len(data_cache), + data_cache._capacity, + format_file_size(sum(len(chunk) for _, chunk in data_cache.items())), + ) + + # -- file contents ------------------------------------------------------------ + + def read(self, ino, offset, size, *, pos_key=None): + """Return *size* bytes at *offset* of the file with inode number *ino*.""" + item = self.get_item(ino) + return self.reader.read(item.get("chunks") or [], offset, size, pos_key=pos_key) + + +class DataReader: + """Reads file content out of the chunk lists of items. + + Recently used chunks are kept decrypted in an in-memory cache, so the many small, + sequential reads a file system does for one big file do not fetch and decrypt the + same chunk over and over. This complements, but does not duplicate, the repository's + own pack cache: that one caches the raw (still encrypted and compressed) on-disk + bytes, whereas we cache the output of the per-chunk RepoObj.parse() that runs on + every fetch even on a pack cache hit. So the two layers save different work + (repository I/O vs. per-chunk decrypt + decompress). + + All repository access happens under *lock* (borgstore connections are not thread-safe), + but the data is returned to the caller outside of it, so a slow consumer (e.g. a slow + network client) does not block other users of the same repository. The caches are + thread-safe by themselves, see LRUCache. + """ + + # how many chunk positions to remember for sequential reads (1 per open file) + POSITIONS = 8 + + def __init__(self, pipeline, *, lock=None, capacity=None, allow_damaged_files=False): + self.pipeline = pipeline + self.lock = lock if lock is not None else threading.RLock() + capacity = data_cache_capacity() if capacity is None else capacity + logger.debug("vfs: data cache capacity: %d chunks", capacity) + self.data_cache = LRUCache(capacity=capacity) # chunk id -> decrypted chunk data + self.allow_damaged_files = allow_damaged_files + # remembers where a sequential reader is within the chunk list, so a read does + # not have to walk the list from the beginning again (that would be quadratic). + self._last_pos = LRUCache(capacity=self.POSITIONS) + + def read(self, chunks, offset, size, *, pos_key=None): + """Return *size* bytes at *offset* of the content described by the *chunks* list.""" + return b"".join(self.iter_data(chunks, offset, size, pos_key=pos_key)) + + def iter_data(self, chunks, offset, size, *, pos_key=None): + """Yield the bytes at [offset, offset+size) of the content described by *chunks*. + + *pos_key* identifies the reader (e.g. a file handle) for the sequential-read + optimization. Raises ChunkMissing if a chunk is not in the repository (unless + allow_damaged_files is set - then all-zero data is returned for it). + """ + if size <= 0: + return + chunk_no, chunk_offset = self._get_pos(pos_key) if pos_key is not None else (0, 0) + if chunk_offset > offset: + # this is not a sequential read, so we lost track and start from the beginning again + chunk_no, chunk_offset = 0, 0 + offset -= chunk_offset + # note: using index iteration to avoid frequently copying big (sub)lists by slicing + for idx in range(chunk_no, len(chunks)): + entry = chunks[idx] + if entry.size <= offset: + offset -= entry.size + chunk_offset += entry.size + chunk_no += 1 + continue + n = min(size, entry.size - offset) + data = self._get_chunk(entry, fully_read=offset + n >= entry.size) + yield data[offset : offset + n] + offset = 0 + size -= n + if not size: + if pos_key is not None: + self._set_pos(pos_key, (chunk_no, chunk_offset)) + break + + def _get_chunk(self, entry, fully_read=False): + """Return the decrypted data of chunk *entry*, using (and updating) the data cache. + + A chunk that is *fully_read* is not worth keeping: a sequential reader is done + with it, so it is dropped from the cache instead of evicting something else. + """ + with self.lock: + try: + data = self.data_cache[entry.id] + except KeyError: + pass + else: + if fully_read: + del self.data_cache[entry.id] + return data + data = next( + self.pipeline.fetch_many([entry], ro_type=ROBJ_FILE_STREAM, replacement_chunk=self.allow_damaged_files) + ) + if data is None: + raise ChunkMissing(entry.id) + if not fully_read: + self.data_cache[entry.id] = data + return data + + # the read positions are only a hint: a read that starts before the remembered + # position just starts over, so a racing update costs at most a rescan of the + # chunk list. + + def _get_pos(self, pos_key): + return self._last_pos.get(pos_key, (0, 0)) + + def _set_pos(self, pos_key, pos): + self._last_pos[pos_key] = pos + + def forget(self, pos_key): + """Forget the read position of *pos_key* (call this when a file is closed).""" + self._last_pos.pop(pos_key, None) + + +def data_cache_capacity(): + # number of decrypted chunks to keep cached, see borg mount --help. + # default: number of CPUs (a small, non-zero value). + capacity = int(os.environ.get("BORG_MOUNT_DATA_CACHE_ENTRIES", os.cpu_count() or 1)) + return max(1, capacity) + + +def build_item_filter(args): + """Build the item filter selecting the paths/patterns given on the command line.""" + # lazy import: pulling in the archiver package at module import time would be heavy + # (it defines all subcommands) and risks an import cycle. + from .archiver._common import build_matcher, build_filter + + # omitting args.pattern_roots here, restricting to paths only by cli args.paths: + matcher = build_matcher(getattr(args, "patterns", None) or [], getattr(args, "paths", None) or []) + return build_filter(matcher, getattr(args, "strip_components", 0)) + + +def pop_option(options, key, present, not_present, wanted_type, int_base=0): + """Remove mount option *key* from the *options* list and return its value.""" + assert isinstance(options, list) # we mutate this + for idx, option in enumerate(options): + if option == key: + options.pop(idx) + return present + if option.startswith(key + "="): + options.pop(idx) + value = option.split("=", 1)[1] + if wanted_type is bool: + v = value.lower() + if v in ("y", "yes", "true", "1"): + return True + if v in ("n", "no", "false", "0"): + return False + raise ValueError("unsupported value in option: %s" % option) + if wanted_type is int: + try: + return int(value, base=int_base) + except ValueError: + raise ValueError("unsupported value in option: %s" % option) from None + try: + return wanted_type(value) + except ValueError: + raise ValueError("unsupported value in option: %s" % option) from None + else: + return not_present + + +def parse_mount_options(args, mountpoint, mount_options): + """Process the "borg mount" options; returns (libfuse options, VFSOptions). + + The borg specific options are removed from the option list here and end up in the + VFSOptions, everything else is passed on to libfuse. + """ + # default_permissions enables permission checking by the kernel. Without + # this, any umask (or uid/gid) would not have an effect and this could + # cause security issues if used with allow_other mount option. + # When not using allow_other or allow_root, access is limited to the + # mounting user anyway. + options = ["fsname=borgfs", "ro", "default_permissions"] + if mount_options: + options.extend(mount_options.split(",")) + if is_darwin: + # macFUSE supports a volname mount option to give what finder displays on desktop / in directory list. + volname = pop_option(options, "volname", "", "", str) + # if the user did not specify it, we make something up, + # because otherwise it would be "macFUSE Volume 0 (Python)", #7690. + volname = volname or f"{os.path.basename(mountpoint)} (borgfs)" + options.append(f"volname={volname}") + ignore_permissions = pop_option(options, "ignore_permissions", True, False, bool) + if ignore_permissions: + # in case users have a use-case that requires NOT giving "default_permissions", + # this is enabled by the custom "ignore_permissions" mount option which just + # removes "default_permissions" again: + pop_option(options, "default_permissions", True, False, bool) + vfs_options = VFSOptions( + allow_damaged_files=pop_option(options, "allow_damaged_files", True, False, bool), + versions=pop_option(options, "versions", True, False, bool), + uid_forced=pop_option(options, "uid", None, None, int), + gid_forced=pop_option(options, "gid", None, None, int), + umask=pop_option(options, "umask", 0, 0, int, int_base=8), # umask is octal, e.g. 222 or 0222 + numeric_ids=getattr(args, "numeric_ids", False), + strip_components=getattr(args, "strip_components", 0), + item_filter=build_item_filter(args), + ) + dir_uid = vfs_options.uid_forced if vfs_options.uid_forced is not None else os.getuid() + dir_gid = vfs_options.gid_forced if vfs_options.gid_forced is not None else os.getgid() + dir_user = uid2user(dir_uid) + dir_group = gid2group(dir_gid) + if not isinstance(dir_user, str): + raise Error( + f"uid {dir_uid} can not be resolved to a username. " + f"Please check that the corresponding user exists or do not specify a uid mount option." + ) + if not isinstance(dir_group, str): + raise Error( + f"gid {dir_gid} can not be resolved to a group name. " + f"Please check that the corresponding group exists or do not specify a gid mount option." + ) + vfs_options.dir_item = Item( + mode=DEFAULT_DIR_MODE & ~vfs_options.umask, + mtime=int(time.time() * 1e9), + user=dir_user, + group=dir_group, + uid=dir_uid, + gid=dir_gid, + ) + return options, vfs_options diff --git a/src/borg/webdav.py b/src/borg/webdav.py index da2791d999..85435e2d3c 100644 --- a/src/borg/webdav.py +++ b/src/borg/webdav.py @@ -13,15 +13,14 @@ import html import mimetypes -import os import re import stat import tarfile import threading -import unicodedata from datetime import datetime, timezone from email.utils import formatdate from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import NamedTuple from urllib.parse import parse_qs, quote, unquote_to_bytes from xml.etree import ElementTree as ET # nosec B405 # only used to *build* response XML, never to parse input from xml.parsers import expat @@ -29,177 +28,26 @@ from . import __version__ from .archive import Archive from .constants import * # NOQA -from .helpers import bin_to_hex, remove_surrogates, HardLinkManager -from .helpers.lrucache import LRUCache +from .helpers import remove_surrogates, HardLinkManager from .logger import create_logger +from .vfs import ArchiveVFS, ChunkMissing, DEFAULT_DIR_MODE logger = create_logger(__name__) -DEFAULT_DIR_MODE = 0o40755 +class Resource(NamedTuple): + """What this server needs to know about one file system object it serves. -class Node: - """A node in an archive's directory tree: a directory (children is a dict) or a leaf.""" - - __slots__ = ("mode", "mtime", "size", "chunks", "target", "children", "nfc_names") - - def __init__(self, mode, mtime=0, size=0, chunks=None, target=None, children=None): - self.mode = mode - self.mtime = mtime # ns - self.size = size - self.chunks = chunks - self.target = target # symlink target - self.children = children # {name(str): Node} for directories, None otherwise - self.nfc_names = None # lazily built by lookup_child(), see there - - @property - def is_dir(self): - return self.children is not None - - -def nfc(name): - """NFC-normalize *name*; names with surrogates (non-UTF-8 bytes) pass through unchanged.""" - return unicodedata.normalize("NFC", name) - - -def lookup_child(node, name): - """Return (child_name, child_node) for *name* in directory *node*. Raises KeyError. - - An exact match always wins. If there is none, we retry comparing Unicode NFC forms: - file systems disagree about normalization (macOS decomposes, so its WebDAV client asks - for "gru..." where the archive stores "grü..."), and an exact-only - lookup would answer 404 for a file the client just saw in our listing. - - The NFC index is built lazily (only for directories that actually get such a request) - and skips names that are ambiguous, i.e. where several different names share one NFC - form - those keep requiring an exact match, so we never silently serve the wrong file. - """ - try: - return name, node.children[name] - except KeyError: - pass - if node.nfc_names is None: - index = {} - for child_name in node.children: - key = nfc(child_name) - index[key] = None if key in index else child_name # None marks an ambiguous key - node.nfc_names = index - child_name = node.nfc_names.get(nfc(name)) - if child_name is None: - raise KeyError(name) - return child_name, node.children[child_name] - - -class ArchiveVFS: - """Read-only view of the archives selected by the archive filter args. - - The top level maps (deduplicated) archive names to lazily built directory trees. - All repository access happens under repo_lock, because borgstore connections are - not thread-safe. + The tree itself lives in the (shared) archive VFS - this is just the flat view + of one node of it that building a listing, a PROPFIND response or a download + needs, so we do not look up the same item metadata over and over. """ - def __init__(self, manifest, args, repo_lock): - self.manifest = manifest - self.repo_lock = repo_lock - archives = manifest.archives.list_considering(args) - # deduplicate archive names (archives of a series all share the same name) - name_counter = {} - for archive in archives: - name_counter[archive.name] = name_counter.get(archive.name, 0) + 1 - self.archives = {} # display name -> ArchiveInfo - for archive in archives: - name = archive.name - if name_counter[name] > 1: - name += f"-{bin_to_hex(archive.id):.8}" - self.archives[name] = archive - self._trees = {} # display name -> (root Node, DownloadPipeline) - self.nfc_archives = None # lazily built by lookup_archive(), see there - timestamps = [archive.ts for archive in archives] - self.root_mtime = int(max(timestamps).timestamp() * 1e9) if timestamps else 0 - - def get_root(self, name): - """Return (root Node, pipeline) for archive *name*, building the tree on first access.""" - try: - return self._trees[name] - except KeyError: - pass - archive_info = self.archives[name] # may raise KeyError -> 404 - with self.repo_lock: - if name not in self._trees: # re-check under lock - self._trees[name] = self._build_tree(archive_info) - return self._trees[name] - - def _build_tree(self, archive_info): - logger.debug("webdav: building tree for archive %s ...", remove_surrogates(archive_info.name)) - archive = Archive(self.manifest, archive_info.id) - archive_mtime = int(archive_info.ts.timestamp() * 1e9) - root = Node(DEFAULT_DIR_MODE, mtime=archive_mtime, children={}) - for item in archive.iter_items(): - segments = [s for s in item.path.split("/") if s] - if not segments: - continue - node = root - for segment in segments[:-1]: - child = node.children.get(segment) - if child is None or not child.is_dir: - # intermediate directory not (yet) seen as an item: synthesize it - child = Node(DEFAULT_DIR_MODE, mtime=archive_mtime, children={}) - node.children[segment] = child - node = child - name = segments[-1] - if stat.S_ISDIR(item.mode): - existing = node.children.get(name) - if existing is not None and existing.is_dir: - # was synthesized before the dir item was seen: update metadata in place - existing.mode = item.mode - existing.mtime = item.mtime - else: - node.children[name] = Node(item.mode, mtime=item.mtime, children={}) - else: - node.children[name] = Node( - item.mode, - mtime=item.mtime, - size=item.get_size(), - chunks=item.get("chunks"), - target=item.get("target"), - ) - return root, archive.pipeline - - def resolve(self, segments): - """Resolve path segments (first one is the archive name) to (Node, pipeline, canonical). - - *canonical* are the segments as they are actually stored in the archive: a client may - address a path in a different Unicode normalization than the archive uses (see - lookup_child()), and the stored spelling is what path-based operations (e.g. the tar - download, which matches item paths) have to use. - - Raises KeyError if not found. - """ - archive_name = self.lookup_archive(segments[0]) - root, pipeline = self.get_root(archive_name) - node = root - canonical = [archive_name] - for segment in segments[1:]: - if not node.is_dir: - raise KeyError(segment) - name, node = lookup_child(node, segment) - canonical.append(name) - return node, pipeline, canonical - - def lookup_archive(self, name): - """Return the stored archive name matching *name*. Raises KeyError if there is none.""" - if name in self.archives: - return name - if self.nfc_archives is None: - index = {} - for archive_name in self.archives: - key = nfc(archive_name) - index[key] = None if key in index else archive_name - self.nfc_archives = index - archive_name = self.nfc_archives.get(nfc(name)) - if archive_name is None: - raise KeyError(name) - return archive_name + is_dir: bool + mode: int + mtime: int # ns + size: int + target: str | None # symlink target def strip_crlf(value): @@ -240,9 +88,9 @@ def guess_content_type(name): return mimetypes.guess_type(remove_surrogates(name), strict=False)[0] or "application/octet-stream" -def make_etag(node): +def make_etag(res): # archive contents are immutable, so mtime+size identify the content well enough. - return f'"{node.mtime:x}-{node.size:x}"' + return f'"{res.mtime:x}-{res.size:x}"' def parse_byte_range(header, size): @@ -357,33 +205,33 @@ def start_element(name, attrs): return "allprop", None # allprop, maybe with an include element -def make_prop_element(tag, name, node): - """Build the XML element for live property *tag* of resource *node*. +def make_prop_element(tag, name, res): + """Build the XML element for live property *tag* of resource *res*. Returns None if the property is not defined for this resource. """ elem = ET.Element(tag) if tag == "{DAV:}resourcetype": - if node.is_dir: + if res.is_dir: ET.SubElement(elem, "{DAV:}collection") elif tag == "{DAV:}displayname": elem.text = remove_surrogates(name) elif tag == "{DAV:}getlastmodified": - elem.text = http_date(node.mtime) + elem.text = http_date(res.mtime) elif tag == "{DAV:}creationdate": - elem.text = iso8601(node.mtime) + elem.text = iso8601(res.mtime) elif tag == "{DAV:}getcontentlength": - if node.is_dir: + if res.is_dir: return None - elem.text = str(node.size) + elem.text = str(res.size) elif tag == "{DAV:}getcontenttype": - if node.is_dir: + if res.is_dir: return None elem.text = guess_content_type(name) elif tag == "{DAV:}getetag": - if node.is_dir: + if res.is_dir: return None - elem.text = make_etag(node) + elem.text = make_etag(res) elif tag in ("{DAV:}supportedlock", "{DAV:}lockdiscovery"): pass # empty elements: locking is not supported else: @@ -394,22 +242,22 @@ def make_prop_element(tag, name, node): def render_multistatus(resources, mode, requested): """Render a PROPFIND result as a multistatus XML document (bytes). - *resources* is a list of (href, displayname, node) tuples, *mode* / *requested* + *resources* is a list of (href, displayname, Resource) tuples, *mode* / *requested* are the parse_propfind() results. """ multistatus = ET.Element("{DAV:}multistatus") - for href, name, node in resources: + for href, name, res in resources: response = ET.SubElement(multistatus, "{DAV:}response") ET.SubElement(response, "{DAV:}href").text = href found = ET.Element("{DAV:}prop") missing = ET.Element("{DAV:}prop") if mode == "propname": for tag in DAV_PROPS: - if make_prop_element(tag, name, node) is not None: + if make_prop_element(tag, name, res) is not None: ET.SubElement(found, tag) else: for tag in DAV_PROPS if mode == "allprop" else requested: - elem = make_prop_element(tag, name, node) + elem = make_prop_element(tag, name, res) if elem is not None: found.append(elem) else: @@ -549,9 +397,7 @@ class WebDAVHandler(BaseHTTPRequestHandler): sys_version = "" # do not tell clients about the python version we use # set on the handler class by make_server(): - vfs = None - repo_lock = None - data_cache = None # LRUCache: chunk id -> decrypted chunk data, shared across requests + vfs = None # the shared archive VFS, see vfs.py def version_string(self): # the base class would append sys_version, giving a trailing space if it is empty. @@ -652,11 +498,12 @@ def _handle_get_head(self, head): try: # use the canonical (as stored in the archive) segments from here on, so that a # request in a different Unicode normalization still names the stored items. - node, pipeline, segments = self.vfs.resolve(segments) + segments, node = self.vfs.resolve(segments) except KeyError: self.send_error(404) return - if node.is_dir: + res = self._resource(node) + if res.is_dir: if not dir_syntax: self._redirect_to_dir(segments) return @@ -664,15 +511,22 @@ def _handle_get_head(self, head): self._send_tar(segments, head) return self._send_dir_listing(segments, node, head) - elif stat.S_ISREG(node.mode): - self._send_file(segments[-1], node, pipeline, head) - elif stat.S_ISLNK(node.mode): + elif stat.S_ISREG(res.mode): + self._send_file(segments[-1], node, res, head) + elif stat.S_ISLNK(res.mode): self.send_error( - 403, explain=f"symbolic link (target: {remove_surrogates(node.target or '?')}), not downloadable" + 403, explain=f"symbolic link (target: {remove_surrogates(res.target or '?')}), not downloadable" ) else: self.send_error(403, explain="special file, not downloadable") + def _resource(self, node): + """Return the Resource describing *node*.""" + item = self.vfs.get_item(node.ino) + return Resource( + is_dir=node.is_dir, mode=item.mode, mtime=item.mtime, size=item.get_size(), target=item.get("target") + ) + def _handle_propfind(self): body = self._read_body() if body is None: @@ -704,29 +558,32 @@ def _handle_propfind(self): self.wfile.write(result) def _propfind_resources(self, segments, depth): - """Return the [(href, displayname, node), ...] a PROPFIND on *segments* refers to.""" + """Return the [(href, displayname, Resource), ...] a PROPFIND on *segments* refers to.""" resources = [] if not segments: # server root: the list of archives - resources.append(("/", "/", Node(DEFAULT_DIR_MODE, mtime=self.vfs.root_mtime, children={}))) + resources.append(("/", "/", Resource(True, DEFAULT_DIR_MODE, self.vfs.root_mtime, 0, None))) if depth == "1": for name in sorted(self.vfs.archives): archive_info = self.vfs.archives[name] - node = Node(DEFAULT_DIR_MODE, mtime=int(archive_info.ts.timestamp() * 1e9), children={}) - resources.append(("/" + encode_path(name) + "/", name, node)) + mtime_ns = int(archive_info.ts.timestamp() * 1e9) + res = Resource(True, DEFAULT_DIR_MODE, mtime_ns, 0, None) + resources.append(("/" + encode_path(name) + "/", name, res)) return resources - node, _, segments = self.vfs.resolve(segments) # may raise KeyError - if not (node.is_dir or stat.S_ISREG(node.mode)): + segments, node = self.vfs.resolve(segments) # may raise KeyError + res = self._resource(node) + if not (res.is_dir or stat.S_ISREG(res.mode)): raise KeyError(segments[-1]) # symlinks and special files are not exposed via WebDAV base = "/" + "/".join(encode_path(s) for s in segments) - if not node.is_dir: - return [(base, segments[-1], node)] - resources.append((base + "/", segments[-1], node)) + if not res.is_dir: + return [(base, segments[-1], res)] + resources.append((base + "/", segments[-1], res)) if depth == "1": - for name, child in sorted(node.children.items()): - if child.is_dir: - resources.append((f"{base}/{encode_path(name)}/", name, child)) - elif stat.S_ISREG(child.mode): - resources.append((f"{base}/{encode_path(name)}", name, child)) + for name, child in sorted(self.vfs.children(node), key=lambda kv: kv[0]): + child_res = self._resource(child) + if child_res.is_dir: + resources.append((f"{base}/{encode_path(name)}/", name, child_res)) + elif stat.S_ISREG(child_res.mode): + resources.append((f"{base}/{encode_path(name)}", name, child_res)) # symlinks and special files are not exposed via WebDAV return resources @@ -771,7 +628,8 @@ def _send_dir_listing(self, segments, node, head): f' (preserves metadata)">{DOWNLOAD_ICON_SVG}' ) rows = [make_row("../", "..")] - children = sorted(node.children.items(), key=lambda kv: (not kv[1].is_dir, kv[0])) + children = [(name, self._resource(child)) for name, child in self.vfs.children(node)] + children.sort(key=lambda kv: (not kv[1].is_dir, kv[0])) # directories first for name, child in children: display_name = remove_surrogates(name) if child.is_dir: @@ -787,8 +645,8 @@ def _send_dir_listing(self, segments, node, head): rows.append(make_row(None, display_name, mtime_ns=child.mtime)) self._send_page(render_page(title, rows, heading=heading), head) - def _send_file(self, name, node, pipeline, head): - etag = make_etag(node) + def _send_file(self, name, node, res, head): + etag = make_etag(res) if_none_match = self.headers.get("If-None-Match") if if_none_match: client_tags = [t.strip() for t in if_none_match.split(",")] @@ -802,78 +660,54 @@ def _send_file(self, name, node, pipeline, head): if range_header: if_range = self.headers.get("If-Range") if if_range is None or if_range.strip() == etag: - byte_range = parse_byte_range(range_header, node.size) + byte_range = parse_byte_range(range_header, res.size) if byte_range == "unsatisfiable": self.send_response(416) - self.send_header("Content-Range", f"bytes */{node.size}") + self.send_header("Content-Range", f"bytes */{res.size}") self.send_header("Content-Length", "0") self.end_headers() return if byte_range: start, end = byte_range self.send_response(206) - self.send_header("Content-Range", f"bytes {start}-{end}/{node.size}") + self.send_header("Content-Range", f"bytes {start}-{end}/{res.size}") else: - start, end = 0, node.size - 1 + start, end = 0, res.size - 1 self.send_response(200) self.send_header("Content-Type", guess_content_type(name)) - self.send_header("Content-Length", str(end - start + 1 if node.size else 0)) - self.send_header("Last-Modified", http_date(node.mtime)) + self.send_header("Content-Length", str(end - start + 1 if res.size else 0)) + self.send_header("Last-Modified", http_date(res.mtime)) self.send_header("ETag", etag) self.send_header("Accept-Ranges", "bytes") self.send_header("X-Content-Type-Options", "nosniff") content_disposition = self._content_disposition(name) # CR/LF-sanitized, see there self.send_header("Content-Disposition", content_disposition) self.end_headers() - if head or node.size == 0: + if head or res.size == 0: return # no body to send (and Content-Length is 0 for an empty file) - if not node.chunks: + chunks = self.vfs.get_item(node.ino).get("chunks") + if not chunks: # anomaly: a non-empty file with no chunks list (e.g. corrupted metadata). We # already sent Content-Length > 0, so abort the connection instead of leaving # the client waiting forever for body bytes that will never arrive. logger.error( - "webdav: file %s has size %d but no chunks, aborting the connection.", - remove_surrogates(name), - node.size, + "webdav: file %s has size %d but no chunks, aborting the connection.", remove_surrogates(name), res.size ) self.close_connection = True return - # select only the chunks overlapping the requested range, so nothing else - # gets fetched and decrypted (the chunk sizes are known in advance). - selected, first_offset, pos = [], 0, 0 - for entry in node.chunks: - if pos + entry.size <= start: - pos += entry.size - continue - if pos > end: - break - if not selected: - first_offset = start - pos - selected.append(entry) - pos += entry.size - remaining = end - start + 1 - for entry in selected: - if remaining <= 0: - break - # serialize repository and data cache access, but write to the client outside - # the lock, so one slow client cannot block other requests for the whole download. - with self.repo_lock: - data = self._get_chunk(pipeline, entry) - if data is None: - # chunk missing in repository - never serve silently corrupted data: - # abort the connection, the client sees a short read (Content-Length mismatch). - logger.error( - "webdav: chunk missing while serving %s, aborting the connection.", remove_surrogates(name) - ) - self.close_connection = True - return - if first_offset: - data = data[first_offset:] - first_offset = 0 - if len(data) > remaining: - data = data[:remaining] - self.wfile.write(data) - remaining -= len(data) + # the reader fetches only the chunks overlapping the requested range (the chunk + # sizes are known in advance), each one under the repository lock - but we write + # to the client outside of it, so one slow client cannot block other requests. + # pos_key: a mounted file system reads a big file with many sequential range + # requests - remember where in the chunk list they got us. + try: + for data in self.vfs.reader.iter_data(chunks, start, end - start + 1, pos_key=node.ino): + self.wfile.write(data) + except ChunkMissing: + # chunk missing in repository - never serve silently corrupted data: + # abort the connection, the client sees a short read (Content-Length mismatch). + logger.error("webdav: chunk missing while serving %s, aborting the connection.", remove_surrogates(name)) + self.close_connection = True def _send_chunked(self, data): """Write one HTTP/1.1 chunked-transfer-encoding block; *data* must be non-empty.""" @@ -895,8 +729,8 @@ def _send_tar(self, segments, head): The tar size is not known in advance (PAX header sizes vary), so we stream it with chunked transfer encoding. Repository access (item iteration and chunk - fetching) is serialized under repo_lock, but each chunk is written to the client - outside the lock - so a slow client cannot block other requests, and the + fetching) is serialized under the VFS lock, but each chunk is written to the + client outside the lock - so a slow client cannot block other requests, and the LockRefresher can keep the repository lock alive during a long download. """ # lazy import: pulling in the archiver package at module import time would be @@ -927,16 +761,15 @@ def want(item): if head: return - with self.repo_lock: + with self.vfs.lock: archive = Archive(self.vfs.manifest, self.vfs.archives[archive_name].id) - pipeline = archive.pipeline item_iter = archive.iter_items(want) hlm = HardLinkManager(id_type=bytes, info_type=str) # hlid -> (stripped) path of the first link complete = False try: while True: - with self.repo_lock: + with self.vfs.lock: try: item = next(item_iter) except StopIteration: @@ -948,7 +781,7 @@ def want(item): continue # unsupported item type, skipped (with a warning) tarinfo.pax_headers = item_to_paxheaders("PAX", item) self._send_chunked(tarinfo.tobuf(tarfile.PAX_FORMAT, tarfile.ENCODING, "surrogateescape")) - if needs_content and not self._send_tar_content(item, tarinfo.size, pipeline): + if needs_content and not self._send_tar_content(item, tarinfo.size): return # a chunk was missing: leave the chunked stream unterminated (see finally) # end-of-archive marker (two zero blocks), then terminate the chunked stream. self._send_chunked(b"\0" * (tarfile.BLOCKSIZE * 2)) @@ -960,7 +793,7 @@ def want(item): # tar is truncated (never present a corrupt archive as if it were complete). self.close_connection = True - def _send_tar_content(self, item, size, pipeline): + def _send_tar_content(self, item, size): """Stream *item*'s file content into the tar, padded to a 512-byte block boundary. Returns False (after logging) if the content cannot be produced in full (a chunk @@ -977,48 +810,21 @@ def _send_tar_content(self, item, size, pipeline): size, ) return False - for entry in chunks: - # fetch under the lock, write to the client outside it (see _send_tar). - with self.repo_lock: - data = next(pipeline.fetch_many([entry], ro_type=ROBJ_FILE_STREAM, replacement_chunk=False)) - if data is None: - logger.error( - "webdav: chunk missing while streaming tar member %s, aborting the connection.", - remove_surrogates(item.path), - ) - return False - self._send_chunked(data) + try: + # fetches under the lock, writes to the client outside it (see _send_tar). + for data in self.vfs.reader.iter_data(chunks, 0, size): + self._send_chunked(data) + except ChunkMissing: + logger.error( + "webdav: chunk missing while streaming tar member %s, aborting the connection.", + remove_surrogates(item.path), + ) + return False padding = (tarfile.BLOCKSIZE - size % tarfile.BLOCKSIZE) % tarfile.BLOCKSIZE if padding: self._send_chunked(b"\0" * padding) return True - def _get_chunk(self, pipeline, entry): - """Return the decrypted data of chunk *entry*, using the shared data cache. - - Returns None if the chunk is missing in the repository. Must be called while - holding repo_lock (it guards both the repository and the data cache, and the - data cache's LRUCache is not thread-safe). The cache is keyed by chunk id - (a content hash), so it is valid across archives and requests. Reads of a big - file over a mounted file system tend to come as many small sequential range - requests hitting the same chunk - the cache avoids re-fetching and re-decrypting - it every time. - - This complements, but does not duplicate, the repository's own pack cache: - fetch_many() -> Repository.get_many() caches whole packs (raw, still encrypted - and compressed on-disk bytes), whereas we cache the *decrypted+decompressed* - chunk data, i.e. the output of the per-chunk RepoObj.parse() that runs on every - fetch even on a pack cache hit. So the two layers save different work (repo I/O - vs. per-chunk decrypt+decompress). - """ - try: - return self.data_cache[entry.id] - except KeyError: - data = next(pipeline.fetch_many([entry], ro_type=ROBJ_FILE_STREAM, replacement_chunk=False)) - if data is not None: - self.data_cache[entry.id] = data - return data - @staticmethod def _content_disposition(name): # File names from an archive can contain any byte except NUL and "/", including @@ -1035,13 +841,6 @@ def _content_disposition(name): return strip_crlf(result) # no-op by construction, but makes the CR/LF safety explicit -def data_cache_capacity(): - # number of decrypted chunks to keep cached; same knob as borg mount uses. - # default: number of CPUs (a small, non-zero value), see also borg mount --help. - capacity = int(os.environ.get("BORG_MOUNT_DATA_CACHE_ENTRIES", os.cpu_count() or 1)) - return max(1, capacity) - - def make_server(manifest, args, bind="127.0.0.1", port=8000): """Create a ThreadingHTTPServer serving the archives selected by *args*. @@ -1049,14 +848,11 @@ def make_server(manifest, args, bind="127.0.0.1", port=8000): server threads is serialized with it (use it for e.g. LockRefresher, too). """ repo_lock = threading.RLock() - vfs = ArchiveVFS(manifest, args, repo_lock) - capacity = data_cache_capacity() - logger.debug("webdav: data cache capacity: %d chunks", capacity) - data_cache = LRUCache(capacity=capacity) + vfs = ArchiveVFS(manifest, args, lock=repo_lock) + vfs.create_filesystem() - handler_class = type("WebDAVHandler", (WebDAVHandler,), dict(vfs=vfs, repo_lock=repo_lock, data_cache=data_cache)) + handler_class = type("WebDAVHandler", (WebDAVHandler,), dict(vfs=vfs)) server = ThreadingHTTPServer((bind, port), handler_class) server.daemon_threads = True server.repo_lock = repo_lock - server.data_cache = data_cache return server