extract: do not abort on corrupted chunks, replace them by all-zero data with a warning, fixes #840

docs/internals/frontends.rst

@@ -997,6 +997,8 @@ Warnings
         {}: {}
     BackupBrokenSymlinkError rc: 112
         {}: {}
+    BackupDamagedChunksError rc: 113
+        {}: {}
 
 Operations
     - cache.close

src/borg/archive.py

@@ -28,7 +28,7 @@
 from .digests import ContentDigester
 from .crypto.low_level import IntegrityError as IntegrityErrorBase
 from .helpers import BackupError, BackupRaceConditionError, BackupItemExcluded
-from .helpers import BackupSymlinkParentError, BackupPathTraversalError
+from .helpers import BackupSymlinkParentError, BackupPathTraversalError, BackupDamagedChunksError
 from .helpers import BackupOSError, BackupPermissionError, BackupFileNotFoundError, BackupIOError, BackupTimeoutError
 from .helpers import HardLinkManager
 from .helpers import archive_hostname, archive_username
@@ -392,8 +392,32 @@ def unpack_many(self, ids, *, filter=None):
                         item.chunks_healthy = [ChunkListEntry(*e) for e in item.chunks_healthy]
                     yield item
 
-    def fetch_many(self, chunks, ro_type=None, replacement_chunk=True):
+    def fetch_many(self, chunks, ro_type=None, replacement_chunk=True, replace_corrupted=False, damaged=None):
+        """
+        Yield the plaintext data of *chunks* (ChunkListEntry objects or bare chunk ids), in order.
+
+        A chunk that is missing from the repository or that is corrupted (does not authenticate,
+        decrypt or decompress) can not be returned as it was. If *replacement_chunk* is set and
+        the chunk size is known (ChunkListEntry), a missing chunk is replaced by an all-zero chunk
+        of the correct size and an error is logged; otherwise it yields None. A corrupted chunk
+        is replaced the same way if additionally *replace_corrupted* is set (extract and mount
+        do that, so a damaged file still comes out with the right size); otherwise it raises
+        IntegrityError, so commands that create new archives or repositories from the data
+        (recreate, transfer) abort instead of storing all-zero data as if it were the content.
+        If *damaged* is a list, the ids of the replaced chunks are appended to it, so the caller
+        can report the affected file.
+        """
         assert ro_type is not None
+
+        def replacement(id, size, problem):
+            # all-zero chunk of the correct size, so the content stream keeps its offsets.
+            logger.error(f"repository object {bin_to_hex(id)} {problem}, returning {size} zero bytes.")
+            if damaged is not None:
+                damaged.append(id)
+            data = zeros[:size]
+            assert len(data) == size, f"replacement chunk size {size} exceeds {len(zeros)}"
+            return data
+
         ids = []
         sizes = []
         if all(isinstance(chunk, ChunkListEntry) for chunk in chunks):
@@ -416,17 +440,22 @@ def fetch_many(self, chunks, ro_type=None, replacement_chunk=True):
             cdata = next(fetched)
             if cdata is None:
                 if replacement_chunk and size is not None:
-                    logger.error(f"repository object {bin_to_hex(id)} missing, returning {size} zero bytes.")
-                    data = zeros[:size]  # return an all-zero replacement chunk of correct size
+                    data = replacement(id, size, "missing")
                 else:
                     logger.error(f"repository object {bin_to_hex(id)} missing, returning None.")
                     data = None
             else:
                 try:
                     data = self.parsed_cache[(id, ro_type)]
                 except KeyError:
-                    _, data = self.repo_objs.parse(id, cdata, ro_type=ro_type)
-                    self.parsed_cache[(id, ro_type)] = data
+                    try:
+                        _, data = self.repo_objs.parse(id, cdata, ro_type=ro_type)
+                    except IntegrityErrorBase as err:
+                        if not (replacement_chunk and replace_corrupted and size is not None):
+                            raise
+                        data = replacement(id, size, f"corrupted ({err})")
+                    else:
+                        self.parsed_cache[(id, ro_type)] = data
             assert data is None or size is None or len(data) == size
             yield data
 
