repository: salvage_pack review fixes, refs #10026

src/borg/repoobj.py

@@ -292,18 +292,23 @@ def object_validator(repo_objs):
     return validate
 
 
-def object_authenticator(repo_objs):
+def whole_object_authenticator(repo_objs):
     """Return authenticate(chunk_id, obj): True if obj is the whole repo object with id chunk_id.
 
-    obj is an object's header, metadata slot and data slot. Parsing it checks that the header's sizes
-    add up to len(obj) and verifies the tags of both slots, each computed over the slot and over the
-    header prefix (magic, version, chunk id) and chunk_id as AAD (additional authenticated data: bytes
-    the tag covers without being part of the ciphertext). The data is neither decompressed nor hashed,
-    so a plaintext that does not hash to chunk_id is accepted.
+    obj is an object's header, metadata slot and data slot (object_validator gets the header and the
+    metadata slot only). Parsing it checks that the header's sizes add up to len(obj) and verifies
+    the tags of both slots, each computed over the slot and over the header prefix (magic, version,
+    chunk id) and chunk_id as AAD (additional authenticated data: bytes the tag covers without being
+    part of the ciphertext). Only the tags are checked: an object whose plaintext does not hash to
+    chunk_id is accepted.
 
-    With the authenticated_no_key workaround, the "authenticated-*" modes do not verify the tags, so
-    this accepts any object whose metadata slot unpacks.
+    Raises Error for an "authenticated-*" key with the authenticated_no_key workaround, which skips
+    the tag verification.
     """
+    from .crypto.key import MACKeyBase  # crypto.key imports this module
+
+    if AUTHENTICATED_NO_KEY and isinstance(repo_objs.key, MACKeyBase):
+        raise Error("Objects can not be authenticated with BORG_WORKAROUNDS=authenticated_no_key.")
 
     def authenticate(chunk_id, obj):
         try:

src/borg/repository.py

@@ -31,7 +31,6 @@ from .helpers.lrucache import LRUCache
 from .storelocking import Lock
 from .logger import create_logger
 from .repoobj import RepoObj, OBJ_MAGIC
-from .crypto import key as crypto_key
 from .crypto.key import is_keyfile, key_factory, store_hash, STORE_HASH_NAME
 
 logger = create_logger(__name__)
@@ -473,7 +472,7 @@ class PackReader:
         end = start + size
         obj = buf[start:end] if end <= len(buf) else self.read(offset, size)
         if not validate(hdr.chunk_id, obj):
-            return "object does not authenticate"
+            return "object header or metadata does not authenticate"
         return None
 
     def _find_header(self, offset, pack_size, validate):
@@ -690,8 +689,8 @@ def remove_missing_pack_entries(chunks, missing_pack_ids):
 SALVAGE_INTACT = "intact"  # the pack's store hash matches its name
 SALVAGE_DONE = "salvaged"  # the pack was replaced by one holding only its authenticated objects
 SALVAGE_NOTHING_AUTHENTICATES = "nothing authenticates"  # no object in the pack authenticates
-SALVAGE_UNSTABLE = "unstable"  # two reads of the pack gave different results
-SALVAGE_READ_ERROR = "read error"  # reading the pack raised OSError
+SALVAGE_READS_DIFFER = "reads differ"  # two reads of the pack disagree
+SALVAGE_READ_ERROR = "read error"  # reading the pack failed
 
 # status: one of the SALVAGE_* outcomes.
 # new_pack_id: id of the replacement pack (SALVAGE_DONE), else None.
@@ -977,9 +976,8 @@ class Repository:
             if cache_size:
                 ns_config["packs/"]["size"] = int(cache_size)
             cache_url = cache_dir.as_uri()
-        # True if BORG_STORE_CACHE set up a local cache of packs: store.load() of a pack may then return the
-        # cached copy instead of reading the backend.
-        self.pack_store_cache = cache_url is not None
+        # True if packs are cached locally (BORG_STORE_CACHE): store.load() of a pack may return the cached copy.
+        self.uses_pack_store_cache = cache_url is not None
 
         propagate_rsh()  # borgstore shall use the same remote shell command as borg
 
