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

docs/internals/packs.rst

@@ -241,6 +241,31 @@ single blob cannot be removed from a pack in place: all of these paths write a n
 pack file without it and then delete the old one, so store-level 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, exactly as in the repair walk above
+  (``repoobj.object_validator``). ``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``. That size decides how far the
+  dropped range reaches, so the index entry serves as a second source for it, one that
+  does not come from the bytes being examined.
+
+Anything else keeps its bytes, for ``borg check --repair`` to re-index. The walk still
+steps by the blob's own size, so a corrupt length field can put it at a wrong offset --
+where nothing passes both checks. Authenticating needs the key, so a caller without one
+drops no gap bytes: ``borg debug delete-obj`` opens the repository without a key and
+therefore reclaims no superseded gap bytes from the pack it rewrites.
+
 
 .. _pack-index-namespace:
 

src/borg/archive.py

@@ -2357,6 +2357,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.
@@ -2377,7 +2378,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]

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/repository.py

@@ -48,6 +48,11 @@
 # meta_size, which only MAX_DATA_SIZE bounds, from triggering a large read.
 MAX_VALIDATED_META_SIZE = 64 * 1024
 
+# how much superseded_gap_ranges reads past an object header, so the metadata slot behind it comes
+# in the same request. A metadata slot is at most 112 bytes over all key modes; a larger one costs
+# one more read.
+GAP_META_READAHEAD = 256
+
 
 def repo_lister(repository, *, limit=None):
     marker = None
@@ -545,7 +550,7 @@ def check_pack_objects(pack_hex, obj_ranges, pack_size):
         )
 
 
-def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
+def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, validate=None):
     """Find the superseded duplicates among a pack's gap bytes (bytes no index entry covers).
 
     A gap holds a chunk copy stored again elsewhere, or objects from a backup that crashed before
@@ -555,10 +560,24 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
     (borg check --repair re-indexes it) or whose entry points back at this offset (its only copy)
     is not reported. A header that does not parse or overruns its gap ends the walk over that gap.
 
+    A duplicate is reported only when both hold:
+
+    - validate accepts its header and metadata slot, authenticating magic, version, chunk id,
+      meta_size and data_size.
+    - its total size equals the index entry's obj_size. The total size sets how far the reported
+      range reaches, so the entry is a second source for it, independent of the object's own bytes.
+
+    Anything else keeps its bytes and the walk continues past it.
+
     obj_ranges: the offset-ordered, validated (obj_offset, obj_size) ranges of the pack's indexed
     objects; the gaps are the byte ranges between (and after) them.
+    validate: validate(chunk_id, obj) -> bool over an object's header and metadata slot, see
+    repoobj.object_validator. None reports nothing.
     Returns the offset-ordered list of (offset, size) ranges holding superseded duplicates.
     """
+    if validate is None:
+        return []
+
     # find the gaps: byte ranges no indexed object covers.
     gaps = []  # (start, end) of each gap, offset-ordered
     cursor = 0
@@ -574,17 +593,23 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
     for gstart, gend in gaps:
         offset = gstart
         while offset < gend:
-            hdr_data = reader.read(offset, hdr_size)
-            if len(hdr_data) < hdr_size:
+            # the header, and the metadata slot behind it in the same request.
+            buf = reader.read(offset, min(gend - offset, hdr_size + GAP_META_READAHEAD))
+            if len(buf) < hdr_size:
                 break
-            hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
-            obj_size = hdr_size + hdr.meta_size + hdr.data_size
-            if hdr.magic != OBJ_MAGIC or offset + obj_size > gend:
+            # gend, not pack_size: an object reaching past this gap is not one of its objects.
+            hdr, _ = PackReader._parse_header(buf[:hdr_size], offset, gend)
+            if hdr is None:
                 break
-            if hdr.chunk_id in chunks:
-                entry = chunks[hdr.chunk_id]
-                if entry.pack_id != pack_id or entry.obj_offset != offset:
+            obj_size = hdr_size + hdr.meta_size + hdr.data_size
+            entry = chunks.get(hdr.chunk_id)
+            superseded = entry is not None and (entry.pack_id != pack_id or entry.obj_offset != offset)
+            if superseded and obj_size == entry.obj_size:
+                # buf starts at offset, so the readahead above usually already holds the slot.
+                if reader._validation_problem(hdr, offset, buf, offset, validate) is None:
                     drop_ranges.append((offset, obj_size))
+            # the walk steps by obj_size before anything authenticates it: a wrong one lands the
+            # walk at a wrong offset, where the two checks above apply again.
             offset += obj_size
     return drop_ranges
 
@@ -1525,12 +1550,15 @@ def put(self, id, data):
         # PackWriter shares this repository's index, so add() triggers the lazy build itself.
         return self._pack_writer.add(id, data)
 
-    def delete(self, id, *, update_index=True):
+    def delete(self, id, *, update_index=True, validate=None):
         """Delete a single repo object by rewriting its pack without it (via compact_pack).
 
         With update_index=True the full chunk index is written back so the next borg process sees the
         deletion; callers that rebuild the index themselves (check --repair) pass update_index=False to
         skip the per-object index rewrite.
+
+        validate: authenticates a gap object before its bytes are dropped, see
+        superseded_gap_ranges. None drops no gap bytes.
         """
         self._lock_refresh()
         entry = self.chunks.get(id)