@@ -947,7 +976,10 @@ def same_item(item, st):
                     # it does not really set hard links due to dry_run, but behave the same as non-dry_run.
                     if "chunks" in item:
                         item_chunks_size = 0
-                        for data in self.pipeline.fetch_many(item.chunks, ro_type=ROBJ_FILE_STREAM):
+                        damaged = []  # ids of missing/corrupted chunks that were replaced by zeros
+                        for data in self.pipeline.fetch_many(
+                            item.chunks, ro_type=ROBJ_FILE_STREAM, replace_corrupted=True, damaged=damaged
+                        ):
                             if pi:
                                 pi.show(increase=len(data), info=[remove_surrogates(item.path)])
                             if stdout:
@@ -963,6 +995,8 @@ def same_item(item, st):
                                         item_size, item_chunks_size
                                     )
                                 )
+                        if damaged:
+                            raise BackupDamagedChunksError(len(damaged))
             return
 
         dest = self.cwd
@@ -1016,7 +1050,10 @@ def make_parent(path):
                     fd = open(path, "wb")
                 try:
                     trailing_hole = False
-                    for data in self.pipeline.fetch_many(item.chunks, ro_type=ROBJ_FILE_STREAM):
+                    damaged = []  # ids of missing/corrupted chunks that were replaced by zeros
+                    for data in self.pipeline.fetch_many(
+                        item.chunks, ro_type=ROBJ_FILE_STREAM, replace_corrupted=True, damaged=damaged
+                    ):
                         if pi:
                             pi.show(increase=len(data), info=[remove_surrogates(item.path)])
                         with backup_io("write"):
@@ -1059,6 +1096,9 @@ def make_parent(path):
                         raise BackupError(
                             f"Size inconsistency detected: size {item_size}, chunks size {item_chunks_size}"
                         )
+                if damaged:
+                    # the file is complete (size, attrs), but parts of its content are all-zero replacements.
+                    raise BackupDamagedChunksError(len(damaged))
             return
         with backup_io:
             # No repository access beyond this point.

src/borg/archiver/extract_cmd.py

@@ -147,6 +147,14 @@ def build_parser_extract(self, subparsers, common_parser, mid_common_parser):
         ``--progress`` can be slower than no progress display, since it makes one additional
         pass over the archive metadata.
 
+        If a file's content chunks are missing from the repository or are corrupted (they fail
+        authentication, decryption or decompression), the extraction does not abort: each such
+        chunk is written as all-zero data of the correct size, an error naming the chunk is logged,
+        the file is reported with a warning and the exit code is a warning. The extracted file thus
+        has the correct size and metadata, but wrong (all-zero) content where the damaged chunks
+        were. Run ``borg check`` to find out which chunks and archives are affected. This also
+        applies to ``--dry-run`` (which thus can be used to find unreadable files) and ``--stdout``.
+
         When using ``--stats``, borg reports the store statistics (lines prefixed with
         "Store") for the extraction: per-operation call counts and timings, the load/store
         data volumes and throughput, and cache hits/misses. This includes ``--dry-run``

src/borg/archiver/mount_cmds.py