@@ -2382,7 +2380,7 @@ class Repository:
         self.store_delete(pack_key)
         return new_pack_id, len(pack_data)
 
-    def salvage_pack(self, pack_id, *, validate, authenticate, chunks, before_old_pack_delete=None):
+    def salvage_pack(self, pack_id, *, validate, authenticate, chunks=None, before_old_pack_delete=None):
         """Replace pack <pack_id> by a pack holding only the objects in it that authenticate.
 
         A pack is named by the store hash (STORE_HASH_NAME) of its content.
@@ -2391,8 +2389,9 @@ class Repository:
             is the repo object with id chunk_id, see repoobj.object_validator. The pack is walked with
             PackReader.iter_headers(validate).
         authenticate: authenticate(chunk_id, obj) -> bool, True if obj (a whole object: header, metadata
-            slot and data slot) is the repo object with id chunk_id, see repoobj.object_authenticator.
-        chunks: the ChunkIndex to update, or None to leave the chunk index unchanged.
+            slot and data slot) is the repo object with id chunk_id, see repoobj.whole_object_authenticator.
+            It must verify the tags of both slots.
+        chunks: the ChunkIndex to update. Default: self.chunks.
         before_old_pack_delete: callable without arguments, called once just before the old pack is deleted,
             see SALVAGE_DONE.
 
@@ -2402,10 +2401,11 @@ class Repository:
         objects' bytes in their old order.
 
         Returns a SalvageResult. The store and the chunk index change only for SALVAGE_DONE:
-        - SALVAGE_UNSTABLE: the loaded bytes hash to the pack's name, or a second load of the pack
-          differs from the first. An object is dropped only if two identical loads both fail it.
-        - SALVAGE_READ_ERROR: reading the pack raised OSError. All reads happen before the first
-          store change.
+        - SALVAGE_READS_DIFFER: the loaded bytes hash to the pack's name although the store hash did
+          not, or a second load of the pack differs from the first, e.g. due to corruption in memory
+          or in transfer. Objects are dropped only if both loads return the same bytes.
+        - SALVAGE_READ_ERROR: reading the pack raised OSError or a store backend error other than
+          StoreObjectNotFound. All reads happen before the first store change.
         - SALVAGE_DONE: the replacement pack is stored, before_old_pack_delete is called, chunks is
           updated, then the old pack is deleted. If the replacement pack has the old pack's name
           (the kept bytes are the undamaged pack), storing it overwrites the old pack, and
@@ -2416,44 +2416,43 @@ class Repository:
         has no entry gets one with flags F_USED and size 0 (the plaintext size is unknown). Entries
         of other packs and F_PENDING entries stay as they are.
 
-        Raises Error before any store access if BORG_WORKAROUNDS=authenticated_no_key is set (then
-        decrypting skips the tag verification, so authenticate accepts damaged objects) or if
-        pack_store_cache is set (then two loads can return the same cached copy). Raises
-        PermissionDenied unless the repo permissions allow compaction (see assert_writable), and
-        StoreObjectNotFound if the pack is missing. Requires the exclusive lock.
+        Raises Error before any store access if uses_pack_store_cache is set (then two loads can
+        return the same cached copy). Raises PermissionDenied unless the repo permissions allow
+        compaction (see assert_writable), and StoreObjectNotFound if the pack is missing. Requires the
+        exclusive lock.
         """
-        if crypto_key.AUTHENTICATED_NO_KEY:
-            raise Error(
-                "Pack salvage refused: with BORG_WORKAROUNDS=authenticated_no_key objects are not authenticated."
-            )
-        if self.pack_store_cache:
+        if self.uses_pack_store_cache:
             raise Error("Pack salvage refused: with BORG_STORE_CACHE, pack reads may return the cached copy.")
         self._lock_refresh()
