Merge pull request #1303 from enkore/issue/1302

fuse: EIO on damaged files unless told not to (-o allow_damaged_files)

borg/archiver.py

@@ -1268,6 +1268,13 @@ class Archiver:
         browsing an archive or restoring individual files. Unless the ``--foreground``
         option is given the command will run in the background until the filesystem
         is ``umounted``.
+
+        For mount options, see the fuse(8) manual page. Additional mount options
+        supported by borg:
+
+        - allow_damaged_files: by default damaged files (where missing chunks were
+          replaced with runs of zeros by borg check --repair) are not readable and
+          return EIO (I/O error). Set this option to read such files.
         """)
         subparser = subparsers.add_parser('mount', parents=[common_parser],
                                           description=self.do_mount.__doc__,

borg/fuse.py

@@ -6,11 +6,16 @@ import os
 import stat
 import tempfile
 import time
-from .archive import Archive
-from .helpers import daemonize, bigint_to_int
 from distutils.version import LooseVersion
+
 import msgpack
 
+from .archive import Archive
+from .helpers import daemonize, bigint_to_int
+from .logger import create_logger
+logger = create_logger()
+
+
 # Does this version of llfuse support ns precision?
 have_fuse_xtime_ns = hasattr(llfuse.EntryAttributes, 'st_mtime_ns')
 
@@ -42,6 +47,9 @@ class ItemCache:
 class FuseOperations(llfuse.Operations):
     """Export archive as a fuse filesystem
     """
+
+    allow_damaged_files = False
+
     def __init__(self, key, repository, manifest, archive, cached_repo):
         super().__init__()
         self._inode_count = 0
@@ -68,6 +76,32 @@ class FuseOperations(llfuse.Operations):
                 self.contents[1][os.fsencode(archive_name)] = archive_inode
                 self.pending_archives[archive_inode] = Archive(repository, key, manifest, archive_name)
 
+    def mount(self, mountpoint, mount_options, foreground=False):
+        """Mount filesystem on *mountpoint* with *mount_options*."""
+        options = ['fsname=borgfs', 'ro']
+        if mount_options:
+            options.extend(mount_options.split(','))
+        try:
+            options.remove('allow_damaged_files')
+            self.allow_damaged_files = True
+        except ValueError:
+            pass
+        llfuse.init(self, mountpoint, options)
+        if not foreground:
+            daemonize()
+
+        # 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
+        # nasty consequences, imagine the user has e.g. an active rsync mirror
+        # job - seeing the mountpoint empty, rsync would delete everything in the
+        # mirror.
+        umount = False
+        try:
+            signal = fuse_main()
+            umount = (signal is None)  # no crash and no signal -> umount request
+        finally:
+            llfuse.close(umount)
+
     def process_archive(self, archive, prefix=[]):
         """Build fuse inode hierarchy from archive metadata
         """
@@ -225,6 +259,15 @@ class FuseOperations(llfuse.Operations):
         return self.getattr(inode)
 
     def open(self, inode, flags, ctx=None):
+        if not self.allow_damaged_files:
+            item = self.get_item(inode)
+            if b'chunks_healthy' in item:
+                # Processed archive items don't carry the path anymore; for converting the inode
+                # to the path we'd either have to store the inverse of the current structure,
+                # or search the entire archive. So we just don't print it. It's easy to correlate anyway.
+                logger.warning('File has damaged (all-zero) chunks. Try running borg check --repair. '
+                               'Mount with allow_damaged_files to read damaged files.')
+                raise llfuse.FUSEError(errno.EIO)
         return inode
 
     def opendir(self, inode, ctx=None):
@@ -256,23 +299,3 @@ class FuseOperations(llfuse.Operations):
     def readlink(self, inode, ctx=None):
         item = self.get_item(inode)
         return os.fsencode(item[b'source'])
-
-    def mount(self, mountpoint, extra_options, foreground=False):
-        options = ['fsname=borgfs', 'ro']
-        if extra_options:
-            options.extend(extra_options.split(','))
-        llfuse.init(self, mountpoint, options)
-        if not foreground:
-            daemonize()
-
-        # 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
-        # nasty consequences, imagine the user has e.g. an active rsync mirror
-        # job - seeing the mountpoint empty, rsync would delete everything in the
-        # mirror.
-        umount = False
-        try:
-            signal = fuse_main()
-            umount = (signal is None)  # no crash and no signal -> umount request
-        finally:
-            llfuse.close(umount)

borg/testsuite/__init__.py

@@ -94,9 +94,12 @@ class BaseTestCase(unittest.TestCase):
             self._assert_dirs_equal_cmp(sub_diff)
 
     @contextmanager
-    def fuse_mount(self, location, mountpoint):
+    def fuse_mount(self, location, mountpoint, mount_options=None):
         os.mkdir(mountpoint)
-        self.cmd('mount', location, mountpoint, fork=True)
+        args = ['mount', location, mountpoint]
+        if mount_options:
+            args += '-o', mount_options
+        self.cmd(*args, fork=True)
         self.wait_for_mount(mountpoint)
         yield
         if sys.platform.startswith('linux'):

borg/testsuite/archiver.py

@@ -234,6 +234,13 @@ class ArchiverTestCaseBase(BaseTestCase):
     def create_src_archive(self, name):
         self.cmd('create', self.repository_location + '::' + name, src_dir)
 
+    def open_archive(self, name):
+        repository = Repository(self.repository_path)
+        with repository:
+            manifest, key = Manifest.load(repository)
+            archive = Archive(repository, key, manifest, name)
+        return archive, repository
+
 
 class ArchiverTestCase(ArchiverTestCaseBase):
 
@@ -1037,6 +1044,31 @@ class ArchiverTestCase(ArchiverTestCaseBase):
             sto = os.stat(out_fn)
             assert stat.S_ISFIFO(sto.st_mode)
 
+    @unittest.skipUnless(has_llfuse, 'llfuse not installed')
+    def test_fuse_allow_damaged_files(self):
+        self.cmd('init', self.repository_location)
+        self.create_src_archive('archive')
+        # Get rid of a chunk and repair it
+        archive, repository = self.open_archive('archive')
+        with repository:
+            for item in archive.iter_items():
+                if item[b'path'].endswith('testsuite/archiver.py'):
+                    repository.delete(item[b'chunks'][-1][0])
+                    path = item[b'path']  # store full path for later
+                    break
+            else:
+                assert False  # missed the file
+            repository.commit()
+        self.cmd('check', '--repair', self.repository_location, exit_code=0)
+
+        mountpoint = os.path.join(self.tmpdir, 'mountpoint')
+        with self.fuse_mount(self.repository_location + '::archive', mountpoint):
+            with pytest.raises(OSError) as excinfo:
+                open(os.path.join(mountpoint, path))
+            assert excinfo.value.errno == errno.EIO
+        with self.fuse_mount(self.repository_location + '::archive', mountpoint, 'allow_damaged_files'):
+            open(os.path.join(mountpoint, path)).close()
+
     def verify_aes_counter_uniqueness(self, method):
         seen = set()  # Chunks already seen
         used = set()  # counter values already used
@@ -1117,13 +1149,6 @@ class ArchiverCheckTestCase(ArchiverTestCaseBase):
             self.create_src_archive('archive1')
             self.create_src_archive('archive2')
 
-    def open_archive(self, name):
-        repository = Repository(self.repository_path)
-        with repository:
-            manifest, key = Manifest.load(repository)
-            archive = Archive(repository, key, manifest, name)
-        return archive, repository
-
     def test_check_usage(self):
         output = self.cmd('check', '-v', self.repository_location, exit_code=0)
         self.assert_in('Starting repository check', output)