diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 0f8d6acec9..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 @@ -91,10 +94,34 @@ A reader locates the next blob by advancing:: next_blob_offset = current_blob_offset + REPOOBJ_HEADER_SIZE + meta_size + data_size -The per-blob magic limits the blast radius of corrupted length fields: if -``meta_size`` or ``data_size`` is damaged, the scanner loses at most one blob. -Once it finds the next ``OBJ_MAGIC`` sequence it resumes. Other corruption -(payload bit flips) is caught by AEAD on that blob without losing position. +``iter_headers()`` checks every header it walks: it must have ``OBJ_MAGIC``, a +supported version, and sizes that keep the blob inside the pack. A header that +fails these checks means a corrupt pack, and ``IntegrityError`` is raised. + +The per-blob magic limits the blast radius of corrupted length fields. The +repair walk (``iter_headers(validate=...)``, used when ``borg check --repair`` +rebuilds the chunks index from the packs) scans for the next blob and resumes +there, so the blobs after the damaged part of the pack are still found. The scan +starts just past the last blob the walk accepted: a corrupted length field that +keeps the blob inside the pack leaves the header valid, and is noticed only at +the misaligned offset it points to. The blob carrying it is dropped when a +recovered blob starts inside the extent it claims - its length field is wrong, +so it can not be read back. + +``OBJ_MAGIC`` occurs inside the payloads as well, and in the ``none-*`` and +``authenticated-*`` modes the payloads are user content stored as it is, so a +backed up file can contain something shaped like a blob. A candidate is +therefore accepted only when its metadata slot verifies against the header AAD +described above; the header and that slot, a few hundred bytes, are what the +scan reads. Verifying needs the key, so a repair that cannot read the manifest +walks without scanning. + +In the ``none-*`` modes the tag is an unkeyed checksum, so the scan accepts any +well-formed blob, including one a backed up file contains. + +``data_size`` is not part of the AAD, so accepting a candidate authenticates +its chunk id, and its size only as far as the blob fits into the pack. Bit flips +in the data are caught when the blob is read, on that blob alone. Blobs follow one another contiguously with no padding:: @@ -191,6 +218,28 @@ determines via mark-and-sweep that none of a pack's blobs are referenced by any archive, it removes the whole file. Individual blobs cannot be removed without rewriting the entire pack, so deletion always operates at pack granularity. +Gap bytes +~~~~~~~~~ + +A pack can hold bytes that no chunks index entry covers -- its *gaps*: a copy of a chunk +that was stored again elsewhere, or blobs from a backup that crashed before writing its +index. Rewriting a pack (``compact_pack``, ``transform_pack``) walks the gaps and drops +the blobs among them that are *superseded*: whose chunk id the index maps to a copy at +another location, which by the id/content invariant holds the same plaintext. + +A gap blob is dropped only when both hold: + +* its header and metadata slot authenticate (``repoobj.object_validator``), verifying its + magic, version, chunk id and ``meta_size``. ``OBJ_MAGIC`` plus a well-formed header is + not evidence that bytes are a blob: in the ``none-*`` and ``authenticated-*`` modes the + payloads are user content stored as it is, so a backed up file can contain one. +* its total size equals the index entry's ``obj_size``. ``data_size`` lies outside what + the authentication covers and sets how far the dropped range reaches, so the entry + serves as a second source for it. + +Anything else keeps its bytes, for ``borg check --repair`` to re-index. Authenticating +needs the key, so a caller without one drops no gap bytes. + .. _pack-index-namespace: diff --git a/src/borg/archive.py b/src/borg/archive.py index 23dcc7ed57..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 @@ -1947,7 +1947,18 @@ def check( # so we do not rebuild it from the packs (reading every pack is far too slow for a routine check). # --repair does rebuild from the packs (slow_rebuild=repair), working from the real packs so it # can detect and fix archives that reference chunks whose pack has gone missing. - self.chunks = build_chunkindex_from_repo(self.repository, slow_rebuild=repair, write_immediately=False) + # --repair also passes validate, which makes the rebuild resync past a corrupt object header. + # Validating needs the key, so read it here. manifest_only=True, because the other source + # make_key reads keys from is self.chunks, which is only built below. + if repair and self.key is None: + try: + self.key = self.make_key(repository, manifest_only=True) + except IntegrityError as err: + logger.warning(f"{err}. Packs with a corrupt object header can not be repaired.") + validate = object_validator(RepoObj(self.key)) if repair and self.key is not None else None + self.chunks = build_chunkindex_from_repo( + self.repository, slow_rebuild=repair, validate=validate, write_immediately=False + ) if self.key is None: self.key = self.make_key(repository) self.repo_objs = RepoObj(self.key) @@ -2081,6 +2092,7 @@ def verify_data(self): if defect_chunks: if self.repair: logger.warning("Found defect chunks, removing them from the repository.") + validate = object_validator(self.repo_objs) for defect_chunk in defect_chunks: # remote repo (ssh): retry might help for strange network / NIC / RAM errors # as the chunk will be retransmitted from remote server. @@ -2101,7 +2113,7 @@ def verify_data(self): # failed twice -> remove this defect chunk. delete rewrites its pack without it, # keeping the other chunks. update_index=False: finish() rebuilds the index from # the rewritten packs anyway, so a per-chunk full index write would be wasted. - self.repository.delete(defect_chunk, update_index=False) + self.repository.delete(defect_chunk, update_index=False, validate=validate) self.chunks_modified = True # drop it from our own index too, so rebuild_archives reports the file it belongs to. del self.chunks[defect_chunk] @@ -2480,7 +2492,12 @@ def finish(self): # the packs changed, so the index no longer matches them: rebuild it from the packs # and persist it. logger.info("Rebuilding and writing the repository chunks index.") - build_chunkindex_from_repo(self.repository, slow_rebuild=True, write_immediately=True) + build_chunkindex_from_repo( + self.repository, + slow_rebuild=True, + validate=object_validator(self.repo_objs), + write_immediately=True, + ) else: # the packs are unchanged, so the index still matches them: persist it as is. logger.info("Writing the rebuilt repository chunks index.") 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/cache.py b/src/borg/cache.py index b018879b38..6602802839 100644 --- a/src/borg/cache.py +++ b/src/borg/cache.py @@ -852,10 +852,18 @@ def repack_chunkindex(repository): def build_chunkindex_from_repo( - repository, *, slow_rebuild=False, fragments_only=False, write_immediately=False, init_flags=ChunkIndex.F_USED + repository, + *, + slow_rebuild=False, + fragments_only=False, + validate=None, + write_immediately=False, + init_flags=ChunkIndex.F_USED, ): # fragments_only: build the index from the index/ fragments only, returning None if they cannot be # read completely, and never write to the repo. + # validate: handed to PackReader.iter_headers when rebuilding from the packs, making it resync + # past a corrupt object header rather than raise IntegrityError. assert not (slow_rebuild and fragments_only) assert not (fragments_only and write_immediately) # fragments_only never writes to the repo # first, try to build a fresh, mostly complete chunk index from centrally stored index fragments: @@ -947,7 +955,7 @@ def build_chunkindex_from_repo( repository._lock_refresh() pi.show(increase=1) pack_id = hex_to_bin(info.name) - for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(): + for chunk_id, obj_offset, obj_size in PackReader(repository.store, pack_id).iter_headers(validate=validate): num_chunks += 1 chunks[chunk_id] = ChunkIndexEntry( flags=init_flags, size=0, pack_id=pack_id, obj_offset=obj_offset, obj_size=obj_size 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 a007f7561f..ddb0e33e4c 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,14 @@ # 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 + +# 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 @@ -362,46 +370,124 @@ 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(chunk_id, obj) confirms the header and metadata slot at that position. + """ + 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, so a bigger one is a false match on + # OBJ_MAGIC inside a payload. + if obj_size <= MAX_DATA_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) + 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) 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 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: 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(scan_from, pack_size, validate) + if next_offset is None: + logger.warning( + 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"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 - 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 + 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): @@ -424,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 @@ -434,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 @@ -453,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 @@ -1398,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) @@ -1413,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. @@ -1421,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. @@ -1478,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 @@ -1632,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 @@ -1646,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 @@ -1684,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/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index 34a9ee5cb3..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 ( @@ -718,6 +719,33 @@ 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 chunks index from a pack whose object header is damaged. + + 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": + 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 "continuing at the object at offset" 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). @@ -735,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/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/cache_test.py b/src/borg/testsuite/cache_test.py index 872499c197..a6fdda8685 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; with validate, the objects after it are 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 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 damaged, so its id is unknown + 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..e0020e12c9 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 @@ -12,10 +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 ..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 @@ -116,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): @@ -494,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 @@ -518,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 @@ -1868,6 +1882,169 @@ 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 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_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)) + 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_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("