+        if chunks is None:
+            chunks = self.chunks
         self.assert_writable()
         pack_hex = bin_to_hex(pack_id)
         pack_key = "packs/" + pack_hex
 
-        def untouched(status):
+        def unchanged(status):
             return SalvageResult(status, None, [], 0, [])
 
         try:
             if self.store.hash(pack_key, algorithm=STORE_HASH_NAME) == pack_hex:
-                return untouched(SALVAGE_INTACT)
+                return unchanged(SALVAGE_INTACT)
             pack_contents = self.store.load(pack_key)
+            if store_hash(pack_contents).digest() == pack_id:
+                return unchanged(SALVAGE_READS_DIFFER)
             reader = PackReader(pack_id=pack_id, pack_contents=pack_contents)
             kept_old = []  # (chunk_id, old offset, size) of each kept object, offset-ordered
             for chunk_id, offset, size in reader.iter_headers(validate=validate):
                 if authenticate(chunk_id, reader.read(offset, size)):
                     kept_old.append((chunk_id, offset, size))
             if not kept_old:
-                return untouched(SALVAGE_NOTHING_AUTHENTICATES)
-            if store_hash(pack_contents).digest() == pack_id:
-                return untouched(SALVAGE_UNSTABLE)
+                return unchanged(SALVAGE_NOTHING_AUTHENTICATES)
             if self.store.load(pack_key) != pack_contents:
-                return untouched(SALVAGE_UNSTABLE)
-        except OSError as exc:
+                return unchanged(SALVAGE_READS_DIFFER)
+        except StoreObjectNotFound:
+            raise
+        except (OSError, StoreBackendError) as exc:
             logger.warning(f"pack {pack_hex}: {exc}, not salvaging it.")
-            return untouched(SALVAGE_READ_ERROR)
+            return unchanged(SALVAGE_READ_ERROR)
 
         new_pack_data = b"".join(pack_contents[offset : offset + size] for _, offset, size in kept_old)
         # equal to pack_id if the kept bytes are the undamaged pack, e.g. when the damage is appended bytes.
@@ -2469,34 +2468,33 @@ class Repository:
         if before_old_pack_delete is not None and new_pack_id != pack_id:
             before_old_pack_delete()
 
+        new_by_old_offset = {old[1]: new for old, new in zip(kept_old, kept)}
+        new_by_id = {}  # chunk_id -> its first kept copy
+        for new in kept:
+            new_by_id.setdefault(new[0], new)
+        # collect first: the index must not be mutated while iterating it.
+        listed = [
+            (chunk_id, entry.obj_offset)
+            for chunk_id, entry in chunks.iteritems()
+            if entry.pack_id == pack_id and not (entry.flags & ChunkIndex.F_PENDING)
+        ]
+        new_locations = []
         removed_ids = []
-        if chunks is not None:
-            new_by_old_offset = {old[1]: new for old, new in zip(kept_old, kept)}
-            new_by_id = {}  # chunk_id -> its first kept copy
-            for new in kept:
-                new_by_id.setdefault(new[0], new)
-            # collect first: the index must not be mutated while iterating it.
-            listed = [
-                (chunk_id, entry.obj_offset)
-                for chunk_id, entry in chunks.iteritems()
-                if entry.pack_id == pack_id and not (entry.flags & ChunkIndex.F_PENDING)
-            ]
-            new_locations = []
-            for chunk_id, old_offset in listed:
-                new = new_by_old_offset.get(old_offset)
-                if new is None or new[0] != chunk_id:
-                    new = new_by_id.get(chunk_id)
-                if new is None:
-                    del chunks[chunk_id]
-                    removed_ids.append(chunk_id)
-                else:
-                    new_locations.append((chunk_id, new_pack_id, new[1], new[2]))
-            chunks.update_pack_info(new_locations)
-            for chunk_id, (_, offset, size) in new_by_id.items():
-                if chunk_id not in chunks:
-                    chunks[chunk_id] = ChunkIndexEntry(
-                        flags=ChunkIndex.F_USED, size=0, pack_id=new_pack_id, obj_offset=offset, obj_size=size
-                    )
+        for chunk_id, old_offset in listed:
+            new = new_by_old_offset.get(old_offset)
+            if new is None or new[0] != chunk_id:
+                new = new_by_id.get(chunk_id)
+            if new is None:
+                del chunks[chunk_id]
+                removed_ids.append(chunk_id)
+            else:
+                new_locations.append((chunk_id, new_pack_id, new[1], new[2]))
+        chunks.update_pack_info(new_locations)
+        for chunk_id, (_, offset, size) in new_by_id.items():
+            if chunk_id not in chunks:
+                chunks[chunk_id] = ChunkIndexEntry(
+                    flags=ChunkIndex.F_USED, size=0, pack_id=new_pack_id, obj_offset=offset, obj_size=size
+                )
 
         if new_pack_id != pack_id:  # else storing the replacement pack overwrote the old one
             self.store_delete(pack_key)

