Merge pull request #10167 from mr-raj12/superseded-gap-authenticate-10093

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

docs/internals/packs.rst

@@ -244,6 +244,26 @@ 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
+~~~~~~~~~
+
+The *gaps* of a pack are its byte ranges that no chunks index entry covers. They hold chunk
+copies that were stored again in another pack, and blobs of a backup that crashed before
+writing its index. A gap blob is *superseded* when the index maps its chunk id to another
+location. Equal chunk ids mean equal plaintext, so a superseded blob is redundant, whatever
+the stored size of the indexed copy (compression and obfuscation padding change it).
+
+Rewriting a pack (``compact_pack``, ``transform_pack``) drops the superseded gap blobs whose
+header and metadata slot validate, checked as in the repair walk above
+(``repoobj.object_validator``), and copies all other gap bytes into the new pack.
+Validation covers ``meta_size`` and ``data_size``, so a dropped range is exactly one blob.
+Without a validator (``validate=None``), no gap bytes are dropped.
+
+The walk over a gap steps from header to header by the blob size each header states. It ends
+at a header that does not parse or that reaches past the gap. The rest of that gap is kept,
+and so is a superseded blob that does not validate; both are logged as a warning with the
+pack id and the offset.
+
 
 .. _pack-index-namespace:
 

src/borg/archive.py

@@ -23,7 +23,7 @@ logger = create_logger()
 from . import xattr
 from .chunkers import get_chunker, Chunk, release_chunk_data
 from .cache import ChunkListEntry, build_chunkindex_from_repo, write_chunkindex_to_repo
-from .crypto.key import key_factory, UnsupportedPayloadError
+from .crypto.key import key_from_repository
 from .constants import *  # NOQA
 from .digests import ContentDigester
 from .crypto.low_level import IntegrityError as IntegrityErrorBase
@@ -2303,41 +2303,16 @@ class ArchiveChecker:
         return self.repair or not self.error_found
 
     def make_key(self, repository, manifest_only=False):
-        attempt = 0
+        """Return the key loaded by key_from_repository.
 
-        #  try the manifest first!
-        try:
-            cdata = repository.get_manifest()
-        except NoManifestError:
-            pass
-        else:
-            try:
-                return key_factory(repository, cdata)
-            except UnsupportedPayloadError:
-                # we get here, if the cdata we got has a corrupted key type byte
-                pass  # ignore it, just continue trying
+        manifest_only: read only the manifest, else also the objects of the chunk ids in self.chunks.
+        """
 
-        if not manifest_only:
-            for chunkid, _ in self.chunks.iteritems():
-                attempt += 1
-                if attempt > 999:
-                    # we did a lot of attempts, but could not create the key via key_factory, give up.
-                    break
-                cdata = repository.get(chunkid)
-                try:
-                    return key_factory(repository, cdata)
-                except UnsupportedPayloadError:
-                    # we get here, if the cdata we got has a corrupted key type byte
-                    pass  # ignore it, just try the next chunk
+        def chunk_ids():  # reads self.chunks only if the manifest does not identify the key type
+            for id, _ in self.chunks.iteritems():
+                yield id
 
-        if attempt == 0:
-            if manifest_only:
-                msg = "make_key: failed to create the key (tried only the manifest)"
-            else:
-                msg = "make_key: repository has no chunks at all!"
-        else:
-            msg = "make_key: failed to create the key (tried %d chunks)" % attempt
-        raise IntegrityError(msg)
+        return key_from_repository(repository, () if manifest_only else chunk_ids())
 
     def verify_data(self):
         logger.info("Starting cryptographic data integrity verification...")
@@ -2378,6 +2353,7 @@ class ArchiveChecker:
         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.
@@ -2398,7 +2374,7 @@ class ArchiveChecker:
                         # 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 ..hashindex import ChunkIndex
 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 @@ class ArchiveGarbageCollector:
                 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/debug_cmd.py

@@ -3,6 +3,7 @@ import textwrap
 
 from ..archive import Archive
 from ..constants import *  # NOQA
+from ..crypto.key import key_from_repository, KeyfileInvalidError, RepoKeyNotFoundError, UnsupportedKeyFormatError
 from ..helpers import msgpack
 from ..helpers import FilesystemPathSpec
 from ..helpers import sysinfo
@@ -10,16 +11,36 @@ from ..helpers import bin_to_hex, hex_to_bin, prepare_dump_dict
 from ..helpers import dash_open
 from ..helpers import StableDict
 from ..helpers import archivename_validator, CompressionSpec
-from ..helpers import CommandError, RTError
+from ..helpers import CommandError, IntegrityError, RTError
 from ..helpers.argparsing import ArgumentParser
 from ..manifest import Manifest
 from ..platform import get_process_id
 from ..repository import Repository, LIST_SCAN_LIMIT, repo_lister
-from ..repoobj import RepoObj
+from ..repoobj import RepoObj, object_validator
 
 from ._common import with_repository, Highlander
 from ._common import process_epilog
 
+from ..logger import create_logger
+
+logger = create_logger()
+
+
+def gap_validator(repository):
+    """Return repoobj.object_validator for the key of repository, or None if there is no key to use.
+
+    The key is loaded with key_from_repository. There is no key to use if no stored object identifies
+    the key type (IntegrityError), no key is found (RepoKeyNotFoundError), or the key is invalid
+    (KeyfileInvalidError, UnsupportedKeyFormatError); a warning is logged then. Other errors, e.g. a
+    wrong passphrase, propagate.
+    """
+    try:
+        key = key_from_repository(repository)
+    except (IntegrityError, RepoKeyNotFoundError, KeyfileInvalidError, UnsupportedKeyFormatError) as err:
+        logger.warning(f"Could not set up the key, so rewritten packs keep their superseded gap bytes: {err}")
+        return None
+    return object_validator(RepoObj(key))
+
 
 class DebugMixIn:
     def do_debug_info(self, args):
@@ -111,7 +132,6 @@ class DebugMixIn:
     @with_repository(manifest=False)
     def do_debug_dump_repo_objs(self, args, repository):
         """Dumps (decrypted, decompressed) repository objects."""
-        from ..crypto.key import key_factory
 
         def decrypt_dump(id, cdata):
             if cdata is not None:
@@ -123,12 +143,7 @@ class DebugMixIn:
             with open(filename, "wb") as fd:
                 fd.write(data)
 
-        # set up the key without depending on a manifest obj
-        result = repository.list(limit=1, marker=None)
-        id, _ = result[0]
-        cdata = repository.get(id)
-        key = key_factory(repository, cdata)
-        repo_objs = RepoObj(key)
+        repo_objs = RepoObj(key_from_repository(repository))
         for id, stored_size in repo_lister(repository, limit=LIST_SCAN_LIMIT):
             cdata = repository.get(id)
             decrypt_dump(id, cdata)
@@ -161,14 +176,7 @@ class DebugMixIn:
         if not wanted:
             raise CommandError("search term needs to be hex:123abc or str:foobar style")
 
-        from ..crypto.key import key_factory
-
-        # set up the key without depending on a manifest obj
-        result = repository.list(limit=1, marker=None)
-        id, _ = result[0]
-        cdata = repository.get(id)
-        key = key_factory(repository, cdata)
-        repo_objs = RepoObj(key)
+        repo_objs = RepoObj(key_from_repository(repository))
 
         last_data = b""
         last_id = None
