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 @@
 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 @@ def check(
         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
-
-        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
-
-        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)
+        manifest_only: read only the manifest, else also the objects of the chunk ids in self.chunks.
+        """
+
+        def chunk_ids():  # reads self.chunks only if the manifest does not identify the key type
+            for id, _ in self.chunks.iteritems():
+                yield id
+
+        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 @@ 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.
@@ -2398,7 +2374,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/debug_cmd.py

@@ -3,23 +3,44 @@
 
 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
 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 @@ def do_debug_dump_manifest(self, args, repository, manifest):
     @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 @@ def decrypt_dump(id, cdata):
             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 @@ def print_finding(info, wanted, data, offset):
         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 @@ def do_debug_put_obj(self, args, repository):
     @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 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/crypto/key.py

@@ -3,6 +3,7 @@
 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 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