src/borg/testsuite/archiver/check_cmd_test.py

@@ -1101,7 +1101,8 @@ def test_repair_resyncs_pack_with_corrupt_object_header(archivers, request, dama
         repository.store_store(key, pack)
 
     output = cmd(archiver, "check", "--repair", "--debug", exit_code=0)
-    problem = {"magic": "no object header", "data_size": "object does not authenticate"}[damaged_field]
+    problems = {"magic": "no object header", "data_size": "object header or metadata does not authenticate"}
+    problem = problems[damaged_field]
     assert f"{problem} at offset {damaged_offset}" in output
     assert f"continuing at the object at offset {next_offset}" in output  # the rebuild resumed at the next object
     # the resync dropped an object, so the summary reports a problem.

src/borg/testsuite/repoobj_test.py

@@ -12,8 +12,8 @@ from ..repoobj import (
     OBJ_MAGIC,
     OBJ_VERSION,
     RepoObj,
-    object_authenticator,
     object_validator,
+    whole_object_authenticator,
 )
 from ..legacy.repoobj import RepoObj1
 from ..compress import LZ4
@@ -524,7 +524,7 @@ def test_object_validator_accepts_every_compression(compression):
 
 
 @pytest.mark.parametrize("key_class", [AuthenticatedKey, CHPOKey, AESOCBKey])
-def test_object_authenticator_checks_both_slots(key_class):
+def test_whole_object_authenticator_checks_both_slots(key_class):
     # a flipped byte in the metadata slot or in the data slot, a wrong chunk id or a truncated object
     # fails authentication, for each key family.
     key = key_class(None)
@@ -534,7 +534,7 @@ def test_object_authenticator_checks_both_slots(key_class):
     data = b"payload" * 100
     chunk_id = repo_objs.id_hash(data)
     obj = repo_objs.format(chunk_id, {}, data, ro_type=ROBJ_FILE_STREAM)
-    authenticate = object_authenticator(repo_objs)
+    authenticate = whole_object_authenticator(repo_objs)
     assert authenticate(chunk_id, obj)
     assert authenticate(chunk_id, memoryview(obj))
     for pos in (RepoObj.obj_header.size, len(obj) - 1):  # first byte of the metadata slot, last of the data slot
@@ -545,15 +545,29 @@ def test_object_authenticator_checks_both_slots(key_class):
     assert not authenticate(chunk_id, obj[:-1])
 
 
-def test_object_authenticator_does_not_check_the_id(aead_key):
-    # the data is not decompressed and not hashed: an object whose plaintext does not match its id,
-    # but whose slots authenticate, is accepted.
+@pytest.mark.parametrize("key_class", [AuthenticatedKey, CHPOKey, AESOCBKey])
+def test_whole_object_authenticator_refuses_authenticated_no_key(key_class, monkeypatch):
+    # the workaround skips the tag verification of the "authenticated-*" keys only.
+    key = key_class(None)
+    key.init_from_random_data()
+    key.init_ciphers()
+    monkeypatch.setattr("borg.repoobj.AUTHENTICATED_NO_KEY", True)
+    if key_class is AuthenticatedKey:
+        with pytest.raises(Error, match="authenticated_no_key"):
+            whole_object_authenticator(RepoObj(key))
+    else:
+        assert callable(whole_object_authenticator(RepoObj(key)))
+
+
+def test_whole_object_authenticator_does_not_check_the_id(aead_key):
+    # only the tags are checked: an object whose slots authenticate but whose plaintext does not match
+    # its id is accepted.
     repo_objs = RepoObj(aead_key)
     chunk_id = repo_objs.id_hash(b"foobar" * 10)
-    assert object_authenticator(repo_objs)(chunk_id, wrong_content_object(repo_objs, chunk_id))
+    assert whole_object_authenticator(repo_objs)(chunk_id, wrong_content_object(repo_objs, chunk_id))
 
 
-def test_object_authenticator_propagates_an_unexpected_exception(monkeypatch):
+def test_whole_object_authenticator_propagates_an_unexpected_exception(monkeypatch):
     # only a failed authentication or unpacking gives False, any other exception propagates.
     repo_objs = RepoObj(make_test_key())
     data = b"payload" * 100
@@ -565,4 +579,4 @@ def test_object_authenticator_propagates_an_unexpected_exception(monkeypatch):
 
     monkeypatch.setattr(repo_objs, "parse", parse_raising_valueerror)
     with pytest.raises(ValueError):
-        object_authenticator(repo_objs)(chunk_id, obj)
+        whole_object_authenticator(repo_objs)(chunk_id, obj)

src/borg/testsuite/repository_test.py

@@ -23,8 +23,8 @@ from ..platform import get_process_id
 from ..repository import Repository, MAX_DATA_SIZE, MAX_VALIDATED_META_SIZE, propagate_rsh, rest_serve_command
 from ..repository import PackWriter, PackReader, PackTracker, superseded_gap_ranges
 from ..repository import SALVAGE_DONE, SALVAGE_INTACT, SALVAGE_NOTHING_AUTHENTICATES
-from ..repository import SALVAGE_READ_ERROR, SALVAGE_UNSTABLE
-from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION, object_authenticator, object_validator
+from ..repository import SALVAGE_READ_ERROR, SALVAGE_READS_DIFFER, StoreBackendError, StoreObjectNotFound
+from ..repoobj import RepoObj, OBJ_MAGIC, OBJ_VERSION, object_validator, whole_object_authenticator
 from . import make_test_key, set_test_key_on_open
 from .hashindex_test import H
 from .repoobj_test import CHUNK_ID_OFFSET, DATA_SIZE_OFFSET, META_SIZE_OFFSET
@@ -3272,7 +3272,10 @@ def test_superseded_gap_ranges_warns_where_it_keeps_bytes(tmp_path, caplog):
         assert gap_ranges(bytes(damaged) + garbage, chunks, validate) == []
         assert gap_ranges(b"\0" * 3, chunks, validate) == []
 
-    assert f"pack {pack_hex}: object does not authenticate at offset 0 in a gap, keeping its bytes." in caplog.text
+    assert (
+        f"pack {pack_hex}: object header or metadata does not authenticate at offset 0 in a gap, keeping its bytes."
+        in caplog.text
+    )
     assert (
         f"pack {pack_hex}: no object header at offset {len(rejected)} in a gap, "
         f"keeping the remaining {len(garbage)} bytes of the gap." in caplog.text
@@ -3442,7 +3445,7 @@ def test_create_failure_leaves_no_store_behind(tmp_path, monkeypatch):
     assert os.path.exists(os.path.join(location, "config", "config"))
 
 
-def store_salvage_pack(repository, objs, *, listed, flip=(), tail=b""):
+def store_damaged_pack(repository, objs, *, listed, flip=(), tail=b""):
     """Store objs as one pack named by the store hash of their bytes, then damage it.
 
     The byte at each position in flip is flipped and tail is appended, so a flip or a tail makes the
@@ -3468,16 +3471,14 @@ def store_salvage_pack(repository, objs, *, listed, flip=(), tail=b""):
     return pack_id, offsets
 
 
-def offsets_end(objs, i):
+def last_byte_offset(objs, i):
     # position of the last byte of objs[i] in a pack of objs: the last byte of its data slot.
     return sum(len(obj) for _, obj in objs[: i + 1]) - 1
 
 
 def salvage(repository, repo_objs, pack_id, **kwargs):
-    kwargs.setdefault("chunks", repository.chunks)
-    return repository.salvage_pack(
-        pack_id, validate=object_validator(repo_objs), authenticate=object_authenticator(repo_objs), **kwargs
-    )
+    kwargs.setdefault("authenticate", whole_object_authenticator(repo_objs))
+    return repository.salvage_pack(pack_id, validate=object_validator(repo_objs), **kwargs)
 
 
 def store_contents(repository):
@@ -3503,7 +3504,7 @@ def three_objects(repo_objs):
 def test_salvage_pack_leaves_an_intact_pack_alone(salvage_repository):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3))
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3))
     packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
     called = []
 
@@ -3522,7 +3523,7 @@ def test_salvage_pack_drops_an_object_failing_authentication(salvage_repository)
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
     (id0, obj0), (id1, obj1), (id2, obj2) = objs
-    pack_id, offsets = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+    pack_id, offsets = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
     list(salvage_repository.get_many([id0]))  # loads the pack into the pack cache
 
     result = salvage(salvage_repository, repo_objs, pack_id)
@@ -3544,7 +3545,7 @@ def test_salvage_pack_keeps_an_object_the_index_does_not_list(salvage_repository
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
     (id0, obj0), (id1, obj1), (id2, obj2) = objs
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=[0, 2], flip=[offsets_end(objs, 2)])
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=[0, 2], flip=[last_byte_offset(objs, 2)])
 
     result = salvage(salvage_repository, repo_objs, pack_id)
 
@@ -3560,8 +3561,8 @@ def test_salvage_pack_keeps_an_object_the_index_does_not_list(salvage_repository
 def test_salvage_pack_leaves_a_pack_alone_if_nothing_authenticates(salvage_repository):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    flip = [offsets_end(objs, i) for i in range(3)]
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=flip)
+    flip = [last_byte_offset(objs, i) for i in range(3)]
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=flip)
     packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
 
     result = salvage(salvage_repository, repo_objs, pack_id)
@@ -3576,7 +3577,7 @@ def test_salvage_pack_leaves_a_pack_alone_if_nothing_authenticates(salvage_repos
 def test_salvage_pack_drops_uncovered_trailing_bytes(salvage_repository, tail):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), tail=tail)
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), tail=tail)
     index_before = index_contents(salvage_repository.chunks)
     called = []
 
@@ -3595,33 +3596,14 @@ def test_salvage_pack_drops_uncovered_trailing_bytes(salvage_repository, tail):
         assert bytes(salvage_repository.get(chunk_id)) == obj
 
 
-def test_salvage_pack_refuses_under_authenticated_no_key(salvage_repository, monkeypatch):
-    # with the workaround, decrypting skips the tag verification, so a damaged object would authenticate.
-    repo_objs = plain_repo_objs()
-    objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
-    damaged = bytearray(objs[1][1])
-    damaged[-1] ^= 0xFF
-    packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
-    monkeypatch.setattr("borg.crypto.key.AUTHENTICATED_NO_KEY", True)
-    assert object_authenticator(repo_objs)(objs[1][0], bytes(damaged))
-
-    with pytest.raises(Error, match="authenticated_no_key"):
-        salvage(salvage_repository, repo_objs, pack_id)
-
-    assert store_contents(salvage_repository) == packs_before
-    assert index_contents(salvage_repository.chunks) == index_before
-
-
 def test_salvage_pack_refuses_with_a_pack_store_cache(tmp_path, monkeypatch):
-    # with BORG_STORE_CACHE, store.load() of a pack can return the cached copy, so a second read is not
-    # a second read of the pack.
+    # with BORG_STORE_CACHE, both loads of a pack can return the same cached copy.
     monkeypatch.setenv("BORG_STORE_CACHE", os.fspath(tmp_path / "cache"))
     with Repository(os.fspath(tmp_path / "repo"), exclusive=True, create=True) as repository:
         repository.chunks = ChunkIndex()
         repo_objs = plain_repo_objs()
         objs = three_objects(repo_objs)
-        pack_id, _ = store_salvage_pack(repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+        pack_id, _ = store_damaged_pack(repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
         with pytest.raises(Error, match="BORG_STORE_CACHE"):
             salvage(repository, repo_objs, pack_id)
         assert bin_to_hex(pack_id) in store_contents(repository)
@@ -3630,17 +3612,20 @@ def test_salvage_pack_refuses_with_a_pack_store_cache(tmp_path, monkeypatch):
 def test_salvage_pack_refuses_without_write_permission(salvage_repository):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
+    packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
     salvage_repository.permissions = {"packs": "lrw", "index": "lrwWD"}  # packs/ has no delete
     with pytest.raises(Repository.PermissionDenied):
         salvage(salvage_repository, repo_objs, pack_id)
+    assert store_contents(salvage_repository) == packs_before
+    assert index_contents(salvage_repository.chunks) == index_before
 
 
 def test_salvage_pack_leaves_a_pack_alone_if_two_reads_differ(salvage_repository, monkeypatch):
     # the first load has an extra flipped byte in object 0, the second load does not.
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
     packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
     load = salvage_repository.store.load
     loads = []
@@ -3650,7 +3635,7 @@ def test_salvage_pack_leaves_a_pack_alone_if_two_reads_differ(salvage_repository
         loads.append(name)
         if len(loads) == 1:
             data = bytearray(data)
-            data[offsets_end(objs, 0)] ^= 0xFF
+            data[last_byte_offset(objs, 0)] ^= 0xFF
             data = bytes(data)
         return data
 
@@ -3658,7 +3643,7 @@ def test_salvage_pack_leaves_a_pack_alone_if_two_reads_differ(salvage_repository
 
     result = salvage(salvage_repository, repo_objs, pack_id)
 
-    assert result.status == SALVAGE_UNSTABLE
+    assert result.status == SALVAGE_READS_DIFFER
     assert len(loads) == 2
     monkeypatch.undo()
     assert store_contents(salvage_repository) == packs_before
@@ -3666,28 +3651,39 @@ def test_salvage_pack_leaves_a_pack_alone_if_two_reads_differ(salvage_repository
 
 
 def test_salvage_pack_leaves_a_pack_that_reads_intact_alone(salvage_repository, monkeypatch):
-    # the store hash does not match the name, but the loaded bytes do: the reads disagree.
+    # the store hash does not match the name, but the loaded bytes do: the reads disagree, and no object
+    # is authenticated.
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3))
-    packs_before = store_contents(salvage_repository)
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3))
+    packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
     monkeypatch.setattr(salvage_repository.store, "hash", lambda name, algorithm: "0" * 64)
+    authenticated = []
 
-    result = salvage(salvage_repository, repo_objs, pack_id)
+    def authenticate(chunk_id, obj):
+        authenticated.append(chunk_id)
+        return False
 
-    assert result.status == SALVAGE_UNSTABLE
+    result = salvage(salvage_repository, repo_objs, pack_id, authenticate=authenticate)
+
+    assert result.status == SALVAGE_READS_DIFFER
+    assert authenticated == []
     assert store_contents(salvage_repository) == packs_before
+    assert index_contents(salvage_repository.chunks) == index_before
 
 
 @pytest.mark.parametrize("method", ["hash", "load"])
-def test_salvage_pack_leaves_a_pack_alone_on_a_read_error(salvage_repository, monkeypatch, method):
+@pytest.mark.parametrize(
+    "error", [OSError(5, "Input/output error"), StoreBackendError("server error")], ids=["OSError", "BackendError"]
+)
+def test_salvage_pack_leaves_a_pack_alone_on_a_read_error(salvage_repository, monkeypatch, method, error):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
     packs_before, index_before = store_contents(salvage_repository), index_contents(salvage_repository.chunks)
 
     def failing(*args, **kwargs):
-        raise OSError(5, "Input/output error")
+        raise error
 
     monkeypatch.setattr(salvage_repository.store, method, failing)
 
@@ -3699,18 +3695,25 @@ def test_salvage_pack_leaves_a_pack_alone_on_a_read_error(salvage_repository, mo
     assert index_contents(salvage_repository.chunks) == index_before
 
 
-def test_salvage_pack_without_chunks_leaves_the_index_alone(salvage_repository):
+def test_salvage_pack_raises_if_the_pack_is_missing(salvage_repository):
+    repo_objs = plain_repo_objs()
+    with pytest.raises(StoreObjectNotFound):
+        salvage(salvage_repository, repo_objs, bytes(32))
+
+
+def test_salvage_pack_updates_the_given_chunk_index(salvage_repository):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
-    index_before = index_contents(salvage_repository.chunks)
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
+    chunks = salvage_repository.chunks
+    salvage_repository.chunks = ChunkIndex()
 
-    result = salvage(salvage_repository, repo_objs, pack_id, chunks=None)
+    result = salvage(salvage_repository, repo_objs, pack_id, chunks=chunks)
 
     assert result.status == SALVAGE_DONE
-    assert result.removed_ids == []
-    assert index_contents(salvage_repository.chunks) == index_before
-    assert set(store_contents(salvage_repository)) == {bin_to_hex(result.new_pack_id)}
+    assert result.removed_ids == [objs[1][0]]
+    assert chunks[objs[0][0]].pack_id == result.new_pack_id
+    assert index_contents(salvage_repository.chunks) == {}
 
 
 def test_salvage_pack_indexes_duplicates(salvage_repository):
@@ -3720,11 +3723,11 @@ def test_salvage_pack_indexes_duplicates(salvage_repository):
     id_a, obj_a = real_chunk(repo_objs, b"A" * 100)
     id_b, obj_b = real_chunk(repo_objs, b"B" * 100)
     id_c, obj_c = real_chunk(repo_objs, b"C" * 100)
-    other_pack_id, _ = store_salvage_pack(salvage_repository, [(id_a, obj_a)], listed=[0])
+    other_pack_id, _ = store_damaged_pack(salvage_repository, [(id_a, obj_a)], listed=[0])
     objs = [(id_a, obj_a), (id_b, obj_b), (id_b, obj_b), (id_c, obj_c)]
     # id_a is unlisted here, id_b is listed at its damaged first copy, id_c is listed and its only copy is damaged.
-    pack_id, _ = store_salvage_pack(
-        salvage_repository, objs, listed=[1, 3], flip=[offsets_end(objs, 1), offsets_end(objs, 3)]
+    pack_id, _ = store_damaged_pack(
+        salvage_repository, objs, listed=[1, 3], flip=[last_byte_offset(objs, 1), last_byte_offset(objs, 3)]
     )
 
     result = salvage(salvage_repository, repo_objs, pack_id)
@@ -3743,7 +3746,7 @@ def test_salvage_pack_order_of_changes(salvage_repository, monkeypatch):
     repo_objs = plain_repo_objs()
     objs = three_objects(repo_objs)
     id0 = objs[0][0]
-    pack_id, _ = store_salvage_pack(salvage_repository, objs, listed=range(3), flip=[offsets_end(objs, 1)])
+    pack_id, _ = store_damaged_pack(salvage_repository, objs, listed=range(3), flip=[last_byte_offset(objs, 1)])
     events = []
     store_store, store_delete = salvage_repository.store_store, salvage_repository.store_delete