@@ -146,9 +146,9 @@ def build_parser_mount_umount(self, subparsers, common_parser, mid_common_parser
 
         - ``versions``: when used with a repository mount, this gives a merged, versioned
           view of the files in the archives. EXPERIMENTAL; layout may change in the future.
-        - ``allow_damaged_files``: by default, damaged files (where chunks are missing)
-          will return EIO (I/O error) when trying to read the related parts of the file.
-          Set this option to replace the missing parts with all-zero bytes.
+        - ``allow_damaged_files``: by default, damaged files (where chunks are missing or
+          corrupted) will return EIO (I/O error) when trying to read the related parts of the
+          file. Set this option to replace the damaged parts with all-zero bytes.
         - ``ignore_permissions``: for security reasons the ``default_permissions`` mount
           option is internally enforced by Borg. ``ignore_permissions`` can be given to
           not enforce ``default_permissions``.

src/borg/fuse.py

@@ -61,6 +61,7 @@ def async_wrapper(fn):
 
 from .helpers import daemonizing, signal_handler
 from .storelocking import LockRefresher
+from .crypto.low_level import IntegrityError as IntegrityErrorBase
 from .vfs import ArchiveVFS, ChunkMissing, parse_mount_options
 
 BLOCK_SIZE = 512  # Standard filesystem block size for st_blocks and statfs
@@ -220,6 +221,10 @@ def read(self, fh, offset, size):
             return self.vfs.read(fh, offset, size, pos_key=fh)
         except ChunkMissing:
             raise llfuse.FUSEError(errno.EIO) from None
+        except IntegrityErrorBase as err:
+            # a corrupted chunk (unless allow_damaged_files replaced it by zeros): report it, EIO for the read.
+            logger.error("mount: %s", err)
+            raise llfuse.FUSEError(errno.EIO) from None
 
     def _readdir_entries(self, fh):
         node = self._dir_node(fh)

src/borg/helpers/__init__.py

@@ -20,6 +20,7 @@
 from .errors import BackupPermissionError, BackupIOError, BackupFileNotFoundError, BackupTimeoutError
 from .errors import BackupSymlinkParentError, BackupPathTraversalError, BackupHardlinkSourceError
 from .errors import BackupBrokenSymlinkError
+from .errors import BackupDamagedChunksError
 from .fs import ensure_dir, join_base_dir
 from .fs import get_security_dir, get_keys_dir, get_base_dir, get_cache_dir, get_config_dir, get_runtime_dir
 from .fs import dir_is_tagged, dir_is_cachedir, remove_dotdot_prefixes, make_path_safe, scandir_inorder

src/borg/helpers/errors.py

@@ -239,5 +239,22 @@ class BackupBrokenSymlinkError(BackupError):
     exit_mcode = 112
 
 
+class BackupDamagedChunksError(BackupError):
+    """{}: {}"""
+
+    # Raised by extract after a file's content was written: some of its chunks were missing from the
+    # repository or corrupted and DownloadPipeline.fetch_many replaced them by all-zero data of the
+    # correct size, so the file has its right size but wrong content in those places. Reported as a
+    # per-file warning (path: message); borg check reports which chunks / archives are affected.
+    exit_mcode = 113
+
+    def __init__(self, count):
+        super().__init__(count)
+        self.count = count
+
+    def __str__(self):
+        return f"{self.count} chunk(s) missing or corrupted in the repository, replaced by all-zero data"
+
+
 class BackupItemExcluded(Exception):
     """Used internally to skip an item from processing when it is excluded."""

src/borg/hlfuse.py

@@ -24,6 +24,7 @@
 
 from .helpers import daemonizing, signal_handler
 from .storelocking import LockRefresher
+from .crypto.low_level import IntegrityError as IntegrityErrorBase
 from .vfs import ArchiveVFS, ChunkMissing, parse_mount_options
 
 BLOCK_SIZE = 512  # Standard filesystem block size for st_blocks and statfs
@@ -184,6 +185,10 @@ def read(self, path, size, offset, fi):
             return self.vfs.read(node.ino, offset, size, pos_key=fi.fh)
         except ChunkMissing:
             raise hlfuse.FuseOSError(errno.EIO) from None
+        except IntegrityErrorBase as err:
+            # a corrupted chunk (unless allow_damaged_files replaced it by zeros): report it, EIO for the read.
+            logger.error("mount: %s", err)
+            raise hlfuse.FuseOSError(errno.EIO) from None
 
     def readdir(self, path, fh=None):
         node = self._find_node(path)

src/borg/testsuite/archiver/check_cmd_test.py

@@ -12,6 +12,7 @@
 from ...cache import delete_chunkindex_from_repo
 from ...constants import *  # NOQA
 from ...helpers import bin_to_hex, msgpack, CommandError, CorruptPack, Error, IntegrityError, sig_int
+from ...helpers import BackupDamagedChunksError
 from ...manifest import Archives, Manifest
 from ...repoobj import RepoObj
 from ...repository import PackTracker, Repository
@@ -1033,10 +1034,12 @@ def test_verify_data_wrong_chunk_content(archivers, request, monkeypatch):
     assert f"{bin_to_hex(chunk.id)}, integrity error" in output
     assert "id verification failed" in output
 
-    # with "read" in BORG_ASSERT_ID, reads check it too:
+    # with "read" in BORG_ASSERT_ID, reads check it too: extract treats the chunk as corrupted, i.e.
+    # it extracts all-zero data instead and reports the file with a warning.
     monkeypatch.setenv("BORG_ASSERT_ID", "read")
-    with pytest.raises(IntegrityError):  # local (not forked): the Error propagates instead of setting the rc
-        cmd(archiver, "extract", "archive1")
+    output = cmd(archiver, "extract", "archive1", exit_code=BackupDamagedChunksError.exit_mcode)
+    assert "id verification failed" in output
+    assert "1 chunk(s) missing or corrupted in the repository, replaced by all-zero data" in output
 
 
 def test_repair_wrong_item_metadata_chunk_content(archivers, request, monkeypatch):

src/borg/testsuite/archiver/extract_cmd_test.py

@@ -19,9 +19,11 @@
 from ...manifest import Manifest
 from ...repository import Repository
 from ...helpers import EXIT_WARNING, BackupIOError, BackupOSError, BackupPermissionError, BackupSymlinkParentError
+from ...helpers import BackupDamagedChunksError
 from ...helpers import bin_to_hex
 from ...helpers import flags_noatime, flags_normal
 from .. import changedir, same_ts_ns, granularity_sleep
+from ..repository_test import corrupt_chunk_on_disk
 from .. import are_symlinks_supported, are_hardlinks_supported, is_utime_fully_supported, is_birthtime_fully_supported
 from ...platform import get_birthtime_ns
 from ...platformflags import is_darwin, is_freebsd, is_win32
@@ -38,6 +40,7 @@
     create_src_archive,
     open_archive,
     src_file,
+    src_dir,
 )
 
 pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary")  # NOQA
