check --repair: resync past corrupt object headers when rebuilding the chunks index, #8476

When check --repair rebuilds the chunks index from the packs, a corrupt object
header now makes iter_headers resync rather than raise: it takes a validate
function and scans forward for the next object, in 1 MiB windows that overlap by
one header so a header on a window boundary is still found. Repository-only
checks pass no validate and keep raising IntegrityError on a corrupt header.

OBJ_MAGIC also occurs inside payloads, so a candidate is accepted only when it
authenticates. For AEAD keys, decrypting the metadata authenticates it against
the header's magic, version and chunk_id, so the walk confirms a chunk id from a
few hundred bytes. Keys that authenticate by chunk_id == id_hash(content)
(id_check_is_authentication) read the whole object and parse() at the "repair"
id place; validate.needs_data selects between the two.

Authentication needs the key, so check --repair makes it before the rebuild with
manifest_only=True. A repair that cannot read the manifest has no key and walks
without resyncing.

Author: mr-raj12

docs/internals/packs.rst

@@ -89,10 +89,30 @@ A reader locates the next blob by advancing::
 
     next_blob_offset = current_blob_offset + REPOOBJ_HEADER_SIZE + meta_size + data_size
 
-The per-blob magic limits the blast radius of corrupted length fields: if
-``meta_size`` or ``data_size`` is damaged, the scanner loses at most one blob.
-Once it finds the next ``OBJ_MAGIC`` sequence it resumes. Other corruption
-(payload bit flips) is caught by AEAD on that blob without losing position.
+``iter_headers()`` checks every header it walks: it must have ``OBJ_MAGIC``, a
+supported version, and sizes that keep the blob inside the pack. A header that
+fails these checks means a corrupt pack, and ``IntegrityError`` is raised.
+
+The per-blob magic limits the blast radius of corrupted length fields. The
+repair walk (``iter_headers(validate=...)``, used when ``borg check --repair``
+rebuilds the chunks index from the packs) scans forward for the next blob and
+resumes there, so the blobs after the damaged part of the pack are still found.
+
+``OBJ_MAGIC`` occurs inside the payloads as well, and in ``none`` and
+``authenticated`` mode the payloads are user content stored as it is, so a
+backed up file can contain something shaped like a blob. The scan therefore
+accepts a candidate only if it parses. For the AEAD keys it reads the header and
+the encrypted metadata, a few hundred bytes: decrypting the metadata
+authenticates it together with the header's magic, version and chunk_id, which
+are its AAD (additional authenticated data: authenticated with the ciphertext,
+but not encrypted). The other keys authenticate by ``chunk_id == id_hash(content)``
+(``KeyBase.id_check_is_authentication``), which needs the blob's data, so for
+those the scan reads the whole blob. The key is needed either way; a repair that
+cannot read the manifest walks without scanning.
+
+``data_size`` is not part of that AAD, so accepting a candidate authenticates
+its chunk id, and its size only as far as the blob fits into the pack. Bit flips
+in the data are caught when the blob is read, on that blob alone.
 
 Blobs follow one another contiguously with no padding::
 

src/borg/archive.py

@@ -1862,6 +1862,32 @@ def __next__(self):
             return next(self._unpacker)
 
 
+def resync_validator(repo_objs):
+    """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id.
+
+    obj holds an object's header and encrypted metadata, plus its encrypted data when
+    validate.needs_data is set. For most keys, decrypting the metadata authenticates it against the
+    header (magic, version, chunk_id), so the metadata alone decides. Keys that authenticate by
+    chunk_id == id_hash(content) (id_check_is_authentication) need the data; for them
+    validate.needs_data is set and parse() checks that id at the "repair" id place.
+    """
+    needs_data = repo_objs.key.id_check_is_authentication
+
+    def validate(chunk_id, obj):
+        try:
+            if needs_data:
+                repo_objs.parse(chunk_id, obj, ro_type=ROBJ_DONTCARE, assert_id_place="repair")
+            else:
+                repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE)
+        except Exception:
+            # authentication, id check, msgpack or decompression can each raise on non-object bytes.
+            return False
+        return True
+
+    validate.needs_data = needs_data
+    return validate
+
+
 class ArchiveChecker:
     # Bound how many missing file chunks rebuild_archives buffers for its end-of-run report,
     # so checking a badly damaged repo with very many missing chunks can not exhaust memory.
