diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 5f97d22f34..a9b7e7cb7c 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -244,6 +244,26 @@ single blob cannot be removed from a pack in place: all of these paths write a n pack file without it and then delete the old one, so store-level deletion always operates at pack granularity. +Gap bytes +~~~~~~~~~ + +The *gaps* of a pack are its byte ranges that no chunks index entry covers. They hold chunk +copies that were stored again in another pack, and blobs of a backup that crashed before +writing its index. A gap blob is *superseded* when the index maps its chunk id to another +location. Equal chunk ids mean equal plaintext, so a superseded blob is redundant, whatever +the stored size of the indexed copy (compression and obfuscation padding change it). + +Rewriting a pack (``compact_pack``, ``transform_pack``) drops the superseded gap blobs whose +header and metadata slot validate, checked as in the repair walk above +(``repoobj.object_validator``), and copies all other gap bytes into the new pack. +Validation covers ``meta_size`` and ``data_size``, so a dropped range is exactly one blob. +Without a validator (``validate=None``), no gap bytes are dropped. + +The walk over a gap steps from header to header by the blob size each header states. It ends +at a header that does not parse or that reaches past the gap. The rest of that gap is kept, +and so is a superseded blob that does not validate; both are logged as a warning with the +pack id and the offset. + .. _pack-index-namespace: diff --git a/src/borg/archive.py b/src/borg/archive.py index c30a24b6b7..6c3dde5c19 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -23,7 +23,7 @@ from . import xattr from .chunkers import get_chunker, Chunk, release_chunk_data from .cache import ChunkListEntry, build_chunkindex_from_repo, write_chunkindex_to_repo -from .crypto.key import key_factory, UnsupportedPayloadError +from .crypto.key import key_from_repository from .constants import * # NOQA from .digests import ContentDigester from .crypto.low_level import IntegrityError as IntegrityErrorBase @@ -2303,41 +2303,16 @@ def check( return self.repair or not self.error_found def make_key(self, repository, manifest_only=False): - attempt = 0 + """Return the key loaded by key_from_repository. - # try the manifest first! - try: - cdata = repository.get_manifest() - except NoManifestError: - pass - else: - try: - return key_factory(repository, cdata) - except UnsupportedPayloadError: - # we get here, if the cdata we got has a corrupted key type byte - pass # ignore it, just continue trying - - if not manifest_only: - for chunkid, _ in self.chunks.iteritems(): - attempt += 1 - if attempt > 999: - # we did a lot of attempts, but could not create the key via key_factory, give up. - break - cdata = repository.get(chunkid) - try: - return key_factory(repository, cdata) - except UnsupportedPayloadError: - # we get here, if the cdata we got has a corrupted key type byte - pass # ignore it, just try the next chunk - - if attempt == 0: - if manifest_only: - msg = "make_key: failed to create the key (tried only the manifest)" - else: - msg = "make_key: repository has no chunks at all!" - else: - msg = "make_key: failed to create the key (tried %d chunks)" % attempt - raise IntegrityError(msg) + manifest_only: read only the manifest, else also the objects of the chunk ids in self.chunks. + """ + + def chunk_ids(): # reads self.chunks only if the manifest does not identify the key type + for id, _ in self.chunks.iteritems(): + yield id + + return key_from_repository(repository, () if manifest_only else chunk_ids()) def verify_data(self): logger.info("Starting cryptographic data integrity verification...") @@ -2378,6 +2353,7 @@ def verify_data(self): if defect_chunks: if self.repair: logger.warning("Found defect chunks, removing them from the repository.") + validate = object_validator(self.repo_objs) for defect_chunk in defect_chunks: # remote repo (ssh): retry might help for strange network / NIC / RAM errors # as the chunk will be retransmitted from remote server. @@ -2398,7 +2374,7 @@ def verify_data(self): # failed twice -> remove this defect chunk. delete rewrites its pack without it, # keeping the other chunks. update_index=False: finish() rebuilds the index from # the rewritten packs anyway, so a per-chunk full index write would be wasted. - self.repository.delete(defect_chunk, update_index=False) + self.repository.delete(defect_chunk, update_index=False, validate=validate) self.chunks_modified = True # drop it from our own index too, so rebuild_archives reports the file it belongs to. del self.chunks[defect_chunk] diff --git a/src/borg/archiver/compact_cmd.py b/src/borg/archiver/compact_cmd.py index 551ba3e7a5..cc6e880088 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/debug_cmd.py b/src/borg/archiver/debug_cmd.py index 0acf60a4c9..cb924e7974 100644 --- a/src/borg/archiver/debug_cmd.py +++ b/src/borg/archiver/debug_cmd.py @@ -3,6 +3,7 @@ from ..archive import Archive from ..constants import * # NOQA +from ..crypto.key import key_from_repository, KeyfileInvalidError, RepoKeyNotFoundError, UnsupportedKeyFormatError from ..helpers import msgpack from ..helpers import FilesystemPathSpec from ..helpers import sysinfo @@ -10,16 +11,36 @@ from ..helpers import dash_open from ..helpers import StableDict from ..helpers import archivename_validator, CompressionSpec -from ..helpers import CommandError, RTError +from ..helpers import CommandError, IntegrityError, RTError from ..helpers.argparsing import ArgumentParser from ..manifest import Manifest from ..platform import get_process_id from ..repository import Repository, LIST_SCAN_LIMIT, repo_lister -from ..repoobj import RepoObj +from ..repoobj import RepoObj, object_validator from ._common import with_repository, Highlander from ._common import process_epilog +from ..logger import create_logger + +logger = create_logger() + + +def gap_validator(repository): + """Return repoobj.object_validator for the key of repository, or None if there is no key to use. + + The key is loaded with key_from_repository. There is no key to use if no stored object identifies + the key type (IntegrityError), no key is found (RepoKeyNotFoundError), or the key is invalid + (KeyfileInvalidError, UnsupportedKeyFormatError); a warning is logged then. Other errors, e.g. a + wrong passphrase, propagate. + """ + try: + key = key_from_repository(repository) + except (IntegrityError, RepoKeyNotFoundError, KeyfileInvalidError, UnsupportedKeyFormatError) as err: + logger.warning(f"Could not set up the key, so rewritten packs keep their superseded gap bytes: {err}") + return None + return object_validator(RepoObj(key)) + class DebugMixIn: def do_debug_info(self, args): @@ -111,7 +132,6 @@ def do_debug_dump_manifest(self, args, repository, manifest): @with_repository(manifest=False) def do_debug_dump_repo_objs(self, args, repository): """Dumps (decrypted, decompressed) repository objects.""" - from ..crypto.key import key_factory def decrypt_dump(id, cdata): if cdata is not None: @@ -123,12 +143,7 @@ def decrypt_dump(id, cdata): with open(filename, "wb") as fd: fd.write(data) - # set up the key without depending on a manifest obj - result = repository.list(limit=1, marker=None) - id, _ = result[0] - cdata = repository.get(id) - key = key_factory(repository, cdata) - repo_objs = RepoObj(key) + repo_objs = RepoObj(key_from_repository(repository)) for id, stored_size in repo_lister(repository, limit=LIST_SCAN_LIMIT): cdata = repository.get(id) decrypt_dump(id, cdata) @@ -161,14 +176,7 @@ def print_finding(info, wanted, data, offset): if not wanted: raise CommandError("search term needs to be hex:123abc or str:foobar style") - from ..crypto.key import key_factory - - # set up the key without depending on a manifest obj - result = repository.list(limit=1, marker=None) - id, _ = result[0] - cdata = repository.get(id) - key = key_factory(repository, cdata) - repo_objs = RepoObj(key) + repo_objs = RepoObj(key_from_repository(repository)) last_data = b"" last_id = None @@ -288,14 +296,19 @@ def do_debug_put_obj(self, args, repository): @with_repository(manifest=False, exclusive=True) def do_debug_delete_obj(self, args, repository): """Deletes the objects with the given IDs from the repository.""" + ids = [] for hex_id in args.ids: try: - id = hex_to_bin(hex_id, length=32) + ids.append((hex_id, hex_to_bin(hex_id, length=32))) except ValueError: + ids.append((hex_id, None)) + validate = gap_validator(repository) if any(id is not None for _, id in ids) else None + for hex_id, id in ids: + if id is None: print("object id %s is invalid." % hex_id) else: try: - repository.delete(id) + repository.delete(id, validate=validate) except Repository.ObjectNotFound: print("object %s not found." % hex_id) else: 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/crypto/key.py b/src/borg/crypto/key.py index 23a5f8fa63..5641c12d5a 100644 --- a/src/borg/crypto/key.py +++ b/src/borg/crypto/key.py @@ -3,6 +3,7 @@ import os import textwrap from hashlib import sha256 +from itertools import islice from math import ceil from pathlib import Path from typing import Any, Literal, ClassVar, Optional @@ -24,7 +25,7 @@ from ..helpers import msgpack from ..helpers import workarounds from ..item import Key, EncryptedKey -from ..manifest import Manifest +from ..manifest import Manifest, NoManifestError from ..platform import SaveFile from ..repoobj import RepoObj, RepoObj1 @@ -222,7 +223,17 @@ def identify_key(manifest_data): raise UnsupportedPayloadError(key_type) -def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj): +def identify_stored_key(manifest_chunk, *, ro_cls=RepoObj): + """Return (key class, data slot) of the stored object manifest_chunk. + + A stored object is an object header, a metadata slot and a data slot (see RepoObj). The first + byte of the data slot is the key type byte, which selects the key class. + + manifest_chunk: the stored object, e.g. the manifest. + ro_cls: the RepoObj class that parses manifest_chunk. + Raises IntegrityError if manifest_chunk is damaged (see ro_cls.extract_crypted_data), and + UnsupportedPayloadError if the key type byte selects no key class usable with ro_cls. + """ manifest_data = ro_cls.extract_crypted_data(manifest_chunk) assert manifest_data, "manifest data must not be zero bytes long" key_cls = identify_key(manifest_data) @@ -232,11 +243,49 @@ def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj): # tagged envelope modes (see MACKeyBase). The legacy key classes only exist to read borg # 1.x repositories (ro_cls is RepoObj1 then), e.g. for "borg transfer --from-borg1". raise UnsupportedPayloadError(manifest_data[0]) + return key_cls, manifest_data + + +def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj): + key_cls, manifest_data = identify_stored_key(manifest_chunk, ro_cls=ro_cls) key = key_cls.detect(repository, manifest_data, other=other) key.stored_type = manifest_data[0] return key +def key_from_repository(repository, ids=None): + """Return the key of repository, loaded from the first stored object that identifies the key type. + + Stored objects are read in this order: the manifest, then the objects of the chunk ids in ids, at + most 999 of them. An object identifies the key type if identify_stored_key accepts it. Errors + loading the key, e.g. a wrong passphrase, propagate. + + repository: the Repository whose key is loaded. + ids: iterable of chunk ids. None: the chunk ids in the chunks index of repository. + Raises IntegrityError if no object read identifies the key type. + """ + max_objects = 999 + + def stored_objects(): + try: + yield repository.get_manifest() + except NoManifestError: + pass + chunk_ids = (id for id, _ in repository.list(limit=max_objects)) if ids is None else ids + for id in islice(chunk_ids, max_objects): + yield repository.get(id) + + count = 0 + for cdata in stored_objects(): + count += 1 + try: + identify_stored_key(cdata) + except (IntegrityError, UnsupportedPayloadError): + continue + return key_factory(repository, cdata) + raise IntegrityError(f"no stored object identifies the key type ({count} objects read)") + + def uses_same_chunker_secret(other_key, key): """is the chunker secret the same?""" # avoid breaking the deduplication by a different chunker secret diff --git a/src/borg/repository.py b/src/borg/repository.py index fa1fb6b495..ac81dae7e8 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -545,20 +545,33 @@ def check_pack_objects(pack_hex, obj_ranges, pack_size): ) -def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size): - """Find the superseded duplicates among a pack's gap bytes (bytes no index entry covers). - - A gap holds a chunk copy stored again elsewhere, or objects from a backup that crashed before - writing its index. Walk each gap's object headers: an object whose chunk id the index maps to a - different location is a superseded duplicate (the id is a keyed MAC of the plaintext, so equal - ids mean equal content) and its bytes are redundant. An object whose id is not in the index - (borg check --repair re-indexes it) or whose entry points back at this offset (its only copy) - is not reported. A header that does not parse or overruns its gap ends the walk over that gap. - - obj_ranges: the offset-ordered, validated (obj_offset, obj_size) ranges of the pack's indexed - objects; the gaps are the byte ranges between (and after) them. - Returns the offset-ordered list of (offset, size) ranges holding superseded duplicates. +def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size, *, validate): + """Return the offset-ordered (offset, size) ranges of the superseded duplicates in a pack's gaps. + + A gap is a byte range of the pack that no chunks index entry covers. A superseded duplicate is + an object in a gap whose chunk id the index maps to another location. Equal chunk ids mean + equal plaintext, so its bytes are redundant, whatever the stored size of the indexed copy. + + Each gap is walked from object header to object header, stepping by the object size the header + states. An object is reported when its chunk id is indexed at another location and validate + accepts it. The walk over a gap ends at a header that does not parse or that reaches past the + gap. Objects validate rejects are kept, and so is the rest of a gap where the walk ends early; + both are logged as a warning with the pack id and the offset. + + reader: PackReader of the pack. + chunks: the chunks index (chunk id -> ChunkIndexEntry). + pack_id: id of the pack. + obj_ranges: the offset-ordered (obj_offset, obj_size) ranges of the pack's indexed objects, + non-overlapping and within the pack. The gaps are the byte ranges between and after them. + pack_size: size of the pack in bytes. + validate: validate(chunk_id, obj) -> bool, see repoobj.object_validator. obj is the object's + header and metadata slot (the meta_size metadata bytes after the header). True means chunk + id, meta_size and data_size are verified, so the reported range is exactly the object. + With None, nothing is reported. """ + if validate is None: + return [] + # find the gaps: byte ranges no indexed object covers. gaps = [] # (start, end) of each gap, offset-ordered cursor = 0 @@ -569,22 +582,40 @@ def superseded_gap_ranges(reader, chunks, pack_id, obj_ranges, pack_size): if cursor < pack_size: gaps.append((cursor, pack_size)) + pack_hex = bin_to_hex(pack_id) drop_ranges = [] # (obj_offset, obj_size) of superseded duplicates, offset-ordered hdr_size = RepoObj.obj_header.size for gstart, gend in gaps: offset = gstart while offset < gend: - hdr_data = reader.read(offset, hdr_size) - if len(hdr_data) < hdr_size: + # one read for the header and the metadata slot after it. + buf = reader.read(offset, min(gend - offset, META_READ_SIZE)) + if len(buf) < hdr_size: + hdr, problem = None, f"{len(buf)} bytes, too few for an object header," + else: + hdr, problem = PackReader._parse_header(buf[:hdr_size], offset, pack_size) + if hdr is not None and offset + hdr_size + hdr.meta_size + hdr.data_size > gend: + hdr, problem = None, "object reaching past its gap" + if hdr is None: + logger.warning( + f"pack {pack_hex}: {problem} at offset {offset} in a gap, " + f"keeping the remaining {gend - offset} bytes of the gap." + ) break - hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data)) obj_size = hdr_size + hdr.meta_size + hdr.data_size - if hdr.magic != OBJ_MAGIC or offset + obj_size > gend: - break - if hdr.chunk_id in chunks: - entry = chunks[hdr.chunk_id] - if entry.pack_id != pack_id or entry.obj_offset != offset: + entry = chunks.get(hdr.chunk_id) + if entry is not None and (entry.pack_id != pack_id or entry.obj_offset != offset): + problem = reader._validation_problem(hdr, offset, buf, offset, validate) + if problem is None: drop_ranges.append((offset, obj_size)) + else: + logger.warning(f"pack {pack_hex}: {problem} at offset {offset} in a gap, keeping its bytes.") + # TODO: obj_size is verified only for objects validate accepts. A wrong obj_size + # usually ends the walk at a header that does not parse, so the superseded duplicates in + # the rest of the gap are kept on every rewrite. PackReader._find_header could resync to + # the next object validate accepts, but in the none-* and authenticated-* modes that can + # be a copy of an object inside another object's unencrypted data, whose range can cover + # bytes of the gap objects after it, which would then be dropped. offset += obj_size return drop_ranges @@ -1533,12 +1564,14 @@ 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, *, validate, update_index=True): """Delete a single repo object by rewriting its pack without it (via compact_pack). With update_index=True the full chunk index is written back so the next borg process sees the deletion; callers that rebuild the index themselves (check --repair) pass update_index=False to skip the per-object index rewrite. + + validate: passed to compact_pack. """ self._lock_refresh() entry = self.chunks.get(id) @@ -1548,7 +1581,7 @@ 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. @@ -1556,22 +1589,21 @@ 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, validate, chunks=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. + validate: passed to superseded_gap_ranges, whose ranges are dropped. chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index updates to. Must be the index keep_ids and drop_ids were derived from. Default: self.chunks. Together, keep_ids and drop_ids must cover every object the chunk index lists for this pack; an unlisted indexed object would keep its bytes in the new pack but its index entry would go - stale when the old pack is deleted. Bytes that no index entry covers appear as gaps between the - listed objects: a gap object whose chunk id is in the index is a superseded duplicate (its - authoritative copy is elsewhere) and is dropped; a gap object whose id is not in the index is - copied into the new pack unchanged, to be handled by "borg check --repair". An overlap between - listed objects, or an object claiming to end past the pack file, means index corruption and - raises IntegrityError. + stale when the old pack is deleted. Gaps (byte ranges no index entry covers) are copied into + the new pack, except the superseded duplicates superseded_gap_ranges reports. An overlap + between listed objects, or an object claiming to end past the pack file, means index + corruption and raises IntegrityError. The new pack is the old pack minus the dropped objects, built via store.defrag; kept objects are repointed in the chunk index and dropped objects' chunk index entries are removed. @@ -1613,7 +1645,7 @@ 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 @@ -1767,7 +1799,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, *, validate, chunks=None, before_change=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 @@ -1777,17 +1809,17 @@ def transform_pack(self, pack_id, ids, transform, *, chunks=None, before_change= transform: called as transform(chunk_id, obj_bytes) with an object's stored bytes; returns the replacement bytes, or obj_bytes itself (the identical bytes object) to keep the object unchanged. The chunk id (and thus the plaintext) must not change; sizes may. + validate: passed to superseded_gap_ranges, whose ranges are dropped. chunks: the ChunkIndex to look up the objects' pack locations in and to apply the index updates to. Must be the index was derived from. Default: self.chunks. before_change: called once, just before the first store modification; use it to invalidate stored chunk indexes for crash safety (see #9748). Not called when the pack is kept. - The whole pack file is loaded into memory (bounded by the pack size limit). Gap bytes - (bytes no index entry covers) are handled like in compact_pack: an object superseded by a - copy stored elsewhere is dropped, all other unindexed bytes are copied into the new pack - unchanged, to be handled by "borg check --repair". An overlap between indexed objects, or - an object claiming to end past the pack file, means index corruption and raises - IntegrityError, before anything is written. + The whole pack file is loaded into memory (bounded by the pack size limit). Gaps (byte ranges + no index entry covers) are copied into the new pack, except the superseded duplicates + superseded_gap_ranges reports. An overlap between indexed objects, or an object claiming to + end past the pack file, means index corruption and raises IntegrityError, before anything is + written. If every object is kept and no gap bytes are dropped, the store and the chunk index are not touched at all. Otherwise the new pack (named sha256 of its content) is stored, the indexed @@ -1819,7 +1851,7 @@ 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/__init__.py b/src/borg/testsuite/archiver/__init__.py index e5eea8d491..d753cd7a4e 100644 --- a/src/borg/testsuite/archiver/__init__.py +++ b/src/borg/testsuite/archiver/__init__.py @@ -235,7 +235,7 @@ def write_wrong_content_chunk(archive, repository, chunk_id, *, ro_type=ROBJ_FIL if wrong_data is None: data = read_chunk(archive, repository, chunk_id, ro_type=ro_type) wrong_data = bytes([data[0] ^ 0xFF]) + data[1:] - repository.delete(chunk_id) # put() would not replace an id the repo already has + repository.delete(chunk_id, validate=None) # put() would not replace an id the repo already has repository.put(chunk_id, repo_objs.format(chunk_id, {}, wrong_data, ro_type=ro_type)) repository.flush() diff --git a/src/borg/testsuite/archiver/check_cmd_test.py b/src/borg/testsuite/archiver/check_cmd_test.py index d71624f548..d2539d679f 100644 --- a/src/borg/testsuite/archiver/check_cmd_test.py +++ b/src/borg/testsuite/archiver/check_cmd_test.py @@ -342,7 +342,7 @@ def test_missing_file_chunk(archivers, request): if item.path.endswith(src_file): valid_chunks = item.chunks killed_chunk = valid_chunks[-1] - repository.delete(killed_chunk.id) + repository.delete(killed_chunk.id, validate=None) break else: pytest.fail("should not happen") # convert 'fail' @@ -398,7 +398,7 @@ def test_missing_file_chunk_report_truncated(archiver): continue chunk_id = item.chunks[-1].id if chunk_id not in killed_ids: - repository.delete(chunk_id) + repository.delete(chunk_id, validate=None) killed_ids.append(chunk_id) if len(killed_ids) >= 3: break @@ -430,7 +430,7 @@ def test_missing_file_chunk_refs_truncated(archivers, request): for item in archive.iter_items(): if item.path.endswith("samefile0"): killed_id = item.chunks[0].id - repository.delete(killed_id) + repository.delete(killed_id, validate=None) break assert killed_id is not None @@ -445,7 +445,7 @@ def test_missing_archive_item_chunk(archivers, request): check_cmd_setup(archiver) archive, repository = open_archive(archiver.repository_path, "archive1") with repository: - repository.delete(archive.item_ids[0]) + repository.delete(archive.item_ids[0], validate=None) cmd(archiver, "check", exit_code=1) cmd(archiver, "check", "--repair", exit_code=0) cmd(archiver, "check", exit_code=0) @@ -456,7 +456,7 @@ def test_missing_archive_metadata(archivers, request): check_cmd_setup(archiver) archive, repository = open_archive(archiver.repository_path, "archive1") with repository: - repository.delete(archive.id) + repository.delete(archive.id, validate=None) cmd(archiver, "check", exit_code=1) cmd(archiver, "check", "--repair", exit_code=0) cmd(archiver, "check", exit_code=0) @@ -524,7 +524,7 @@ def test_check_format_missing_archive_metadata(archivers, request): check_cmd_setup(archiver) archive, repository = open_archive(archiver.repository_path, "archive1") with repository: - repository.delete(archive.id) + repository.delete(archive.id, validate=None) archive_id_hex = bin_to_hex(archive.id) output = cmd(archiver, "check", "-v", "--archives-only", "--format", "{archive} {comment}", exit_code=1) # the archive directory entry has no name for it, only the id, which {archive} {comment} would not show. @@ -543,7 +543,7 @@ def test_missing_manifest(archivers, request): if isinstance(repository, Repository): repository.store_delete("config/manifest") else: - repository.delete(Manifest.MANIFEST_ID) + repository.delete(Manifest.MANIFEST_ID, validate=None) cmd(archiver, "check", exit_code=1) output = cmd(archiver, "check", "-v", "--repair", exit_code=0) assert "archive1" in output diff --git a/src/borg/testsuite/archiver/compact_cmd_test.py b/src/borg/testsuite/archiver/compact_cmd_test.py index f7f6efb93a..39482a8718 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,11 @@ pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary") # NOQA +def gc_manifest(repository): + """Return a manifest stand-in for ArchiveGarbageCollector, whose repo_objs use a none-sha256 key.""" + return SimpleNamespace(repo_objs=RepoObj(ChecksumKey(repository))) + + @pytest.mark.parametrize("stats", (True, False)) def test_compact_empty_repository(archivers, request, stats): archiver = request.getfixturevalue(archivers) @@ -228,7 +236,7 @@ def test_compact_packs_respects_threshold(tmp_path): flags = ChunkIndex.F_USED if H(i) in used else ChunkIndex.F_NONE repository.chunks[H(i)] = entry._replace(flags=flags) - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=40) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=40) gc.chunks = repository.chunks gc.compact_packs() @@ -253,42 +261,47 @@ def test_compact_superseded_duplicate(tmp_path): location = os.fspath(tmp_path / "repo") with Repository(location, exclusive=True, create=True) as repository: + manifest = gc_manifest(repository) + repo_objs = manifest.repo_objs + # objects formatted by repo_objs, so the superseded copy of X validates. + w, x, y = b"WWWW", b"XXXX", b"YYYY" + w_id, x_id, y_id = (repo_objs.id_hash(data) for data in (w, x, y)) repository._pack_writer.max_count = 4 # one flush() -> one pack # pack A: three objects W, X, Y (X will be the one later superseded by a copy in pack B) - for cid, data in [(H(0), b"WWWW"), (H(1), b"XXXX"), (H(2), b"YYYY")]: - repository.put(cid, fchunk(data, chunk_id=cid)) + for cid, data in [(w_id, w), (x_id, x), (y_id, y)]: + repository.put(cid, repo_objs.format(cid, {}, data, ro_type=ROBJ_FILE_STREAM)) repository.flush() - pack_a = repository.chunks[H(0)].pack_id + pack_a = repository.chunks[w_id].pack_id pack_a_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(pack_a)) - x_size = repository.chunks[H(1)].obj_size # X's copy in pack A becomes a superseded gap - y_size = repository.chunks[H(2)].obj_size + x_size = repository.chunks[x_id].obj_size # X's copy in pack A becomes a superseded gap + y_size = repository.chunks[y_id].obj_size # pack B: a second copy of X only, in its own pack (as a concurrent writer would have produced). - repository.put(H(1), fchunk(b"XXXX", chunk_id=H(1))) + repository.put(x_id, repo_objs.format(x_id, {}, x, ro_type=ROBJ_FILE_STREAM)) repository.flush() - pack_b = repository.chunks[H(1)].pack_id + pack_b = repository.chunks[x_id].pack_id assert pack_b != pack_a # after the (simulated) fragment merge, the index points X at pack B; pack A's X bytes are now # a superseded, unindexed span. put() already repointed the index to pack B, so nothing to do. # mark usage: W and X used, Y unused. pack A is now mixed (W used, X superseded gap, Y unused). - used = {H(0), H(1)} - for i in range(3): - entry = repository.chunks[H(i)] - flags = ChunkIndex.F_USED if H(i) in used else ChunkIndex.F_NONE - repository.chunks[H(i)] = entry._replace(flags=flags) + used = {w_id, x_id} + for cid in (w_id, x_id, y_id): + entry = repository.chunks[cid] + flags = ChunkIndex.F_USED if cid in used else ChunkIndex.F_NONE + repository.chunks[cid] = entry._replace(flags=flags) - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, manifest, stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() # W still readable; X still readable from pack B; Y (the unused indexed object) dropped. - assert pdchunk(repository.get(H(0))) == b"WWWW" - assert pdchunk(repository.get(H(1))) == b"XXXX" - assert repository.get(H(2), raise_missing=False) is None + assert repo_objs.parse(w_id, repository.get(w_id), ro_type=ROBJ_FILE_STREAM)[1] == w + assert repo_objs.parse(x_id, repository.get(x_id), ro_type=ROBJ_FILE_STREAM)[1] == x + assert repository.get(y_id, raise_missing=False) is None # pack A rewritten, shrunk by Y's bytes (unused indexed) plus X's superseded gap: only W remains. assert bin_to_hex(pack_a) not in [info.name for info in repository.store_list("packs")] - new_pack = repository.chunks[H(0)].pack_id + new_pack = repository.chunks[w_id].pack_id new_size = next(i.size for i in repository.store_list("packs") if i.name == bin_to_hex(new_pack)) assert new_size == pack_a_size - y_size - x_size @@ -312,7 +325,7 @@ def test_compact_keeps_orphan_pack(tmp_path): repository.store_store(orphan_key, b"orphan pack bytes") assert "ab" * 32 in [info.name for info in repository.store_list("packs")] - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() @@ -342,7 +355,7 @@ def test_compact_keeps_unindexed_waste(tmp_path): # ... but H(1)'s big object becomes an unindexed superseded span (well over threshold if counted). del repository.chunks[H(1)] - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() @@ -377,7 +390,7 @@ def test_compact_reclaims_indexed_waste_only(tmp_path): repository.chunks[H(2)] = repository.chunks[H(2)]._replace(flags=ChunkIndex.F_USED) del repository.chunks[H(3)] # its bytes remain in unindexed_pack as unindexed data - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() @@ -439,7 +452,7 @@ def test_compact_keeps_stale_index_entries(tmp_path): repository.chunks[H(0)] = repository.chunks[H(0)]._replace(flags=ChunkIndex.F_USED) repository.store_delete("packs/" + bin_to_hex(gone_pack)) # delete the pack file the index still references - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() @@ -460,7 +473,7 @@ def test_compact_skips_oversized_index_entry(tmp_path): entry = repository.chunks[H(0)] repository.chunks[H(0)] = entry._replace(flags=ChunkIndex.F_USED, obj_size=entry.obj_size + 10000) - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() @@ -492,7 +505,7 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch): total_bytes = sum(repository.store.info("packs/" + name).size for name in packs_before) assert total_bytes >= repository.pack_max_size # combined size crosses the merge threshold - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() assert gc.store_changed is True # the merge changed the store @@ -510,7 +523,7 @@ def test_compact_packs_merges_tiny_packs(tmp_path, monkeypatch): # a merged full-size pack is no longer tiny (the tiny limit is pack_max_size // 2 here), so a # second compact finds nothing to merge and leaves the store unchanged. - gc2 = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc2 = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc2.chunks = repository.chunks gc2.compact_packs() assert gc2.store_changed is False @@ -537,7 +550,7 @@ def test_compact_packs_below_merge_size_gate_leaves_tiny_packs(tmp_path, monkeyp packs_before = {info.name for info in repository.store_list("packs")} assert len(packs_before) == 3 - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() assert gc.store_changed is False # combined tiny bytes stay far below one full pack: leave them alone @@ -566,7 +579,7 @@ def test_compact_packs_below_all_packs_gate_changes_nothing(tmp_path): packs_before = {info.name for info in repository.store_list("packs")} assert len(packs_before) == 2 - gc = ArchiveGarbageCollector(repository, manifest=None, stats=False, threshold=10) + gc = ArchiveGarbageCollector(repository, gc_manifest(repository), stats=False, threshold=10) gc.chunks = repository.chunks gc.compact_packs() assert gc.store_changed is False # below the all-packs gate: nothing was touched diff --git a/src/borg/testsuite/archiver/debug_cmds_test.py b/src/borg/testsuite/archiver/debug_cmds_test.py index b5659aa103..7ed83fe248 100644 --- a/src/borg/testsuite/archiver/debug_cmds_test.py +++ b/src/borg/testsuite/archiver/debug_cmds_test.py @@ -2,10 +2,17 @@ import os import pstats +import pytest + +from ...cache import write_chunkindex_to_repo from ...constants import * # NOQA +from ...helpers import bin_to_hex +from ...helpers.passphrase import PassphraseWrong +from ...manifest import Manifest from .. import changedir from ..compress_test import Compressor -from . import cmd, create_test_files, create_regular_file, generate_archiver_tests, RK_ENCRYPTION +from . import cmd, create_test_files, create_regular_file, generate_archiver_tests, open_repository +from . import KF_ENCRYPTION, KF_LOCATION, RK_ENCRYPTION pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,binary") # NOQA @@ -75,6 +82,8 @@ def test_debug_put_get_delete_obj(archivers, request): output = cmd(archiver, "debug", "delete-obj", id_hash) assert "deleted" in output + # put-obj stored the file's bytes, not a repo object, so the key type is read from the manifest. + assert "Could not set up the key" not in output # the object is gone now: deleting it again reports it is not there output = cmd(archiver, "debug", "delete-obj", id_hash) @@ -84,6 +93,93 @@ def test_debug_put_get_delete_obj(archivers, request): assert "is invalid" in output +def put_pack_with_superseded_gap(archiver): + """Store objects W, X and Y in one pack and X again in a second pack, then write the chunks index. + + The index maps X to the second pack, so X's copy in the first pack is a superseded duplicate. + Returns ((w_id, x_id, y_id), (w_size, x_size, y_size)), the object sizes in the first pack. + """ + with open_repository(archiver) as repository: + repo_objs = Manifest.load(repository, Manifest.NO_OPERATION_CHECK).repo_objs + datas = (b"W" * 100, b"X" * 100, b"Y" * 100) + ids = tuple(repo_objs.id_hash(data) for data in datas) + objs = [repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM) for id, data in zip(ids, datas)] + for id, obj in zip(ids, objs): + repository.put(id, obj) + repository.flush() + pack_id = repository.chunks[ids[0]].pack_id + assert all(repository.chunks[id].pack_id == pack_id for id in ids) + sizes = tuple(repository.chunks[id].obj_size for id in ids) + repository.put(ids[1], objs[1]) # the index now points X at this second copy + repository.flush() + assert repository.chunks[ids[1]].pack_id != pack_id + write_chunkindex_to_repo(repository, repository.chunks, incremental=False, force_write=True, delete_other=True) + return ids, sizes + + +def pack_size_of(archiver, id): + with open_repository(archiver) as repository: + pack_name = bin_to_hex(repository.chunks[id].pack_id) + return next(info.size for info in repository.store_list("packs") if info.name == pack_name) + + +def test_debug_delete_obj_drops_superseded_gap(archivers, request): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + (w_id, x_id, y_id), (_, _, y_size) = put_pack_with_superseded_gap(archiver) + + output = cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id)) + + assert "deleted" in output + assert "Could not set up the key" not in output + assert pack_size_of(archiver, y_id) == y_size # W and the superseded copy of X are gone + with open_repository(archiver) as repository: + assert repository.get(w_id, raise_missing=False) is None + assert repository.get(x_id) is not None # served from its second copy + + +def test_debug_delete_obj_without_a_key_keeps_superseded_gap(archivers, request): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", KF_ENCRYPTION, KF_LOCATION) + (w_id, x_id, y_id), (_, x_size, y_size) = put_pack_with_superseded_gap(archiver) + for name in os.listdir(archiver.keys_path): + os.unlink(os.path.join(archiver.keys_path, name)) + + output = cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id)) + + assert "Could not set up the key" in output + assert "deleted" in output + assert pack_size_of(archiver, y_id) == x_size + y_size # only W is gone, the gap is kept + + +def test_debug_delete_obj_with_a_wrong_passphrase_deletes_nothing(archivers, request, monkeypatch): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + (w_id, x_id, y_id), (w_size, x_size, y_size) = put_pack_with_superseded_gap(archiver) + monkeypatch.setenv("BORG_PASSPHRASE", "wrong") + + if archiver.FORK_DEFAULT: + cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id), exit_code=PassphraseWrong.exit_mcode) + else: + with pytest.raises(PassphraseWrong): + cmd(archiver, "debug", "delete-obj", bin_to_hex(w_id)) + + assert pack_size_of(archiver, y_id) == w_size + x_size + y_size + with open_repository(archiver) as repository: + assert repository.get(w_id, raise_missing=False) is not None + + +def test_debug_delete_obj_with_invalid_ids_only_sets_up_no_key(archivers, request, monkeypatch): + archiver = request.getfixturevalue(archivers) + cmd(archiver, "repo-create", RK_ENCRYPTION) + monkeypatch.setenv("BORG_PASSPHRASE", "wrong") + + output = cmd(archiver, "debug", "delete-obj", "invalid") + + assert "is invalid" in output + assert "Could not set up the key" not in output + + def test_debug_id_hash_format_put_get_parse_obj(archivers, request): """Test format-obj and parse-obj commands.""" archiver = request.getfixturevalue(archivers) diff --git a/src/borg/testsuite/archiver/extract_cmd_test.py b/src/borg/testsuite/archiver/extract_cmd_test.py index f40edea26e..cbcf3b897d 100644 --- a/src/borg/testsuite/archiver/extract_cmd_test.py +++ b/src/borg/testsuite/archiver/extract_cmd_test.py @@ -1036,7 +1036,7 @@ def test_extract_file_with_missing_chunk(archivers, request): for item in archive.iter_items(): if item.path.endswith(src_file): chunk = item.chunks[-1] - repository.delete(chunk.id) + repository.delete(chunk.id, validate=None) break else: assert False # missed the file diff --git a/src/borg/testsuite/archiver/mount_cmds_test.py b/src/borg/testsuite/archiver/mount_cmds_test.py index ed444c89a7..5e39dd6beb 100644 --- a/src/borg/testsuite/archiver/mount_cmds_test.py +++ b/src/borg/testsuite/archiver/mount_cmds_test.py @@ -313,7 +313,7 @@ def test_fuse_allow_damaged_files(archivers, request): with repository: for item in archive.iter_items(): if item.path.endswith(src_file): - repository.delete(item.chunks[-1].id) + repository.delete(item.chunks[-1].id, validate=None) path = item.path # store full path for later break else: diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py index 739c799202..bac2fab466 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): @@ -264,7 +264,7 @@ def test_transform_pack_keeps_unindexed_gap(tmp_path): replacements = {H(0): fchunk(b"W" * 100, chunk_id=H(0)), H(2): fchunk(b"y", chunk_id=H(2))} calls = [] new_pack_id, new_size = repository.transform_pack( - pack_id, [H(0), H(2)], transform_via(replacements), before_change=lambda: calls.append(1) + pack_id, [H(0), H(2)], transform_via(replacements), before_change=lambda: calls.append(1), validate=None ) assert new_pack_id != pack_id assert calls == [1] # before_change called (once), the store was modified @@ -303,7 +303,10 @@ def test_transform_pack_drops_superseded_gap(tmp_path): assert pack_b != pack_a w_new = fchunk(b"W" * 100, chunk_id=H(0)) - new_pack_id, new_size = repository.transform_pack(pack_a, [H(0)], transform_via({H(0): w_new})) + # fchunk objects have no encrypted metadata slot (see fchunk), so accept_all stands in for validate. + new_pack_id, new_size = repository.transform_pack( + pack_a, [H(0)], transform_via({H(0): w_new}), validate=accept_all + ) assert new_pack_id != pack_a assert new_size == len(w_new) # only W remains, X's superseded bytes were dropped assert pdchunk(repository.get(H(0))) == b"W" * 100 @@ -324,7 +327,7 @@ def test_transform_pack_unchanged_pack_untouched(tmp_path): calls = [] new_pack_id, new_size = repository.transform_pack( - pack_id, [H(0), H(1)], transform_via({}), before_change=lambda: calls.append(1) + pack_id, [H(0), H(1)], transform_via({}), before_change=lambda: calls.append(1), validate=None ) assert new_pack_id == pack_id assert calls == [] # nothing changed, so before_change was never called diff --git a/src/borg/testsuite/archiver/webdav_cmd_test.py b/src/borg/testsuite/archiver/webdav_cmd_test.py index 7b6f957265..e2e53b1ae0 100644 --- a/src/borg/testsuite/archiver/webdav_cmd_test.py +++ b/src/borg/testsuite/archiver/webdav_cmd_test.py @@ -627,7 +627,7 @@ def test_webdav_damaged_file(archivers, request): archive = Archive(manifest, manifest.archives.get("test").id) for item in archive.iter_items(): if item.path.endswith("big"): - repository.delete(item.chunks[-1].id) # get rid of a chunk of "big" + repository.delete(item.chunks[-1].id, validate=None) # get rid of a chunk of "big" break else: assert False # missed the file diff --git a/src/borg/testsuite/crypto/key_test.py b/src/borg/testsuite/crypto/key_test.py index e104fb54cb..831b9a3b6b 100644 --- a/src/borg/testsuite/crypto/key_test.py +++ b/src/borg/testsuite/crypto/key_test.py @@ -16,13 +16,15 @@ from ...crypto.key import ID_HMAC_SHA_256, ID_BLAKE2b_256, ID_BLAKE3_256 from ...crypto.key import UnsupportedManifestError, UnsupportedKeyFormatError, UnsupportedPayloadError from ...crypto.key import RepoKeyNotFoundError -from ...crypto.key import identify_key +from ...crypto.key import identify_key, key_from_repository from ...crypto.low_level import IntegrityError as IntegrityErrorBase from ...helpers import Error from ...helpers import IntegrityError from ...helpers import Location from ...helpers import msgpack -from ...constants import KEY_ALGORITHMS, KeyBlobStorage, KeyType, ROBJ_MANIFEST +from ...manifest import NoManifestError +from ...repoobj import RepoObj +from ...constants import KEY_ALGORITHMS, KeyBlobStorage, KeyType, ROBJ_FILE_STREAM, ROBJ_MANIFEST from ...helpers import hex_to_bin, bin_to_hex @@ -665,3 +667,72 @@ def test_argon2_wrong_passphrase_returns_none(monkeypatch): saved = repository.store_key.call_args.args[0] _, saved_b64 = keyfile_parse(saved) assert key.decrypt_key_file(a2b_base64(saved_b64), "wrong passphrase") is None + + +class StoredObjectsRepository: + """A repository with a stored manifest (None: no manifest) and objects by chunk id.""" + + def __init__(self, manifest, objects): + self.manifest = manifest + self.objects = objects + + def get_manifest(self): + if self.manifest is None: + raise NoManifestError + return self.manifest + + def get(self, id): + return self.objects[id] + + def list(self, limit=None): + return [(id, len(obj)) for id, obj in self.objects.items()][:limit] + + +def stored_object(key_cls, id): + return RepoObj(key_cls(MagicMock(id=bytes(32)))).format(id, {}, b"data", ro_type=ROBJ_FILE_STREAM) + + +def with_key_type(cdata, key_type): + """Return cdata with its key type byte, the first byte of the data slot, set to key_type.""" + offset = len(cdata) - len(RepoObj.extract_crypted_data(cdata)) + return cdata[:offset] + bytes([key_type]) + cdata[offset + 1 :] + + +def test_key_from_repository_reads_the_key_type_from_the_manifest(): + objects = {b"o" * 32: stored_object(ChecksumKey, b"o" * 32)} + repository = StoredObjectsRepository(stored_object(Blake3ChecksumKey, bytes(32)), objects) + assert isinstance(key_from_repository(repository), Blake3ChecksumKey) + + +def test_key_from_repository_skips_objects_that_do_not_identify_a_key_type(): + good = stored_object(ChecksumKey, b"g" * 32) + objects = { + b"d" * 32: b"damaged", + b"u" * 32: with_key_type(good, KeyType.DROPPED_BLAKE3AUTHENTICATED), + b"g" * 32: good, + } + repository = StoredObjectsRepository(b"damaged manifest", objects) + assert isinstance(key_from_repository(repository), ChecksumKey) + + +def test_key_from_repository_raises_if_no_object_identifies_the_key_type(): + objects = {b"g" * 32: stored_object(ChecksumKey, b"g" * 32)} + with pytest.raises(IntegrityError): + key_from_repository(StoredObjectsRepository(None, objects), ids=()) # the manifest only + with pytest.raises(IntegrityError): + key_from_repository(StoredObjectsRepository(None, {b"d" * 32: b"damaged"})) + + +def test_key_from_repository_loads_the_key_once(monkeypatch): + ids = (b"a" * 32, b"b" * 32) + repository = StoredObjectsRepository(None, {id: stored_object(ChecksumKey, id) for id in ids}) + detected = [] + + def detect(repository, manifest_data, *, other=False): + detected.append(manifest_data) + raise IntegrityError("the key can not be loaded") + + monkeypatch.setattr(ChecksumKey, "detect", detect) + with pytest.raises(IntegrityError, match="the key can not be loaded"): + key_from_repository(repository) + assert len(detected) == 1 diff --git a/src/borg/testsuite/repoobj_test.py b/src/borg/testsuite/repoobj_test.py index 89232595b4..8c761da61d 100644 --- a/src/borg/testsuite/repoobj_test.py +++ b/src/borg/testsuite/repoobj_test.py @@ -19,8 +19,9 @@ from ..legacy.repoobj import RepoObj1 from ..compress import LZ4 -# offsets of the size fields in the object header. -META_SIZE_OFFSET = len(OBJ_MAGIC) + 1 + 32 # the magic, the version byte and the chunk id precede it +# offsets of object header fields. +CHUNK_ID_OFFSET = len(OBJ_MAGIC) + 1 # after the magic and the version byte +META_SIZE_OFFSET = CHUNK_ID_OFFSET + 32 DATA_SIZE_OFFSET = META_SIZE_OFFSET + 4 diff --git a/src/borg/testsuite/repository_test.py b/src/borg/testsuite/repository_test.py index 7fa3c4ab9f..d9815621f3 100644 --- a/src/borg/testsuite/repository_test.py +++ b/src/borg/testsuite/repository_test.py @@ -16,12 +16,12 @@ from ..constants import MAX_CLOCK_SKEW, ROBJ_FILE_STREAM from ..crypto.key import CHPOKey, ChecksumKey from ..helpers import IntegrityError, Location, bin_to_hex -from ..hashindex import ChunkIndex +from ..hashindex import ChunkIndex, ChunkIndexEntry from ..repository import Repository, MAX_DATA_SIZE, MAX_VALIDATED_META_SIZE, propagate_rsh, rest_serve_command -from ..repository import PackWriter, PackReader, PackTracker +from ..repository import PackWriter, PackReader, PackTracker, superseded_gap_ranges from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION, object_validator from .hashindex_test import H -from .repoobj_test import DATA_SIZE_OFFSET, META_SIZE_OFFSET +from .repoobj_test import CHUNK_ID_OFFSET, DATA_SIZE_OFFSET, META_SIZE_OFFSET def test_rest_serve_command_local(): @@ -187,7 +187,7 @@ def test_consistency(repo_fixtures, request): repository.flush() assert pdchunk(repository.get(H(0))) == b"bar" # delete removes the object the index points at; the stale earlier copies are not resurrected. - repository.delete(H(0)) + repository.delete(H(0), validate=None) with pytest.raises(Repository.ObjectNotFound): repository.get(H(0)) @@ -202,7 +202,7 @@ def test_delete_with_stale_earlier_object_in_pack(repo_fixtures, request): repository.put(H(1), fchunk(b"bbb")) # fills the pack, flushing both objects repository.put(H(0), fchunk(b"ccc")) # re-put: H(0)'s index entry moves to a new pack repository.flush() - repository.delete(H(1)) + repository.delete(H(1), validate=None) with pytest.raises(Repository.ObjectNotFound): repository.get(H(1)) assert pdchunk(repository.get(H(0))) == b"ccc" # H(0) still served from its new pack @@ -391,7 +391,9 @@ def test_compact_pack_copy_forward(repo_fixtures, request): assert repository.chunks[H(1)].pack_id == old_pack_id assert repository.chunks[H(2)].pack_id == old_pack_id - new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids={H(1)}) + new_pack_id, dropped = repository.compact_pack( + old_pack_id, keep_ids={H(0), H(2)}, drop_ids={H(1)}, validate=None + ) assert new_pack_id is not None and new_pack_id != old_pack_id assert dropped == len(chunk1) # reported freed bytes for --stats @@ -412,7 +414,9 @@ def test_compact_pack_drops_whole_pack(repo_fixtures, request): with repository: old_pack_id = repository.chunks[H(0)].pack_id - new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids=set(), drop_ids={H(0), H(1)}) + new_pack_id, dropped = repository.compact_pack( + old_pack_id, keep_ids=set(), drop_ids={H(0), H(1)}, validate=None + ) assert new_pack_id is None # every byte dropped: no replacement pack assert dropped == len(chunk0) + len(chunk1) @@ -432,7 +436,7 @@ def test_compact_pack_keep_all_is_noop(repo_fixtures, request): old_pack_id = repository.chunks[H(0)].pack_id new_pack_id, dropped = repository.compact_pack( - old_pack_id, keep_ids={H(1), H(0)}, drop_ids=set() + old_pack_id, keep_ids={H(1), H(0)}, drop_ids=set(), validate=None ) # out of order assert new_pack_id == old_pack_id @@ -455,7 +459,7 @@ def test_compact_pack_keeps_gap(repo_fixtures, request): old_pack_id = repository.chunks[H(0)].pack_id del repository.chunks[H(1)] # H(1)'s bytes stay in the pack but are now unindexed (a gap) - new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(2)}, drop_ids={H(0)}) + new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(2)}, drop_ids={H(0)}, validate=None) assert new_pack_id is not None and new_pack_id != old_pack_id assert pdchunk(repository.get(H(2))) == b"DATA2" @@ -477,7 +481,7 @@ def test_compact_pack_keeps_trailing_bytes(repo_fixtures, request): old_pack_id = repository.chunks[H(0)].pack_id del repository.chunks[H(2)] # trailing unindexed bytes - new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(1)}, drop_ids={H(0)}) + new_pack_id, _ = repository.compact_pack(old_pack_id, keep_ids={H(1)}, drop_ids={H(0)}, validate=None) assert new_pack_id is not None and new_pack_id != old_pack_id assert pdchunk(repository.get(H(1))) == b"DATA1" @@ -499,7 +503,9 @@ def test_compact_pack_drops_superseded_gap(repo_fixtures, request): old_pack_id = repository.chunks[H(0)].pack_id repository.chunks[H(1)] = repository.chunks[H(1)]._replace(pack_id=H(9)) # authoritative copy elsewhere - new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set()) + new_pack_id, dropped = repository.compact_pack( + old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set(), validate=accept_all + ) assert new_pack_id is not None and new_pack_id != old_pack_id assert dropped == len(chunk1) # the superseded gap's bytes are counted as freed @@ -523,7 +529,9 @@ def test_compact_pack_keeps_self_referencing_gap(repo_fixtures, request): with repository: old_pack_id = repository.chunks[H(0)].pack_id - new_pack_id, dropped = repository.compact_pack(old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set()) + new_pack_id, dropped = repository.compact_pack( + old_pack_id, keep_ids={H(0), H(2)}, drop_ids=set(), validate=accept_all + ) assert new_pack_id == old_pack_id # nothing dropped, defrag reproduced the same pack assert dropped == 0 # the self-referencing gap is kept, nothing freed @@ -544,7 +552,7 @@ def test_compact_pack_detects_overlap(repo_fixtures, request): repository.chunks[H(1)] = entry._replace(obj_offset=0) # now overlaps H(0) at offset 0 with pytest.raises(IntegrityError): - repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set()) + repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set(), validate=None) assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")] @@ -561,7 +569,7 @@ def test_compact_pack_detects_past_eof(repo_fixtures, request): repository.chunks[H(1)] = entry._replace(obj_size=entry.obj_size + 1000) # now claims to end past EOF with pytest.raises(IntegrityError): - repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set()) + repository.compact_pack(old_pack_id, keep_ids={H(0), H(1)}, drop_ids=set(), validate=None) assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")] @@ -583,7 +591,7 @@ def short_read(*args, **kwargs): monkeypatch.setattr(repository.store, "defrag", short_read) with pytest.raises(IntegrityError): - repository.compact_pack(old_pack_id, keep_ids={H(0)}, drop_ids={H(1)}) + repository.compact_pack(old_pack_id, keep_ids={H(0)}, drop_ids={H(1)}, validate=None) assert bin_to_hex(old_pack_id) in [info.name for info in repository.store_list("packs")] assert H(1) in repository.chunks # still indexed: aborted before deleting the dropped id @@ -723,7 +731,7 @@ def test_max_data_size(repo_fixtures, request): assert pdchunk(repository.get(H(0))) == max_data with pytest.raises(IntegrityError): repository.put(H(1), fchunk(max_data + b"x")) - repository.delete(H(0)) + repository.delete(H(0), validate=None) def check(repository, repo_path, repair=False, status=True): @@ -2254,3 +2262,121 @@ def test_pack_reader_in_memory_read_returns_view(): assert bytes(view) == obj2 pack[len(obj1)] ^= 0xFF # a write to pack_contents is visible through the view assert view[0] == obj2[0] ^ 0xFF + + +THIS_PACK = H(98) # id of the pack whose gaps are walked +OTHER_PACK = H(99) # id of the pack holding the indexed copies + + +def gap_pack(repo_objs, datas): + """Return (objects, chunks): datas formatted as repo objects, and an index mapping each to OTHER_PACK. + + superseded_gap_ranges reports such an object if validate accepts it. + """ + objs = [repo_objs.format(repo_objs.id_hash(data), {}, data, ro_type=ROBJ_FILE_STREAM) for data in datas] + chunks = {repo_objs.id_hash(data): ChunkIndexEntry(0, 0, OTHER_PACK, 0, len(obj)) for data, obj in zip(datas, objs)} + return objs, chunks + + +def gap_ranges(pack, chunks, validate): + # no indexed objects: the whole pack is one gap. + reader = PackReader(pack_contents=pack) + return superseded_gap_ranges(reader, chunks, THIS_PACK, [], len(pack), validate=validate) + + +def test_superseded_gap_ranges_reports_an_authenticated_duplicate(tmp_path): + repo_objs = aead_repo_objs(tmp_path) + (obj,), chunks = gap_pack(repo_objs, [b"superseded"]) + + assert gap_ranges(obj, chunks, object_validator(repo_objs)) == [(0, len(obj))] + assert gap_ranges(obj, chunks, None) == [] # no validator, nothing to report + + +def test_superseded_gap_ranges_rejects_a_forged_chunk_id(tmp_path): + # the header's chunk id is replaced by the id of another indexed chunk, which the metadata + # slot's tag does not authenticate. + repo_objs = aead_repo_objs(tmp_path) + (obj,), chunks = gap_pack(repo_objs, [b"superseded"]) + victim = repo_objs.id_hash(b"a chunk stored elsewhere") + chunks[victim] = ChunkIndexEntry(0, 0, OTHER_PACK, 0, len(obj)) + forged = bytearray(obj) + forged[CHUNK_ID_OFFSET : CHUNK_ID_OFFSET + 32] = victim + + assert gap_ranges(bytes(forged), chunks, object_validator(repo_objs)) == [] + + +def test_superseded_gap_ranges_rejects_an_inflated_data_size(tmp_path): + # data_size is increased to also cover the next object. validate rejects it: data_size must + # match csize, the data size recorded in the authenticated metadata. + repo_objs = aead_repo_objs(tmp_path) + (obj, behind), chunks = gap_pack(repo_objs, [b"superseded", b"innocent bystander"]) + inflated = bytearray(obj + behind) + (data_size,) = struct.unpack("