From 6f805da28e4b661b3fa7fa6f309714629635978c Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Thu, 13 Aug 2026 03:30:06 +0530 Subject: [PATCH 1/4] check --repair: resync past corrupt object headers when rebuilding the chunks index, #8476 When check --repair rebuilds the chunks index from the packs, a corrupt object header now makes iter_headers resync rather than raise: it takes a validate function and scans forward for the next object, in 1 MiB windows that overlap by one header so a header on a window boundary is still found. Repository-only checks pass no validate and keep raising IntegrityError on a corrupt header. OBJ_MAGIC also occurs inside payloads, so a candidate is accepted only when it authenticates. For AEAD keys, decrypting the metadata authenticates it against the header's magic, version and chunk_id, so the walk confirms a chunk id from a few hundred bytes. Keys that authenticate by chunk_id == id_hash(content) (id_check_is_authentication) read the whole object and parse() at the "repair" id place; validate.needs_data selects between the two. Authentication needs the key, so check --repair makes it before the rebuild with manifest_only=True. A repair that cannot read the manifest has no key and walks without resyncing. --- docs/internals/packs.rst | 28 +++- src/borg/archive.py | 39 ++++- src/borg/cache.py | 11 +- src/borg/repository.py | 103 +++++++++--- src/borg/testsuite/archiver/check_cmd_test.py | 28 ++++ src/borg/testsuite/cache_test.py | 23 ++- src/borg/testsuite/repository_test.py | 147 ++++++++++++++++++ 7 files changed, 350 insertions(+), 29 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 0f8d6acec9..5e43d1a23f 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -91,10 +91,30 @@ A reader locates the next blob by advancing:: next_blob_offset = current_blob_offset + REPOOBJ_HEADER_SIZE + meta_size + data_size -The per-blob magic limits the blast radius of corrupted length fields: if -``meta_size`` or ``data_size`` is damaged, the scanner loses at most one blob. -Once it finds the next ``OBJ_MAGIC`` sequence it resumes. Other corruption -(payload bit flips) is caught by AEAD on that blob without losing position. +``iter_headers()`` checks every header it walks: it must have ``OBJ_MAGIC``, a +supported version, and sizes that keep the blob inside the pack. A header that +fails these checks means a corrupt pack, and ``IntegrityError`` is raised. + +The per-blob magic limits the blast radius of corrupted length fields. The +repair walk (``iter_headers(validate=...)``, used when ``borg check --repair`` +rebuilds the chunks index from the packs) scans forward for the next blob and +resumes there, so the blobs after the damaged part of the pack are still found. + +``OBJ_MAGIC`` occurs inside the payloads as well, and in ``none`` and +``authenticated`` mode the payloads are user content stored as it is, so a +backed up file can contain something shaped like a blob. The scan therefore +accepts a candidate only if it parses. For the AEAD keys it reads the header and +the encrypted metadata, a few hundred bytes: decrypting the metadata +authenticates it together with the header's magic, version and chunk_id, which +are its AAD (additional authenticated data: authenticated with the ciphertext, +but not encrypted). The other keys authenticate by ``chunk_id == id_hash(content)`` +(``KeyBase.id_check_is_authentication``), which needs the blob's data, so for +those the scan reads the whole blob. The key is needed either way; a repair that +cannot read the manifest walks without scanning. + +``data_size`` is not part of that AAD, so accepting a candidate authenticates +its chunk id, and its size only as far as the blob fits into the pack. Bit flips +in the data are caught when the blob is read, on that blob alone. Blobs follow one another contiguously with no padding:: diff --git a/src/borg/archive.py b/src/borg/archive.py index 23dcc7ed57..3eb7b53d63 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -1893,6 +1893,32 @@ def __next__(self): return next(self._unpacker) +def resync_validator(repo_objs): + """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id. + + obj holds an object's header and encrypted metadata, plus its encrypted data when + validate.needs_data is set. For most keys, decrypting the metadata authenticates it against the + header (magic, version, chunk_id), so the metadata alone decides. Keys that authenticate by + chunk_id == id_hash(content) (id_check_is_authentication) need the data; for them + validate.needs_data is set and parse() checks that id at the "repair" id place. + """ + needs_data = repo_objs.key.id_check_is_authentication + + def validate(chunk_id, obj): + try: + if needs_data: + repo_objs.parse(chunk_id, obj, ro_type=ROBJ_DONTCARE, assert_id_place="repair") + else: + repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) + except Exception: + # authentication, id check, msgpack or decompression can each raise on non-object bytes. + return False + return True + + validate.needs_data = needs_data + return validate + + class ArchiveChecker: # Bound how many missing file chunks rebuild_archives buffers for its end-of-run report, # so checking a badly damaged repo with very many missing chunks can not exhaust memory. @@ -1947,7 +1973,18 @@ def check( # so we do not rebuild it from the packs (reading every pack is far too slow for a routine check). # --repair does rebuild from the packs (slow_rebuild=repair), working from the real packs so it # can detect and fix archives that reference chunks whose pack has gone missing. - self.chunks = build_chunkindex_from_repo(self.repository, slow_rebuild=repair, write_immediately=False) + # Under --repair, validate lets the rebuild resync past a corrupt object header (see resync_validator). + # It authenticates objects with the key, so make the key first; manifest_only=True makes make_key use + # the manifest, not self.chunks, which is still unset here. + if self.key is None: + try: + self.key = self.make_key(repository, manifest_only=True) + except IntegrityError as err: + logger.warning(f"{err}. Packs with a corrupt object header can not be repaired.") + validate = resync_validator(RepoObj(self.key)) if repair and self.key is not None else None + self.chunks = build_chunkindex_from_repo( + self.repository, slow_rebuild=repair, validate=validate, write_immediately=False + ) if self.key is None: self.key = self.make_key(repository) self.repo_objs = RepoObj(self.key) diff --git a/src/borg/cache.py b/src/borg/cache.py index b018879b38..3a67df31c7 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -852,7 +852,13 @@ def repack_chunkindex(repository): def build_chunkindex_from_repo( - repository, *, slow_rebuild=False, fragments_only=False, write_immediately=False, init_flags=ChunkIndex.F_USED + repository, + *, + slow_rebuild=False, + fragments_only=False, + validate=None, + write_immediately=False, + init_flags=ChunkIndex.F_USED, ): # fragments_only: build the index from the index/ fragments only, returning None if they cannot be # read completely, and never write to the repo. @@ -947,7 +953,8 @@ def build_chunkindex_from_repo( repository._lock_refresh() pi.show(increase=1) pack_id = hex_to_bin(info.name) - for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(): + # validate makes iter_headers resync past a corrupt object header and index the objects after it. + for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(validate=validate): num_chunks += 1 chunks[chunk_id] = ChunkIndexEntry( flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size diff --git a/src/borg/repository.py b/src/borg/repository.py index a007f7561f..c7b1484d38 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -30,7 +30,7 @@ from .storelocking import Lock from .logger import create_logger from .manifest import NoManifestError -from .repoobj import RepoObj, OBJ_MAGIC +from .repoobj import RepoObj, OBJ_MAGIC, SUPPORTED_OBJ_VERSIONS from .crypto.key import is_keyfile logger = create_logger(__name__) @@ -38,6 +38,9 @@ # an object name is its sha256 as 64 lowercase hex digits. _valid_object_name = re.compile(r"[0-9a-f]{64}").fullmatch +# how much of a pack PackReader reads at once when searching for the next object header. +RESYNC_WINDOW_SIZE = 1024 * 1024 + def repo_lister(repository, *, limit=None): marker = None @@ -362,24 +365,73 @@ def read(self, offset, size): return self.store.load(self.key, offset=offset, size=size) def size(self): - """Return the pack size in bytes; for a store-backed pack this is one metadata lookup.""" + """Return the pack size in bytes (a store metadata lookup, unless the pack is in memory).""" if self.pack_contents is not None: return len(self.pack_contents) return self.store.info(self.key).size - def iter_headers(self): + @staticmethod + def _parse_header(hdr_data, offset, pack_size): + """Return the ObjHeader in hdr_data if it is a valid header at offset, None otherwise. + + Valid means: OBJ_MAGIC, a supported version, and an object that fits into the pack. + """ + hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data)) + if hdr.magic != OBJ_MAGIC or hdr.version not in SUPPORTED_OBJ_VERSIONS: + return None + if offset + RepoObj.obj_header.size + hdr.meta_size + hdr.data_size > pack_size: + return None + return hdr + + def _find_header(self, offset, pack_size, validate): + """Scan forward from offset for the next object validate accepts, return its offset or None. + + A pack has no framing besides the object headers, so this searches for OBJ_MAGIC. That byte + sequence also occurs inside payloads, so a candidate is accepted only when its header parses + and validate confirms it. + """ + hdr_size = RepoObj.obj_header.size + while offset + hdr_size <= pack_size: + # a window at a time, so the scan costs one store request per RESYNC_WINDOW_SIZE bytes. + buf = bytes(self.read(offset, min(RESYNC_WINDOW_SIZE, pack_size - offset))) + if len(buf) < hdr_size: + break + pos = 0 + while True: + pos = buf.find(OBJ_MAGIC, pos) + if pos < 0 or pos + hdr_size > len(buf): + break # not in this window, or a header overlapping its end: the next window has it + hdr = self._parse_header(buf[pos : pos + hdr_size], offset + pos, pack_size) + if hdr is not None: + obj_size = hdr_size + hdr.meta_size + hdr.data_size + # an object is at most MAX_DATA_SIZE bytes (Repository.put), so a larger candidate is a + # false match on OBJ_MAGIC in a payload. + if obj_size <= MAX_DATA_SIZE: + size = obj_size if validate.needs_data else hdr_size + hdr.meta_size + end = pos + size + # the window holds these bytes, unless the candidate crosses its end. + obj = buf[pos:end] if end <= len(buf) else self.read(offset + pos, size) + if validate(hdr.chunk_id, obj): + return offset + pos + pos += 1 + # step by the window less one header, so a magic straddling the boundary is still found. + offset += max(len(buf) - (hdr_size - 1), 1) + return None + + def iter_headers(self, validate=None): """Yield (chunk_id, offset, size) for each object by walking the fixed object headers. - Only the headers are read, not the payloads, so locating every object costs one short - range read per object (or just a slice, when the pack is already in memory), plus one - store metadata lookup for the pack size. + The walk reads a header per object: one short range read each (or a slice, for a pack in + memory), plus one store metadata lookup for the pack size. + + A header must have OBJ_MAGIC, a supported version and describe an object that fits into + the pack, otherwise the pack is corrupt and IntegrityError is raised. A read shorter than + a header ends the walk: that is the end of the pack. - Each full header must have OBJ_MAGIC and describe an object that fits into the pack, - otherwise the pack is corrupt and IntegrityError is raised. Ending the walk instead - would be worse than raising: the chunks index rebuilt from these headers would just be - missing the rest of the pack, and borg check --repair would then "fix" the archives by - dropping chunks that are there. - A trailing partial header is the clean end of the pack, not corruption. + validate(chunk_id, obj): returns whether obj is a repo object with id chunk_id, where obj is + its header and metadata, plus its data when validate.needs_data is set. When validate is + given, a corrupt header makes the walk resync: it scans for the next object validate accepts + (see _find_header), continues there, and logs the skipped bytes. """ pack_hex = bin_to_hex(self.pack_id) if self.pack_id is not None else "" pack_size = self.size() @@ -389,17 +441,26 @@ def iter_headers(self): hdr_data = self.read(offset, hdr_size) if len(hdr_data) < hdr_size: break # clean EOF, or trailing partial bytes - hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data)) - if hdr.magic != OBJ_MAGIC: - raise IntegrityError( - f'pack {pack_hex}: no object header at offset {offset} (pack corruption), run "borg check"' + hdr = self._parse_header(hdr_data, offset, pack_size) + if hdr is None: + if validate is None: + raise IntegrityError( + f'pack {pack_hex}: invalid object header at offset {offset} (pack corruption), run "borg check"' + ) + next_offset = self._find_header(offset + 1, pack_size, validate) + if next_offset is None: + logger.warning( + f"pack {pack_hex}: invalid object header at offset {offset} and none after it, " + f"skipping the remaining {pack_size - offset} bytes." + ) + break + logger.warning( + f"pack {pack_hex}: invalid object header at offset {offset}, " + f"skipping {next_offset - offset} bytes to the next one." ) + offset = next_offset + continue obj_size = hdr_size + hdr.meta_size + hdr.data_size - if offset + obj_size > pack_size: - raise IntegrityError( - f"pack {pack_hex}: object extends past end of file at offset {offset} " - f'(pack corruption), run "borg check"' - ) yield hdr.chunk_id, offset, obj_size offset += obj_size diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 34a9ee5cb3..911548f100 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -718,6 +718,34 @@ def test_extra_chunks(archivers, request): cmd(archiver, "check", "-v", exit_code=0) # check does not deal with orphans anymore +def test_repair_resyncs_pack_with_corrupt_object_header(archivers, request): + """--repair rebuilds the index from a pack whose object header is damaged. + + A damaged header makes the walk lose the object boundaries, so the rebuild scans for the next + object that authenticates and carries on there. That needs the key, which --repair makes before + the rebuild. Repairing the pack itself is a separate step, see #10026. + """ + archiver = request.getfixturevalue(archivers) + if archiver.get_kind() != "local": + pytest.skip("inspects the store directly") + check_cmd_setup(archiver) + cmd(archiver, "check", exit_code=0) + + with Repository(archiver.repository_location, exclusive=True) as repository: + # damage the header of the second object of a pack that holds more than two. + by_pack = {} + for chunk_id, entry in repository.chunks.items(): + by_pack.setdefault(entry.pack_id, []).append((entry.obj_offset, chunk_id)) + pack_id, objs = next((p, sorted(o)) for p, o in by_pack.items() if len(o) > 2) + damaged_offset, _ = objs[1] + key = "packs/" + bin_to_hex(pack_id) + repository.store_store(key, corrupt(repository.store_load(key), damaged_offset)) + + output = cmd(archiver, "check", "--repair", "--debug", exit_code=0) + assert f"invalid object header at offset {damaged_offset}" in output + assert "bytes to the next one" in output # the rebuild resumed at the next object + + def test_repair_finish_flushes_pack_writer(archivers, request): """finish() stores chunks re-added during --repair before it (re)builds the index (#10055). diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index 872499c197..9944aae820 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -26,7 +26,8 @@ ) from ..hashindex import ChunkIndex, ChunkIndexEntry from ..crypto.key import AESOCBKey -from ..helpers import safe_ns +from ..helpers import bin_to_hex, safe_ns +from ..helpers import IntegrityError from ..helpers.msgpack import int_to_timestamp from ..manifest import Manifest from ..repository import Repository @@ -505,6 +506,26 @@ def test_close_consolidates_fragments_across_sessions(tmp_path, monkeypatch): assert cid in index +def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path): + """A corrupt object header fails the rebuild, but with repair=True the rest of the pack is indexed.""" + from .repository_test import accept_all, fchunk + + obj1 = bytearray(fchunk(b"first", chunk_id=H(90))) + obj2 = fchunk(b"second", chunk_id=H(91)) + obj1[0] ^= 0xFF # break the magic of the first object's header + pack_id = H(92) + with Repository(os.fspath(tmp_path / "repository"), exclusive=True, create=True) as repository: + repository.store_store("packs/" + bin_to_hex(pack_id), bytes(obj1) + obj2) + with pytest.raises(IntegrityError): + build_chunkindex_from_repo(repository, slow_rebuild=True) + # accept_all: accepts every candidate, so this exercises the plumbing only. + index = build_chunkindex_from_repo(repository, slow_rebuild=True, validate=accept_all) + assert H(91) in index # found by resyncing past the damaged header + assert H(90) not in index # its header is gone, so the object can not be indexed + assert index[H(91)].pack_id == pack_id + assert index[H(91)].obj_offset == len(obj1) + + def test_repack_leaves_sealed_untouched_and_reconstructs(tmp_path, monkeypatch): """Sealed (>= MIN) fragments survive a repack; build_chunkindex_from_repo reconstructs the index.""" monkeypatch.setattr(cache_mod, "CHUNKINDEX_FRAGMENT_ENTRIES_MIN", 1000) diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 2352e5ad79..11aab716c9 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -13,6 +13,11 @@ from ..constants import MAX_CLOCK_SKEW from ..helpers import IntegrityError, Location, bin_to_hex from ..hashindex import ChunkIndex +from .. import repository as repository_module +from ..archive import resync_validator +from ..compress import CNONE +from ..constants import ROBJ_FILE_STREAM +from ..crypto.key import CHPOKey, PlaintextKey from ..repository import Repository, MAX_DATA_SIZE, propagate_rsh, rest_serve_command, PackWriter, PackReader from ..repository import PackTracker from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION @@ -1868,6 +1873,148 @@ def test_pack_reader_raises_on_object_past_end_of_pack_through_store(tmp_path): list(reader.iter_headers()) +def test_pack_reader_raises_on_unsupported_version(): + obj = bytearray(fchunk(b"data", chunk_id=H(7))) + obj[len(OBJ_MAGIC)] = 0xEE # version byte + with pytest.raises(IntegrityError): + list(PackReader(pack_contents=bytes(obj)).iter_headers()) + + +def accept_all(chunk_id, obj): + # validate stand-in: accepts every candidate. + return True + + +accept_all.needs_data = False + + +def test_pack_reader_resync_skips_to_next_object(): + # after a corrupt header the walk continues at the next object. + obj1 = bytearray(fchunk(b"payload-one", meta=b"meta1", chunk_id=H(1))) + obj2 = fchunk(b"payload-two", meta=b"meta2", chunk_id=H(2)) + obj1[0] ^= 0xFF # break the magic of the first object's header + reader = PackReader(pack_contents=bytes(obj1) + obj2) + assert list(reader.iter_headers(validate=accept_all)) == [(H(2), len(obj1), len(obj2))] + + +def test_pack_reader_resync_recovers_from_corrupted_size(): + # a header whose sizes point past the pack, so the next object is found by scanning. + obj1 = bytearray(fchunk(b"payload-one", meta=b"meta1", chunk_id=H(1))) + obj2 = fchunk(b"payload-two", meta=b"meta2", chunk_id=H(2)) + obj3 = fchunk(b"payload-three", chunk_id=H(3)) + # the header's data_size field (magic 8, version 1, chunk_id 32, meta_size 4, data_size 4), + # set to a value reaching far past the end of the pack: + obj1[45:49] = b"\xff\xff\xff\x00" + pack = bytes(obj1) + obj2 + obj3 + reader = PackReader(pack_contents=pack) + assert list(reader.iter_headers(validate=accept_all)) == [ + (H(2), len(obj1), len(obj2)), + (H(3), len(obj1) + len(obj2), len(obj3)), + ] + + +def test_pack_reader_resync_ignores_magic_in_payload(): + # both headers are broken, so the scan runs into the OBJ_MAGIC in obj2's payload before obj3. + obj1 = bytearray(fchunk(b"data", chunk_id=H(1))) + obj2 = bytearray(fchunk(OBJ_MAGIC + b"looks like a header, is not", chunk_id=H(2))) + obj3 = fchunk(b"payload-three", chunk_id=H(3)) + obj1[0] ^= 0xFF + obj2[0] ^= 0xFF + reader = PackReader(pack_contents=bytes(obj1) + bytes(obj2) + obj3) + assert list(reader.iter_headers(validate=accept_all)) == [(H(3), len(obj1) + len(obj2), len(obj3))] + + +def test_pack_reader_resync_finds_header_across_window_boundary(monkeypatch): + # the next header straddles a scan window boundary. + monkeypatch.setattr(repository_module, "RESYNC_WINDOW_SIZE", 64) + obj1 = bytearray(fchunk(b"x" * 100, chunk_id=H(1))) + obj2 = fchunk(b"payload-two", chunk_id=H(2)) + obj1[0] ^= 0xFF + reader = PackReader(pack_contents=bytes(obj1) + obj2) + assert list(reader.iter_headers(validate=accept_all)) == [(H(2), len(obj1), len(obj2))] + + +def test_pack_reader_resync_no_further_header(): + # no object after the damage: the walk ends with what it found. + obj = fchunk(b"data", chunk_id=H(1)) + pack = obj + b"\xaa" * 200 + reader = PackReader(pack_contents=pack) + assert list(reader.iter_headers(validate=accept_all)) == [(H(1), 0, len(obj))] + + +def aead_repo_objs(tmp_path): + # a RepoObj with an AEAD key, whose metadata authenticates on its own. + repository = Repository(str(tmp_path / "repo"), create=True) + key = CHPOKey(repository) + key.init_from_random_data() + key.init_ciphers() + return RepoObj(key) + + +def test_pack_reader_resync_rejects_metadata_that_does_not_authenticate(tmp_path): + # bytes with a well-formed header whose metadata does not decrypt: the scan must walk past them. + repo_objs = aead_repo_objs(tmp_path) + data = b"the real next object" + real_id = repo_objs.id_hash(data) + obj1 = bytearray(repo_objs.format(repo_objs.id_hash(b"first"), {}, b"first", ro_type=ROBJ_FILE_STREAM)) + obj1[0] ^= 0xFF # break obj1's header, so the walk has to resync + garbage = fchunk(b"payload", meta=b"not encrypted metadata", chunk_id=H(9)) + obj2 = repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM) + reader = PackReader(pack_contents=bytes(obj1) + garbage + obj2) + headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + assert headers == [(real_id, len(obj1) + len(garbage), len(obj2))] + + +def test_pack_reader_resync_accepts_an_object_with_corrupt_data(tmp_path): + # the AEAD keys authenticate the metadata, so the scan resyncs at an object with damaged data. + # Reading that object reports the damage. + repo_objs = aead_repo_objs(tmp_path) + data = b"the real next object" + real_id = repo_objs.id_hash(data) + obj1 = bytearray(repo_objs.format(repo_objs.id_hash(b"first"), {}, b"first", ro_type=ROBJ_FILE_STREAM)) + obj1[0] ^= 0xFF # break obj1's header, so the walk has to resync + obj2 = bytearray(repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM)) + obj2[-1] ^= 0xFF # damage the encrypted data, leaving the header and the metadata intact + reader = PackReader(pack_contents=bytes(obj1) + bytes(obj2)) + headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + assert headers == [(real_id, len(obj1), len(obj2))] + with pytest.raises(IntegrityError): + repo_objs.parse(real_id, bytes(obj2), ro_type=ROBJ_FILE_STREAM) + + +def test_pack_reader_resync_rejects_user_content_that_looks_like_an_object(tmp_path): + # In "none" mode with no compression, user content lands in the pack as it is, so a backed up + # file can contain something shaped like an object. Those keys authenticate by the id check over + # the content, so the scan reads whole candidates. + repository = Repository(str(tmp_path / "repo"), create=True) + repo_objs = RepoObj(PlaintextKey(repository)) + assert resync_validator(repo_objs).needs_data + repo_objs.compressor = CNONE() + decoy = bytearray(repo_objs.format(repo_objs.id_hash(b"decoy"), {}, b"decoy", ro_type=ROBJ_FILE_STREAM)) + decoy[-1] ^= 0xFF # its content no longer hashes to the id in its header + content = bytes(decoy) # a user stores exactly those bytes in a file + obj1 = bytearray(repo_objs.format(repo_objs.id_hash(content), {}, content, ro_type=ROBJ_FILE_STREAM)) + assert content in obj1 # the decoy is in the pack verbatim + obj1[0] ^= 0xFF # break obj1's header, so the walk resyncs and runs into the decoy + data = b"the real next object" + real_id = repo_objs.id_hash(data) + obj2 = repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM) + reader = PackReader(pack_contents=bytes(obj1) + obj2) + headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + assert headers == [(real_id, len(obj1), len(obj2))] + + +def test_pack_reader_resync_through_store(tmp_path): + obj1 = bytearray(fchunk(b"FIRST", chunk_id=H(47))) + obj2 = fchunk(b"SECOND", chunk_id=H(48)) + obj1[0] ^= 0xFF + pack_id = H(50) + with Repository(str(tmp_path / "repo"), exclusive=True, create=True) as repository: + repository.store_store("packs/" + bin_to_hex(pack_id), bytes(obj1) + obj2) + reader = PackReader(repository.store, pack_id) + assert list(reader.iter_headers(validate=accept_all)) == [(H(48), len(obj1), len(obj2))] + + def test_pack_reader_size(tmp_path): obj = fchunk(b"data", meta=b"meta", chunk_id=H(6)) assert PackReader(pack_contents=obj).size() == len(obj) From d196c0a9565723564f565efde412c6fdfebd2e0a Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Sat, 15 Aug 2026 19:28:07 +0530 Subject: [PATCH 2/4] check --repair: validate a resync candidate from its metadata slot alone, #8476 Every key mode covers the object header by the metadata slot's AAD, so parse_meta confirms a candidate and validate.needs_data is gone. --- docs/internals/packs.rst | 25 ++++++++--------- src/borg/archive.py | 28 ++++++++----------- src/borg/cache.py | 3 +- src/borg/repository.py | 16 +++++------ src/borg/testsuite/archiver/check_cmd_test.py | 7 ++--- src/borg/testsuite/cache_test.py | 6 ++-- src/borg/testsuite/repository_test.py | 20 ++++++------- 7 files changed, 49 insertions(+), 56 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 5e43d1a23f..0fdf58cb59 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -100,19 +100,18 @@ repair walk (``iter_headers(validate=...)``, used when ``borg check --repair`` rebuilds the chunks index from the packs) scans forward for the next blob and resumes there, so the blobs after the damaged part of the pack are still found. -``OBJ_MAGIC`` occurs inside the payloads as well, and in ``none`` and -``authenticated`` mode the payloads are user content stored as it is, so a -backed up file can contain something shaped like a blob. The scan therefore -accepts a candidate only if it parses. For the AEAD keys it reads the header and -the encrypted metadata, a few hundred bytes: decrypting the metadata -authenticates it together with the header's magic, version and chunk_id, which -are its AAD (additional authenticated data: authenticated with the ciphertext, -but not encrypted). The other keys authenticate by ``chunk_id == id_hash(content)`` -(``KeyBase.id_check_is_authentication``), which needs the blob's data, so for -those the scan reads the whole blob. The key is needed either way; a repair that -cannot read the manifest walks without scanning. - -``data_size`` is not part of that AAD, so accepting a candidate authenticates +``OBJ_MAGIC`` occurs inside the payloads as well, and in the ``none-*`` and +``authenticated-*`` modes the payloads are user content stored as it is, so a +backed up file can contain something shaped like a blob. A candidate is +therefore accepted only when its metadata slot verifies against the header AAD +described above; the header and that slot, a few hundred bytes, are what the +scan reads. Verifying needs the key, so a repair that cannot read the manifest +walks without scanning. + +In the ``none-*`` modes the tag is an unkeyed checksum, so the scan accepts any +well-formed blob, including one a backed up file contains. + +``data_size`` is not part of the AAD, so accepting a candidate authenticates its chunk id, and its size only as far as the blob fits into the pack. Bit flips in the data are caught when the blob is read, on that blob alone. diff --git a/src/borg/archive.py b/src/borg/archive.py index 3eb7b53d63..d2117ed522 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -1896,26 +1896,22 @@ def __next__(self): def resync_validator(repo_objs): """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id. - obj holds an object's header and encrypted metadata, plus its encrypted data when - validate.needs_data is set. For most keys, decrypting the metadata authenticates it against the - header (magic, version, chunk_id), so the metadata alone decides. Keys that authenticate by - chunk_id == id_hash(content) (id_check_is_authentication) need the data; for them - validate.needs_data is set and parse() checks that id at the "repair" id place. + obj is an object's header plus its metadata slot. Parsing that slot verifies its tag, which is + computed over the header's magic, version and chunk id as well (AAD, additional authenticated + data: bytes the tag covers without being part of the ciphertext). + + In the "none-*" modes the tag is an unkeyed checksum, so validate accepts any well-formed + object, including one that a backed up file contains. """ - needs_data = repo_objs.key.id_check_is_authentication def validate(chunk_id, obj): try: - if needs_data: - repo_objs.parse(chunk_id, obj, ro_type=ROBJ_DONTCARE, assert_id_place="repair") - else: - repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) + repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) except Exception: - # authentication, id check, msgpack or decompression can each raise on non-object bytes. + # arbitrary bytes fail the tag, the msgpack unpacking or the length checks. return False return True - validate.needs_data = needs_data return validate @@ -1973,10 +1969,10 @@ def check( # so we do not rebuild it from the packs (reading every pack is far too slow for a routine check). # --repair does rebuild from the packs (slow_rebuild=repair), working from the real packs so it # can detect and fix archives that reference chunks whose pack has gone missing. - # Under --repair, validate lets the rebuild resync past a corrupt object header (see resync_validator). - # It authenticates objects with the key, so make the key first; manifest_only=True makes make_key use - # the manifest, not self.chunks, which is still unset here. - if self.key is None: + # --repair also passes validate, which makes the rebuild resync past a corrupt object header. + # Validating needs the key, so read it here. manifest_only=True, because the other source + # make_key reads keys from is self.chunks, which is only built below. + if repair and self.key is None: try: self.key = self.make_key(repository, manifest_only=True) except IntegrityError as err: diff --git a/src/borg/cache.py b/src/borg/cache.py index 3a67df31c7..6602802839 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -862,6 +862,8 @@ def build_chunkindex_from_repo( ): # fragments_only: build the index from the index/ fragments only, returning None if they cannot be # read completely, and never write to the repo. + # validate: handed to PackReader.iter_headers when rebuilding from the packs, making it resync + # past a corrupt object header rather than raise IntegrityError. assert not (slow_rebuild and fragments_only) assert not (fragments_only and write_immediately) # fragments_only never writes to the repo # first, try to build a fresh, mostly complete chunk index from centrally stored index fragments: @@ -953,7 +955,6 @@ def build_chunkindex_from_repo( repository._lock_refresh() pi.show(increase=1) pack_id = hex_to_bin(info.name) - # validate makes iter_headers resync past a corrupt object header and index the objects after it. for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(validate=validate): num_chunks += 1 chunks[chunk_id] = ChunkIndexEntry( diff --git a/src/borg/repository.py b/src/borg/repository.py index c7b1484d38..1e8afd504e 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -388,7 +388,7 @@ def _find_header(self, offset, pack_size, validate): A pack has no framing besides the object headers, so this searches for OBJ_MAGIC. That byte sequence also occurs inside payloads, so a candidate is accepted only when its header parses - and validate confirms it. + and validate(chunk_id, obj) confirms the header and metadata slot at that position. """ hdr_size = RepoObj.obj_header.size while offset + hdr_size <= pack_size: @@ -404,10 +404,10 @@ def _find_header(self, offset, pack_size, validate): hdr = self._parse_header(buf[pos : pos + hdr_size], offset + pos, pack_size) if hdr is not None: obj_size = hdr_size + hdr.meta_size + hdr.data_size - # an object is at most MAX_DATA_SIZE bytes (Repository.put), so a larger candidate is a - # false match on OBJ_MAGIC in a payload. + # an object is at most MAX_DATA_SIZE bytes, so a bigger one is a false match on + # OBJ_MAGIC inside a payload. if obj_size <= MAX_DATA_SIZE: - size = obj_size if validate.needs_data else hdr_size + hdr.meta_size + size = hdr_size + hdr.meta_size # the bytes validate looks at end = pos + size # the window holds these bytes, unless the candidate crosses its end. obj = buf[pos:end] if end <= len(buf) else self.read(offset + pos, size) @@ -428,10 +428,10 @@ def iter_headers(self, validate=None): the pack, otherwise the pack is corrupt and IntegrityError is raised. A read shorter than a header ends the walk: that is the end of the pack. - validate(chunk_id, obj): returns whether obj is a repo object with id chunk_id, where obj is - its header and metadata, plus its data when validate.needs_data is set. When validate is - given, a corrupt header makes the walk resync: it scans for the next object validate accepts - (see _find_header), continues there, and logs the skipped bytes. + validate(chunk_id, obj) tells whether obj - an object's header and metadata slot - is the + repo object with id chunk_id. Given one, a corrupt header makes the walk resync instead: + it scans for the next object validate accepts, logs how many bytes that skipped and + continues there. """ pack_hex = bin_to_hex(self.pack_id) if self.pack_id is not None else "" pack_size = self.size() diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 911548f100..76d2e2cc9b 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -719,11 +719,10 @@ def test_extra_chunks(archivers, request): def test_repair_resyncs_pack_with_corrupt_object_header(archivers, request): - """--repair rebuilds the index from a pack whose object header is damaged. + """--repair rebuilds the chunks index from a pack whose object header is damaged. - A damaged header makes the walk lose the object boundaries, so the rebuild scans for the next - object that authenticates and carries on there. That needs the key, which --repair makes before - the rebuild. Repairing the pack itself is a separate step, see #10026. + A damaged header loses the object boundaries, so the rebuild scans for the next object that + authenticates and continues there. Authenticating needs the key, which --repair reads first. """ archiver = request.getfixturevalue(archivers) if archiver.get_kind() != "local": diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py index 9944aae820..a6fdda8685 100644 --- a/src/borg/testsuite/cache_test.py +++ b/src/borg/testsuite/cache_test.py @@ -507,7 +507,7 @@ def test_close_consolidates_fragments_across_sessions(tmp_path, monkeypatch): def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path): - """A corrupt object header fails the rebuild, but with repair=True the rest of the pack is indexed.""" + """A corrupt object header fails the rebuild; with validate, the objects after it are indexed.""" from .repository_test import accept_all, fchunk obj1 = bytearray(fchunk(b"first", chunk_id=H(90))) @@ -518,10 +518,10 @@ def test_build_chunkindex_repair_resyncs_after_corrupt_header(tmp_path): repository.store_store("packs/" + bin_to_hex(pack_id), bytes(obj1) + obj2) with pytest.raises(IntegrityError): build_chunkindex_from_repo(repository, slow_rebuild=True) - # accept_all: accepts every candidate, so this exercises the plumbing only. + # accept_all takes any candidate, so this covers the plumbing, not the authentication. index = build_chunkindex_from_repo(repository, slow_rebuild=True, validate=accept_all) assert H(91) in index # found by resyncing past the damaged header - assert H(90) not in index # its header is gone, so the object can not be indexed + assert H(90) not in index # its header is damaged, so its id is unknown assert index[H(91)].pack_id == pack_id assert index[H(91)].obj_offset == len(obj1) diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 11aab716c9..6bd0bf022f 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -17,7 +17,7 @@ from ..archive import resync_validator from ..compress import CNONE from ..constants import ROBJ_FILE_STREAM -from ..crypto.key import CHPOKey, PlaintextKey +from ..crypto.key import CHPOKey, ChecksumKey from ..repository import Repository, MAX_DATA_SIZE, propagate_rsh, rest_serve_command, PackWriter, PackReader from ..repository import PackTracker from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION @@ -1885,9 +1885,6 @@ def accept_all(chunk_id, obj): return True -accept_all.needs_data = False - - def test_pack_reader_resync_skips_to_next_object(): # after a corrupt header the walk continues at the next object. obj1 = bytearray(fchunk(b"payload-one", meta=b"meta1", chunk_id=H(1))) @@ -1982,16 +1979,17 @@ def test_pack_reader_resync_accepts_an_object_with_corrupt_data(tmp_path): repo_objs.parse(real_id, bytes(obj2), ro_type=ROBJ_FILE_STREAM) -def test_pack_reader_resync_rejects_user_content_that_looks_like_an_object(tmp_path): - # In "none" mode with no compression, user content lands in the pack as it is, so a backed up - # file can contain something shaped like an object. Those keys authenticate by the id check over - # the content, so the scan reads whole candidates. +def test_pack_reader_resync_rejects_damaged_user_content_without_a_key(tmp_path): + # In "none-*" mode with no compression, user content lands in the pack as it is, so a backed up + # file can contain something shaped like an object. The metadata slot's checksum covers the + # object header, so damaged candidate bytes are still ruled out - what these modes can not rule + # out is an intact object put into a file on purpose, there being no secret to tell them apart. repository = Repository(str(tmp_path / "repo"), create=True) - repo_objs = RepoObj(PlaintextKey(repository)) - assert resync_validator(repo_objs).needs_data + repo_objs = RepoObj(ChecksumKey(repository)) repo_objs.compressor = CNONE() decoy = bytearray(repo_objs.format(repo_objs.id_hash(b"decoy"), {}, b"decoy", ro_type=ROBJ_FILE_STREAM)) - decoy[-1] ^= 0xFF # its content no longer hashes to the id in its header + hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(bytes(decoy[: RepoObj.obj_header.size]))) + decoy[RepoObj.obj_header.size + hdr.meta_size - 1] ^= 0xFF # damage its metadata slot content = bytes(decoy) # a user stores exactly those bytes in a file obj1 = bytearray(repo_objs.format(repo_objs.id_hash(content), {}, content, ro_type=ROBJ_FILE_STREAM)) assert content in obj1 # the decoy is in the pack verbatim From ab1762f1a8d3c291475d38c7475127958e9bbb32 Mon Sep 17 00:00:00 2001 From: Mrityunjay Raj Date: Fri, 21 Aug 2026 01:01:02 +0530 Subject: [PATCH 3/4] check --repair: resync from the last accepted object header, #8476 A meta_size or data_size corrupted to a value that keeps the object inside the pack leaves the header valid, so the walk only notices at the misaligned offset it jumps to; scanning from there loses the intact objects in between. An object is dropped when a recovered object starts inside it, its size field being wrong. --- docs/internals/packs.rst | 9 +++-- src/borg/archive.py | 7 +++- src/borg/repository.py | 34 +++++++++++++++---- src/borg/testsuite/archiver/check_cmd_test.py | 4 ++- src/borg/testsuite/repository_test.py | 31 ++++++++++++++++- 5 files changed, 73 insertions(+), 12 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 0fdf58cb59..39e9b7918c 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -97,8 +97,13 @@ fails these checks means a corrupt pack, and ``IntegrityError`` is raised. The per-blob magic limits the blast radius of corrupted length fields. The repair walk (``iter_headers(validate=...)``, used when ``borg check --repair`` -rebuilds the chunks index from the packs) scans forward for the next blob and -resumes there, so the blobs after the damaged part of the pack are still found. +rebuilds the chunks index from the packs) scans for the next blob and resumes +there, so the blobs after the damaged part of the pack are still found. The scan +starts just past the last blob the walk accepted: a corrupted length field that +keeps the blob inside the pack leaves the header valid, and is noticed only at +the misaligned offset it points to. The blob carrying it is dropped when a +recovered blob starts inside the extent it claims - its length field is wrong, +so it can not be read back. ``OBJ_MAGIC`` occurs inside the payloads as well, and in the ``none-*`` and ``authenticated-*`` modes the payloads are user content stored as it is, so a diff --git a/src/borg/archive.py b/src/borg/archive.py index d2117ed522..52eed5152b 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -2513,7 +2513,12 @@ def finish(self): # the packs changed, so the index no longer matches them: rebuild it from the packs # and persist it. logger.info("Rebuilding and writing the repository chunks index.") - build_chunkindex_from_repo(self.repository, slow_rebuild=True, write_immediately=True) + build_chunkindex_from_repo( + self.repository, + slow_rebuild=True, + validate=resync_validator(self.repo_objs), + write_immediately=True, + ) else: # the packs are unchanged, so the index still matches them: persist it as is. logger.info("Writing the rebuilt repository chunks index.") diff --git a/src/borg/repository.py b/src/borg/repository.py index 1e8afd504e..aec087c8a1 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -430,13 +430,17 @@ def iter_headers(self, validate=None): validate(chunk_id, obj) tells whether obj - an object's header and metadata slot - is the repo object with id chunk_id. Given one, a corrupt header makes the walk resync instead: - it scans for the next object validate accepts, logs how many bytes that skipped and - continues there. + it scans from just past the last object it accepted for the next object validate accepts + and continues there. It scans from there because a corrupt meta_size or data_size leaves + the header valid and moves the walk into a later object. An object is dropped when a + recovered object starts inside it: its size field is wrong. """ pack_hex = bin_to_hex(self.pack_id) if self.pack_id is not None else "" pack_size = self.size() hdr_size = RepoObj.obj_header.size offset = 0 + scan_from = 1 # just past the last accepted object's header + pending = None # the last object seen, yielded once the walk gets past its end while True: hdr_data = self.read(offset, hdr_size) if len(hdr_data) < hdr_size: @@ -447,22 +451,38 @@ def iter_headers(self, validate=None): raise IntegrityError( f'pack {pack_hex}: invalid object header at offset {offset} (pack corruption), run "borg check"' ) - next_offset = self._find_header(offset + 1, pack_size, validate) + next_offset = self._find_header(scan_from, pack_size, validate) if next_offset is None: logger.warning( - f"pack {pack_hex}: invalid object header at offset {offset} and none after it, " - f"skipping the remaining {pack_size - offset} bytes." + f"pack {pack_hex}: invalid object header at offset {offset} and no object after " + f"offset {scan_from}, skipping the remaining {pack_size - offset} bytes." ) break logger.warning( f"pack {pack_hex}: invalid object header at offset {offset}, " - f"skipping {next_offset - offset} bytes to the next one." + f"continuing at the object at offset {next_offset}." ) + if pending is not None: + _, pending_offset, pending_size = pending + if next_offset < pending_offset + pending_size: + # the recovered object lies inside the bytes this one claims, so its size + # field is wrong: it can not be read back, and index entries that overlap + # are index corruption. + logger.warning( + f"pack {pack_hex}: object at offset {pending_offset} has a wrong size, dropping it." + ) + pending = None offset = next_offset + scan_from = next_offset + 1 continue obj_size = hdr_size + hdr.meta_size + hdr.data_size - yield hdr.chunk_id, offset, obj_size + if pending is not None: + yield pending + pending = (hdr.chunk_id, offset, obj_size) + scan_from = offset + 1 offset += obj_size + if pending is not None: + yield pending def check_pack_objects(pack_hex, obj_ranges, pack_size): diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 76d2e2cc9b..57dfede8d0 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -10,6 +10,7 @@ from ...constants import * # NOQA from ...helpers import bin_to_hex, msgpack, CommandError, Error, IntegrityError, sig_int from ...manifest import Archives, Manifest +from ...repoobj import RepoObj from ...repository import PackTracker, Repository from ..repository_test import fchunk, corrupt_chunk_on_disk from . import ( @@ -742,7 +743,7 @@ def test_repair_resyncs_pack_with_corrupt_object_header(archivers, request): output = cmd(archiver, "check", "--repair", "--debug", exit_code=0) assert f"invalid object header at offset {damaged_offset}" in output - assert "bytes to the next one" in output # the rebuild resumed at the next object + assert "continuing at the object at offset" in output # the rebuild resumed at the next object def test_repair_finish_flushes_pack_writer(archivers, request): @@ -762,6 +763,7 @@ def test_repair_finish_flushes_pack_writer(archivers, request): checker.repair = True checker.repository = repository checker.key = checker.make_key(repository) + checker.repo_objs = RepoObj(checker.key) checker.manifest = Manifest.load(repository, (Manifest.Operation.CHECK,), key=checker.key) # re-adding a chunk makes the chunks index no longer match the packs, so finish() rebuilds it. checker.chunks_modified = True diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 6bd0bf022f..2820e03be4 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -1,6 +1,7 @@ import io import logging import os +import struct import sys import time from collections import namedtuple @@ -1894,7 +1895,7 @@ def test_pack_reader_resync_skips_to_next_object(): assert list(reader.iter_headers(validate=accept_all)) == [(H(2), len(obj1), len(obj2))] -def test_pack_reader_resync_recovers_from_corrupted_size(): +def test_pack_reader_resync_recovers_from_size_past_the_pack_end(): # a header whose sizes point past the pack, so the next object is found by scanning. obj1 = bytearray(fchunk(b"payload-one", meta=b"meta1", chunk_id=H(1))) obj2 = fchunk(b"payload-two", meta=b"meta2", chunk_id=H(2)) @@ -1910,6 +1911,34 @@ def test_pack_reader_resync_recovers_from_corrupted_size(): ] +def test_pack_reader_resync_recovers_from_size_pointing_into_the_pack(): + # a data_size corrupted to a value that keeps the object inside the pack leaves the header + # valid, so the walk jumps 60 bytes into obj3. Scanning from obj1 finds obj2 and obj3; obj1 is + # dropped, obj2 starting inside the extent it claims. + obj1 = bytearray(fchunk(b"A" * 100, meta=b"m1", chunk_id=H(1))) + obj2 = fchunk(b"B" * 100, meta=b"m2", chunk_id=H(2)) + obj3 = fchunk(b"C" * 100, meta=b"m3", chunk_id=H(3)) + obj1[45:49] = struct.pack(" Date: Fri, 21 Aug 2026 10:02:33 +0530 Subject: [PATCH 4/4] compact: authenticate a gap object before dropping its bytes, #10093 superseded_gap_ranges computed the byte ranges compact drops from object headers as they are: magic, chunk id, meta_size and data_size were all taken on trust. A wrong data_size in a header whose chunk id is indexed elsewhere extended the dropped range past the object into the gap bytes behind it, and those were dropped without ever being looked at. Gaps are where the only copy of a chunk can sit (a backup that crashed before writing its index, or a stale index), so that is not free. Parse the header with PackReader._parse_header, so a header that does not parse or overruns its gap ends the walk over that gap. Read the metadata slot in the same request and require: - validate accepts the header and metadata slot. The slot's tag covers magic, version and chunk id as AAD, and meta_size through the length of the ciphertext slice it selects. - the object's total size equals the index entry's obj_size. data_size is outside what the tag covers and it sets how far the reported range reaches; the entry is a second source for it. Anything else keeps its bytes and the walk continues past it, so a corrupt length field can desync the walk but can no longer drop anything. validate is threaded through delete(), compact_pack() and transform_pack() from the callers that have a key (compact, repo-compress, check --repair). Without it no gap bytes are dropped at all, so "borg debug delete-obj", which opens the repository without a key, stops reclaiming superseded gap bytes. resync_validator moves from archive.py to repoobj.py as object_validator: it is plain RepoObj parsing, and repository.py's callers must not import archive.py for it. --- docs/internals/packs.rst | 25 +++++ src/borg/archive.py | 31 +----- src/borg/archiver/compact_cmd.py | 6 +- src/borg/archiver/repo_compress_cmd.py | 9 +- src/borg/repoobj.py | 23 +++- src/borg/repository.py | 85 ++++++++++----- .../testsuite/archiver/compact_cmd_test.py | 70 +++++++----- .../archiver/repo_compress_cmd_test.py | 7 +- src/borg/testsuite/repository_test.py | 101 +++++++++++++++--- 9 files changed, 260 insertions(+), 97 deletions(-) diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 39e9b7918c..df110c9d2a 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -64,6 +64,9 @@ the AAD; tampering with either still fails the check, because it changes the len slice being read. A forged ``chunk_id``, version, or magic byte therefore fails authentication in ``RepoObj.parse()``/``parse_meta()``. +``parse_meta()`` reads the metadata slot only, so of the two length fields it covers +``meta_size`` alone; ``data_size`` is covered by ``parse()``. + ``encrypted_meta`` and ``encrypted_data`` each add a one-byte slot tag on top of the shared header AAD -- ``b"M"`` for ``encrypted_meta``, ``b"D"`` for ``encrypted_data`` -- binding each ciphertext to its slot. This stops an attacker controlling repo storage from swapping the two ciphertexts (adjusting @@ -215,6 +218,28 @@ determines via mark-and-sweep that none of a pack's blobs are referenced by any archive, it removes the whole file. Individual blobs cannot be removed without rewriting the entire pack, so deletion always operates at pack granularity. +Gap bytes +~~~~~~~~~ + +A pack can hold bytes that no chunks index entry covers -- its *gaps*: a copy of a chunk +that was stored again elsewhere, or blobs from a backup that crashed before writing its +index. Rewriting a pack (``compact_pack``, ``transform_pack``) walks the gaps and drops +the blobs among them that are *superseded*: whose chunk id the index maps to a copy at +another location, which by the id/content invariant holds the same plaintext. + +A gap blob is dropped only when both hold: + +* its header and metadata slot authenticate (``repoobj.object_validator``), verifying its + magic, version, chunk id and ``meta_size``. ``OBJ_MAGIC`` plus a well-formed header is + not evidence that bytes are a blob: in the ``none-*`` and ``authenticated-*`` modes the + payloads are user content stored as it is, so a backed up file can contain one. +* its total size equals the index entry's ``obj_size``. ``data_size`` lies outside what + the authentication covers and sets how far the dropped range reaches, so the entry + serves as a second source for it. + +Anything else keeps its bytes, for ``borg check --repair`` to re-index. Authenticating +needs the key, so a caller without one drops no gap bytes. + .. _pack-index-namespace: diff --git a/src/borg/archive.py b/src/borg/archive.py index 52eed5152b..25d9c062fb 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -52,7 +52,7 @@ from . import platform from .platform import acl_get, acl_set, set_flags, get_flags, set_times, swidth from .repository import Repository, NoManifestError -from .repoobj import RepoObj +from .repoobj import RepoObj, object_validator # macOS: SF_DATALESS marks dataless placeholder files (e.g. cloud files not materialized locally). # Reading such files triggers downloading their content. stat.SF_DATALESS is only available @@ -1893,28 +1893,6 @@ def __next__(self): return next(self._unpacker) -def resync_validator(repo_objs): - """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id. - - obj is an object's header plus its metadata slot. Parsing that slot verifies its tag, which is - computed over the header's magic, version and chunk id as well (AAD, additional authenticated - data: bytes the tag covers without being part of the ciphertext). - - In the "none-*" modes the tag is an unkeyed checksum, so validate accepts any well-formed - object, including one that a backed up file contains. - """ - - def validate(chunk_id, obj): - try: - repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) - except Exception: - # arbitrary bytes fail the tag, the msgpack unpacking or the length checks. - return False - return True - - return validate - - class ArchiveChecker: # Bound how many missing file chunks rebuild_archives buffers for its end-of-run report, # so checking a badly damaged repo with very many missing chunks can not exhaust memory. @@ -1977,7 +1955,7 @@ def check( self.key = self.make_key(repository, manifest_only=True) except IntegrityError as err: logger.warning(f"{err}. Packs with a corrupt object header can not be repaired.") - validate = resync_validator(RepoObj(self.key)) if repair and self.key is not None else None + validate = object_validator(RepoObj(self.key)) if repair and self.key is not None else None self.chunks = build_chunkindex_from_repo( self.repository, slow_rebuild=repair, validate=validate, write_immediately=False ) @@ -2114,6 +2092,7 @@ def verify_data(self): if defect_chunks: if self.repair: logger.warning("Found defect chunks, removing them from the repository.") + validate = object_validator(self.repo_objs) for defect_chunk in defect_chunks: # remote repo (ssh): retry might help for strange network / NIC / RAM errors # as the chunk will be retransmitted from remote server. @@ -2134,7 +2113,7 @@ def verify_data(self): # failed twice -> remove this defect chunk. delete rewrites its pack without it, # keeping the other chunks. update_index=False: finish() rebuilds the index from # the rewritten packs anyway, so a per-chunk full index write would be wasted. - self.repository.delete(defect_chunk, update_index=False) + self.repository.delete(defect_chunk, update_index=False, validate=validate) self.chunks_modified = True # drop it from our own index too, so rebuild_archives reports the file it belongs to. del self.chunks[defect_chunk] @@ -2516,7 +2495,7 @@ def finish(self): build_chunkindex_from_repo( self.repository, slow_rebuild=True, - validate=resync_validator(self.repo_objs), + validate=object_validator(self.repo_objs), write_immediately=True, ) else: diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 00e78680ec..8bd4daf57a 100644 --- a/src/borg/archiver/compact_cmd.py +++ b/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) diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py index 273abc787e..c97e2c81a9 100644 --- a/src/borg/archiver/repo_compress_cmd.py +++ b/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,6 +105,7 @@ 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 @@ -111,7 +113,12 @@ def recompress(self): # 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 diff --git a/src/borg/repoobj.py b/src/borg/repoobj.py index 61e5b541d4..45ee0007c3 100644 --- a/src/borg/repoobj.py +++ b/src/borg/repoobj.py @@ -75,7 +75,8 @@ def get_assert_id_places(): # Size of the header prefix used as AEAD AAD (additional authenticated data: authenticated together # with the ciphertext, but not itself encrypted) for OBJ_VERSION_HEADER_AAD objects: magic(8) + # version(1) + chunk_id(32). meta_size and data_size are excluded, since they are only known after -# encryption; a change to either still fails authentication, by changing the ciphertext slice length. +# encryption; a change to either still fails authentication, by changing the ciphertext slice +# length. parse() covers both, parse_meta() only meta_size - it does not read the data slot. REPOOBJ_HEADER_AAD_SIZE = len(OBJ_MAGIC) + 1 + 32 META_AAD_TAG = b"M" @@ -277,5 +278,25 @@ def parse( return meta_compressed if want_compressed else meta, data_compressed if want_compressed else data +def object_validator(repo_objs): + """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id. + + obj is an object's header and metadata slot, trailing bytes allowed. Parsing the slot verifies + its tag, which covers the header's magic, version, chunk id and meta_size, but not data_size. + + In the "none-*" modes the tag is an unkeyed checksum, so any well-formed object validates. + """ + + def validate(chunk_id, obj): + try: + repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) + except Exception: + # arbitrary bytes fail the tag, the msgpack unpacking or the length checks. + return False + return True + + return validate + + # Backward compatibility: RepoObj1 has moved to borg.legacy.repoobj from .legacy.repoobj import RepoObj1 # noqa: F401 diff --git a/src/borg/repository.py b/src/borg/repository.py index aec087c8a1..ddb0e33e4c 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -41,6 +41,11 @@ # how much of a pack PackReader reads at once when searching for the next object header. RESYNC_WINDOW_SIZE = 1024 * 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 @@ -505,7 +510,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 @@ -515,10 +520,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 and + meta_size. + - its total size equals the index entry's obj_size. data_size is outside what validate covers, + and it sets how far the reported range reaches; the entry is a second source for it. + + 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 @@ -534,17 +553,26 @@ 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 + if obj_size > MAX_DATA_SIZE: + break # larger than any object: not a header + 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: + meta_end = hdr_size + hdr.meta_size + obj = buf[:meta_end] if meta_end <= len(buf) else reader.read(offset, meta_end) + if validate(hdr.chunk_id, obj): drop_ranges.append((offset, obj_size)) + # obj_size is unauthenticated: a wrong one lands the walk at a wrong offset, where the + # two checks above apply again. offset += obj_size return drop_ranges @@ -1479,12 +1507,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) @@ -1494,7 +1525,7 @@ 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. @@ -1502,22 +1533,24 @@ def delete(self, id, *, update_index=True): 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 , keeping and dropping , 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. @@ -1559,7 +1592,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 @@ -1713,7 +1746,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 , passing each indexed object's bytes through . ids: the chunk ids of this pack's objects. Must cover every object the chunk index lists @@ -1727,13 +1760,15 @@ def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change= updates to. Must be the index 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 @@ -1765,7 +1800,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 diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index f7f6efb93a..c40cc8835f 100644 --- a/src/borg/testsuite/archiver/compact_cmd_test.py +++ b/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,14 @@ pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary") # NOQA +def gc_manifest(repository): + """Manifest stand-in: repo_objs is the only attribute ArchiveGarbageCollector uses. + + ChecksumKey is the "none-sha256" mode, which formats and authenticates objects without key material. + """ + 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 +239,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 +264,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 + # formatted objects: dropping the duplicate's bytes needs its metadata slot to authenticate. + 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 +328,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 +358,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 +393,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 +455,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 +476,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 +508,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 +526,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 +553,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 +582,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 diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 739c799202..162197130a 100644 --- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py +++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py @@ -12,7 +12,7 @@ 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): @@ -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 authenticating metadata slot, 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 diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 2820e03be4..e0020e12c9 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -13,15 +13,14 @@ from ..cache import write_chunkindex_invalid from ..constants import MAX_CLOCK_SKEW from ..helpers import IntegrityError, Location, bin_to_hex -from ..hashindex import ChunkIndex +from ..hashindex import ChunkIndex, ChunkIndexEntry from .. import repository as repository_module -from ..archive import resync_validator from ..compress import CNONE from ..constants import ROBJ_FILE_STREAM from ..crypto.key import CHPOKey, ChecksumKey from ..repository import Repository, MAX_DATA_SIZE, propagate_rsh, rest_serve_command, PackWriter, PackReader -from ..repository import PackTracker -from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION +from ..repository import PackTracker, superseded_gap_ranges +from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION, object_validator from .hashindex_test import H @@ -122,6 +121,11 @@ def pdchunk(chunk): return pchunk(chunk)[0] +def accept_all(chunk_id, obj): + # validate stand-in: accepts every candidate. + return True + + def test_basic_operations(repo_fixtures, request): with get_repository_from_fixture(repo_fixtures, request) as repository: for x in range(100): @@ -500,7 +504,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 @@ -524,7 +530,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 @@ -1881,11 +1889,6 @@ def test_pack_reader_raises_on_unsupported_version(): list(PackReader(pack_contents=bytes(obj)).iter_headers()) -def accept_all(chunk_id, obj): - # validate stand-in: accepts every candidate. - return True - - def test_pack_reader_resync_skips_to_next_object(): # after a corrupt header the walk continues at the next object. obj1 = bytearray(fchunk(b"payload-one", meta=b"meta1", chunk_id=H(1))) @@ -1987,7 +1990,7 @@ def test_pack_reader_resync_rejects_metadata_that_does_not_authenticate(tmp_path garbage = fchunk(b"payload", meta=b"not encrypted metadata", chunk_id=H(9)) obj2 = repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM) reader = PackReader(pack_contents=bytes(obj1) + garbage + obj2) - headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + headers = list(reader.iter_headers(validate=object_validator(repo_objs))) assert headers == [(real_id, len(obj1) + len(garbage), len(obj2))] @@ -2002,7 +2005,7 @@ def test_pack_reader_resync_accepts_an_object_with_corrupt_data(tmp_path): obj2 = bytearray(repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM)) obj2[-1] ^= 0xFF # damage the encrypted data, leaving the header and the metadata intact reader = PackReader(pack_contents=bytes(obj1) + bytes(obj2)) - headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + headers = list(reader.iter_headers(validate=object_validator(repo_objs))) assert headers == [(real_id, len(obj1), len(obj2))] with pytest.raises(IntegrityError): repo_objs.parse(real_id, bytes(obj2), ro_type=ROBJ_FILE_STREAM) @@ -2027,7 +2030,7 @@ def test_pack_reader_resync_rejects_damaged_user_content_without_a_key(tmp_path) real_id = repo_objs.id_hash(data) obj2 = repo_objs.format(real_id, {}, data, ro_type=ROBJ_FILE_STREAM) reader = PackReader(pack_contents=bytes(obj1) + obj2) - headers = list(reader.iter_headers(validate=resync_validator(repo_objs))) + headers = list(reader.iter_headers(validate=object_validator(repo_objs))) assert headers == [(real_id, len(obj1), len(obj2))] @@ -2062,3 +2065,73 @@ 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) # the pack whose gaps are walked +OTHER_PACK = H(99) # the pack the index points a superseded duplicate's authoritative copy at + + +def gap_pack(repo_objs, datas): + """Build a pack of repo objects, plus a chunks index that supersedes every one of them. + + datas: each object's plaintext, in pack order. + Returns (objects, chunks): chunks maps each object's id to an entry in OTHER_PACK of the same + size, so superseded_gap_ranges reports an object exactly when it passes both of its checks. + """ + 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): + # the whole pack is one gap: no indexed object of THIS_PACK covers any of it. + 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 chunk id sits in cleartext in the header. Overwriting it with the id of an indexed chunk + # of the same size passes the size check, so only the metadata slot's tag rules the object out. + 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[len(OBJ_MAGIC) + 1 : len(OBJ_MAGIC) + 1 + 32] = victim # the header's chunk_id field + + assert gap_ranges(bytes(forged), chunks, object_validator(repo_objs)) == [] + + +def test_superseded_gap_ranges_rejects_a_size_the_index_contradicts(tmp_path): + # data_size is authenticated by nothing the metadata slot covers. Inflating it would stretch the + # reported range over the bytes behind the object; the index entry's obj_size rules it out. + repo_objs = aead_repo_objs(tmp_path) + (obj, behind), chunks = gap_pack(repo_objs, [b"superseded", b"innocent bystander"]) + hdr_size = RepoObj.obj_header.size + inflated = bytearray(obj + behind) + data_size_at = hdr_size - 4 + (data_size,) = struct.unpack("