@@ -288,14 +296,19 @@ class DebugMixIn:
     @with_repository(manifest=False, exclusive=True)
     def do_debug_delete_obj(self, args, repository):
         """Deletes the objects with the given IDs from the repository."""
+        ids = []
         for hex_id in args.ids:
             try:
-                id = hex_to_bin(hex_id, length=32)
+                ids.append((hex_id, hex_to_bin(hex_id, length=32)))
             except ValueError:
+                ids.append((hex_id, None))
+        validate = gap_validator(repository) if any(id is not None for _, id in ids) else None
+        for hex_id, id in ids:
+            if id is None:
                 print("object id %s is invalid." % hex_id)
             else:
                 try:
-                    repository.delete(id)
+                    repository.delete(id, validate=validate)
                 except Repository.ObjectNotFound:
                     print("object %s not found." % hex_id)
                 else:

src/borg/archiver/repo_compress_cmd.py

@@ -11,6 +11,7 @@ from ..helpers import sig_int, ProgressIndicatorPercent, Error, CompressionSpec
 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,6 +105,7 @@ class PackRecompressor:
         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
@@ -111,7 +113,12 @@ class PackRecompressor:
             # 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/crypto/key.py

@@ -3,6 +3,7 @@ import hmac
 import os
 import textwrap
 from hashlib import sha256
+from itertools import islice
 from math import ceil
 from pathlib import Path
 from typing import Any, Literal, ClassVar, Optional
@@ -24,7 +25,7 @@ from ..helpers.passphrase import Passphrase, PasswordRetriesExceeded, Passphrase
 from ..helpers import msgpack
 from ..helpers import workarounds
 from ..item import Key, EncryptedKey
-from ..manifest import Manifest
+from ..manifest import Manifest, NoManifestError
 from ..platform import SaveFile
 from ..repoobj import RepoObj, RepoObj1
 
@@ -222,7 +223,17 @@ def identify_key(manifest_data):
     raise UnsupportedPayloadError(key_type)
 
 
-def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
+def identify_stored_key(manifest_chunk, *, ro_cls=RepoObj):
+    """Return (key class, data slot) of the stored object manifest_chunk.
+
+    A stored object is an object header, a metadata slot and a data slot (see RepoObj). The first
+    byte of the data slot is the key type byte, which selects the key class.
+
+    manifest_chunk: the stored object, e.g. the manifest.
+    ro_cls: the RepoObj class that parses manifest_chunk.
+    Raises IntegrityError if manifest_chunk is damaged (see ro_cls.extract_crypted_data), and
+    UnsupportedPayloadError if the key type byte selects no key class usable with ro_cls.
+    """
     manifest_data = ro_cls.extract_crypted_data(manifest_chunk)
     assert manifest_data, "manifest data must not be zero bytes long"
     key_cls = identify_key(manifest_data)
@@ -232,11 +243,49 @@ def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
         # tagged envelope modes (see MACKeyBase). The legacy key classes only exist to read borg
         # 1.x repositories (ro_cls is RepoObj1 then), e.g. for "borg transfer --from-borg1".
         raise UnsupportedPayloadError(manifest_data[0])
+    return key_cls, manifest_data
+
+
+def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
+    key_cls, manifest_data = identify_stored_key(manifest_chunk, ro_cls=ro_cls)
     key = key_cls.detect(repository, manifest_data, other=other)
     key.stored_type = manifest_data[0]
     return key
 
 
+def key_from_repository(repository, ids=None):
+    """Return the key of repository, loaded from the first stored object that identifies the key type.
+
+    Stored objects are read in this order: the manifest, then the objects of the chunk ids in ids, at
+    most 999 of them. An object identifies the key type if identify_stored_key accepts it. Errors
+    loading the key, e.g. a wrong passphrase, propagate.
+
+    repository: the Repository whose key is loaded.
+    ids: iterable of chunk ids. None: the chunk ids in the chunks index of repository.
+    Raises IntegrityError if no object read identifies the key type.
+    """
+    max_objects = 999
+
+    def stored_objects():
+        try:
+            yield repository.get_manifest()
+        except NoManifestError:
+            pass
+        chunk_ids = (id for id, _ in repository.list(limit=max_objects)) if ids is None else ids
+        for id in islice(chunk_ids, max_objects):
+            yield repository.get(id)
+
+    count = 0
+    for cdata in stored_objects():
+        count += 1
+        try:
+            identify_stored_key(cdata)
+        except (IntegrityError, UnsupportedPayloadError):
+            continue
+        return key_factory(repository, cdata)
+    raise IntegrityError(f"no stored object identifies the key type ({count} objects read)")
+
+
 def uses_same_chunker_secret(other_key, key):
     """is the chunker secret the same?"""
     # avoid breaking the deduplication by a different chunker secret

src/borg/repository.py

@@ -545,20 +545,33 @@ def check_pack_objects(pack_hex, obj_ranges, pack_size):
         )
 
 
-def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
-    """Find the superseded duplicates among a pack's gap bytes (bytes no index entry covers).
+def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, validate):
+    """Return the offset-ordered (offset, size) ranges of the superseded duplicates in a pack's gaps.
 