@@ -1540,30 +1568,32 @@ def delete(self, id, *, update_index=True):
         # keep every object the chunk index lists for this pack, except the one being deleted.
         keep_ids = {cid for cid, e in self.chunks.iteritems() if e.pack_id == pack_id}
         keep_ids.discard(id)
-        self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id})
+        self.compact_pack(pack_id, keep_ids=keep_ids, drop_ids={id}, validate=validate)
         if update_index:
             # close() only persists new entries incrementally, so write the full index here to record
             # the removal for the next borg process.
             from .cache import write_chunkindex_to_repo
 
             write_chunkindex_to_repo(self, self.chunks, incremental=False, force_write=True, delete_other=True)
 
-    def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None):
+    def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None, validate=None):
         """Rewrite pack <pack_id>, keeping <keep_ids> and dropping <drop_ids>, then delete the old pack.
 
         keep_ids: chunk ids in this pack to copy into the new pack.
         drop_ids: chunk ids in this pack to discard. Must not overlap keep_ids.
         chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index
             updates to. Must be the index keep_ids and drop_ids were derived from. Default: self.chunks.
+        validate: authenticates a gap object before its bytes are dropped, see
+            superseded_gap_ranges. None drops no gap bytes. Default: None.
 
         Together, keep_ids and drop_ids must cover every object the chunk index lists for this pack;
         an unlisted indexed object would keep its bytes in the new pack but its index entry would go
         stale when the old pack is deleted. Bytes that no index entry covers appear as gaps between the
-        listed objects: a gap object whose chunk id is in the index is a superseded duplicate (its
-        authoritative copy is elsewhere) and is dropped; a gap object whose id is not in the index is
-        copied into the new pack unchanged, to be handled by "borg check --repair". An overlap between
-        listed objects, or an object claiming to end past the pack file, means index corruption and
-        raises IntegrityError.
+        listed objects: a gap object that authenticates as a superseded duplicate (its authoritative
+        copy is elsewhere) is dropped, every other gap byte is copied into the new pack unchanged, to
+        be handled by "borg check --repair" - see superseded_gap_ranges. An overlap between listed
+        objects, or an object claiming to end past the pack file, means index corruption and raises
+        IntegrityError.
 
         The new pack is the old pack minus the dropped objects, built via store.defrag; kept objects are
         repointed in the chunk index and dropped objects' chunk index entries are removed.
@@ -1605,7 +1635,7 @@ def compact_pack(self, pack_id, *, keep_ids: set, drop_ids: set, chunks=None):
         # toward the rewrite threshold and a wholly superseded orphan pack can be dropped outright.
         drop_ranges = [(offset, size) for offset, _, size, keep in located if not keep]
         reader = PackReader(store=self.store, pack_id=pack_id)
-        drop_ranges += superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size)
+        drop_ranges += superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, validate=validate)
         drop_ranges.sort()
         dropped_bytes = sum(size for _, size in drop_ranges)  # on-disk bytes this rewrite frees, for --stats
 
@@ -1759,7 +1789,7 @@ def merge_packs(self, pack_ids, *, chunks=None, max_size=None):
             pi.show(increase=1)
         pi.finish()
 
-    def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change=None):
+    def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change=None, validate=None):
         """Rewrite pack <pack_id>, passing each indexed object's bytes through <transform>.
 
         ids: the chunk ids of this pack's objects. Must cover every object the chunk index lists
@@ -1773,13 +1803,15 @@ def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change=
             updates to. Must be the index <ids> was derived from. Default: self.chunks.
         before_change: called once, just before the first store modification; use it to invalidate
             stored chunk indexes for crash safety (see #9748). Not called when the pack is kept.
+        validate: authenticates a gap object before its bytes are dropped, see
+            superseded_gap_ranges. None drops no gap bytes. Default: None.
 
         The whole pack file is loaded into memory (bounded by the pack size limit). Gap bytes
-        (bytes no index entry covers) are handled like in compact_pack: an object superseded by a
-        copy stored elsewhere is dropped, all other unindexed bytes are copied into the new pack
-        unchanged, to be handled by "borg check --repair". An overlap between indexed objects, or
-        an object claiming to end past the pack file, means index corruption and raises
-        IntegrityError, before anything is written.
+        (bytes no index entry covers) are handled like in compact_pack: an object that authenticates
+        as superseded by a copy stored elsewhere is dropped, all other unindexed bytes are copied
+        into the new pack unchanged, to be handled by "borg check --repair". An overlap between
+        indexed objects, or an object claiming to end past the pack file, means index corruption and
+        raises IntegrityError, before anything is written.
 
         If every object is kept and no gap bytes are dropped, the store and the chunk index are not
         touched at all. Otherwise the new pack (named sha256 of its content) is stored, the indexed
@@ -1811,7 +1843,7 @@ def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change=
         located.sort()
         obj_ranges = [(offset, size) for offset, _, size in located]
         check_pack_objects(pack_hex, obj_ranges, pack_size)
-        drop_ranges = superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size)
+        drop_ranges = superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, validate=validate)
 
         # assemble the new pack in offset order: transformed objects, dropped ranges skipped, all
         # other bytes copied verbatim. the two range lists never overlap (drops lie in gaps), so a