@@ -1913,7 +1939,18 @@ def check(
         # so we do not rebuild it from the packs (reading every pack is far too slow for a routine check).
         # --repair does rebuild from the packs (slow_rebuild=repair), working from the real packs so it
         # can detect and fix archives that reference chunks whose pack has gone missing.
-        self.chunks = build_chunkindex_from_repo(self.repository, slow_rebuild=repair, write_immediately=False)
+        # Under --repair, validate lets the rebuild resync past a corrupt object header (see resync_validator).
+        # It authenticates objects with the key, so make the key first; manifest_only=True makes make_key use
+        # the manifest, not self.chunks, which is still unset here.
+        if self.key is None:
+            try:
+                self.key = self.make_key(repository, manifest_only=True)
+            except IntegrityError as err:
+                logger.warning(f"{err}. Packs with a corrupt object header can not be repaired.")
+        validate = resync_validator(RepoObj(self.key)) if repair and self.key is not None else None
+        self.chunks = build_chunkindex_from_repo(
+            self.repository, slow_rebuild=repair, validate=validate, write_immediately=False
+        )
         if self.key is None:
             self.key = self.make_key(repository)
         self.repo_objs = RepoObj(self.key)

src/borg/cache.py

@@ -839,7 +839,7 @@ def repack_chunkindex(repository):
 
 
 def build_chunkindex_from_repo(
-    repository, *, slow_rebuild=False, write_immediately=False, init_flags=ChunkIndex.F_USED
+    repository, *, slow_rebuild=False, validate=None, write_immediately=False, init_flags=ChunkIndex.F_USED
 ):
     # first, try to build a fresh, mostly complete chunk index from centrally stored index fragments:
     if not slow_rebuild:
@@ -906,7 +906,8 @@ def build_chunkindex_from_repo(
         # PackReader uses the store directly, so refresh the lock here; a full rebuild can be slow.
         repository._lock_refresh()
         pack_id = hex_to_bin(info.name)
-        for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers():
+        # validate makes iter_headers resync past a corrupt object header and index the objects after it.
+        for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(validate=validate):
             num_chunks += 1
             chunks[chunk_id] = ChunkIndexEntry(
                 flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size

src/borg/repository.py

@@ -30,14 +30,17 @@
 from .storelocking import Lock
 from .logger import create_logger
 from .manifest import NoManifestError
-from .repoobj import RepoObj, OBJ_MAGIC
+from .repoobj import RepoObj, OBJ_MAGIC, SUPPORTED_OBJ_VERSIONS
 from .crypto.key import is_keyfile
 
 logger = create_logger(__name__)
 
 # an object name is its sha256 as 64 lowercase hex digits.
 _valid_object_name = re.compile(r"[0-9a-f]{64}").fullmatch
 
+# how much of a pack PackReader reads at once when searching for the next object header.
+RESYNC_WINDOW_SIZE = 1024 * 1024
+
 
 def repo_lister(repository, *, limit=None):
     marker = None
@@ -362,24 +365,73 @@ def read(self, offset, size):
         return self.store.load(self.key, offset=offset, size=size)
 
     def size(self):
-        """Return the pack size in bytes; for a store-backed pack this is one metadata lookup."""
+        """Return the pack size in bytes (a store metadata lookup, unless the pack is in memory)."""
         if self.pack_contents is not None:
             return len(self.pack_contents)
         return self.store.info(self.key).size
 
-    def iter_headers(self):
+    @staticmethod
+    def _parse_header(hdr_data, offset, pack_size):
+        """Return the ObjHeader in hdr_data if it is a valid header at offset, None otherwise.
+
+        Valid means: OBJ_MAGIC, a supported version, and an object that fits into the pack.
+        """
+        hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
+        if hdr.magic != OBJ_MAGIC or hdr.version not in SUPPORTED_OBJ_VERSIONS:
+            return None
+        if offset + RepoObj.obj_header.size + hdr.meta_size + hdr.data_size > pack_size:
+            return None
+        return hdr
+
+    def _find_header(self, offset, pack_size, validate):
+        """Scan forward from offset for the next object validate accepts, return its offset or None.
+
+        A pack has no framing besides the object headers, so this searches for OBJ_MAGIC. That byte
+        sequence also occurs inside payloads, so a candidate is accepted only when its header parses
+        and validate confirms it.
+        """
+        hdr_size = RepoObj.obj_header.size
+        while offset + hdr_size <= pack_size:
+            # a window at a time, so the scan costs one store request per RESYNC_WINDOW_SIZE bytes.
+            buf = bytes(self.read(offset, min(RESYNC_WINDOW_SIZE, pack_size - offset)))
+            if len(buf) < hdr_size:
+                break
+            pos = 0
+            while True:
+                pos = buf.find(OBJ_MAGIC, pos)
+                if pos < 0 or pos + hdr_size > len(buf):
+                    break  # not in this window, or a header overlapping its end: the next window has it
+                hdr = self._parse_header(buf[pos : pos + hdr_size], offset + pos, pack_size)
+                if hdr is not None:
+                    obj_size = hdr_size + hdr.meta_size + hdr.data_size
+                    # an object is at most MAX_DATA_SIZE bytes (Repository.put), so a larger candidate is a
+                    # false match on OBJ_MAGIC in a payload.
+                    if obj_size <= MAX_DATA_SIZE:
+                        size = obj_size if validate.needs_data else hdr_size + hdr.meta_size
+                        end = pos + size
+                        # the window holds these bytes, unless the candidate crosses its end.
+                        obj = buf[pos:end] if end <= len(buf) else self.read(offset + pos, size)
+                        if validate(hdr.chunk_id, obj):
+                            return offset + pos
+                pos += 1
+            # step by the window less one header, so a magic straddling the boundary is still found.
+            offset += max(len(buf) - (hdr_size - 1), 1)
+        return None
+
+    def iter_headers(self, validate=None):
         """Yield (chunk_id, offset, size) for each object by walking the fixed object headers.
 
-        Only the headers are read, not the payloads, so locating every object costs one short
-        range read per object (or just a slice, when the pack is already in memory), plus one
-        store metadata lookup for the pack size.
+        The walk reads a header per object: one short range read each (or a slice, for a pack in
+        memory), plus one store metadata lookup for the pack size.
+
+        A header must have OBJ_MAGIC, a supported version and describe an object that fits into
+        the pack, otherwise the pack is corrupt and IntegrityError is raised. A read shorter than
+        a header ends the walk: that is the end of the pack.
 
-        Each full header must have OBJ_MAGIC and describe an object that fits into the pack,
-        otherwise the pack is corrupt and IntegrityError is raised. Ending the walk instead
-        would be worse than raising: the chunks index rebuilt from these headers would just be
-        missing the rest of the pack, and borg check --repair would then "fix" the archives by
-        dropping chunks that are there.
-        A trailing partial header is the clean end of the pack, not corruption.
+        validate(chunk_id, obj): returns whether obj is a repo object with id chunk_id, where obj is
+        its header and metadata, plus its data when validate.needs_data is set. When validate is
+        given, a corrupt header makes the walk resync: it scans for the next object validate accepts
+        (see _find_header), continues there, and logs the skipped bytes.
         """
         pack_hex = bin_to_hex(self.pack_id) if self.pack_id is not None else "<no id>"
         pack_size = self.size()
@@ -389,17 +441,26 @@ def iter_headers(self):
             hdr_data = self.read(offset, hdr_size)
             if len(hdr_data) < hdr_size:
                 break  # clean EOF, or trailing partial bytes
-            hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
-            if hdr.magic != OBJ_MAGIC:
-                raise IntegrityError(
-                    f'pack {pack_hex}: no object header at offset {offset} (pack corruption), run "borg check"'
+            hdr = self._parse_header(hdr_data, offset, pack_size)
+            if hdr is None:
+                if validate is None:
+                    raise IntegrityError(
+                        f'pack {pack_hex}: invalid object header at offset {offset} (pack corruption), run "borg check"'
+                    )
+                next_offset = self._find_header(offset + 1, pack_size, validate)
+                if next_offset is None:
+                    logger.warning(
+                        f"pack {pack_hex}: invalid object header at offset {offset} and none after it, "
+                        f"skipping the remaining {pack_size - offset} bytes."
+                    )
+                    break
+                logger.warning(
+                    f"pack {pack_hex}: invalid object header at offset {offset}, "
+                    f"skipping {next_offset - offset} bytes to the next one."
                 )
+                offset = next_offset
+                continue
             obj_size = hdr_size + hdr.meta_size + hdr.data_size
-            if offset + obj_size > pack_size:
-                raise IntegrityError(
-                    f"pack {pack_hex}: object extends past end of file at offset {offset} "
-                    f'(pack corruption), run "borg check"'
-                )
             yield hdr.chunk_id, offset, obj_size
             offset += obj_size
 

src/borg/testsuite/archiver/check_cmd_test.py

@@ -688,6 +688,34 @@ def test_extra_chunks(archivers, request):
     cmd(archiver, "check", "-v", exit_code=0)  # check does not deal with orphans anymore
 
 
+def test_repair_resyncs_pack_with_corrupt_object_header(archivers, request):
+    """--repair rebuilds the index from a pack whose object header is damaged.
+
+    A damaged header makes the walk lose the object boundaries, so the rebuild scans for the next
+    object that authenticates and carries on there. That needs the key, which --repair makes before
+    the rebuild. Repairing the pack itself is a separate step, see #10026.
+    """
+    archiver = request.getfixturevalue(archivers)
+    if archiver.get_kind() != "local":
+        pytest.skip("inspects the store directly")
+    check_cmd_setup(archiver)
+    cmd(archiver, "check", exit_code=0)
+
+    with Repository(archiver.repository_location, exclusive=True) as repository:
+        # damage the header of the second object of a pack that holds more than two.
+        by_pack = {}
+        for chunk_id, entry in repository.chunks.items():
+            by_pack.setdefault(entry.pack_id, []).append((entry.obj_offset, chunk_id))
+        pack_id, objs = next((p, sorted(o)) for p, o in by_pack.items() if len(o) > 2)
+        damaged_offset, _ = objs[1]
+        key = "packs/" + bin_to_hex(pack_id)
+        repository.store_store(key, corrupt(repository.store_load(key), damaged_offset))
+
+    output = cmd(archiver, "check", "--repair", "--debug", exit_code=0)
+    assert f"invalid object header at offset {damaged_offset}" in output
+    assert "bytes to the next one" in output  # the rebuild resumed at the next object
+
+
 def test_repair_finish_flushes_pack_writer(archivers, request):
     """finish() stores chunks re-added during --repair before it drops the index (#10055).
 

src/borg/testsuite/cache_test.py

@@ -25,7 +25,8 @@
 )
 from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..crypto.key import AESOCBKey
-from ..helpers import safe_ns
+from ..helpers import bin_to_hex, safe_ns
+from ..helpers import IntegrityError
 from ..helpers.msgpack import int_to_timestamp
 from ..manifest import Manifest
 from ..repository import Repository
@@ -475,6 +476,26 @@ def test_close_consolidates_fragments_across_sessions(tmp_path, monkeypatch):
             assert cid in index
 
 
+def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path):
+    """A corrupt object header fails the rebuild, but with repair=True the rest of the pack is indexed."""
+    from .repository_test import accept_all, fchunk
+
+    obj1 = bytearray(fchunk(b"first", chunk_id=H(90)))
+    obj2 = fchunk(b"second", chunk_id=H(91))
+    obj1[0] ^= 0xFF  # break the magic of the first object's header
+    pack_id = H(92)
+    with Repository(os.fspath(tmp_path / "repository"), exclusive=True, create=True) as repository:
+        repository.store_store("packs/" + bin_to_hex(pack_id), bytes(obj1) + obj2)
+        with pytest.raises(IntegrityError):
+            build_chunkindex_from_repo(repository, slow_rebuild=True)
+        # accept_all: accepts every candidate, so this exercises the plumbing only.
+        index = build_chunkindex_from_repo(repository, slow_rebuild=True, validate=accept_all)
+        assert H(91) in index  # found by resyncing past the damaged header
+        assert H(90) not in index  # its header is gone, so the object can not be indexed
+        assert index[H(91)].pack_id == pack_id
+        assert index[H(91)].obj_offset == len(obj1)
+
+
 def test_repack_leaves_sealed_untouched_and_reconstructs(tmp_path, monkeypatch):
     """Sealed (>= MIN) fragments survive a repack; build_chunkindex_from_repo reconstructs the index."""
     monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000)