compact: authenticate a gap object before dropping its bytes, #10093

docs/internals/packs.rst

@@ -64,6 +64,9 @@ the AAD; tampering with either still fails the check, because it changes the len
 slice being read. A forged ``chunk_id``, version, or magic byte therefore fails
 authentication in ``RepoObj.parse()``/``parse_meta()``.
 
+``parse_meta()`` reads the metadata slot only, so of the two length fields it covers
+``meta_size`` alone; ``data_size`` is covered by ``parse()``.
+
 ``encrypted_meta`` and ``encrypted_data`` each add a one-byte slot tag on top of the shared header
 AAD -- ``b"M"`` for ``encrypted_meta``, ``b"D"`` for ``encrypted_data`` -- binding each ciphertext to
 its slot. This stops an attacker controlling repo storage from swapping the two ciphertexts (adjusting
@@ -91,10 +94,34 @@ 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 for the next blob and resumes
+there, so the blobs after the damaged part of the pack are still found. The scan
+starts just past the last blob the walk accepted: a corrupted length field that
+keeps the blob inside the pack leaves the header valid, and is noticed only at
+the misaligned offset it points to. The blob carrying it is dropped when a
+recovered blob starts inside the extent it claims - its length field is wrong,
+so it can not be read back.
+
+``OBJ_MAGIC`` occurs inside the payloads as well, and in the ``none-*`` and
+``authenticated-*`` modes the payloads are user content stored as it is, so a
+backed up file can contain something shaped like a blob. A candidate is
+therefore accepted only when its metadata slot verifies against the header AAD
+described above; the header and that slot, a few hundred bytes, are what the
+scan reads. Verifying needs the key, so a repair that cannot read the manifest
+walks without scanning.
+
+In the ``none-*`` modes the tag is an unkeyed checksum, so the scan accepts any
+well-formed blob, including one a backed up file contains.
+
+``data_size`` is not part of the 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::
 
@@ -191,6 +218,28 @@ determines via mark-and-sweep that none of a pack's blobs are referenced by any
 archive, it removes the whole file. Individual blobs cannot be removed without
 rewriting the entire pack, so deletion always operates at pack granularity.
 
+Gap bytes
+~~~~~~~~~
+
+A pack can hold bytes that no chunks index entry covers -- its *gaps*: a copy of a chunk
+that was stored again elsewhere, or blobs from a backup that crashed before writing its
+index. Rewriting a pack (``compact_pack``, ``transform_pack``) walks the gaps and drops
+the blobs among them that are *superseded*: whose chunk id the index maps to a copy at
+another location, which by the id/content invariant holds the same plaintext.
+
+A gap blob is dropped only when both hold:
+
+* its header and metadata slot authenticate (``repoobj.object_validator``), verifying its
+  magic, version, chunk id and ``meta_size``. ``OBJ_MAGIC`` plus a well-formed header is
+  not evidence that bytes are a blob: in the ``none-*`` and ``authenticated-*`` modes the
+  payloads are user content stored as it is, so a backed up file can contain one.
+* its total size equals the index entry's ``obj_size``. ``data_size`` lies outside what
+  the authentication covers and sets how far the dropped range reaches, so the entry
+  serves as a second source for it.
+
+Anything else keeps its bytes, for ``borg check --repair`` to re-index. Authenticating
+needs the key, so a caller without one drops no gap bytes.
+
 
 .. _pack-index-namespace:
 

src/borg/archive.py

@@ -52,7 +52,7 @@
 from . import platform
 from .platform import acl_get, acl_set, set_flags, get_flags, set_times, swidth
 from .repository import Repository, NoManifestError
-from .repoobj import RepoObj
+from .repoobj import RepoObj, object_validator
 
 # macOS: SF_DATALESS marks dataless placeholder files (e.g. cloud files not materialized locally).
 # Reading such files triggers downloading their content. stat.SF_DATALESS is only available
@@ -1947,7 +1947,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)
+        # --repair also passes validate, which makes the rebuild resync past a corrupt object header.
+        # Validating needs the key, so read it here. manifest_only=True, because the other source
+        # make_key reads keys from is self.chunks, which is only built below.
+        if repair and 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 = object_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)
@@ -2081,6 +2092,7 @@ def verify_data(self):
         if defect_chunks:
             if self.repair:
                 logger.warning("Found defect chunks, removing them from the repository.")
+                validate = object_validator(self.repo_objs)
                 for defect_chunk in defect_chunks:
                     # remote repo (ssh): retry might help for strange network / NIC / RAM errors
                     # as the chunk will be retransmitted from remote server.
@@ -2101,7 +2113,7 @@ def verify_data(self):
                         # failed twice -> remove this defect chunk. delete rewrites its pack without it,
                         # keeping the other chunks. update_index=False: finish() rebuilds the index from
                         # the rewritten packs anyway, so a per-chunk full index write would be wasted.
-                        self.repository.delete(defect_chunk, update_index=False)
+                        self.repository.delete(defect_chunk, update_index=False, validate=validate)
                         self.chunks_modified = True
                         # drop it from our own index too, so rebuild_archives reports the file it belongs to.
                         del self.chunks[defect_chunk]
@@ -2480,7 +2492,12 @@ def finish(self):
                 # the packs changed, so the index no longer matches them: rebuild it from the packs
                 # and persist it.
                 logger.info("Rebuilding and writing the repository chunks index.")
-                build_chunkindex_from_repo(self.repository, slow_rebuild=True, write_immediately=True)
+                build_chunkindex_from_repo(
+                    self.repository,
+                    slow_rebuild=True,
+                    validate=object_validator(self.repo_objs),
+                    write_immediately=True,
+                )
             else:
                 # the packs are unchanged, so the index still matches them: persist it as is.
                 logger.info("Writing the rebuilt repository chunks index.")