@@ -1026,23 +1029,62 @@ def test_dry_run_extraction_flags(archivers, request):
     assert not os.listdir("output"), "Output directory should be empty after dry-run"
 
 
-def test_extract_file_with_missing_chunk(archivers, request):
-    archiver = request.getfixturevalue(archivers)
+def _damage_last_chunk(archiver, damage):
+    """Create an archive of src_dir and damage the last chunk of src_file in the repository.
+
+    *damage* is "missing" (the chunk gets deleted) or "corrupted" (a byte of the chunk is flipped
+    in its pack, so it does not authenticate any more). Returns (item path, damaged chunk).
+    """
     cmd(archiver, "repo-create", RK_ENCRYPTION)
     create_src_archive(archiver, "archive")
-    # Get rid of a chunk
     archive, repository = open_archive(archiver.repository_path, "archive")
     with repository:
         for item in archive.iter_items():
             if item.path.endswith(src_file):
                 chunk = item.chunks[-1]
-                repository.delete(chunk.id)
-                break
+                if damage == "missing":
+                    repository.delete(chunk.id)
+                else:
+                    corrupt_chunk_on_disk(repository, chunk.id)
+                return item.path, chunk
         else:
             assert False  # missed the file
-    output = cmd(archiver, "extract", "archive")
-    # TODO: this is a bit dirty still: no warning/error rc, no filename output for the damaged file.
-    assert f"repository object {bin_to_hex(chunk.id)} missing, returning {chunk.size} zero bytes." in output
+
+
+@pytest.mark.parametrize("damage", ["missing", "corrupted"])
+def test_extract_file_with_damaged_chunk(archivers, request, damage):
+    # a missing or corrupted chunk does not abort the extraction: it is replaced by all-zero data
+    # of the correct size, the file gets a warning and the rc is a warning, the other files are fine.
+    archiver = request.getfixturevalue(archivers)
+    path, chunk = _damage_last_chunk(archiver, damage)
+    with open(os.path.join(os.path.dirname(src_dir), src_file), "rb") as f:
+        original = f.read()
+    with changedir("output"):
+        output = cmd(archiver, "extract", "archive", exit_code=BackupDamagedChunksError.exit_mcode)
+        assert f"repository object {bin_to_hex(chunk.id)} {damage}" in output
+        assert f"returning {chunk.size} zero bytes." in output
+        assert f"{path}: 1 chunk(s) missing or corrupted in the repository, replaced by all-zero data" in output
+        with open(path, "rb") as f:
+            extracted = f.read()
+        assert len(extracted) == len(original)
+        assert extracted[: -chunk.size] == original[: -chunk.size]
+        assert extracted[-chunk.size :] == bytes(chunk.size)
+        # the other files of the archive were extracted normally
+        with open(os.path.join(os.path.dirname(path), "extract_cmd.py"), "rb") as f1:
+            with open(os.path.join(src_dir, "extract_cmd.py"), "rb") as f2:
+                assert f1.read() == f2.read()
+
+
+@pytest.mark.parametrize("damage", ["missing", "corrupted"])
+def test_extract_dry_run_with_damaged_chunk(archivers, request, damage):
+    # --dry-run reads and verifies all data, so it reports damaged files the same way (and can be
+    # used to find them), without writing anything.
+    archiver = request.getfixturevalue(archivers)
+    path, chunk = _damage_last_chunk(archiver, damage)
+    with changedir("output"):
+        output = cmd(archiver, "extract", "--dry-run", "archive", exit_code=BackupDamagedChunksError.exit_mcode)
+        assert f"{path}: 1 chunk(s) missing or corrupted in the repository, replaced by all-zero data" in output
+        assert not os.listdir(".")
 
 
 def test_extract_existing_directory(archivers, request):