-    A gap holds a chunk copy stored again elsewhere, or objects from a backup that crashed before
-    writing its index. Walk each gap's object headers: an object whose chunk id the index maps to a
-    different location is a superseded duplicate (the id is a keyed MAC of the plaintext, so equal
-    ids mean equal content) and its bytes are redundant. An object whose id is not in the index
-    (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 gap is a byte range of the pack that no chunks index entry covers. A superseded duplicate is
+    an object in a gap whose chunk id the index maps to another location. Equal chunk ids mean
+    equal plaintext, so its bytes are redundant, whatever the stored size of the indexed copy.
 
-    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.
-    Returns the offset-ordered list of (offset, size) ranges holding superseded duplicates.
+    Each gap is walked from object header to object header, stepping by the object size the header
+    states. An object is reported when its chunk id is indexed at another location and validate
+    accepts it. The walk over a gap ends at a header that does not parse or that reaches past the
+    gap. Objects validate rejects are kept, and so is the rest of a gap where the walk ends early;
+    both are logged as a warning with the pack id and the offset.
+
+    reader: PackReader of the pack.
+    chunks: the chunks index (chunk id -> ChunkIndexEntry).
+    pack_id: id of the pack.
+    obj_ranges: the offset-ordered (obj_offset, obj_size) ranges of the pack's indexed objects,
+        non-overlapping and within the pack. The gaps are the byte ranges between and after them.
+    pack_size: size of the pack in bytes.
+    validate: validate(chunk_id, obj) -> bool, see repoobj.object_validator. obj is the object's
+        header and metadata slot (the meta_size metadata bytes after the header). True means chunk
+        id, meta_size and data_size are verified, so the reported range is exactly the object.
+        With None, nothing is reported.
     """
+    if validate is None:
+        return []
+
     # find the gaps: byte ranges no indexed object covers.
     gaps = []  # (start, end) of each gap, offset-ordered
     cursor = 0
@@ -569,22 +582,40 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size):
     if cursor < pack_size:
         gaps.append((cursor, pack_size))
 
+    pack_hex = bin_to_hex(pack_id)
     drop_ranges = []  # (obj_offset, obj_size) of superseded duplicates, offset-ordered
     hdr_size = RepoObj.obj_header.size
     for gstart, gend in gaps:
         offset = gstart
         while offset < gend:
-            hdr_data = reader.read(offset, hdr_size)
-            if len(hdr_data) < hdr_size:
+            # one read for the header and the metadata slot after it.
+            buf = reader.read(offset, min(gend - offset, META_READ_SIZE))
+            if len(buf) < hdr_size:
+                hdr, problem = None, f"{len(buf)} bytes, too few for an object header,"
+            else:
+                hdr, problem = PackReader._parse_header(buf[:hdr_size], offset, pack_size)
+            if hdr is not None and offset + hdr_size + hdr.meta_size + hdr.data_size > gend:
+                hdr, problem = None, "object reaching past its gap"
+            if hdr is None:
+                logger.warning(
+                    f"pack {pack_hex}: {problem} at offset {offset} in a gap, "
+                    f"keeping the remaining {gend - offset} bytes of the gap."
+                )
                 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:
-                break
-            if hdr.chunk_id in chunks:
-                entry = chunks[hdr.chunk_id]
-                if entry.pack_id != pack_id or entry.obj_offset != offset:
+            entry = chunks.get(hdr.chunk_id)
+            if entry is not None and (entry.pack_id != pack_id or entry.obj_offset != offset):
+                problem = reader._validation_problem(hdr, offset, buf, offset, validate)
+                if problem is None:
                     drop_ranges.append((offset, obj_size))
+                else:
+                    logger.warning(f"pack {pack_hex}: {problem} at offset {offset} in a gap, keeping its bytes.")
+            # TODO: obj_size is verified only for objects validate accepts. A wrong obj_size
+            # usually ends the walk at a header that does not parse, so the superseded duplicates in
+            # the rest of the gap are kept on every rewrite. PackReader._find_header could resync to
+            # the next object validate accepts, but in the none-* and authenticated-* modes that can
+            # be a copy of an object inside another object's unencrypted data, whose range can cover
+            # bytes of the gap objects after it, which would then be dropped.
             offset += obj_size
     return drop_ranges
 
@@ -1533,12 +1564,14 @@ class Repository:
         # 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, *, validate, update_index=True):
         """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: passed to compact_pack.
         """
         self._lock_refresh()
         entry = self.chunks.get(id)
@@ -1548,7 +1581,7 @@ class Repository:
         # 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.
@@ -1556,22 +1589,21 @@ class Repository:
 
             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, validate, chunks=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.
+        validate: passed to superseded_gap_ranges, whose ranges are dropped.
         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.
 
         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.
+        stale when the old pack is deleted. Gaps (byte ranges no index entry covers) are copied into
+        the new pack, except the superseded duplicates superseded_gap_ranges reports. 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.
@@ -1613,7 +1645,7 @@ class Repository:
         # 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
 
@@ -1767,7 +1799,7 @@ class Repository:
             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, *, validate, chunks=None, before_change=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
@@ -1777,17 +1809,17 @@ class Repository:
         transform: called as transform(chunk_id, obj_bytes) with an object's stored bytes; returns
             the replacement bytes, or obj_bytes itself (the identical bytes object) to keep the
             object unchanged. The chunk id (and thus the plaintext) must not change; sizes may.
+        validate: passed to superseded_gap_ranges, whose ranges are dropped.
         chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index
             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.
 
-        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.
+        The whole pack file is loaded into memory (bounded by the pack size limit). Gaps (byte ranges
+        no index entry covers) are copied into the new pack, except the superseded duplicates
+        superseded_gap_ranges reports. 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
@@ -1819,7 +1851,7 @@ class Repository:
         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

src/borg/testsuite/archiver/__init__.py

@@ -235,7 +235,7 @@ def write_wrong_content_chunk(archive, repository, chunk_id, *, ro_type=ROBJ_FIL
     if wrong_data is None:
         data = read_chunk(archive, repository, chunk_id, ro_type=ro_type)
         wrong_data = bytes([data[0] ^ 0xFF]) + data[1:]
-    repository.delete(chunk_id)  # put() would not replace an id the repo already has
+    repository.delete(chunk_id, validate=None)  # put() would not replace an id the repo already has
     repository.put(chunk_id, repo_objs.format(chunk_id, {}, wrong_data, ro_type=ro_type))
     repository.flush()
 

src/borg/testsuite/archiver/check_cmd_test.py

@@ -342,7 +342,7 @@ def test_missing_file_chunk(archivers, request):
             if item.path.endswith(src_file):
                 valid_chunks = item.chunks
                 killed_chunk = valid_chunks[-1]
-                repository.delete(killed_chunk.id)
+                repository.delete(killed_chunk.id, validate=None)
                 break
         else:
             pytest.fail("should not happen")  # convert 'fail'
@@ -398,7 +398,7 @@ def test_missing_file_chunk_report_truncated(archiver):
                 continue
             chunk_id = item.chunks[-1].id
             if chunk_id not in killed_ids:
-                repository.delete(chunk_id)
+                repository.delete(chunk_id, validate=None)
                 killed_ids.append(chunk_id)
             if len(killed_ids) >= 3:
                 break
@@ -430,7 +430,7 @@ def test_missing_file_chunk_refs_truncated(archivers, request):
         for item in archive.iter_items():
             if item.path.endswith("samefile0"):
                 killed_id = item.chunks[0].id
-                repository.delete(killed_id)
+                repository.delete(killed_id, validate=None)
                 break
     assert killed_id is not None
 
@@ -445,7 +445,7 @@ def test_missing_archive_item_chunk(archivers, request):
     check_cmd_setup(archiver)
     archive, repository = open_archive(archiver.repository_path, "archive1")
     with repository:
-        repository.delete(archive.item_ids[0])
+        repository.delete(archive.item_ids[0], validate=None)
     cmd(archiver, "check", exit_code=1)
     cmd(archiver, "check", "--repair", exit_code=0)
     cmd(archiver, "check", exit_code=0)
@@ -456,7 +456,7 @@ def test_missing_archive_metadata(archivers, request):
     check_cmd_setup(archiver)
     archive, repository = open_archive(archiver.repository_path, "archive1")
     with repository:
-        repository.delete(archive.id)
+        repository.delete(archive.id, validate=None)
     cmd(archiver, "check", exit_code=1)
     cmd(archiver, "check", "--repair", exit_code=0)
     cmd(archiver, "check", exit_code=0)
@@ -524,7 +524,7 @@ def test_check_format_missing_archive_metadata(archivers, request):
     check_cmd_setup(archiver)
     archive, repository = open_archive(archiver.repository_path, "archive1")
     with repository:
-        repository.delete(archive.id)
+        repository.delete(archive.id, validate=None)
     archive_id_hex = bin_to_hex(archive.id)
     output = cmd(archiver, "check", "-v", "--archives-only", "--format", "{archive} {comment}", exit_code=1)
     # the archive directory entry has no name for it, only the id, which {archive} {comment} would not show.
@@ -543,7 +543,7 @@ def test_missing_manifest(archivers, request):
         if isinstance(repository, Repository):
             repository.store_delete("config/manifest")
         else:
-            repository.delete(Manifest.MANIFEST_ID)
+            repository.delete(Manifest.MANIFEST_ID, validate=None)
     cmd(archiver, "check", exit_code=1)
     output = cmd(archiver, "check", "-v", "--repair", exit_code=0)
     assert "archive1" in output

src/borg/testsuite/archiver/compact_cmd_test.py

@@ -1,11 +1,14 @@
 import os
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
 from ...constants import *  # NOQA
 from ...helpers import get_cache_dir, bin_to_hex, sig_int, Error
 from ...hashindex import ChunkIndex
+from ...crypto.key import ChecksumKey
+from ...repoobj import RepoObj
 from ...repository import Repository
 from ...cache import files_cache_name, discover_files_cache_names, list_chunkindex_hashes
 from ...cache import delete_chunkindex_from_repo, write_chunkindex_to_repo
@@ -19,6 +22,11 @@ from ..repository_test import H, fchunk, pdchunk
 pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary")  # NOQA
 
 
+def gc_manifest(repository):
+    """Return a manifest stand-in for ArchiveGarbageCollector, whose repo_objs use a none-sha256 key."""
+    return SimpleNamespace(repo_objs=RepoObj(ChecksumKey(repository)))
+
+
 @pytest.mark.parametrize("stats", (True, False))
 def test_compact_empty_repository(archivers, request, stats):
     archiver = request.getfixturevalue(archivers)
@@ -228,7 +236,7 @@ def test_compact_packs_respects_threshold(tmp_path):
             flags = ChunkIndex.F_USED if H(i) in used else ChunkIndex.F_NONE
             repository.chunks[H(i)] = entry._replace(flags=flags)
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=40)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=40)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -253,42 +261,47 @@ def test_compact_superseded_duplicate(tmp_path):
 
     location = os.fspath(tmp_path / "repo")
     with Repository(location, exclusive=True, create=True) as repository:
+        manifest = gc_manifest(repository)
+        repo_objs = manifest.repo_objs
+        # objects formatted by repo_objs, so the superseded copy of X validates.
+        w, x, y = b"WWWW", b"XXXX", b"YYYY"
+        w_id, x_id, y_id = (repo_objs.id_hash(data) for data in (w, x, y))
         repository._pack_writer.max_count = 4  # one flush() -> one pack
         # pack A: three objects W, X, Y (X will be the one later superseded by a copy in pack B)
-        for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX"), (H(2), b"YYYY")]:
-            repository.put(cid, fchunk(data, chunk_id=cid))
+        for cid, data in [(w_id, w), (x_id, x), (y_id, y)]:
+            repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM))
         repository.flush()
-        pack_a = repository.chunks[H(0)].pack_id
+        pack_a = repository.chunks[w_id].pack_id
         pack_a_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(pack_a))
-        x_size = repository.chunks[H(1)].obj_size  # X's copy in pack A becomes a superseded gap
-        y_size = repository.chunks[H(2)].obj_size
+        x_size = repository.chunks[x_id].obj_size  # X's copy in pack A becomes a superseded gap
+        y_size = repository.chunks[y_id].obj_size
 
         # pack B: a second copy of X only, in its own pack (as a concurrent writer would have produced).
-        repository.put(H(1), fchunk(b"XXXX", chunk_id=H(1)))
+        repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM))
         repository.flush()
-        pack_b = repository.chunks[H(1)].pack_id
+        pack_b = repository.chunks[x_id].pack_id
         assert pack_b != pack_a
         # after the (simulated) fragment merge, the index points X at pack B; pack A's X bytes are now
         # a superseded, unindexed span. put() already repointed the index to pack B, so nothing to do.
 
         # mark usage: W and X used, Y unused. pack A is now mixed (W used, X superseded gap, Y unused).
-        used = {H(0), H(1)}
-        for i in range(3):
-            entry = repository.chunks[H(i)]
-            flags = ChunkIndex.F_USED if H(i) in used else ChunkIndex.F_NONE
-            repository.chunks[H(i)] = entry._replace(flags=flags)
+        used = {w_id, x_id}
+        for cid in (w_id, x_id, y_id):
+            entry = repository.chunks[cid]
+            flags = ChunkIndex.F_USED if cid in used else ChunkIndex.F_NONE
+            repository.chunks[cid] = entry._replace(flags=flags)
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
         # W still readable; X still readable from pack B; Y (the unused indexed object) dropped.
-        assert pdchunk(repository.get(H(0))) == b"WWWW"
-        assert pdchunk(repository.get(H(1))) == b"XXXX"
-        assert repository.get(H(2), raise_missing=False) is None
+        assert repo_objs.parse(w_id, repository.get(w_id), ro_type=ROBJ_FILE_STREAM)[1] == w
+        assert repo_objs.parse(x_id, repository.get(x_id), ro_type=ROBJ_FILE_STREAM)[1] == x
+        assert repository.get(y_id, raise_missing=False) is None
         # pack A rewritten, shrunk by Y's bytes (unused indexed) plus X's superseded gap: only W remains.
         assert bin_to_hex(pack_a) not in [info.name for info in repository.store_list("packs")]
-        new_pack = repository.chunks[H(0)].pack_id
+        new_pack = repository.chunks[w_id].pack_id
         new_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(new_pack))
         assert new_size == pack_a_size - y_size - x_size
 
@@ -312,7 +325,7 @@ def test_compact_keeps_orphan_pack(tmp_path):
         repository.store_store(orphan_key, b"orphan pack bytes")
         assert "ab" * 32 in [info.name for info in repository.store_list("packs")]
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -342,7 +355,7 @@ def test_compact_keeps_unindexed_waste(tmp_path):
         # ... but H(1)'s big object becomes an unindexed superseded span (well over threshold if counted).
         del repository.chunks[H(1)]
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -377,7 +390,7 @@ def test_compact_reclaims_indexed_waste_only(tmp_path):
         repository.chunks[H(2)] = repository.chunks[H(2)]._replace(flags=ChunkIndex.F_USED)
         del repository.chunks[H(3)]  # its bytes remain in unindexed_pack as unindexed data
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -439,7 +452,7 @@ def test_compact_keeps_stale_index_entries(tmp_path):
         repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED)
         repository.store_delete("packs/" + bin_to_hex(gone_pack))  # delete the pack file the index still references
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -460,7 +473,7 @@ def test_compact_skips_oversized_index_entry(tmp_path):
         entry = repository.chunks[H(0)]
         repository.chunks[H(0)] = entry._replace(flags=ChunkIndex.F_USED, obj_size=entry.obj_size + 10000)
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
 
@@ -492,7 +505,7 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch):
         total_bytes = sum(repository.store.info("packs/" + name).size for name in packs_before)
         assert total_bytes >= repository.pack_max_size  # combined size crosses the merge threshold
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
         assert gc.store_changed is True  # the merge changed the store
@@ -510,7 +523,7 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch):
 
         # a merged full-size pack is no longer tiny (the tiny limit is pack_max_size // 2 here), so a
         # second compact finds nothing to merge and leaves the store unchanged.
-        gc2 = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc2 = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc2.chunks = repository.chunks
         gc2.compact_packs()
         assert gc2.store_changed is False
@@ -537,7 +550,7 @@ def test_compact_packs_below_merge_size_gate_leaves_tiny_packs(tmp_path, monkeyp
         packs_before = {info.name for info in repository.store_list("packs")}
         assert len(packs_before) == 3
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
         assert gc.store_changed is False  # combined tiny bytes stay far below one full pack: leave them alone
@@ -566,7 +579,7 @@ def test_compact_packs_below_all_packs_gate_changes_nothing(tmp_path):
         packs_before = {info.name for info in repository.store_list("packs")}
         assert len(packs_before) == 2
 
-        gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10)
+        gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10)
         gc.chunks = repository.chunks
         gc.compact_packs()
         assert gc.store_changed is False  # below the all-packs gate: nothing was touched

src/borg/testsuite/archiver/debug_cmds_test.py

@@ -2,10 +2,17 @@ import json
 import os
 import pstats
 
+import pytest
+
+from ...cache import write_chunkindex_to_repo
 from ...constants import *  # NOQA
+from ...helpers import bin_to_hex
+from ...helpers.passphrase import PassphraseWrong
+from ...manifest import Manifest
 from .. import changedir
 from ..compress_test import Compressor
-from . import cmd, create_test_files, create_regular_file, generate_archiver_tests, RK_ENCRYPTION
+from . import cmd, create_test_files, create_regular_file, generate_archiver_tests, open_repository
+from . import KF_ENCRYPTION, KF_LOCATION, RK_ENCRYPTION
 
 pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,binary")  # NOQA
 
@@ -75,6 +82,8 @@ def test_debug_put_get_delete_obj(archivers, request):
 
     output = cmd(archiver, "debug", "delete-obj", id_hash)
     assert "deleted" in output
+    # put-obj stored the file's bytes, not a repo object, so the key type is read from the manifest.
+    assert "Could not set up the key" not in output
 
     # the object is gone now: deleting it again reports it is not there
     output = cmd(archiver, "debug", "delete-obj", id_hash)
@@ -84,6 +93,93 @@ def test_debug_put_get_delete_obj(archivers, request):
     assert "is invalid" in output
 
 
+def put_pack_with_superseded_gap(archiver):
+    """Store objects W, X and Y in one pack and X again in a second pack, then write the chunks index.
+
+    The index maps X to the second pack, so X's copy in the first pack is a superseded duplicate.
+    Returns ((w_id, x_id, y_id), (w_size, x_size, y_size)), the object sizes in the first pack.
+    """
+    with open_repository(archiver) as repository:
+        repo_objs = Manifest.load(repository, Manifest.NO_OPERATION_CHECK).repo_objs
+        datas = (b"W" * 100, b"X" * 100, b"Y" * 100)
+        ids = tuple(repo_objs.id_hash(data) for data in datas)
+        objs = [repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM) for id, data in zip(ids, datas)]
+        for id, obj in zip(ids, objs):
+            repository.put(id, obj)
+        repository.flush()
+        pack_id = repository.chunks[ids[0]].pack_id
+        assert all(repository.chunks[id].pack_id == pack_id for id in ids)
+        sizes = tuple(repository.chunks[id].obj_size for id in ids)
+        repository.put(ids[1], objs[1])  # the index now points X at this second copy
+        repository.flush()
+        assert repository.chunks[ids[1]].pack_id != pack_id
+        write_chunkindex_to_repo(repository, repository.chunks, incremental=False, force_write=True, delete_other=True)
+    return ids, sizes
+
+
+def pack_size_of(archiver, id):
+    with open_repository(archiver) as repository:
+        pack_name = bin_to_hex(repository.chunks[id].pack_id)
+        return next(info.size for info in repository.store_list("packs") if info.name == pack_name)
+
+
+def test_debug_delete_obj_drops_superseded_gap(archivers, request):
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    (w_id, x_id, y_id), (_, _, y_size) = put_pack_with_superseded_gap(archiver)
+
+    output = cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id))
+
+    assert "deleted" in output
+    assert "Could not set up the key" not in output
+    assert pack_size_of(archiver, y_id) == y_size  # W and the superseded copy of X are gone
+    with open_repository(archiver) as repository:
+        assert repository.get(w_id, raise_missing=False) is None
+        assert repository.get(x_id) is not None  # served from its second copy
+
+
+def test_debug_delete_obj_without_a_key_keeps_superseded_gap(archivers, request):
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", KF_ENCRYPTION, KF_LOCATION)
+    (w_id, x_id, y_id), (_, x_size, y_size) = put_pack_with_superseded_gap(archiver)
+    for name in os.listdir(archiver.keys_path):
+        os.unlink(os.path.join(archiver.keys_path, name))
+
+    output = cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id))
+
+    assert "Could not set up the key" in output
+    assert "deleted" in output
+    assert pack_size_of(archiver, y_id) == x_size + y_size  # only W is gone, the gap is kept
+
+
+def test_debug_delete_obj_with_a_wrong_passphrase_deletes_nothing(archivers, request, monkeypatch):
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    (w_id, x_id, y_id), (w_size, x_size, y_size) = put_pack_with_superseded_gap(archiver)
+    monkeypatch.setenv("BORG_PASSPHRASE", "wrong")
+
+    if archiver.FORK_DEFAULT:
+        cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id), exit_code=PassphraseWrong.exit_mcode)
+    else:
+        with pytest.raises(PassphraseWrong):
+            cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id))
+
+    assert pack_size_of(archiver, y_id) == w_size + x_size + y_size
+    with open_repository(archiver) as repository:
+        assert repository.get(w_id, raise_missing=False) is not None
+
+
+def test_debug_delete_obj_with_invalid_ids_only_sets_up_no_key(archivers, request, monkeypatch):
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    monkeypatch.setenv("BORG_PASSPHRASE", "wrong")
+
+    output = cmd(archiver, "debug", "delete-obj", "invalid")
+
+    assert "is invalid" in output
+    assert "Could not set up the key" not in output
+
+
 def test_debug_id_hash_format_put_get_parse_obj(archivers, request):
     """Test format-obj and parse-obj commands."""
     archiver = request.getfixturevalue(archivers)

src/borg/testsuite/archiver/extract_cmd_test.py

@@ -1036,7 +1036,7 @@ def test_extract_file_with_missing_chunk(archivers, request):
         for item in archive.iter_items():
             if item.path.endswith(src_file):
                 chunk = item.chunks[-1]
-                repository.delete(chunk.id)
+                repository.delete(chunk.id, validate=None)
                 break
         else:
             assert False  # missed the file

src/borg/testsuite/archiver/mount_cmds_test.py

@@ -313,7 +313,7 @@ def test_fuse_allow_damaged_files(archivers, request):
     with repository:
         for item in archive.iter_items():
             if item.path.endswith(src_file):
-                repository.delete(item.chunks[-1].id)
+                repository.delete(item.chunks[-1].id, validate=None)
                 path = item.path  # store full path for later
                 break
         else:

src/borg/testsuite/archiver/repo_compress_cmd_test.py

@@ -12,7 +12,7 @@ from ...compress import ZSTD, ZLIB, LZ4, CNONE
 from ...archiver.repo_compress_cmd import PackRecompressor
 
 from . import create_regular_file, cmd, RK_ENCRYPTION
-from ..repository_test import H, fchunk, pdchunk
+from ..repository_test import H, accept_all, fchunk, pdchunk
 
 
 def test_repo_compress(archiver):
@@ -264,7 +264,7 @@ def test_transform_pack_keeps_unindexed_gap(tmp_path):
         replacements = {H(0): fchunk(b"W" * 100, chunk_id=H(0)), H(2): fchunk(b"y", chunk_id=H(2))}
         calls = []
         new_pack_id, new_size = repository.transform_pack(
-            pack_id, [H(0), H(2)], transform_via(replacements), before_change=lambda: calls.append(1)
+            pack_id, [H(0), H(2)], transform_via(replacements), before_change=lambda: calls.append(1), validate=None
         )
         assert new_pack_id != pack_id
         assert calls == [1]  # before_change called (once), the store was modified
@@ -303,7 +303,10 @@ def test_transform_pack_drops_superseded_gap(tmp_path):
         assert pack_b != pack_a
 
         w_new = fchunk(b"W" * 100, chunk_id=H(0))
-        new_pack_id, new_size = repository.transform_pack(pack_a, [H(0)], transform_via({H(0): w_new}))
+        # fchunk objects have no encrypted metadata slot (see fchunk), so accept_all stands in for validate.
+        new_pack_id, new_size = repository.transform_pack(
+            pack_a, [H(0)], transform_via({H(0): w_new}), validate=accept_all
+        )
         assert new_pack_id != pack_a
         assert new_size == len(w_new)  # only W remains, X's superseded bytes were dropped
         assert pdchunk(repository.get(H(0))) == b"W" * 100
@@ -324,7 +327,7 @@ def test_transform_pack_unchanged_pack_untouched(tmp_path):
 
         calls = []
         new_pack_id, new_size = repository.transform_pack(
-            pack_id, [H(0), H(1)], transform_via({}), before_change=lambda: calls.append(1)
+            pack_id, [H(0), H(1)], transform_via({}), before_change=lambda: calls.append(1), validate=None
         )
         assert new_pack_id == pack_id
         assert calls == []  # nothing changed, so before_change was never called

src/borg/testsuite/archiver/webdav_cmd_test.py

@@ -627,7 +627,7 @@ def test_webdav_damaged_file(archivers, request):
         archive = Archive(manifest, manifest.archives.get("test").id)
         for item in archive.iter_items():
             if item.path.endswith("big"):
-                repository.delete(item.chunks[-1].id)  # get rid of a chunk of "big"
+                repository.delete(item.chunks[-1].id, validate=None)  # get rid of a chunk of "big"
                 break
         else:
             assert False  # missed the file

src/borg/testsuite/crypto/key_test.py

@@ -16,13 +16,15 @@ from ...crypto.key import AES_OCB_MAX_SESSION_BLOCKS
 from ...crypto.key import ID_HMAC_SHA_256, ID_BLAKE2b_256, ID_BLAKE3_256
 from ...crypto.key import UnsupportedManifestError, UnsupportedKeyFormatError, UnsupportedPayloadError
 from ...crypto.key import RepoKeyNotFoundError
-from ...crypto.key import identify_key
+from ...crypto.key import identify_key, key_from_repository
 from ...crypto.low_level import IntegrityError as IntegrityErrorBase
 from ...helpers import Error
 from ...helpers import IntegrityError
 from ...helpers import Location
 from ...helpers import msgpack
-from ...constants import KEY_ALGORITHMS, KeyBlobStorage, KeyType, ROBJ_MANIFEST
+from ...manifest import NoManifestError
+from ...repoobj import RepoObj
+from ...constants import KEY_ALGORITHMS, KeyBlobStorage, KeyType, ROBJ_FILE_STREAM, ROBJ_MANIFEST
 from ...helpers import hex_to_bin, bin_to_hex
 
 
@@ -665,3 +667,72 @@ def test_argon2_wrong_passphrase_returns_none(monkeypatch):
     saved = repository.store_key.call_args.args[0]
     _, saved_b64 = keyfile_parse(saved)
     assert key.decrypt_key_file(a2b_base64(saved_b64), "wrong passphrase") is None
+
+
+class StoredObjectsRepository:
+    """A repository with a stored manifest (None: no manifest) and objects by chunk id."""
+
+    def __init__(self, manifest, objects):
+        self.manifest = manifest
+        self.objects = objects
+
+    def get_manifest(self):
+        if self.manifest is None:
+            raise NoManifestError
+        return self.manifest
+
+    def get(self, id):
+        return self.objects[id]
+
+    def list(self, limit=None):
+        return [(id, len(obj)) for id, obj in self.objects.items()][:limit]
+
+
+def stored_object(key_cls, id):
+    return RepoObj(key_cls(MagicMock(id=bytes(32)))).format(id, {}, b"data", ro_type=ROBJ_FILE_STREAM)
+
+
+def with_key_type(cdata, key_type):
+    """Return cdata with its key type byte, the first byte of the data slot, set to key_type."""
+    offset = len(cdata) - len(RepoObj.extract_crypted_data(cdata))
+    return cdata[:offset] + bytes([key_type]) + cdata[offset + 1 :]
+
+
+def test_key_from_repository_reads_the_key_type_from_the_manifest():
+    objects = {b"o" * 32: stored_object(ChecksumKey, b"o" * 32)}
+    repository = StoredObjectsRepository(stored_object(Blake3ChecksumKey, bytes(32)), objects)
+    assert isinstance(key_from_repository(repository), Blake3ChecksumKey)
+
+
+def test_key_from_repository_skips_objects_that_do_not_identify_a_key_type():
+    good = stored_object(ChecksumKey, b"g" * 32)
+    objects = {
+        b"d" * 32: b"damaged",
+        b"u" * 32: with_key_type(good, KeyType.DROPPED_BLAKE3AUTHENTICATED),
+        b"g" * 32: good,
+    }
+    repository = StoredObjectsRepository(b"damaged manifest", objects)
+    assert isinstance(key_from_repository(repository), ChecksumKey)
+
+
+def test_key_from_repository_raises_if_no_object_identifies_the_key_type():
+    objects = {b"g" * 32: stored_object(ChecksumKey, b"g" * 32)}
+    with pytest.raises(IntegrityError):
+        key_from_repository(StoredObjectsRepository(None, objects), ids=())  # the manifest only
+    with pytest.raises(IntegrityError):
+        key_from_repository(StoredObjectsRepository(None, {b"d" * 32: b"damaged"}))
+
+
+def test_key_from_repository_loads_the_key_once(monkeypatch):
+    ids = (b"a" * 32, b"b" * 32)
+    repository = StoredObjectsRepository(None, {id: stored_object(ChecksumKey, id) for id in ids})
+    detected = []
+
+    def detect(repository, manifest_data, *, other=False):
+        detected.append(manifest_data)
+        raise IntegrityError("the key can not be loaded")
+
+    monkeypatch.setattr(ChecksumKey, "detect", detect)
+    with pytest.raises(IntegrityError, match="the key can not be loaded"):
+        key_from_repository(repository)
+    assert len(detected) == 1

src/borg/testsuite/repoobj_test.py

@@ -19,8 +19,9 @@ from ..repoobj import (
 from ..legacy.repoobj import RepoObj1
 from ..compress import LZ4
 
-# offsets of the size fields in the object header.
-META_SIZE_OFFSET = len(OBJ_MAGIC) + 1 + 32  # the magic, the version byte and the chunk id precede it
+# offsets of object header fields.
+CHUNK_ID_OFFSET = len(OBJ_MAGIC) + 1  # after the magic and the version byte
+META_SIZE_OFFSET = CHUNK_ID_OFFSET + 32
 DATA_SIZE_OFFSET = META_SIZE_OFFSET + 4
 
 

src/borg/testsuite/repository_test.py

@@ -16,12 +16,12 @@ from ..compress import CNONE
 from ..constants import MAX_CLOCK_SKEW, ROBJ_FILE_STREAM
 from ..crypto.key import CHPOKey, ChecksumKey
 from ..helpers import IntegrityError, Location, bin_to_hex
-from ..hashindex import ChunkIndex
+from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..repository import Repository, MAX_DATA_SIZE, MAX_VALIDATED_META_SIZE, propagate_rsh, rest_serve_command
-from ..repository import PackWriter, PackReader, PackTracker
+from ..repository import PackWriter, PackReader, PackTracker, superseded_gap_ranges
 from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION, object_validator
 from .hashindex_test import H
-from .repoobj_test import DATA_SIZE_OFFSET, META_SIZE_OFFSET
+from .repoobj_test import CHUNK_ID_OFFSET, DATA_SIZE_OFFSET, META_SIZE_OFFSET
 
 
 def test_rest_serve_command_local():
@@ -187,7 +187,7 @@ def test_consistency(repo_fixtures, request):
         repository.flush()
         assert pdchunk(repository.get(H(0))) == b"bar"
         # delete removes the object the index points at; the stale earlier copies are not resurrected.
-        repository.delete(H(0))
+        repository.delete(H(0), validate=None)
         with pytest.raises(Repository.ObjectNotFound):
             repository.get(H(0))
 
@@ -202,7 +202,7 @@ def test_delete_with_stale_earlier_object_in_pack(repo_fixtures, request):
         repository.put(H(1), fchunk(b"bbb"))  # fills the pack, flushing both objects
         repository.put(H(0), fchunk(b"ccc"))  # re-put: H(0)'s index entry moves to a new pack
         repository.flush()
-        repository.delete(H(1))
+        repository.delete(H(1), validate=None)
         with pytest.raises(Repository.ObjectNotFound):
             repository.get(H(1))
         assert pdchunk(repository.get(H(0))) == b"ccc"  # H(0) still served from its new pack
@@ -391,7 +391,9 @@ def test_compact_pack_copy_forward(repo_fixtures, request):
         assert repository.chunks[H(1)].pack_id == old_pack_id
         assert repository.chunks[H(2)].pack_id == old_pack_id
 
-        new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids={H(1)})
+        new_pack_id, dropped = repository.compact_pack(
+            old_pack_id, keep_ids={H(0), H(2)}, drop_ids={H(1)}, validate=None
+        )
 
         assert new_pack_id is not None and new_pack_id != old_pack_id
         assert dropped == len(chunk1)  # reported freed bytes for --stats
@@ -412,7 +414,9 @@ def test_compact_pack_drops_whole_pack(repo_fixtures, request):
     with repository:
         old_pack_id = repository.chunks[H(0)].pack_id
 
-        new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids=set(), drop_ids={H(0), H(1)})
+        new_pack_id, dropped = repository.compact_pack(
+            old_pack_id, keep_ids=set(), drop_ids={H(0), H(1)}, validate=None
+        )
         assert new_pack_id is None  # every byte dropped: no replacement pack
         assert dropped == len(chunk0) + len(chunk1)
 
@@ -432,7 +436,7 @@ def test_compact_pack_keep_all_is_noop(repo_fixtures, request):
         old_pack_id = repository.chunks[H(0)].pack_id
 
         new_pack_id, dropped = repository.compact_pack(
-            old_pack_id, keep_ids={H(1), H(0)}, drop_ids=set()
+            old_pack_id, keep_ids={H(1), H(0)}, drop_ids=set(), validate=None
         )  # out of order
 
         assert new_pack_id == old_pack_id
@@ -455,7 +459,7 @@ def test_compact_pack_keeps_gap(repo_fixtures, request):
         old_pack_id = repository.chunks[H(0)].pack_id
         del repository.chunks[H(1)]  # H(1)'s bytes stay in the pack but are now unindexed (a gap)
 
-        new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(2)}, drop_ids={H(0)})
+        new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(2)}, drop_ids={H(0)}, validate=None)
 
         assert new_pack_id is not None and new_pack_id != old_pack_id
         assert pdchunk(repository.get(H(2))) == b"DATA2"
@@ -477,7 +481,7 @@ def test_compact_pack_keeps_trailing_bytes(repo_fixtures, request):
         old_pack_id = repository.chunks[H(0)].pack_id
         del repository.chunks[H(2)]  # trailing unindexed bytes
 
-        new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(1)}, drop_ids={H(0)})
+        new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(1)}, drop_ids={H(0)}, validate=None)
 
         assert new_pack_id is not None and new_pack_id != old_pack_id
         assert pdchunk(repository.get(H(1))) == b"DATA1"
@@ -499,7 +503,9 @@ def test_compact_pack_drops_superseded_gap(repo_fixtures, request):
         old_pack_id = repository.chunks[H(0)].pack_id
         repository.chunks[H(1)] = repository.chunks[H(1)]._replace(pack_id=H(9))  # authoritative copy elsewhere
 
-        new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set())
+        new_pack_id, dropped = repository.compact_pack(
+            old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set(), validate=accept_all
+        )
 
         assert new_pack_id is not None and new_pack_id != old_pack_id
         assert dropped == len(chunk1)  # the superseded gap's bytes are counted as freed
@@ -523,7 +529,9 @@ def test_compact_pack_keeps_self_referencing_gap(repo_fixtures, request):
     with repository:
         old_pack_id = repository.chunks[H(0)].pack_id
 
-        new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set())
+        new_pack_id, dropped = repository.compact_pack(
+            old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set(), validate=accept_all
+        )
 
         assert new_pack_id == old_pack_id  # nothing dropped, defrag reproduced the same pack
         assert dropped == 0  # the self-referencing gap is kept, nothing freed
@@ -544,7 +552,7 @@ def test_compact_pack_detects_overlap(repo_fixtures, request):
         repository.chunks[H(1)] = entry._replace(obj_offset=0)  # now overlaps H(0) at offset 0
 
         with pytest.raises(IntegrityError):
-            repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set())
+            repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set(), validate=None)
         assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")]
 
 
@@ -561,7 +569,7 @@ def test_compact_pack_detects_past_eof(repo_fixtures, request):
         repository.chunks[H(1)] = entry._replace(obj_size=entry.obj_size + 1000)  # now claims to end past EOF
 
         with pytest.raises(IntegrityError):
-            repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set())
+            repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set(), validate=None)
         assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")]
 
 
@@ -583,7 +591,7 @@ def test_compact_pack_translates_read_range_error(repo_fixtures, request, monkey
 
         monkeypatch.setattr(repository.store, "defrag", short_read)
         with pytest.raises(IntegrityError):
-            repository.compact_pack(old_pack_id, keep_ids={H(0)}, drop_ids={H(1)})
+            repository.compact_pack(old_pack_id, keep_ids={H(0)}, drop_ids={H(1)}, validate=None)
         assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")]
         assert H(1) in repository.chunks  # still indexed: aborted before deleting the dropped id
 
@@ -723,7 +731,7 @@ def test_max_data_size(repo_fixtures, request):
         assert pdchunk(repository.get(H(0))) == max_data
         with pytest.raises(IntegrityError):
             repository.put(H(1), fchunk(max_data + b"x"))
-        repository.delete(H(0))
+        repository.delete(H(0), validate=None)
 
 
 def check(repository, repo_path, repair=False, status=True):
@@ -2254,3 +2262,121 @@ def test_pack_reader_in_memory_read_returns_view():
     assert bytes(view) == obj2
     pack[len(obj1)] ^= 0xFF  # a write to pack_contents is visible through the view
     assert view[0] == obj2[0] ^ 0xFF
+
+
+THIS_PACK = H(98)  # id of the pack whose gaps are walked
+OTHER_PACK = H(99)  # id of the pack holding the indexed copies
+
+
+def gap_pack(repo_objs, datas):
+    """Return (objects, chunks): datas formatted as repo objects, and an index mapping each to OTHER_PACK.
+
+    superseded_gap_ranges reports such an object if validate accepts it.
+    """
+    objs = [repo_objs.format(repo_objs.id_hash(data), {}, data, ro_type=ROBJ_FILE_STREAM) for data in datas]
+    chunks = {repo_objs.id_hash(data): ChunkIndexEntry(0, 0, OTHER_PACK, 0, len(obj)) for data, obj in zip(datas, objs)}
+    return objs, chunks
+
+
+def gap_ranges(pack, chunks, validate):
+    # no indexed objects: the whole pack is one gap.
+    reader = PackReader(pack_contents=pack)
+    return superseded_gap_ranges(reader, chunks, THIS_PACK, [], len(pack), validate=validate)
+
+
+def test_superseded_gap_ranges_reports_an_authenticated_duplicate(tmp_path):
+    repo_objs = aead_repo_objs(tmp_path)
+    (obj,), chunks = gap_pack(repo_objs, [b"superseded"])
+
+    assert gap_ranges(obj, chunks, object_validator(repo_objs)) == [(0, len(obj))]
+    assert gap_ranges(obj, chunks, None) == []  # no validator, nothing to report
+
+
+def test_superseded_gap_ranges_rejects_a_forged_chunk_id(tmp_path):
+    # the header's chunk id is replaced by the id of another indexed chunk, which the metadata
+    # slot's tag does not authenticate.
+    repo_objs = aead_repo_objs(tmp_path)
+    (obj,), chunks = gap_pack(repo_objs, [b"superseded"])
+    victim = repo_objs.id_hash(b"a chunk stored elsewhere")
+    chunks[victim] = ChunkIndexEntry(0, 0, OTHER_PACK, 0, len(obj))
+    forged = bytearray(obj)
+    forged[CHUNK_ID_OFFSET : CHUNK_ID_OFFSET + 32] = victim
+
+    assert gap_ranges(bytes(forged), chunks, object_validator(repo_objs)) == []
+
+
+def test_superseded_gap_ranges_rejects_an_inflated_data_size(tmp_path):
+    # data_size is increased to also cover the next object. validate rejects it: data_size must
+    # match csize, the data size recorded in the authenticated metadata.
+    repo_objs = aead_repo_objs(tmp_path)
+    (obj, behind), chunks = gap_pack(repo_objs, [b"superseded", b"innocent bystander"])
+    inflated = bytearray(obj + behind)
+    (data_size,) = struct.unpack("<I", inflated[DATA_SIZE_OFFSET : DATA_SIZE_OFFSET + 4])
+    inflated[DATA_SIZE_OFFSET : DATA_SIZE_OFFSET + 4] = struct.pack("<I", data_size + len(behind))
+
+    assert gap_ranges(bytes(inflated), chunks, object_validator(repo_objs)) == []
+
+
+def test_superseded_gap_ranges_reports_a_differently_sized_authoritative_copy(tmp_path):
+    # the indexed copy has another obj_size, as with other compression or obfuscation padding.
+    repo_objs = aead_repo_objs(tmp_path)
+    (obj,), chunks = gap_pack(repo_objs, [b"superseded"])
+    (chunk_id,) = chunks
+    chunks[chunk_id] = chunks[chunk_id]._replace(obj_size=len(obj) + 4096)
+
+    assert gap_ranges(obj, chunks, object_validator(repo_objs)) == [(0, len(obj))]
+
+
+def test_superseded_gap_ranges_continues_past_a_rejected_object(tmp_path):
+    # the first object's metadata slot is damaged, its header is intact: the walk steps over it.
+    repo_objs = aead_repo_objs(tmp_path)
+    (rejected, wanted), chunks = gap_pack(repo_objs, [b"tampered", b"superseded"])
+    damaged = bytearray(rejected)
+    hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(bytes(damaged[: RepoObj.obj_header.size])))
+    damaged[RepoObj.obj_header.size + hdr.meta_size - 1] ^= 0xFF  # last byte of the metadata slot
+
+    ranges = gap_ranges(bytes(damaged) + wanted, chunks, object_validator(repo_objs))
+
+    assert ranges == [(len(rejected), len(wanted))]
+
+
+def test_superseded_gap_ranges_warns_where_it_keeps_bytes(tmp_path, caplog):
+    # a rejected object, a header that does not parse and a gap too short for a header.
+    repo_objs = aead_repo_objs(tmp_path)
+    (rejected,), chunks = gap_pack(repo_objs, [b"tampered"])
+    damaged = bytearray(rejected)
+    hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(bytes(damaged[: RepoObj.obj_header.size])))
+    damaged[RepoObj.obj_header.size + hdr.meta_size - 1] ^= 0xFF  # last byte of the metadata slot
+    garbage = b"\0" * 100
+    validate = object_validator(repo_objs)
+    pack_hex = bin_to_hex(THIS_PACK)
+
+    with caplog.at_level(logging.WARNING, logger="borg.repository"):
+        assert gap_ranges(bytes(damaged) + garbage, chunks, validate) == []
+        assert gap_ranges(b"\0" * 3, chunks, validate) == []
+
+    assert f"pack {pack_hex}: object does not authenticate at offset 0 in a gap, keeping its bytes." in caplog.text
+    assert (
+        f"pack {pack_hex}: no object header at offset {len(rejected)} in a gap, "
+        f"keeping the remaining {len(garbage)} bytes of the gap." in caplog.text
+    )
+    assert f"pack {pack_hex}: 3 bytes, too few for an object header, at offset 0 in a gap" in caplog.text
+
+
+def test_superseded_gap_ranges_ends_at_an_object_reaching_past_the_gap(tmp_path, caplog):
+    # the object fits into the pack, but overlaps the indexed object after its gap.
+    repo_objs = aead_repo_objs(tmp_path)
+    (obj,), chunks = gap_pack(repo_objs, [b"superseded"])
+    reader = PackReader(pack_contents=obj + b"\0" * 100)
+    obj_ranges = [(len(obj) - 1, 101)]  # an indexed object overlapping the duplicate's last byte
+
+    with caplog.at_level(logging.WARNING, logger="borg.repository"):
+        ranges = superseded_gap_ranges(
+            reader, chunks, THIS_PACK, obj_ranges, len(obj) + 100, validate=object_validator(repo_objs)
+        )
+
+    assert ranges == []
+    assert (
+        f"pack {bin_to_hex(THIS_PACK)}: object reaching past its gap at offset 0 in a gap, "
+        f"keeping the remaining {len(obj) - 1} bytes of the gap." in caplog.text
+    )