src/borg/archiver/compact_cmd.py

@@ -15,6 +15,7 @@
 from ..helpers import set_ec, EXIT_ERROR, Error, sig_int, format_file_size, bin_to_hex, hex_to_bin, IntegrityError
 from ..helpers import ProgressIndicatorPercent
 from ..manifest import Manifest
+from ..repoobj import object_validator
 from ..repository import Repository
 
 from ..logger import create_logger
@@ -407,12 +408,15 @@ def compact_packs(self):
                 del self.chunks[id]
             progress += 1
             pi.show(progress)  # report after the work, so the final pack lands on 100%
+        validate = object_validator(self.manifest.repo_objs)
         for pid in rewrite_packs:
             if sig_int:
                 break
             # chunks=self.chunks: the index updates (repoint kept objects, remove dropped ones)
             # must land in the index that save_chunk_index() persists (#9850).
-            _, dropped = self.repository.compact_pack(pid, keep_ids=keep[pid], drop_ids=drop[pid], chunks=self.chunks)
+            _, dropped = self.repository.compact_pack(
+                pid, keep_ids=keep[pid], drop_ids=drop[pid], chunks=self.chunks, validate=validate
+            )
             freed += dropped  # unused indexed objects plus superseded duplicates
             progress += 1
             pi.show(progress)

src/borg/archiver/repo_compress_cmd.py

@@ -11,6 +11,7 @@
 from ..helpers import format_file_size, hex_to_bin
 from ..helpers.argparsing import ArgumentParser
 from ..manifest import Manifest
+from ..repoobj import object_validator
 from ..repository import Repository
 
 from ..logger import create_logger
@@ -104,14 +105,20 @@ def recompress(self):
         pi = ProgressIndicatorPercent(
             total=len(packs), msg="Recompressing %3.1f%%", step=0.1, msgid="repo_compress.recompress"
         )
+        validate = object_validator(self.repo_objs)
         for i, (pack_id, pack_size) in enumerate(packs):
             if sig_int:
                 break  # stop cleanly at a pack boundary: save the index below, then raise
             ids = per_pack.get(pack_id)
             # a pack without indexed objects (all-gap) is left for "borg check --repair", see #9868.
             if ids:
                 new_pack_id, new_size = self.repository.transform_pack(
-                    pack_id, ids, self.transform, chunks=self.chunks, before_change=self.invalidate_stored_index
+                    pack_id,
+                    ids,
+                    self.transform,
+                    chunks=self.chunks,
+                    before_change=self.invalidate_stored_index,
+                    validate=validate,
                 )
                 if new_pack_id != pack_id:
                     self.packs_rewritten += 1

src/borg/cache.py

@@ -852,10 +852,18 @@ def repack_chunkindex(repository):
 
 
 def build_chunkindex_from_repo(
-    repository, *, slow_rebuild=False, fragments_only=False, write_immediately=False, init_flags=ChunkIndex.F_USED
+    repository,
+    *,
+    slow_rebuild=False,
+    fragments_only=False,
+    validate=None,
+    write_immediately=False,
+    init_flags=ChunkIndex.F_USED,
 ):
     # fragments_only: build the index from the index/ fragments only, returning None if they cannot be
     # read completely, and never write to the repo.
+    # validate: handed to PackReader.iter_headers when rebuilding from the packs, making it resync
+    # past a corrupt object header rather than raise IntegrityError.
     assert not (slow_rebuild and fragments_only)
     assert not (fragments_only and write_immediately)  # fragments_only never writes to the repo
     # first, try to build a fresh, mostly complete chunk index from centrally stored index fragments:
@@ -947,7 +955,7 @@ def build_chunkindex_from_repo(
         repository._lock_refresh()
         pi.show(increase=1)
         pack_id = hex_to_bin(info.name)
-        for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers():
+        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/repoobj.py

@@ -75,7 +75,8 @@ def get_assert_id_places():
 # Size of the header prefix used as AEAD AAD (additional authenticated data: authenticated together
 # with the ciphertext, but not itself encrypted) for OBJ_VERSION_HEADER_AAD objects: magic(8) +
 # version(1) + chunk_id(32). meta_size and data_size are excluded, since they are only known after
-# encryption; a change to either still fails authentication, by changing the ciphertext slice length.
+# encryption; a change to either still fails authentication, by changing the ciphertext slice
+# length. parse() covers both, parse_meta() only meta_size - it does not read the data slot.
 REPOOBJ_HEADER_AAD_SIZE = len(OBJ_MAGIC) + 1 + 32
 
 META_AAD_TAG = b"M"
@@ -277,5 +278,25 @@ def parse(
         return meta_compressed if want_compressed else meta, data_compressed if want_compressed else data
 
 
+def object_validator(repo_objs):
+    """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id.
+
+    obj is an object's header and metadata slot, trailing bytes allowed. Parsing the slot verifies
+    its tag, which covers the header's magic, version, chunk id and meta_size, but not data_size.
+
+    In the "none-*" modes the tag is an unkeyed checksum, so any well-formed object validates.
+    """
+
+    def validate(chunk_id, obj):
+        try:
+            repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE)
+        except Exception:
+            # arbitrary bytes fail the tag, the msgpack unpacking or the length checks.
+            return False
+        return True
+
+    return validate
+
+
 # Backward compatibility: RepoObj1 has moved to borg.legacy.repoobj
 from .legacy.repoobj import RepoObj1  # noqa: F401