src/borg/testsuite/archiver/mount_cmds_test.py

@@ -21,6 +21,7 @@
 from .. import are_symlinks_supported, are_hardlinks_supported, are_fifos_supported
 from ..platform.platform_test import fakeroot_detected, skipif_not_linux, skipif_fakeroot_detected
 from ..platform.platform_test import skipif_acls_not_working
+from ..repository_test import corrupt_chunk_on_disk
 from . import RK_ENCRYPTION, cmd, assert_dirs_equal, create_regular_file, create_src_archive, open_archive, src_file
 from . import requires_hardlinks, _extract_hardlinks_setup, fuse_mount, create_test_files, generate_archiver_tests
 from . import Archiver
@@ -304,16 +305,20 @@ def test_fuse_archive_dir_format(archivers, request, monkeypatch):
 
 
 @pytest.mark.skipif(not has_any_fuse, reason="FUSE not available")
-def test_fuse_allow_damaged_files(archivers, request):
+@pytest.mark.parametrize("damage", ["missing", "corrupted"])
+def test_fuse_allow_damaged_files(archivers, request, damage):
     archiver = request.getfixturevalue(archivers)
     cmd(archiver, "repo-create", RK_ENCRYPTION)
     create_src_archive(archiver, "archive")
-    # Get rid of a chunk and repair it
+    # damage the last chunk of a file: delete it or corrupt it in its pack (it does not authenticate then)
     archive, repository = open_archive(archiver.repository_path, "archive")
     with repository:
         for item in archive.iter_items():
             if item.path.endswith(src_file):
-                repository.delete(item.chunks[-1].id)
+                if damage == "missing":
+                    repository.delete(item.chunks[-1].id)
+                else:
+                    corrupt_chunk_on_disk(repository, item.chunks[-1].id)
                 path = item.path  # store full path for later
                 break
         else:
@@ -328,7 +333,7 @@ def test_fuse_allow_damaged_files(archivers, request):
 
     with fuse_mount(archiver, mountpoint, "-a", "archive", "-o", "allow_damaged_files"):
         with open(os.path.join(mountpoint, "archive", path), "rb") as f:
-            # no exception raised, missing data will be all-zero
+            # no exception raised, the damaged part will be all-zero
             data = f.read()
         assert data.endswith(b"\0\0")