From 658f345ee7535d577c330f91fd08eeefb03346ca Mon Sep 17 00:00:00 2001
From: Nish_ <120EE0980@nitrkl.ac.in>
Date: Tue, 13 Oct 2026 01:38:09 +0200
Subject: [PATCH 1/6] legacy hashindex: raise ValueError for invalid key/value
 sizes, refs #8649

The borg 1.x index file is data from disk, so do not validate it with
assert. LegacyRepository.prepare_txn catches ValueError and rebuilds the
index, an AssertionError skipped that recovery.

From PR #8682.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/legacy/hashindex.py                |  9 +++--
 src/borg/testsuite/legacy_hashindex_test.py | 37 +++++++++++++++++++++
 2 files changed, 43 insertions(+), 3 deletions(-)
 create mode 100644 src/borg/testsuite/legacy_hashindex_test.py

diff --git a/src/borg/legacy/hashindex.py b/src/borg/legacy/hashindex.py
index 2dad9db392..e0bf9e09df 100644
--- a/src/borg/legacy/hashindex.py
+++ b/src/borg/legacy/hashindex.py
@@ -88,8 +88,10 @@ def _read_fd(self, fd):
         magic, entries, buckets, ksize, vsize = struct.unpack(self.HEADER_FMT, header_bytes)
         if magic != self.MAGIC:
             raise ValueError(f"Invalid file: magic {self.MAGIC.decode()} not found.")
-        assert ksize == self.KEY_SIZE, "invalid key size"
-        assert vsize == self.VALUE_SIZE, "invalid value size"
+        if ksize != self.KEY_SIZE:
+            raise ValueError("Invalid file: invalid key size.")
+        if vsize != self.VALUE_SIZE:
+            raise ValueError("Invalid file: invalid value size.")
         buckets_size = buckets * (ksize + vsize)
         current_pos = fd.tell()
         end_of_file = fd.seek(0, os.SEEK_END)
@@ -105,4 +107,5 @@ def _read_fd(self, fd):
                 continue
             self.ht._set_raw(key, value)
         pos = fd.tell()
-        assert pos == end_of_file
+        if pos != end_of_file:
+            raise ValueError("Invalid file: file size does not match (end of file).")
diff --git a/src/borg/testsuite/legacy_hashindex_test.py b/src/borg/testsuite/legacy_hashindex_test.py
new file mode 100644
index 0000000000..9812160a61
--- /dev/null
+++ b/src/borg/testsuite/legacy_hashindex_test.py
@@ -0,0 +1,37 @@
+"""Tests for borg.legacy.hashindex (NSIndex1)."""
+
+import io
+import struct
+
+import pytest
+
+from ..legacy.hashindex import NSIndex1
+
+
+def written_index():
+    idx = NSIndex1()
+    idx[b"x" * 32] = (1, 2)
+    fd = io.BytesIO()
+    idx.write(fd)
+    return bytearray(fd.getvalue())
+
+
+def test_nsindex1_roundtrip():
+    idx = NSIndex1.read(io.BytesIO(bytes(written_index())))
+    assert idx[b"x" * 32] == (1, 2)
+
+
+@pytest.mark.parametrize(
+    "corrupt",
+    [
+        lambda data: data.__setitem__(slice(16, 17), b"\x10"),  # ksize
+        lambda data: data.__setitem__(slice(17, 18), b"\x04"),  # vsize
+    ],
+    ids=["ksize", "vsize"],
+)
+def test_nsindex1_invalid_header(corrupt):
+    data = written_index()
+    assert struct.calcsize(NSIndex1.HEADER_FMT) == 18
+    corrupt(data)
+    with pytest.raises(ValueError, match="Invalid file"):
+        NSIndex1.read(io.BytesIO(bytes(data)))

From 434cafabebedb1092d607d30683b2a40f67e2297 Mon Sep 17 00:00:00 2001
From: Thomas Waldmann <tw@waldmann-edv.de>
Date: Tue, 13 Oct 2026 01:57:00 +0200
Subject: [PATCH 2/6] item/repoobj/compress: raise IntegrityError for invalid
 repo data, refs #8649

The item.pyx fix_*/want_* helpers and update_internal() methods validate
msgpack data from the repository (items, archive metadata, legacy manifests),
key envelopes and decrypted keys: raise IntegrityError instead of asserting.
borg check's robust item iterator now reports such an item and continues.

RepoObj.parse: an inconsistent csize or an invalid psize in the object
metadata raises IntegrityError.

compress: a decompressed size that does not match the size in the metadata
and the borg 1.x-only formats (legacy zlib, obfuscation header) in a borg 2
object raise DecompressionError.

repo-compress and debug format-obj check the object type they pass to
RepoObj.format (stored object metadata / user supplied JSON).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/archiver/debug_cmd.py                |  4 +
 src/borg/archiver/repo_compress_cmd.py        |  2 +
 src/borg/compress.pyx                         | 11 ++-
 src/borg/item.pyx                             | 74 ++++++++++++-------
 src/borg/legacy/archives.py                   |  1 +
 src/borg/repoobj.py                           | 11 +--
 .../testsuite/archiver/debug_cmds_test.py     | 17 ++++-
 .../archiver/repo_compress_cmd_test.py        | 26 +++++++
 src/borg/testsuite/compress_test.py           | 22 +++++-
 src/borg/testsuite/item_test.py               | 43 ++++++++++-
 src/borg/testsuite/repoobj_test.py            | 43 ++++++++++-
 11 files changed, 216 insertions(+), 38 deletions(-)

diff --git a/src/borg/archiver/debug_cmd.py b/src/borg/archiver/debug_cmd.py
index 18c3dd779b..330a4bf6d6 100644
--- a/src/borg/archiver/debug_cmd.py
+++ b/src/borg/archiver/debug_cmd.py
@@ -241,9 +241,13 @@ def do_debug_format_obj(self, args, repository, manifest):
 
         with open(args.json_path) as f:
             meta = json.load(f)
+        if not isinstance(meta, dict):
+            raise CommandError(f"metadata in {args.json_path} is not a JSON object.")
 
         repo_objs = manifest.repo_objs
         ro_type = meta.pop("type", ROBJ_FILE_STREAM)
+        if not isinstance(ro_type, str) or ro_type == ROBJ_DONTCARE:
+            raise CommandError(f"metadata type {ro_type!r} is invalid.")
         data_encrypted = repo_objs.format(id=id, meta=meta, data=data, ro_type=ro_type)
 
         with open(args.object_path, "wb") as f:
diff --git a/src/borg/archiver/repo_compress_cmd.py b/src/borg/archiver/repo_compress_cmd.py
index 02c0859c1c..32f86c2aed 100644
--- a/src/borg/archiver/repo_compress_cmd.py
+++ b/src/borg/archiver/repo_compress_cmd.py
@@ -184,6 +184,8 @@ def transform(self, id, obj_bytes):
             return obj_bytes
         meta, data = self.repo_objs.parse(id, obj_bytes, ro_type=ROBJ_DONTCARE)
         ro_type = meta.pop("type", None)
+        if not isinstance(ro_type, str) or ro_type == ROBJ_DONTCARE:
+            raise IntegrityError(f"object {bin_to_hex(id)} has an invalid type: {ro_type!r}")
         if self.wanted[2] == -1:
             # if the object was obfuscated, but should not be in future, remove related metadata
             meta.pop("olevel", None)
diff --git a/src/borg/compress.pyx b/src/borg/compress.pyx
index dc2ea31760..0b597b7cdc 100644
--- a/src/borg/compress.pyx
+++ b/src/borg/compress.pyx
@@ -281,7 +281,8 @@ cdef class CompressorBase:
 
     def check_fix_size(self, meta, data):
         if "size" in meta:
-            assert meta["size"] == len(data)
+            if meta["size"] != len(data):
+                raise DecompressionError(f'size mismatch: expected {meta["size"]} bytes, got {len(data)}')
         elif self.legacy_mode:
             meta["size"] = len(data)
         else:
@@ -411,7 +412,7 @@ class LZ4(DecidingCompressor):
             # borg2 stores the exact plaintext size in the (authenticated) object metadata:
             # decompress directly into the result bytes object - no scratch buffer, no copy,
             # no output size guessing. rsize != size means corrupt (or misdescribed) data,
-            # this also covers everything check_fix_size would assert.
+            # this also covers everything check_fix_size would check.
             ret = PyBytes_FromStringAndSize(NULL, size)
             dest = PyBytes_AsString(ret)
             osize = size
@@ -602,7 +603,8 @@ class ZLIB_legacy(CompressorBase):
 
     def decompress(self, meta, data):
         # note: for compatibility no super call, do not strip ID bytes
-        assert self.legacy_mode  # only borg 1.x repos have the legacy ZLIB format
+        if not self.legacy_mode:  # only borg 1.x repos have the legacy ZLIB format
+            raise DecompressionError('legacy zlib format is only supported for borg 1.x repositories')
         assert meta is None
         meta = {}
         meta["ctype"] = ZLIB.ID  # change to non-legacy ZLIB id
@@ -774,7 +776,8 @@ class ObfuscateSize(CompressorBase):
         return meta, obfuscated_data  # for borg2 it is enough that we have the payload size in meta["psize"]
 
     def decompress(self, meta, data):
-        assert self.legacy_mode  # borg2 never dispatches to this, only used for legacy mode
+        if not self.legacy_mode:  # borg2 objects store the inner compressor's ctype, only borg 1.x has this header
+            raise DecompressionError('obfuscation header is only supported for borg 1.x repositories')
         meta, obfuscated_data = super().decompress(meta, data)  # remove obfuscator ID header
         compr_size = self.header_fmt.unpack(obfuscated_data[0:self.header_len])[0]
         compressed_data = obfuscated_data[self.header_len:self.header_len+compr_size]
diff --git a/src/borg/item.pyx b/src/borg/item.pyx
index 2044fc10f6..2d575969b6 100644
--- a/src/borg/item.pyx
+++ b/src/borg/item.pyx
@@ -6,7 +6,7 @@ from libc.string cimport memcmp
 from cpython.bytes cimport PyBytes_AsStringAndSize
 
 from .constants import ITEM_KEYS, ARCHIVE_KEYS
-from .helpers import StableDict
+from .helpers import StableDict, IntegrityError
 from .helpers import format_file_size
 from .helpers.fs import assert_sanitized_path, to_sanitized_path, map_chars, slashify
 from .helpers.msgpack import timestamp_to_int, int_to_timestamp, Timestamp
@@ -22,7 +22,8 @@ def fix_key(data, key, *, errors='strict'):
         value = data.pop(key)
         key = key.decode('utf-8', errors=errors)
         data[key] = value
-    assert isinstance(key, str)
+    if not isinstance(key, str):
+        raise IntegrityError(f'dict key is not a str, but {type(key).__name__}')
     return key
 
 
@@ -44,27 +45,34 @@ def fix_bytes_value(data, key):
     return value
 
 
+def check_list(v, what='list'):
+    """Raise IntegrityError if v is not a tuple or list."""
+    if not isinstance(v, (tuple, list)):
+        raise IntegrityError(f'{what} is not a list, but {type(v).__name__}')
+
+
 def fix_list_of_str(v):
     """Ensure that we have a list of str."""
-    assert isinstance(v, (tuple, list))
+    check_list(v)
     return [want_str(e) for e in v]
 
 
 def fix_list_of_bytes(v):
     """Ensure that we have a list of bytes."""
-    assert isinstance(v, (tuple, list))
+    check_list(v)
     return [want_bytes(e) for e in v]
 
 
 def fix_list_of_chunkentries(v):
     """Ensure that we have a list of valid chunk entries."""
-    assert isinstance(v, (tuple, list))
+    check_list(v, 'chunk list')
     chunks = []
     for ce in v:
-        assert isinstance(ce, (tuple, list))
-        assert len(ce) in (2, 3)  # id, size[, csize]
-        assert isinstance(ce[1], int)
-        assert len(ce) == 2 or isinstance(ce[2], int)
+        check_list(ce, 'chunk entry')
+        if len(ce) not in (2, 3):  # id, size[, csize]
+            raise IntegrityError(f'chunk entry has {len(ce)} elements, expected 2 or 3')
+        if not isinstance(ce[1], int) or (len(ce) == 3 and not isinstance(ce[2], int)):
+            raise IntegrityError(f'chunk entry size is not an int: {ce!r}')
         ce_fixed = [want_bytes(ce[0]), ce[1]]  # list! id, size only, drop csize
         chunks.append(ce_fixed)  # create a list of lists
     return chunks
@@ -72,15 +80,16 @@ def fix_list_of_chunkentries(v):
 
 def fix_tuple_of_str(v):
     """Ensure that we have a tuple of str."""
-    assert isinstance(v, (tuple, list))
+    check_list(v)
     return tuple(want_str(e) for e in v)
 
 
 def fix_tuple_of_str_and_int(v):
     """Ensure that we have a tuple of str or int."""
-    assert isinstance(v, (tuple, list))
+    check_list(v)
     t = tuple(e.decode() if isinstance(e, bytes) else e for e in v)
-    assert all(isinstance(e, (str, int)) for e in t), repr(t)
+    if not all(isinstance(e, (str, int)) for e in t):
+        raise IntegrityError(f'not a tuple of str or int: {t!r}')
     return t
 
 
@@ -91,16 +100,24 @@ def fix_timestamp(v):
     # legacy support
     if isinstance(v, bytes):  # was: bigint_to_int()
         v = int.from_bytes(v, 'little', signed=True)
-    assert isinstance(v, int)
+    if not isinstance(v, int):
+        raise IntegrityError(f'timestamp is not an int, but {type(v).__name__}')
     return int_to_timestamp(v)
 
 
+def check_version(v):
+    """Raise IntegrityError if the version v is not an int."""
+    if not isinstance(v, int):
+        raise IntegrityError(f'version is not an int, but {type(v).__name__}')
+
+
 def want_bytes(v, *, errors='surrogateescape'):
     """Ensure that the value is bytes (encode if necessary)."""
     # legacy support: it being str can be caused by msgpack unpack decoding old data that was packed with use_bin_type=False
     if isinstance(v, str):
         v = v.encode('utf-8', errors=errors)
-    assert isinstance(v, bytes), f'not a bytes object, but {v!r}'
+    if not isinstance(v, bytes):
+        raise IntegrityError(f'not a bytes object, but {v!r}')
     return v
 
 
@@ -108,7 +125,8 @@ def want_str(v, *, errors='surrogateescape'):
     """Ensure that the value is str (decode if necessary)."""
     if isinstance(v, bytes):
         v = v.decode('utf-8', errors=errors)
-    assert isinstance(v, str), f'not a str object, but {v!r}'
+    if not isinstance(v, str):
+        raise IntegrityError(f'not a str object, but {v!r}')
     return v
 
 
@@ -427,7 +445,7 @@ cdef class EncryptedKey(PropDict):
         for k, v in list(d.items()):
             k = fix_key(d, k)
             if k == 'version':
-                assert isinstance(v, int)
+                check_version(v)
             if k in ('algorithm', 'argon2_type', 'label'):
                 v = fix_str_value(d, k)
             if k in ('salt', 'hash', 'data'):
@@ -461,14 +479,15 @@ cdef class Key(PropDict):
         for k, v in list(d.items()):
             k = fix_key(d, k)
             if k == 'version':
-                assert isinstance(v, int)
+                check_version(v)
             if k in ('repository_id', 'crypt_key', 'id_key'):
                 v = fix_bytes_value(d, k)
             self._dict[k] = v
         if 'crypt_key' not in self._dict:  # legacy, we're loading an old v1 key
             k = fix_bytes_value(d, 'enc_key') + fix_bytes_value(d, 'enc_hmac_key')
             assert isinstance(k, bytes), "k == %r" % k
-            assert len(k) in (32 + 32, 32 + 128)  # 256+256 or 256+1024 bits
+            if len(k) not in (32 + 32, 32 + 128):  # 256+256 or 256+1024 bits
+                raise IntegrityError(f'legacy key material has an invalid length: {len(k)}')
             self._dict['crypt_key'] = k
 
 cdef class ArchiveItem(PropDict):
@@ -517,7 +536,7 @@ cdef class ArchiveItem(PropDict):
         for k, v in list(d.items()):
             k = fix_key(d, k)
             if k == 'version':
-                assert isinstance(v, int)
+                check_version(v)
             if k in ('name', 'hostname', 'username', 'comment', 'cwd'):
                 v = fix_str_value(d, k)
             if k in ('start', 'end', 'time', 'time_end'):
@@ -560,29 +579,34 @@ cdef class ManifestItem(PropDict):
         for k, v in list(d.items()):
             k = fix_key(d, k)
             if k == 'version':
-                assert isinstance(v, int)
+                check_version(v)
             if k == 'archives':
                 ad = v
-                assert isinstance(ad, dict)
+                if not isinstance(ad, dict):
+                    raise IntegrityError(f'manifest archives is not a dict, but {type(ad).__name__}')
                 for ak, av in list(ad.items()):
                     ak = fix_key(ad, ak, errors='surrogateescape')
-                    assert isinstance(av, dict)
+                    if not isinstance(av, dict):
+                        raise IntegrityError(f'manifest archive entry is not a dict, but {type(av).__name__}')
                     for ik, iv in list(av.items()):
                         ik = fix_key(av, ik)
                         if ik == 'id':
                             fix_bytes_value(av, 'id')
                         if ik == 'time':
                             fix_str_value(av, 'time')
-                    assert set(av) == {'id', 'time'}
+                    if set(av) != {'id', 'time'}:
+                        raise IntegrityError(f'manifest archive entry has unexpected keys: {sorted(av)!r}')
             if k == 'timestamp':
                 v = fix_str_value(d, k, 'replace')
             if k == 'config':
                 cd = v
-                assert isinstance(cd, dict)
+                if not isinstance(cd, dict):
+                    raise IntegrityError(f'manifest config is not a dict, but {type(cd).__name__}')
                 for ck, cv in list(cd.items()):
                     ck = fix_key(cd, ck)
                     if ck == 'tam_required':
-                        assert isinstance(cv, bool)
+                        if not isinstance(cv, bool):
+                            raise IntegrityError(f'manifest tam_required is not a bool, but {type(cv).__name__}')
             if k == 'item_keys':
                 v = fix_tuple_of_str(v)
             self._dict[k] = v
diff --git a/src/borg/legacy/archives.py b/src/borg/legacy/archives.py
index ce2fc38478..e1e617b514 100644
--- a/src/borg/legacy/archives.py
+++ b/src/borg/legacy/archives.py
@@ -269,6 +269,7 @@ def get_one(self, match, *, match_end=r"\Z", deleted=False):
         return archive_infos[0]
 
     def _set_raw_dict(self, d):
+        # d was validated by ManifestItem.update_internal: str keys, dict values with only "id" and "time".
         for k, v in d.items():
             assert isinstance(k, str)
             assert isinstance(v, dict) and "id" in v and "time" in v
diff --git a/src/borg/repoobj.py b/src/borg/repoobj.py
index 750000f245..3cc89cd1fd 100644
--- a/src/borg/repoobj.py
+++ b/src/borg/repoobj.py
@@ -231,11 +231,12 @@ def parse(
             ctype = meta_compressed["ctype"]
             clevel = meta_compressed["clevel"]
             csize = meta_compressed["csize"]  # always the overall size
-            assert csize == len(data_compressed)
-            psize = meta_compressed.get(
-                "psize", csize
-            )  # obfuscation: psize (payload size) is potentially less than csize.
-            assert psize <= csize
+            if csize != len(data_compressed):
+                raise IntegrityError(f"csize inconsistent: expected {csize} bytes, got {len(data_compressed)}")
+            # obfuscation: psize (payload size) is potentially less than csize.
+            psize = meta_compressed.get("psize", csize)
+            if not isinstance(psize, int) or isinstance(psize, bool) or not 0 <= psize <= csize:
+                raise IntegrityError(f"psize invalid: {psize!r} (csize: {csize})")
             compr_hdr = bytes((ctype, clevel))
             compressor_cls, compression_level = Compressor.detect(compr_hdr)
             compressor = compressor_cls(level=compression_level)
diff --git a/src/borg/testsuite/archiver/debug_cmds_test.py b/src/borg/testsuite/archiver/debug_cmds_test.py
index 13061dfef9..c99e8f4904 100644
--- a/src/borg/testsuite/archiver/debug_cmds_test.py
+++ b/src/borg/testsuite/archiver/debug_cmds_test.py
@@ -7,7 +7,7 @@
 from ...cache import write_chunkindex_to_repo
 from ...constants import *  # NOQA
 from ...crypto.key import RepoKeyNotFoundError
-from ...helpers import bin_to_hex
+from ...helpers import bin_to_hex, CommandError
 from ...helpers.passphrase import PassphraseWrong
 from ...manifest import Manifest
 from .. import changedir
@@ -269,6 +269,21 @@ def test_debug_format_obj_respects_type(archivers, request):
     assert meta_read["type"] == ROBJ_ARCHIVE_STREAM
 
 
+@pytest.mark.parametrize("meta", [[1, 2, 3], {"type": 42}, {"type": ROBJ_DONTCARE}])
+def test_debug_format_obj_invalid_meta(archivers, request, meta):
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    create_regular_file(archiver.input_path, "data.bin", contents=b"some data")
+    create_regular_file(archiver.input_path, "meta.json", contents=json.dumps(meta).encode())
+    id_hash = cmd(archiver, "debug", "id-hash", "input/data.bin").strip()
+    args = ("debug", "format-obj", id_hash, "input/data.bin", "input/meta.json", "output/repoobj.bin")
+    if archiver.FORK_DEFAULT:
+        cmd(archiver, *args, exit_code=CommandError().exit_code)
+    else:
+        with pytest.raises(CommandError):
+            cmd(archiver, *args)
+
+
 def test_debug_dump_archive(archivers, request):
     archiver = request.getfixturevalue(archivers)
     create_regular_file(archiver.input_path, "file1", size=1024 * 80)
diff --git a/src/borg/testsuite/archiver/repo_compress_cmd_test.py b/src/borg/testsuite/archiver/repo_compress_cmd_test.py
index a3481b9215..963303f825 100644
--- a/src/borg/testsuite/archiver/repo_compress_cmd_test.py
+++ b/src/borg/testsuite/archiver/repo_compress_cmd_test.py
@@ -1,5 +1,6 @@
 import os
 import re
+import types
 
 import pytest
 
@@ -10,6 +11,7 @@
 from ...manifest import Manifest
 from ...compress import ZSTD, ZLIB, LZ4, CNONE
 from ...archiver.repo_compress_cmd import PackRecompressor
+from ...repoobj import RepoObj
 
 from .. import make_test_key
 from . import create_regular_file, cmd, open_repository, RK_ENCRYPTION
@@ -426,3 +428,27 @@ def test_transform_pack_unchanged_pack_untouched(tmp_path):
         assert {info.name for info in repository.store_list("packs")} == pack_names_before
         assert pdchunk(repository.get(H(0))) == b"WWWW"
         assert pdchunk(repository.get(H(1))) == b"XXXX"
+
+
+@pytest.mark.parametrize("ro_type", [None, 42, ROBJ_DONTCARE])
+def test_transform_rejects_invalid_type(tmp_path, ro_type):
+    # the type of a stored object comes from the repository, so it is validated before it is reused.
+    def compress(meta, data):
+        meta, data = LZ4().compress(meta, data)
+        if ro_type is None:
+            del meta["type"]
+        else:
+            meta["type"] = ro_type
+        return meta, data
+
+    location = os.fspath(tmp_path / "repo")
+    with Repository(location, exclusive=True, create=True, key_loader=make_test_key) as repository:
+        repo_objs = RepoObj(make_test_key(repository))
+        repo_objs.compressor = types.SimpleNamespace(compress=compress)
+        data = b"foobar" * 10
+        id = repo_objs.id_hash(data)
+        obj = repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM)
+        repo_objs.compressor = CompressionSpec("zstd,3").compressor
+        recompressor = PackRecompressor(repository, types.SimpleNamespace(repo_objs=repo_objs), print_stats=False)
+        with pytest.raises(IntegrityError, match="invalid type"):
+            recompressor.transform(id, obj)
diff --git a/src/borg/testsuite/compress_test.py b/src/borg/testsuite/compress_test.py
index e42f58e784..c1bd6f6abe 100644
--- a/src/borg/testsuite/compress_test.py
+++ b/src/borg/testsuite/compress_test.py
@@ -8,7 +8,7 @@
 from ..compress import get_compressor, Compressor, CNONE, ZLIB, LZ4, LZMA, ZSTD, Auto
 from ..compress import get_zstd_mt_workers, get_zstd_compressor, forget_zstd_compressors
 from ..compress import ZSTD_JOB_SIZE_MIN, ZSTD_MT_MIN_SIZE
-from ..helpers import Error
+from ..helpers import DecompressionError, Error
 from ..helpers import CompressionSpec
 from ..constants import ROBJ_FILE_STREAM, ROBJ_ARCHIVE_META
 from ..helpers.argparsing import ArgumentTypeError
@@ -130,6 +130,26 @@ def test_compressor(c_params):
     assert meta_d["size"] == len(DATA)
 
 
+@pytest.mark.parametrize("name", ["none", "lz4", "zstd", "zlib", "lzma"])
+def test_decompress_size_mismatch(name):
+    c = Compressor(name)
+    meta, data_compressed = c.compress({}, DATA)
+    meta["size"] += 1
+    with pytest.raises(DecompressionError):
+        c.decompress(meta, data_compressed)
+
+
+@pytest.mark.parametrize(
+    "ctype, clevel, data",
+    [(0x78, 0x9C, zlib.compress(DATA)), (0x04, 0xFF, bytes(10))],  # legacy zlib (no ID bytes)  # obfuscation header
+)
+def test_decompress_legacy_only_formats(ctype, clevel, data):
+    # borg2 repo object metadata gives ctype and clevel, see RepoObj.parse.
+    compressor_cls, level = Compressor.detect(bytes((ctype, clevel)))
+    with pytest.raises(DecompressionError):
+        compressor_cls(level=level).decompress({"ctype": ctype, "clevel": clevel, "csize": len(data)}, data)
+
+
 def test_auto():
     compressor_auto_zlib = CompressionSpec("auto,zlib,9").compressor
     compressor_lz4 = CompressionSpec("lz4").compressor
diff --git a/src/borg/testsuite/item_test.py b/src/borg/testsuite/item_test.py
index 044796a72c..845df72b87 100644
--- a/src/borg/testsuite/item_test.py
+++ b/src/borg/testsuite/item_test.py
@@ -2,7 +2,8 @@
 
 from ..cache import ChunkListEntry
 from ..item import MAX_ALIGN_CHUNKS, MAX_ALIGN_WORK, Item, ItemDiff, chunks_contents_equal, chunks_diff_size
-from ..helpers import StableDict
+from ..item import ArchiveItem, EncryptedKey, Key, ManifestItem
+from ..helpers import IntegrityError, StableDict
 from ..helpers.msgpack import Timestamp
 from ..platformflags import is_pypy
 
@@ -64,6 +65,46 @@ def test_item_invalid(invalid_item, error):
         Item(invalid_item)
 
 
+@pytest.mark.parametrize(
+    "cls, internal_dict",
+    [
+        (Item, {42: "a/b/c"}),  # key is neither str nor bytes
+        (Item, {"path": 42}),  # not a str
+        (Item, {"chunks": b"not a list"}),
+        (Item, {"chunks": [b"not a list"]}),
+        (Item, {"chunks": [[b"id"]]}),  # too few elements
+        (Item, {"chunks": [[b"id", "1000"]]}),  # size is not an int
+        (Item, {"chunks": [[b"id", 1000, "10"]]}),  # csize is not an int
+        (Item, {"mtime": "noon"}),  # timestamp is not an int
+        (Item, {"acl_access": 42}),  # not bytes
+        (Item, {"xattrs": {b"user.foo": 42}}),  # value is not bytes
+        (ArchiveItem, {"version": "2"}),
+        (ArchiveItem, {"cmdline": "borg create"}),  # not a list
+        (ArchiveItem, {"chunker_params": ("buzhash", 1.5)}),  # not str or int
+        (ArchiveItem, {"item_ptrs": [42]}),  # not bytes
+        (ManifestItem, {"version": "1"}),
+        (ManifestItem, {"archives": []}),  # not a dict
+        (ManifestItem, {"archives": {"name": []}}),  # entry is not a dict
+        (ManifestItem, {"archives": {"name": {"id": b"id"}}}),  # time is missing
+        (ManifestItem, {"config": []}),  # not a dict
+        (ManifestItem, {"config": {"tam_required": 1}}),  # not a bool
+        (EncryptedKey, {"version": "1"}),
+        (EncryptedKey, {"salt": 42}),  # not bytes
+        (Key, {"version": "2"}),
+        (Key, {"version": 1, "enc_key": b"k" * 32, "enc_hmac_key": b"h" * 16}),  # invalid legacy key length
+    ],
+)
+def test_invalid_internal_dict(cls, internal_dict):
+    # internal_dict comes from msgpack-unpacked repository / key data, so it is validated.
+    with pytest.raises(IntegrityError):
+        cls(internal_dict=internal_dict)
+
+
+def test_legacy_key_internal_dict():
+    key = Key(internal_dict={"version": 1, "enc_key": b"k" * 32, "enc_hmac_key": b"h" * 32})
+    assert key.crypt_key == b"k" * 32 + b"h" * 32
+
+
 def test_item_from_kw():
     item = Item(path="a/b/c", mode=0o666)
     assert item.path == "a/b/c"
diff --git a/src/borg/testsuite/repoobj_test.py b/src/borg/testsuite/repoobj_test.py
index 7ca95f0658..747721a04e 100644
--- a/src/borg/testsuite/repoobj_test.py
+++ b/src/borg/testsuite/repoobj_test.py
@@ -1,9 +1,12 @@
+import types
+import zlib
+
 import pytest
 
 from ..constants import ROBJ_FILE_STREAM, ROBJ_ARCHIVE_META
 from ..crypto.key import AESOCBKey, AuthenticatedKey, CHPOKey, LegacyAuthenticatedKey, LegacyPlaintextKey
 from ..helpers import CompressionSpec
-from ..helpers.errors import Error, IntegrityError
+from ..helpers.errors import DecompressionError, Error, IntegrityError
 from ..repository import Repository
 from ..repoobj import (
     ASSERT_ID_PLACES,
@@ -164,6 +167,44 @@ def test_malformed_object_inconsistent_sizes(key):
         repo_objs.parse(id, hdr, ro_type=ROBJ_FILE_STREAM)
 
 
+@pytest.mark.parametrize(
+    "field, damaged_value",
+    [
+        ("csize", lambda csize: csize + 1),  # csize does not match the data size
+        ("psize", lambda csize: csize + 1),  # psize is bigger than csize
+        ("psize", lambda csize: -1),
+        ("psize", lambda csize: str(csize)),
+    ],
+)
+def test_parse_rejects_inconsistent_sizes(key, field, damaged_value):
+    # authenticated metadata can still be inconsistent, e.g. if written by a broken borg client.
+    def compress(meta, data):
+        meta, data = LZ4().compress(meta, data)
+        meta[field] = damaged_value(meta["csize"])
+        return meta, data
+
+    repo_objs = RepoObj(key)
+    repo_objs.compressor = types.SimpleNamespace(compress=compress)
+    data = b"foobar" * 10
+    id = repo_objs.id_hash(data)
+    cdata = repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM)
+    with pytest.raises(IntegrityError):
+        repo_objs.parse(id, cdata, ro_type=ROBJ_FILE_STREAM)
+
+
+def test_parse_rejects_legacy_compression(key):
+    # the legacy zlib format (no ID bytes) only exists in borg 1.x repositories.
+    repo_objs = RepoObj(key)
+    data = b"foobar" * 10
+    id = repo_objs.id_hash(data)
+    data_compressed = zlib.compress(data)
+    cdata = repo_objs.format(
+        id, {}, data_compressed, compress=False, size=len(data), ctype=0x78, clevel=0x9C, ro_type=ROBJ_FILE_STREAM
+    )
+    with pytest.raises(DecompressionError):
+        repo_objs.parse(id, cdata, ro_type=ROBJ_FILE_STREAM)
+
+
 def test_parse_header(key):
     repo_objs = RepoObj(key)
     data = b"foobar"

From 335dc3bb7ea3d58414edaf8add73446df3d7e52e Mon Sep 17 00:00:00 2001
From: Thomas Waldmann <tw@waldmann-edv.de>
Date: Tue, 13 Oct 2026 01:57:13 +0200
Subject: [PATCH 3/6] archive/cache: do not assert on chunk list sizes and
 archive metadata, refs #8649

The chunk sizes in an archive's chunk lists are repository data:

- DownloadPipeline.fetch_many: a size that does not match the chunk data
  or that is invalid for a replacement chunk raises IntegrityError.
- seen_chunk / reuse_chunk: invalid or inconsistent sizes raise
  IntegrityError.
- files cache: compress_entry raises ValueError for invalid or
  inconsistent sizes, building the files cache from the previous
  archive skips such entries (the files cache is only an optimization).

Archive metadata must have exactly one of item_ptrs / items, check that
where it is loaded. Archives._parse_archive_meta treats invalid metadata
like corrupted metadata, so one bad archive does not break listing it.

import-tar: a BORG tar member with an unsupported BORG.item.version gets
a warning and status "E" and is skipped.

CacheConfig.create: do not fail if a concurrently running borg already
created the cache config.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/archive.py                          | 17 +++++---
 src/borg/archiver/tar_cmds.py                |  7 +++-
 src/borg/cache.py                            | 33 +++++++++++-----
 src/borg/manifest.py                         | 13 ++++---
 src/borg/testsuite/archive_test.py           | 39 ++++++++++++++++++-
 src/borg/testsuite/archiver/tar_cmds_test.py | 16 ++++++++
 src/borg/testsuite/archives_test.py          | 23 +++++++++++
 src/borg/testsuite/cache_test.py             | 41 +++++++++++++++++++-
 8 files changed, 166 insertions(+), 23 deletions(-)

diff --git a/src/borg/archive.py b/src/borg/archive.py
index 00a32902eb..4f90c93a1a 100644
--- a/src/borg/archive.py
+++ b/src/borg/archive.py
@@ -429,17 +429,19 @@ def fetch_many(self, chunks, ro_type=None, replacement_chunk=True, replace_corru
         (recreate, transfer) abort instead of storing all-zero data as if it were the content.
         If *damaged* is a list, the ids of the replaced chunks are appended to it, so the caller
         can report the affected file.
+        A chunk size (ChunkListEntry) that does not match the chunk data, or that is invalid for a
+        replacement chunk, raises IntegrityError.
         """
         assert ro_type is not None
 
         def replacement(id, size, problem):
             # all-zero chunk of the correct size, so the content stream keeps its offsets.
+            if not 0 <= size <= len(zeros):
+                raise IntegrityError(f"chunk {bin_to_hex(id)}: invalid size {size} in chunk list")
             logger.error(f"repository object {bin_to_hex(id)} {problem}, returning {size} zero bytes.")
             if damaged is not None:
                 damaged.append(id)
-            data = zeros[:size]
-            assert len(data) == size, f"replacement chunk size {size} exceeds {len(zeros)}"
-            return data
+            return zeros[:size]
 
         ids = []
         sizes = []
@@ -484,7 +486,10 @@ def replacement(id, size, problem):
                         data = replacement(id, size, f"corrupted ({err})")
                     else:
                         self.parsed_cache[(id, ro_type)] = data
-            assert data is None or size is None or len(data) == size
+            if data is not None and size is not None and len(data) != size:
+                raise IntegrityError(
+                    f"chunk {bin_to_hex(id)}: size {size} in chunk list does not match data size {len(data)}"
+                )
             yield data
 
 
@@ -704,6 +709,8 @@ def _load_meta(self, id):
         metadata = ArchiveItem(internal_dict=archive)
         if metadata.version not in (1, 2):  # legacy: still need to read v1 archives
             raise Exception("Unknown archive metadata version")
+        if ("item_ptrs" in metadata) == ("items" in metadata):
+            raise IntegrityError(f"archive {bin_to_hex(id)}: metadata needs exactly one of item_ptrs, items")
         return metadata
 
     def load(self, id, *, metadata=None):
@@ -2079,7 +2086,7 @@ def __init__(
     def create_helper(self, tarinfo, status=None, type=None):
         ph = tarinfo.pax_headers
         if ph and "BORG.item.version" in ph:
-            assert ph["BORG.item.version"] == "1"
+            assert ph["BORG.item.version"] == "1"  # the caller skips members with other versions
             meta_bin = base64.b64decode(ph["BORG.item.meta"])
             meta_dict = msgpack.unpackb(meta_bin, object_hook=StableDict)
             item = Item(internal_dict=meta_dict)
diff --git a/src/borg/archiver/tar_cmds.py b/src/borg/archiver/tar_cmds.py
index c9e12969ca..5f34df0bad 100644
--- a/src/borg/archiver/tar_cmds.py
+++ b/src/borg/archiver/tar_cmds.py
@@ -587,7 +587,12 @@ def path_components(name):
             else:
                 name = make_path_safe(posixpath.normpath(tarinfo.name))
             # name is the same as the stored item path now, we show it in the file status output.
-            if tarinfo.isreg():
+            borg_item_version = tarinfo.pax_headers.get("BORG.item.version")
+            if borg_item_version not in (None, "1"):
+                # a BORG tar member with an item metadata format we do not know (see export-tar --tar-format=BORG).
+                status = "E"
+                self.print_warning("%s: Unsupported BORG.item.version %s", tarinfo.name, borg_item_version)
+            elif tarinfo.isreg():
                 status = tfo.process_file(tarinfo=tarinfo, status="A", type=stat.S_IFREG, tar=tar)
             elif tarinfo.isdir():
                 status = tfo.process_dir(tarinfo=tarinfo, status="d", type=stat.S_IFDIR)
diff --git a/src/borg/cache.py b/src/borg/cache.py
index fafee9c110..433e6b6243 100644
--- a/src/borg/cache.py
+++ b/src/borg/cache.py
@@ -20,7 +20,7 @@
 
 from borgstore.store import ItemInfo
 
-from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS
+from .constants import CACHE_README, FILES_CACHE_MODE_DISABLED, MAX_DATA_SIZE, ROBJ_FILE_STREAM, TIME_DIFFERS2_NS
 from .constants import CHUNKINDEX_FRAGMENT_ENTRIES_MIN, CHUNKINDEX_FRAGMENT_ENTRIES_MAX
 from .constants import CHUNKINDEX_SMALL_FRAGMENT_CAP, CHUNKINDEX_MERGE_ATTEMPTS, CHUNKINDEX_INVALID_SENTINEL
 from .hashindex import ChunkIndex, ChunkIndexEntry, ChunkIndexEntryFormat
@@ -124,7 +124,9 @@ def exists(self):
         return self.config_path.exists()
 
     def create(self):
-        assert not self.exists()
+        if self.exists():
+            # e.g. created by a concurrently running borg process, the content is the same.
+            return
         config = configparser.ConfigParser(interpolation=None)
         config.add_section("cache")
         config.set("cache", "version", "1")
@@ -244,11 +246,14 @@ def compress_entry(self, entry):
         assert isinstance(entry, FileCacheEntry)
         compressed_chunks = []
         for id, size in entry.chunks:
+            # the chunk sizes come from an archive's chunk list or from the local files cache.
+            if not 0 < size <= MAX_DATA_SIZE:
+                raise ValueError(f"chunk {bin_to_hex(id)}: invalid size {size}")
             cie = self.chunks[id]  # may raise KeyError if chunk id is not in repo
             if cie.size == 0:  # size is not known in the chunks index yet
                 self.chunks[id] = cie._replace(size=size)
-            else:
-                assert size == cie.size, f"{size} != {cie.size}"
+            elif size != cie.size:
+                raise ValueError(f"chunk {bin_to_hex(id)}: size {size} does not match known size {cie.size}")
             idx = self.chunks.k_to_idx(id)
             compressed_chunks.append(idx)
         entry = entry._replace(chunks=compressed_chunks)
@@ -342,10 +347,15 @@ def _build_files_cache(self):
                     digests=item.get("digests"),
                 )
                 # note: if the repo is an a valid state, next line should not fail with KeyError:
-                files[path_hash] = self.compress_entry(entry)
+                try:
+                    files[path_hash] = self.compress_entry(entry)
+                except ValueError as exc:
+                    # invalid chunk sizes in the item's chunk list: skip the entry, the file will just be
+                    # chunked again (the files cache is only an optimization).
+                    logger.debug(f"compress_entry failed for {item.path} [{exc}], skipping.")
         # deal with special snapshot / timestamp granularity case, see FAQ:
         for path_hash in self._newest_path_hashes:
-            del files[path_hash]
+            files.pop(path_hash, None)  # the entry may have been skipped above
         files_cache_logger.debug("FILES-CACHE-BUILD: finished, %d entries loaded.", len(files))
         return files
 
@@ -1233,17 +1243,22 @@ def seen_chunk(self, id, size=None):
         entry = self.chunks.get(id)
         entry_exists = entry is not None
         if entry_exists and size is not None:
+            # *size* may come from an archive's chunk list (e.g. transfer), so it must be checked.
             if entry.size == 0:
                 # AdHocWithFilesCache:
                 # Here *size* is used to update the chunk's size information, which will be zero for existing chunks.
+                if not 0 < size <= MAX_DATA_SIZE:
+                    raise IntegrityError(f"chunk {bin_to_hex(id)}: invalid size {size}")
                 self.chunks[id] = entry._replace(size=size)
-            else:
+            elif size != entry.size:
                 # in case we already have a size information in the entry, check consistency:
-                assert size == entry.size
+                raise IntegrityError(f"chunk {bin_to_hex(id)}: size {size} does not match known size {entry.size}")
         return entry_exists
 
     def reuse_chunk(self, id, size, stats):
-        assert isinstance(size, int) and size > 0
+        # *size* may come from an archive's chunk list (e.g. create --reuse-from, recreate, transfer).
+        if not (isinstance(size, int) and 0 < size <= MAX_DATA_SIZE):
+            raise IntegrityError(f"chunk {bin_to_hex(id)}: invalid size {size!r}")
         stats.update(size, False)
         return ChunkListEntry(id, size)
 
diff --git a/src/borg/manifest.py b/src/borg/manifest.py
index 6a9697375c..39f08e636c 100644
--- a/src/borg/manifest.py
+++ b/src/borg/manifest.py
@@ -20,7 +20,7 @@
     compile_date_pattern,
     DatePatternError,
 )
-from .helpers.errors import Error, CommandError
+from .helpers.errors import Error, CommandError, IntegrityError
 from .crypto.low_level import IntegrityError as IntegrityErrorBase
 from .item import ArchiveItem
 from .patterns import get_regex_from_pattern
@@ -222,7 +222,14 @@ def _parse_archive_meta(self, id: bytes, cdata) -> tuple[dict, ArchiveItem | Non
         else:
             try:
                 _, data = self.manifest.repo_objs.parse(id, cdata, ro_type=ROBJ_ARCHIVE_META)
+                archive_dict = self.manifest.key.unpack_archive(data)
+                archive_item = ArchiveItem(internal_dict=archive_dict)
+                if archive_item.version not in (1, 2):  # legacy: still need to read v1 archives
+                    raise Exception("Unknown archive metadata version")
+                if ("item_ptrs" in archive_item) == ("items" in archive_item):
+                    raise IntegrityError(f"archive {bin_to_hex(id)}: metadata needs exactly one of item_ptrs, items")
             except IntegrityErrorBase:
+                archive_item = None
                 metadata = dict(
                     id=id,
                     name="archive-metadata-has-integrity-error",
@@ -233,10 +240,6 @@ def _parse_archive_meta(self, id: bytes, cdata) -> tuple[dict, ArchiveItem | Non
                     tags=(),
                 )
             else:
-                archive_dict = self.manifest.key.unpack_archive(data)
-                archive_item = ArchiveItem(internal_dict=archive_dict)
-                if archive_item.version not in (1, 2):  # legacy: still need to read v1 archives
-                    raise Exception("Unknown archive metadata version")
                 # callers expect a dict with dict["key"] access, not ArchiveItem.key access.
                 # also, we need to put the id in there.
                 metadata = dict(
diff --git a/src/borg/testsuite/archive_test.py b/src/borg/testsuite/archive_test.py
index a521598fb3..d43ad0b675 100644
--- a/src/borg/testsuite/archive_test.py
+++ b/src/borg/testsuite/archive_test.py
@@ -10,13 +10,13 @@
 
 from . import rejected_dotdot_paths, is_utime_fully_supported, make_test_key, set_test_key_on_open
 from ..cache import ChunkListEntry
-from ..constants import ROBJ_ARCHIVE_STREAM, ROBJ_FILE_STREAM, zeros
+from ..constants import MAX_DATA_SIZE, ROBJ_ARCHIVE_STREAM, ROBJ_FILE_STREAM, zeros
 from ..archive import Archive, CacheChunkBuffer, DownloadPipeline, RobustUnpacker, valid_msgpacked_dict
 from ..archive import ITEM_KEYS, Statistics
 from ..archive import zero_chunk_flags, zero_chunk_id, zero_chunk_ids
 from ..archive import BackupOSError, BackupRaceConditionError, backup_io, backup_io_iter, get_item_uid_gid
 from ..archive import stat_update_check, is_automount, tar_acl_to_borg
-from ..helpers import msgpack, StableDict
+from ..helpers import IntegrityError, msgpack, StableDict
 from ..repoobj import RepoObj
 from ..item import Item, ArchiveItem
 from ..manifest import Archives, Manifest
@@ -316,6 +316,31 @@ def test_download_pipeline_missing_chunk(replacement_chunk):
     assert result == [zeros[: len(data)] if replacement_chunk else None]
 
 
+def test_download_pipeline_size_mismatch():
+    # the size in a chunk list is repository data, too: if it does not match the (authentic) chunk data,
+    # that is an integrity error.
+    key = make_test_key(None)
+    repo_objs = RepoObj(key)
+    data = b"foobar" * 100
+    id = repo_objs.id_hash(data)
+    repository = MockFetchRepo({id: repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM)})
+    pipeline = DownloadPipeline(repository, repo_objs)
+    with pytest.raises(IntegrityError, match="does not match data size"):
+        list(pipeline.fetch_many([ChunkListEntry(id, len(data) + 1)], ro_type=ROBJ_FILE_STREAM))
+
+
+@pytest.mark.parametrize("size", [-1, MAX_DATA_SIZE + 1])
+def test_download_pipeline_missing_chunk_invalid_size(size):
+    # an all-zero replacement chunk can not be made for an invalid size from the chunk list.
+    key = make_test_key(None)
+    repo_objs = RepoObj(key)
+    id = repo_objs.id_hash(b"foobar")
+    repository = MockFetchRepo({id: None})  # the object is gone
+    pipeline = DownloadPipeline(repository, repo_objs)
+    with pytest.raises(IntegrityError, match="invalid size"):
+        list(pipeline.fetch_many([ChunkListEntry(id, size)], ro_type=ROBJ_FILE_STREAM))
+
+
 def test_download_pipeline_zero_chunks_served_locally():
     # repeated all-zero chunks (e.g. from the holes of a sparse file) shall be served
     # directly from the zeros constant, without repository access, see issue #1678.
@@ -882,6 +907,16 @@ def test_reject_non_sanitized_item():
             Item(path=path, user="root", group="root")
 
 
+@pytest.mark.parametrize("items_keys", [{}, {"item_ptrs": [], "items": []}])
+def test_load_meta_needs_exactly_one_of_item_ptrs_items(items_keys):
+    archive = Archive.__new__(Archive)  # only _load_meta is tested, it needs no fully set up Archive
+    archive.repository, archive.repo_objs, archive.key = Mock(), Mock(), Mock()
+    archive.repo_objs.parse.return_value = (None, b"data")
+    archive.key.unpack_archive.return_value = {"version": 2, "name": "a"} | items_keys
+    with pytest.raises(IntegrityError, match="exactly one of item_ptrs, items"):
+        archive._load_meta(b"\x01" * 32)
+
+
 def test_archives_get_by_id_missing_returns_none():
     repo = Mock()
     repo.store_list.return_value = []  # empty store — id will not be found
diff --git a/src/borg/testsuite/archiver/tar_cmds_test.py b/src/borg/testsuite/archiver/tar_cmds_test.py
index 97573567ea..b18d28ed3d 100644
--- a/src/borg/testsuite/archiver/tar_cmds_test.py
+++ b/src/borg/testsuite/archiver/tar_cmds_test.py
@@ -189,6 +189,22 @@ def test_import_tar_nfiles(archivers, request):
     assert info["archives"][0]["stats"]["nfiles"] == 3
 
 
+def test_import_tar_unsupported_borg_item_version(archivers, request):
+    # a BORG tar member with an unknown item metadata version is skipped with a warning.
+    archiver = request.getfixturevalue(archivers)
+    with tarfile.open("input.tar", "w", format=tarfile.PAX_FORMAT) as tar:
+        for name, pax_headers in ("dir/file1", {"BORG.item.version": "2"}), ("dir/file2", {}):
+            data = name.encode()
+            tarinfo = tarfile.TarInfo(name)
+            tarinfo.size = len(data)
+            tarinfo.pax_headers = pax_headers
+            tar.addfile(tarinfo, io.BytesIO(data))
+    cmd(archiver, "repo-create", "--encryption=authenticated-sha256")
+    output = cmd(archiver, "import-tar", "dst", "input.tar", exit_code=EXIT_WARNING)
+    assert "dir/file1: Unsupported BORG.item.version 2" in output
+    assert cmd(archiver, "list", "--short", "dst").splitlines() == ["dir/file2"]
+
+
 def test_import_tar_json(archivers, request):
     """import-tar --json reports the stats of the new archive like create --json does, see #10335."""
     archiver = request.getfixturevalue(archivers)
diff --git a/src/borg/testsuite/archives_test.py b/src/borg/testsuite/archives_test.py
index 563aea50a3..e31abd3806 100644
--- a/src/borg/testsuite/archives_test.py
+++ b/src/borg/testsuite/archives_test.py
@@ -216,6 +216,7 @@ def test_get_archive_meta_success():
         "size": 1024,
         "nfiles": 3,
         "comment": "weekly",
+        "item_ptrs": [],
     }
 
     result, archive_item = ar._get_archive_meta(_id(1))
@@ -243,6 +244,7 @@ def test_get_archive_meta_success_with_tags():
         "username": "",
         "hostname": "",
         "tags": ["beta", "alpha"],
+        "item_ptrs": [],
     }
 
     result, _ = ar._get_archive_meta(_id(1))
@@ -262,6 +264,26 @@ def test_get_archive_meta_bad_version():
         ar._get_archive_meta(_id(1))
 
 
+@pytest.mark.parametrize("items_keys", [{}, {"item_ptrs": [], "items": []}])
+def test_get_archive_meta_needs_exactly_one_of_item_ptrs_items(items_keys):
+    # authentic archive metadata without item pointers (or with both kinds) is treated like corrupted metadata.
+    ar, _, manifest = _archives()
+    manifest.repo_objs.parse.return_value = (None, b"data")
+    manifest.key.unpack_archive.return_value = {
+        "version": 2,
+        "name": "myarchive",
+        "time": TS,
+        "username": "",
+        "hostname": "",
+    } | items_keys
+
+    result, archive_item = ar._get_archive_meta(_id(1))
+
+    assert archive_item is None
+    assert result["exists"] is False
+    assert result["name"] == "archive-metadata-has-integrity-error"
+
+
 def test_infos_gathers_archive_metadata():
     # _infos reads all archive metadata objects with one gather_many call, keeping the order of ids():
     # a missing object (None) and one failing to parse yield placeholder infos with exists=False and
@@ -283,6 +305,7 @@ def parse(id_, cdata, ro_type):
         "time": TS,
         "username": "alice",
         "hostname": "myhost",
+        "item_ptrs": [],
     }
 
     infos, archive_items = zip(*ar._infos())
diff --git a/src/borg/testsuite/cache_test.py b/src/borg/testsuite/cache_test.py
index 89cdf5cd50..4388ad4c57 100644
--- a/src/borg/testsuite/cache_test.py
+++ b/src/borg/testsuite/cache_test.py
@@ -30,11 +30,12 @@
     write_chunkindex_invalid,
     write_chunkindex_to_repo,
 )
+from ..constants import MAX_DATA_SIZE
 from ..hashindex import ChunkIndex, ChunkIndexEntry
 from ..crypto.key import AESOCBKey, AuthenticatedKey
 from ..crypto import sealed_stream
 from ..crypto.sealed_stream import SealedStreamWriter, read_sealed_stream
-from ..helpers import CorruptPack, Error, bin_to_hex, progress, safe_ns
+from ..helpers import CorruptPack, Error, IntegrityError, bin_to_hex, progress, safe_ns
 from ..helpers import msgpack
 from ..helpers.msgpack import int_to_timestamp
 from ..manifest import Manifest
@@ -84,6 +85,36 @@ def test_existing_reuse_after_add_chunk(self, cache):
         assert cache.add_chunk(H(1), {}, b"5678", stats=Statistics()) == (H(1), 4)
         assert cache.reuse_chunk(H(1), 4, Statistics()) == (H(1), 4)
 
+    def test_seen_chunk_size_mismatch(self, cache):
+        # chunk sizes may come from an archive's chunk list (e.g. transfer), they must match the known size.
+        cache.add_chunk(H(3), {}, b"5678", stats=Statistics())
+        assert cache.seen_chunk(H(3), 4)
+        with pytest.raises(IntegrityError, match="does not match known size"):
+            cache.seen_chunk(H(3), 5)
+
+    @pytest.mark.parametrize("size", [0, -1, MAX_DATA_SIZE + 1])
+    def test_seen_chunk_invalid_size(self, cache, size):
+        # H(1) is in the repository, but its size is not known yet (0 in the chunks index).
+        with pytest.raises(IntegrityError, match="invalid size"):
+            cache.seen_chunk(H(1), size)
+
+    @pytest.mark.parametrize("size", [0, -1, MAX_DATA_SIZE + 1, None])
+    def test_reuse_chunk_invalid_size(self, cache, size):
+        with pytest.raises(IntegrityError, match="invalid size"):
+            cache.reuse_chunk(H(1), size, Statistics())
+
+    @pytest.mark.parametrize("sizes", [(4, 5), (0, 0), (-1, -1)])
+    def test_compress_entry_invalid_size(self, cache, sizes):
+        # the first size is added to the chunks index, the second one is in the files cache entry.
+        cache.add_chunk(H(5), {}, b"5678", stats=Statistics())
+        ts = int_to_timestamp(0)
+        entry = FileCacheEntry(age=0, inode=1, size=4, ctime=ts, mtime=ts, chunks=[(H(5), sizes[0])])
+        if sizes[0] > 0:
+            cache.compress_entry(entry)
+        entry = entry._replace(chunks=[(H(5), sizes[1])])
+        with pytest.raises(ValueError):
+            cache.compress_entry(entry)
+
     def test_periodic_index_write_flushes_pending(self, repository, cache):
         # #9900: the periodic index write must flush the pack writer first, so a buffered chunk is
         # persisted with its real pack location rather than the unresolved F_PENDING placeholder.
@@ -217,6 +248,14 @@ def test_files_cache_invalid_content(self, manifest, caplog):
             writer.finish()
         self.check_ignored(cache, caplog, reason="invalid")
 
+    def test_cache_config_create_existing(self, manifest):
+        # a concurrently running borg process may have created the cache config already.
+        cache, _ = self.write_files_cache(manifest)
+        config_path = cache.path / "config"
+        config = config_path.read_text()
+        cache.cache_config.create()
+        assert config_path.read_text() == config
+
     def test_cache_config(self, manifest):
         cache, _ = self.write_files_cache(manifest)
         config_path = cache.path / "config"

From 8c48b8572fac5f798275a6c3d42f83fab2e16031 Mon Sep 17 00:00:00 2001
From: Thomas Waldmann <tw@waldmann-edv.de>
Date: Tue, 13 Oct 2026 01:57:24 +0200
Subject: [PATCH 4/6] locking: validate lock data, release only our own lock,
 refs #8649

storelocking: _parse_lock checks the types of the lock owner fields and
that the timestamp has a timezone. A malformed (but authenticated) lock is
treated as unreadable instead of crashing in platform.process_alive(), so
borg break-lock can remove it.

release() / migrate_lock() select the lock object we hold (my_lock_key)
instead of asserting that only one lock with our process id exists: a
not yet stale leftover of a dead process with the same id (pid reuse,
e.g. in containers) made them fail. migrate_lock raises LockTimeout if
our lock was killed, like refresh does.

fslocking: a lock roster with invalid entries is treated as corrupt
(empty), like a roster that is not valid JSON.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/fslocking.py                   | 18 ++++++
 src/borg/storelocking.py                | 29 ++++++++--
 src/borg/testsuite/fslocking_test.py    | 28 ++++++++++
 src/borg/testsuite/storelocking_test.py | 74 +++++++++++++++++++++++++
 4 files changed, 145 insertions(+), 4 deletions(-)

diff --git a/src/borg/fslocking.py b/src/borg/fslocking.py
index 13bba12df2..5ab8eb6888 100644
--- a/src/borg/fslocking.py
+++ b/src/borg/fslocking.py
@@ -288,6 +288,18 @@ def migrate_lock(self, old_id, new_id):
         self.id, self.unique_name = new_id, new_unique_name
 
 
+def is_roster_entry(entry):
+    """Return whether <entry> is a valid [host, pid, thread] roster entry."""
+    if not (isinstance(entry, list) and len(entry) == 3):
+        return False
+    host, pid, thread = entry
+    if not isinstance(host, str):
+        return False
+    if not all(isinstance(value, int) and not isinstance(value, bool) for value in (pid, thread)):
+        return False
+    return 0 < pid < 2**31
+
+
 class LockRoster:
     """
     A Lock Roster to track shared/exclusive lockers.
@@ -306,6 +318,12 @@ def load(self):
         try:
             with self.path.open() as f:
                 data = json.load(f)
+            if not isinstance(data, dict):
+                raise ValueError("roster is not a dict")
+            for key in (SHARED, EXCLUSIVE):
+                entries = data.get(key, [])
+                if not isinstance(entries, list) or not all(is_roster_entry(entry) for entry in entries):
+                    raise ValueError(f"invalid {key} roster entries")
 
             # Just nuke the stale locks early on load
             if self.kill_stale_locks:
diff --git a/src/borg/storelocking.py b/src/borg/storelocking.py
index b06c0e38ca..397e822db4 100644
--- a/src/borg/storelocking.py
+++ b/src/borg/storelocking.py
@@ -521,6 +521,17 @@ def _parse_lock(self, content):
                 threadid=lock["threadid"],
                 dt=datetime.datetime.fromisoformat(lock["time"]),
             )
+            # authenticated does not mean well-formed: check the types, the owner fields are
+            # passed to platform.process_alive() and the timestamp is compared to aware datetimes.
+            if not isinstance(lock["hostid"], str):
+                raise ValueError(f"invalid hostid: {lock['hostid']!r}")
+            for name in ("processid", "threadid"):
+                if not isinstance(lock[name], int) or isinstance(lock[name], bool):
+                    raise ValueError(f"invalid {name}: {lock[name]!r}")
+            if not 0 < lock["processid"] < 2**31:
+                raise ValueError(f"processid out of range: {lock['processid']!r}")
+            if lock["dt"].utcoffset() is None:
+                raise ValueError(f"timestamp without timezone: {lock['dt']!r}")
         except (IntegrityError, UnicodeDecodeError, ValueError, KeyError, TypeError) as err:
             logger.debug(f"LOCK-PARSE: unreadable lock object: {err!r}.")
             return None
@@ -535,6 +546,16 @@ def _find_locks(self, *, only_exclusive=False, only_mine=False):
                 found_locks.append(lock)
         return found_locks
 
+    def _find_my_lock(self):
+        """
+        Return a list with the lock we currently hold, or an empty list if we hold none (or it is gone).
+
+        Other locks with our id are not ours: they are leftovers of a dead process with the same id
+        (e.g. pid reuse in containers with a fixed BORG_HOST_ID), they get removed when they become stale.
+        """
+        my_key = self.my_lock_key  # single read - a LockRefresher thread may rebind it concurrently
+        return [lock for lock in self._find_locks(only_mine=True) if lock["key"] == my_key]
+
     def acquire(self):
         # goal
         # for exclusive lock: there must be only 1 exclusive lock and no other (exclusive or non-exclusive) locks.
@@ -610,14 +631,13 @@ def _log_blocking_locks(self, locks):
 
     def release(self, *, ignore_not_found=False):
         self.last_refresh_dt = None
-        locks = self._find_locks(only_mine=True)
+        locks = self._find_my_lock()
         if not locks:
             if ignore_not_found:
                 logger.debug("LOCK-RELEASE: trying to release the lock, but none was found.")
                 return
             else:
                 raise NotLocked(self.repository)
-        assert len(locks) == 1
         lock = locks[0]
         logger.debug(f"LOCK-RELEASE: releasing lock: {lock}.")
         self._delete_lock(lock["key"], ignore_not_found=True, update_last_refresh=True)
@@ -642,8 +662,9 @@ def migrate_lock(self, old_id, new_id):
         logger.debug(f"LOCK-MIGRATE: {old_id} -> {new_id}.")
         assert self.id == old_id
         assert len(new_id) == 3
-        old_locks = self._find_locks(only_mine=True)
-        assert len(old_locks) == 1
+        old_locks = self._find_my_lock()
+        if not old_locks:
+            raise LockTimeout(self.repository, LOCK_KILLED_MSG)
         self.id = new_id
         self._create_lock(exclusive=old_locks[0]["exclusive"], update_last_refresh=True)
         self._delete_lock(old_locks[0]["key"], update_last_refresh=False)
diff --git a/src/borg/testsuite/fslocking_test.py b/src/borg/testsuite/fslocking_test.py
index 0d3ff9473f..66f65ad0a5 100644
--- a/src/borg/testsuite/fslocking_test.py
+++ b/src/borg/testsuite/fslocking_test.py
@@ -1,3 +1,4 @@
+import json
 import random
 import time
 from pathlib import Path
@@ -373,3 +374,30 @@ def test_migrate_lock(self, rosterpath):
         roster.migrate_lock(SHARED, old_id, new_id)  # fix the lock
         assert roster.id == new_id
         assert roster.get(SHARED) == {new_id}
+
+    @pytest.mark.parametrize(
+        "data",
+        [
+            [],  # not a dict
+            "foo",
+            {SHARED: 1},  # entries not a list
+            {SHARED: [1]},  # entry not a list
+            {SHARED: [["host", 1]]},  # entry too short
+            {SHARED: [[1, 1, 0]]},  # host not a str
+            {SHARED: [["host", "1", 0]]},  # pid not an int
+            {SHARED: [["host", True, 0]]},  # pid is a bool
+            {EXCLUSIVE: [["host", 1, None]]},  # thread not an int
+            {EXCLUSIVE: [[get_process_id()[0], 0, 0]]},  # pid out of range (local host: the pid gets checked)
+            {EXCLUSIVE: [[get_process_id()[0], 2**40, 0]]},  # dito
+        ],
+    )
+    @pytest.mark.parametrize("kill_stale_locks", [True, False])
+    def test_malformed(self, rosterpath, data, kill_stale_locks):
+        # a malformed roster file is treated like a corrupt one: as empty.
+        rosterpath.write_text(json.dumps(data))
+        roster = LockRoster(rosterpath, id=ID1)
+        roster.kill_stale_locks = kill_stale_locks
+        assert roster.load() == {}
+        assert roster.get(SHARED) == roster.get(EXCLUSIVE) == set()
+        roster.modify(SHARED, ADD)
+        assert roster.get(SHARED) == {ID1}
diff --git a/src/borg/testsuite/storelocking_test.py b/src/borg/testsuite/storelocking_test.py
index cbbfefd99b..de24c46026 100644
--- a/src/borg/testsuite/storelocking_test.py
+++ b/src/borg/testsuite/storelocking_test.py
@@ -463,6 +463,34 @@ def test_migrate_lock(self, lockstore):
         assert old_locks[0]["hostid"] == old_id[0]
         assert new_locks[0]["hostid"] == new_id[0]
 
+    def test_release_with_leftover_lock_of_same_id(self, lockstore):
+        # a not yet stale leftover lock with our id (e.g. pid reuse in a container with a fixed
+        # BORG_HOST_ID) is not ours: release() deletes only our lock and leaves the leftover alone.
+        now = datetime.datetime.now(datetime.UTC)
+        leftover_key = write_raw_lock(lockstore, ID1, exclusive=False, dt=now)
+        lock = Lock(lockstore, exclusive=False, id=ID1).acquire()
+        assert lock.my_lock_key != leftover_key
+        lock.release()
+        assert [info.name for info in lockstore.list("locks")] == [leftover_key]
+        with pytest.raises(NotLocked):
+            lock.release()
+
+    def test_migrate_lock_with_leftover_lock_of_same_id(self, lockstore):
+        now = datetime.datetime.now(datetime.UTC)
+        leftover_key = write_raw_lock(lockstore, ID1, exclusive=False, dt=now)
+        lock = Lock(lockstore, exclusive=False, id=ID1).acquire()
+        lock.migrate_lock(ID1, ID2)
+        keys = {info.name for info in lockstore.list("locks")}
+        assert keys == {leftover_key, lock.my_lock_key}
+        lock.release()
+        assert [info.name for info in lockstore.list("locks")] == [leftover_key]
+
+    def test_migrate_killed_lock(self, lockstore):
+        lock = Lock(lockstore, exclusive=True, id=ID1).acquire()
+        Lock(lockstore, id=ID2).break_lock()  # kill it, like another client would
+        with pytest.raises(LockTimeout):
+            lock.migrate_lock(ID1, ID2)
+
 
 def test_lock_object_is_sealed(lockstore):
     # neither the content nor the name of a lock object tells who uses the repository, see "Lock objects".
@@ -528,6 +556,32 @@ def foreign_key_lock():
     return other_key.encrypt_oneshot(b"", json.dumps(lock).encode(), aad=LOCK_AAD)
 
 
+def malformed_lock(**changes):
+    # a correctly sealed (authenticated) lock object with malformed content.
+    host, pid, thread = get_process_id()  # a local, alive owner: the owner fields get checked
+    now = datetime.datetime.now(datetime.UTC).isoformat(timespec="milliseconds")
+    lock = dict(exclusive=True, hostid=host, processid=pid, threadid=thread, time=now)
+    lock.update(changes)
+    return seal(json.dumps(lock).encode())
+
+
+MALFORMED_LOCKS = [
+    dict(hostid=1),
+    dict(hostid=None),
+    dict(processid="1"),
+    dict(processid=True),
+    dict(processid=1.5),
+    dict(processid=0),
+    dict(processid=-1),
+    dict(processid=2**40),
+    dict(threadid="0"),
+    dict(threadid=False),
+    dict(threadid=None),
+    dict(time="2026-10-13T12:00:00.000"),  # no timezone
+    dict(time=0),
+]
+
+
 class TestUnreadableLock:
     @pytest.mark.parametrize("make_content", [plaintext_lock, foreign_key_lock, lambda: b"garbage"])
     @pytest.mark.parametrize("exclusive", [False, True])
@@ -586,3 +640,23 @@ def load_denied(name, *args, **kwargs):
         assert [info.name for info in lockstore.list("locks")] == [key]  # no lock of ours left behind
         Lock(lockstore, id=ID1).break_lock()
         assert list(lockstore.list("locks")) == []
+
+    @pytest.mark.parametrize("changes", MALFORMED_LOCKS)
+    def test_malformed(self, lockstore, changes):
+        # authenticated does not mean well-formed: fail closed, treat it as an unreadable exclusive lock.
+        key = write_unreadable_lock(lockstore, malformed_lock(**changes))
+        lock = Lock(lockstore, exclusive=False, id=ID1, timeout=0.2)
+        lock.retry_delay_min = lock.retry_delay_max = 0.05
+        locks = lock._get_locks()
+        assert locks[key]["unreadable"] and locks[key]["exclusive"]
+        with pytest.raises(LockTimeout, match=f"unreadable lock object locks/{key}"):
+            lock.acquire()
+        # break-lock can remove it.
+        Lock(lockstore, id=ID1).break_lock()
+        assert list(lockstore.list("locks")) == []
+
+    def test_wellformed(self, lockstore):
+        # the reference for test_malformed: without changes, the lock object is readable.
+        key = write_unreadable_lock(lockstore, malformed_lock())
+        locks = Lock(lockstore, id=ID1)._get_locks()
+        assert not locks[key]["unreadable"]

From 14e730872852217fe34458b6fb30d6b138480b89 Mon Sep 17 00:00:00 2001
From: Thomas Waldmann <tw@waldmann-edv.de>
Date: Tue, 13 Oct 2026 01:57:35 +0200
Subject: [PATCH 5/6] key/legacy: do not assert on key material and borg 1.x
 data, refs #8649

key:
- an empty borg 1.x manifest raises IntegrityError.
- repo-create --copy-crypt-key: a crypt key of the other repository that
  the new encryption mode can not use (e.g. the 32 + 128 bytes key of a
  borg 1.x blake2 mode for authenticated-sha256) raises Error.
- legacy AES-CTR keys: the key class is selected by the unauthenticated
  type byte of the borg 1.x manifest, so check that the key material fits
  the crypto suite (IntegrityError).
- loading a key with invalid crypt_key / id_key lengths raises
  KeyfileInvalidError.

legacy:
- borg serve --from-borg1: if the client closes the connection without
  calling close() (borg 1.x clients do that), close the repository like
  borg 1.4 did instead of failing with an AssertionError.
- the legacy remote checks that "get" results from a borg 1.x server are
  bytes.
- transfer: borg 1.x items with None values or without the required keys
  raise IntegrityError.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/crypto/key.py                      | 29 +++++++++--
 src/borg/legacy/remote.py                   | 10 +++-
 src/borg/legacy/upgrade.py                  | 13 +++--
 src/borg/testsuite/crypto/key_test.py       | 54 ++++++++++++++++++++-
 src/borg/testsuite/legacy_upgrade_test.py   | 19 ++++++++
 src/borg/testsuite/legacyrepository_test.py | 20 +++++---
 6 files changed, 128 insertions(+), 17 deletions(-)

diff --git a/src/borg/crypto/key.py b/src/borg/crypto/key.py
index d2c0a8dc1f..a608db80e1 100644
--- a/src/borg/crypto/key.py
+++ b/src/borg/crypto/key.py
@@ -296,7 +296,8 @@ def legacy_key_factory(repository, manifest_chunk, *, other=False):
     crypted data is the key type byte (see identify_key).
     """
     manifest_data = RepoObj1.extract_crypted_data(manifest_chunk)
-    assert manifest_data, "manifest data must not be zero bytes long"
+    if not manifest_data:
+        raise IntegrityError("Manifest: empty object")
     key_cls = identify_key(manifest_data)
     key = key_cls.detect(repository, manifest_data, other=other)
     key.stored_type = manifest_data[0]
@@ -513,6 +514,9 @@ class AESKeyBase(KeyBase):
 
     CIPHERSUITE: Callable = None  # override in derived class
 
+    # crypt_key length (enc key + MAC key) per CIPHERSUITE, see init_ciphers.
+    CRYPT_KEY_LENGTH_FOR_SUITE = {low_level.AES256_CTR_HMAC_SHA256: 32 + 32, low_level.AES256_CTR_BLAKE2b: 32 + 128}
+
     logically_encrypted = True
 
     def encrypt(self, id, data, aad=b""):
@@ -544,6 +548,10 @@ def init_from_random_data(self):
         self.init_from_given_data(crypt_key=data[0:64], id_key=data[64:96], chunk_seed=chunk_seed)
 
     def init_ciphers(self, manifest_data=None):
+        # the key class is selected by the unauthenticated key type byte of the borg 1.x manifest (see
+        # identify_key), so the key material loaded from the borg key may belong to another crypto suite.
+        if len(self.crypt_key) != self.CRYPT_KEY_LENGTH_FOR_SUITE[self.CIPHERSUITE]:
+            raise IntegrityError("Manifest: key type does not match the key material of the borg key")
         enc_key, enc_hmac_key = self.crypt_key[0:32], self.crypt_key[32:]
         self.cipher = self.CIPHERSUITE(mac_key=enc_hmac_key, enc_key=enc_key, header_len=1, aad_offset=1)
         if manifest_data is None:
@@ -567,6 +575,7 @@ class FlexiKey:
     # have to infer their types across the module import cycle.
     chunk_seed: int
     crypt_key: bytes
+    CRYPT_KEY_LENGTHS: ClassVar[tuple[int, ...]]  # the crypt_key lengths init_from_given_data accepts
     empty_passphrase: bool
     id_key: bytes
     storage: str
@@ -617,6 +626,9 @@ def _load(self, key_data, passphrase):
             key = Key(internal_dict=data)
             if key.version not in (1, 2):  # legacy: item.Key can still process v1 keys
                 raise UnsupportedKeyFormatError()
+            # the key lengths of all modes borg creates and of all borg 1.x modes (blake2: 32 + 128 and 128).
+            if len(key.crypt_key) not in (32 + 32, 32 + 128) or len(key.id_key) not in (32, 128):
+                raise KeyfileInvalidError(self.repository._location.canonical_path(), "(repokey)")
             self.repository_id = key.repository_id
             self.crypt_key = key.crypt_key
             self.id_key = key.id_key
@@ -736,6 +748,13 @@ def create(cls, repository, args, *, other_key=None):
             if other_key.copy_crypt_key:
                 # give the user the option to use the same authenticated encryption (AE) key
                 crypt_key = other_key.crypt_key
+                if len(crypt_key) not in key.CRYPT_KEY_LENGTHS:
+                    # e.g. the 32 + 128 bytes crypt key of a borg 1.x blake2 mode for an authenticated-* mode.
+                    lengths = " or ".join(str(length) for length in key.CRYPT_KEY_LENGTHS)
+                    raise Error(
+                        f"Copying the crypt key ({len(crypt_key)} bytes) to {key.ENC_NAME} is not possible, "
+                        f"it needs a crypt key of {lengths} bytes."
+                    )
             else:
                 # borg transfer re-encrypts all data anyway, thus we can default to a new, random AE key
                 crypt_key = os.urandom(64)
@@ -1189,6 +1208,8 @@ class AuthenticatedKeyBase(MACKeyBase, FlexiKey):
 
     _tag_key: bytes | None = None  # cache for the tag_key property
 
+    CRYPT_KEY_LENGTHS = (32 + 32,)  # see init_from_given_data
+
     @property
     def tag_key(self):
         # derived lazily (and cached): a key instance may also be built by copying the key
@@ -1244,7 +1265,7 @@ def save(self, target, passphrase, algorithm, create=False, label=None, replace=
         self.logically_encrypted = False
 
     def init_from_given_data(self, *, crypt_key, id_key, chunk_seed):
-        assert len(crypt_key) == 32 + 32
+        assert len(crypt_key) in self.CRYPT_KEY_LENGTHS
         assert len(id_key) == 32
         assert isinstance(chunk_seed, int)
         self.crypt_key = crypt_key
@@ -1341,6 +1362,8 @@ class AEADKeyBase(KeyBase):
 
     MAX_IV = 2**48 - 1
 
+    CRYPT_KEY_LENGTHS = (32 + 32, 32 + 128)  # see init_from_given_data
+
     # Maximum amount of data (in 128bit cipher blocks, counting plaintext **and** AAD) that we
     # encrypt using one session key. When this is exceeded, we just start a new session (see
     # new_session): that is cheap (one sha256) and fully transparent, because the sessionID
@@ -1432,7 +1455,7 @@ def decrypt(self, id, data, aad=b""):
             raise IntegrityError(f"Chunk {bin_to_hex(id)}: Could not decrypt [{str(e)}]")
 
     def init_from_given_data(self, *, crypt_key, id_key, chunk_seed):
-        assert len(crypt_key) in (32 + 32, 32 + 128)
+        assert len(crypt_key) in self.CRYPT_KEY_LENGTHS
         assert len(id_key) in (32, 128)
         assert isinstance(chunk_seed, int)
         self.crypt_key = crypt_key
diff --git a/src/borg/legacy/remote.py b/src/borg/legacy/remote.py
index 262621ef14..87b265cfac 100644
--- a/src/borg/legacy/remote.py
+++ b/src/borg/legacy/remote.py
@@ -617,7 +617,11 @@ def get(self, id, read_data=True, raise_missing=True):
 
     def get_many(self, ids, read_data=True, raise_missing=True):
         # note: legacy remote protocol does not support raise_missing parameter, so we ignore it here
-        yield from self.call_many("get", [{"id": id, "read_data": read_data} for id in ids])
+        for result in self.call_many("get", [{"id": id, "read_data": read_data} for id in ids]):
+            # for read_data=False, LegacyRepository.get returns None (a borg 1.x server ignores read_data).
+            if not isinstance(result, bytes) and not (result is None and not read_data):
+                raise UnexpectedRPCDataFormatFromServer(f"get result is {type(result).__name__}, not bytes".encode())
+            yield result
 
     def gather_many(self, ids, raise_missing=True):
         # a legacy repository has no packs, so this is just get_many (same interface as Repository).
@@ -741,7 +745,9 @@ def inner_serve():
 
                 if shutdown_serve:
                     # shutdown wanted! get out of here after sending all log output.
-                    assert self.repository is None
+                    if self.repository is not None:
+                        # the client closed the connection without calling close() (borg 1.x clients do that).
+                        self.close()
                     return
 
                 # process new RPCs
diff --git a/src/borg/legacy/upgrade.py b/src/borg/legacy/upgrade.py
index b059d4fb7f..338f2639e0 100644
--- a/src/borg/legacy/upgrade.py
+++ b/src/borg/legacy/upgrade.py
@@ -3,7 +3,7 @@
 
 from ..constants import REQUIRED_ITEM_KEYS, CH_BUZHASH
 from ..compress import ZLIB, ZLIB_legacy, ObfuscateSize
-from ..helpers import HardLinkManager, join_cmd
+from ..helpers import HardLinkManager, IntegrityError, join_cmd
 from ..item import Item
 from ..logger import create_logger
 from .helpers import borg1_hardlink_with_content, borg1_hardlink_reference
@@ -72,10 +72,17 @@ def upgrade_item(self, *, item):
         for key in "user", "group":
             if key in new_item_dict and new_item_dict[key] is None:
                 del new_item_dict[key]
-        assert not any(value is None for value in new_item_dict.values()), f"found None value in {new_item_dict}"
+        # the item comes from a borg 1.x archive, so it might be malformed.
+        none_keys = sorted(key for key, value in new_item_dict.items() if value is None)
+        if none_keys:
+            where = f"Archive {self.archive.name}, item {item_dict.get('path')!r}"
+            raise IntegrityError(f"{where}: None value for {', '.join(none_keys)}")
         new_item = Item(internal_dict=new_item_dict)
         new_item.get_size(memorize=True)  # if not already present: compute+remember size for items with chunks
-        assert all(key in new_item for key in REQUIRED_ITEM_KEYS)
+        missing_keys = sorted(key for key in REQUIRED_ITEM_KEYS if key not in new_item)
+        if missing_keys:
+            where = f"Archive {self.archive.name}, item {item_dict.get('path')!r}"
+            raise IntegrityError(f"{where}: missing required keys {', '.join(missing_keys)}")
         return new_item
 
     def upgrade_compressed_chunk(self, meta, data):
diff --git a/src/borg/testsuite/crypto/key_test.py b/src/borg/testsuite/crypto/key_test.py
index 10682b8552..7c76a216ac 100644
--- a/src/borg/testsuite/crypto/key_test.py
+++ b/src/borg/testsuite/crypto/key_test.py
@@ -15,7 +15,8 @@
 from ...crypto.key import AES_OCB_MAX_SESSION_BLOCKS
 from ...crypto.key import ID_HMAC_SHA_256, ID_BLAKE2b_256, ID_BLAKE3_256
 from ...crypto.key import UnsupportedManifestError, UnsupportedKeyFormatError
-from ...crypto.key import RepoKeyNotFoundError
+from ...crypto.key import RepoKeyNotFoundError, KeyfileInvalidError
+from ...crypto.key import legacy_key_factory
 from ...crypto.key import identify_key, key_class_for, key_class_of, key_factory, RepositoryKeyInfoMissing
 from ...crypto.key import AVAILABLE_KEY_TYPES
 from ...crypto.low_level import IntegrityError as IntegrityErrorBase
@@ -23,7 +24,7 @@
 from ...helpers import IntegrityError
 from ...helpers import Location
 from ...helpers import msgpack
-from ...constants import KEY_ALGORITHMS, KeyBlobStorage
+from ...constants import KEY_ALGORITHMS, KeyBlobStorage, KeyType
 from ...helpers import hex_to_bin, bin_to_hex
 
 
@@ -191,6 +192,37 @@ def test_keyfile_blake2(self, monkeypatch, keys_dir):
         key = Blake2AESCTRKey.detect(self.MockRepository(), self.keyfile_blake2_cdata)
         assert key.decrypt(self.keyfile_blake2_id, self.keyfile_blake2_cdata) == b"payload"
 
+    @pytest.mark.parametrize(
+        "key_file, cdata, cls, type_byte",
+        (
+            ("keyfile2_key_file", "keyfile2_cdata", Blake2AESCTRKey, KeyType.BLAKE2KEYFILE),
+            ("keyfile_blake2_key_file", "keyfile_blake2_cdata", AESCTRKey, KeyType.KEYFILE),
+        ),
+    )
+    def test_legacy_key_type_mismatch(self, key_file, cdata, cls, type_byte, monkeypatch, keys_dir):
+        # the key class is selected by the unauthenticated key type byte of a borg 1.x manifest,
+        # a wrong one selects a key class that does not match the key material in the borg key.
+        with keys_dir.join("keyfile").open("w") as fd:
+            fd.write(getattr(self, key_file))
+        monkeypatch.setenv("BORG_PASSPHRASE", "passphrase")
+        manifest_data = bytes([type_byte]) + getattr(self, cdata)[1:]
+        with pytest.raises(IntegrityError):
+            cls.detect(self.MockRepository(), manifest_data)
+
+    def test_copy_crypt_key_length_mismatch(self, monkeypatch, keys_dir):
+        with keys_dir.join("keyfile").open("w") as fd:
+            fd.write(self.keyfile_blake2_key_file)
+        monkeypatch.setenv("BORG_PASSPHRASE", "passphrase")
+        other_key = Blake2AESCTRKey.detect(self.MockRepository(), self.keyfile_blake2_cdata)
+        other_key.copy_crypt_key = True
+        assert len(other_key.crypt_key) == 32 + 128
+        # the authenticated-* modes need a crypt key of 32 + 32 bytes.
+        with pytest.raises(Error, match="Copying the crypt key"):
+            AuthenticatedKey.create(self.MockRepository(), self.MockArgs(), other_key=other_key)
+        # the AEAD modes accept both lengths.
+        key = AESOCBKey.create(self.MockRepository(), self.MockArgs(), other_key=other_key)
+        assert key.crypt_key == other_key.crypt_key
+
     def test_legacy_named_keyfile_still_loads(self, monkeypatch, keys_dir):
         monkeypatch.setenv("BORG_PASSPHRASE", "test")
         key = CHPOKey.create(self.MockRepository(), self.kf_args())
@@ -546,6 +578,24 @@ def test_legacy_authenticated_no_key_key_gone(cls, monkeypatch, tmp_path):
     assert bytes(key.decrypt(None, envelope)) == payload
 
 
+def test_legacy_key_factory_empty_manifest():
+    with pytest.raises(IntegrityError, match="empty"):
+        legacy_key_factory(MagicMock(), b"")
+
+
+@pytest.mark.parametrize("crypt_key_len, id_key_len", ((32, 32), (64, 64), (160, 0)))
+def test_load_invalid_key_lengths(crypt_key_len, id_key_len):
+    repository = TestKey.MockRepository()
+    key = CHPOKey(repository)
+    key.repository_id = repository.id
+    key.crypt_key = bytes(crypt_key_len)
+    key.id_key = bytes(id_key_len)
+    key.chunk_seed = 0
+    key_data = key._save("passphrase", algorithm=KEY_ALGORITHMS["argon2"])
+    with pytest.raises(KeyfileInvalidError):
+        CHPOKey(repository)._load(key_data, "passphrase")
+
+
 def test_decrypt_key_file_unsupported_algorithm():
     """We will add more algorithms in the future. We should raise a helpful error."""
     key = CHPOKey(None)
diff --git a/src/borg/testsuite/legacy_upgrade_test.py b/src/borg/testsuite/legacy_upgrade_test.py
index b9884dbca8..4c245a1a61 100644
--- a/src/borg/testsuite/legacy_upgrade_test.py
+++ b/src/borg/testsuite/legacy_upgrade_test.py
@@ -6,8 +6,11 @@
 from struct import Struct
 from unittest.mock import MagicMock
 
+import pytest
+
 from ..compress import ZLIB, ObfuscateSize
 from ..constants import CH_BUZHASH, REQUIRED_ITEM_KEYS
+from ..helpers import IntegrityError
 from ..item import Item
 from ..legacy.upgrade import UpgraderFrom12To20
 
@@ -114,6 +117,22 @@ def test_required_item_keys_always_present():
     assert all(k in result for k in REQUIRED_ITEM_KEYS)
 
 
+def test_missing_required_item_key_raises_integrity_error():
+    u, _, archive = _upgrader()
+    archive.name = "archive1"
+    item = Item(internal_dict={"path": "dir/file", "mode": stat.S_IFREG | 0o644})  # no mtime
+    with pytest.raises(IntegrityError, match="archive1.*dir/file.*missing required keys mtime"):
+        u.upgrade_item(item=item)
+
+
+def test_none_value_raises_integrity_error():
+    u, _, archive = _upgrader()
+    archive.name = "archive1"
+    item = _item(uid=None)
+    with pytest.raises(IntegrityError, match="archive1.*dir/file.*None value for uid"):
+        u.upgrade_item(item=item)
+
+
 # ── upgrade_archive_metadata ──────────────────────────────────────────────────
 
 
diff --git a/src/borg/testsuite/legacyrepository_test.py b/src/borg/testsuite/legacyrepository_test.py
index 8e5b92cc1a..2058814d95 100644
--- a/src/borg/testsuite/legacyrepository_test.py
+++ b/src/borg/testsuite/legacyrepository_test.py
@@ -13,6 +13,7 @@
 from ..fslocking import Lock, LockFailed
 from ..platformflags import is_win32
 from ..legacy.remote import LegacyRemoteRepository, InvalidRPCMethod, bytes_from_borg1
+from ..legacy.remote import UnexpectedRPCDataFormatFromServer
 from ..manifest import Manifest
 from ..legacy.repository import LegacyRepository, LoggedIO
 from ..legacy.repository import MAGIC, MAX_DATA_SIZE, TAG_DELETE, TAG_PUT, TAG_COMMIT
@@ -1011,13 +1012,7 @@ def packb(o):
 )
 setup_logging(is_serve=True)
 server = remote.RepositoryServer(restrict_to_paths=None, restrict_to_repositories=None)
-try:
-    server.serve()
-except AssertionError:
-    # the legacy server expects a close call before stdin gets closed, borg 1.x did not.
-    if server.repository is None:
-        raise
-    server.repository.close()
+server.serve()
 """
 
 
@@ -1051,3 +1046,14 @@ def test_bytes_from_borg1():
     unpacked = msgpack.unpackb(msgpack.packb(value, use_bin_type=False))
     assert all(isinstance(v, str) for v in unpacked)
     assert bytes_from_borg1(unpacked) == value
+
+
+def test_get_many_rejects_non_bytes_from_server(remote_repository, monkeypatch):
+    # the server must send bytes for "get", anything else is an unexpected RPC data format.
+    with remote_repository:
+        with monkeypatch.context() as m:
+            m.setattr(remote_repository, "call_many", lambda cmd, calls: iter([b"data", 42]))
+            results = remote_repository.get_many([H(1), H(2)])
+            assert next(results) == b"data"
+            with pytest.raises(UnexpectedRPCDataFormatFromServer):
+                next(results)

From 431de4892e89c9cf52609251fc79ebaf3582f03c Mon Sep 17 00:00:00 2001
From: Thomas Waldmann <tw@waldmann-edv.de>
Date: Tue, 13 Oct 2026 01:57:47 +0200
Subject: [PATCH 6/6] cli: validate user input instead of asserting later, refs
 #8649

- ChunkerParams: validate list/tuple input (e.g. a YAML list from a config
  file) like the string form. buzhash: require window_size >= 1 and
  window_size + 2^chunk_min + 1 <= 2^chunk_max, buzhash64: require
  window_size >= 1. Such params made the chunker assert, crash (SIGBUS)
  or hang.
- FilesCacheMode: reject a size-only mode, the files cache needs ctime or
  mtime (or disabled).
- create: the slashdot hack pointing to the root directory (//./, ///./)
  works like /./ instead of failing with an AssertionError.
- recreate without --target: check the archive names from the repository
  before changing anything.
- benchmark crud: a failing inner borg command raises CommandError.
- broken version metadata raises RuntimeError.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/borg/__init__.py                          | 11 ++--
 src/borg/archiver/benchmark_cmd.py            | 16 +++--
 src/borg/archiver/create_cmd.py               | 21 +++++--
 src/borg/archiver/recreate_cmd.py             | 11 +++-
 src/borg/helpers/parseformat.py               | 13 +++-
 .../testsuite/archiver/benchmark_cmd_test.py  | 11 ++++
 .../testsuite/archiver/create_cmd_test.py     | 16 +++++
 .../testsuite/helpers/parseformat_test.py     | 61 +++++++++++++++++++
 8 files changed, 141 insertions(+), 19 deletions(-)

diff --git a/src/borg/__init__.py b/src/borg/__init__.py
index 9b38c42b98..e3def430b7 100644
--- a/src/borg/__init__.py
+++ b/src/borg/__init__.py
@@ -5,16 +5,17 @@
 
 __version_tuple__ = parse_version(__version__).release
 
-# assert that all semver components are integers
+# check that all semver components are integers
 # this is mainly to show errors when people repackage poorly
 # and setuptools_scm determines a 0.1.dev... version
-assert not __version__.startswith("0.1.dev") and all(isinstance(v, int) for v in __version_tuple__), (
-    """\
+if __version__.startswith("0.1.dev") or not all(isinstance(v, int) for v in __version_tuple__):
+    raise RuntimeError(
+        """\
 Broken BorgBackup version metadata: %r
 
 Version metadata is obtained dynamically during installation via setuptools_scm;
 please ensure your Git repository has the correct tags, or provide the version
 using SETUPTOOLS_SCM_PRETEND_VERSION in your build script.
 """
-    % __version__
-)
+        % __version__
+    )
diff --git a/src/borg/archiver/benchmark_cmd.py b/src/borg/archiver/benchmark_cmd.py
index fed447cb2f..5f8bc95172 100644
--- a/src/borg/archiver/benchmark_cmd.py
+++ b/src/borg/archiver/benchmark_cmd.py
@@ -22,6 +22,12 @@ class BenchmarkMixIn:
     def do_benchmark_crud(self, args):
         """Benchmark Create, Read, Update, Delete for archives."""
 
+        def check_rc(step, *rcs):
+            # the inner borg commands work on the user's repository: they can fail without a bug in borg.
+            for rc in rcs:
+                if rc != 0:
+                    raise CommandError(f"benchmark crud: {step} failed with rc {rc}.")
+
         def measurement_run(repo, path):
             # Suppress "Done. Run borg compact..." warnings from internal do_delete() calls —
             # they clutter benchmark output and are irrelevant here (repo is temporary).
@@ -48,7 +54,7 @@ def measurement_run(repo, path):
                 )
                 t_end = time.monotonic()
                 dt_create = t_end - t_start
-                assert rc == 0
+                check_rc("create (files cache disabled)", rc)
                 # now build files cache
                 rc1 = get_reset_ec(
                     self.do_create(
@@ -58,7 +64,7 @@ def measurement_run(repo, path):
                 rc2 = get_reset_ec(
                     self.do_delete(self.parse_args([f"--repo={repo}", "delete", "-a", "borg-benchmark-crud2"]))
                 )
-                assert rc1 == rc2 == 0
+                check_rc("create/delete (building the files cache)", rc1, rc2)
                 # measure a no-change update (archive1 is still present)
                 t_start = time.monotonic()
                 rc1 = get_reset_ec(
@@ -71,7 +77,7 @@ def measurement_run(repo, path):
                 rc2 = get_reset_ec(
                     self.do_delete(self.parse_args([f"--repo={repo}", "delete", "-a", "borg-benchmark-crud3"]))
                 )
-                assert rc1 == rc2 == 0
+                check_rc("create/delete (no-change update)", rc1, rc2)
                 # measure extraction (dry-run: without writing result to disk)
                 t_start = time.monotonic()
                 rc = get_reset_ec(
@@ -79,7 +85,7 @@ def measurement_run(repo, path):
                 )
                 t_end = time.monotonic()
                 dt_extract = t_end - t_start
-                assert rc == 0
+                check_rc("extract", rc)
                 # measure archive deletion (of LAST present archive with the data)
                 t_start = time.monotonic()
                 rc = get_reset_ec(
@@ -87,7 +93,7 @@ def measurement_run(repo, path):
                 )
                 t_end = time.monotonic()
                 dt_delete = t_end - t_start
-                assert rc == 0
+                check_rc("delete", rc)
                 return dt_create, dt_update, dt_extract, dt_delete
             finally:
                 archiver_logger.setLevel(original_level)
diff --git a/src/borg/archiver/create_cmd.py b/src/borg/archiver/create_cmd.py
index b1112dce2d..e1d19d6caf 100644
--- a/src/borg/archiver/create_cmd.py
+++ b/src/borg/archiver/create_cmd.py
@@ -44,6 +44,20 @@
 logger = create_logger()
 
 
+def strip_prefix_and_path(path):
+    """
+    Return (strip_prefix, path) for a path given by the user: the slashdot hack strip prefix (see
+    get_strip_prefix) and the normalized path.
+    """
+    strip_prefix = get_strip_prefix(path)
+    path = posixpath.normpath(path)
+    if strip_prefix is not None and not path.strip("/"):
+        # the slashdot hack points to the root directory, e.g. //./ (normpath keeps a leading "//"):
+        # like for /./, there is nothing to strip.
+        strip_prefix = None
+    return strip_prefix, path
+
+
 def stat_root(path):
     """
     stat a recursion root, following it if it is a symlink, see #4737.
@@ -223,9 +237,7 @@ def create_inner(archive, cache, fso):
                     pipe_bin = sys.stdin.buffer
                 pipe = TextIOWrapper(pipe_bin, errors="surrogateescape")
                 for path in iter_separated(pipe, paths_sep):
-                    path = slashify(path)
-                    strip_prefix = get_strip_prefix(path)
-                    path = posixpath.normpath(path)
+                    strip_prefix, path = strip_prefix_and_path(slashify(path))
                     try:
                         with backup_io("stat"):
                             # symlinks given this way are never followed, see #4737.
@@ -280,8 +292,7 @@ def create_inner(archive, cache, fso):
                             fso.stats.files_stats[status] += 1
                         continue
 
-                    strip_prefix = get_strip_prefix(path)
-                    path = posixpath.normpath(path)
+                    strip_prefix, path = strip_prefix_and_path(path)
                     try:
                         with backup_io("stat"):
                             st, followed = stat_root(path)
diff --git a/src/borg/archiver/recreate_cmd.py b/src/borg/archiver/recreate_cmd.py
index 621784b267..6fabdb3da0 100644
--- a/src/borg/archiver/recreate_cmd.py
+++ b/src/borg/archiver/recreate_cmd.py
@@ -3,8 +3,8 @@
 from ..archive import ArchiveRecreater
 from ..constants import *  # NOQA
 from ..helpers import archivename_validator, comment_validator, PathSpec, ChunkerParams, bin_to_hex, CompressionSpec
-from ..helpers import timestamp
-from ..helpers.argparsing import ArgumentParser
+from ..helpers import timestamp, Error
+from ..helpers.argparsing import ArgumentParser, ArgumentTypeError
 
 from ..logger import create_logger
 
@@ -38,6 +38,13 @@ def do_recreate(self, args, repository, manifest, cache):
         )
         archive_infos = manifest.archives.list_considering(args)
         archive_infos = [ai for ai in archive_infos if "@PROT" not in ai.tags]
+        if not args.target:
+            # the recreated archives keep their names from the repository: check them before changing anything.
+            for archive_info in archive_infos:
+                try:
+                    archivename_validator(archive_info.name)
+                except ArgumentTypeError as err:
+                    raise Error(f"{err}, please rename the archive before recreate.") from None
         for archive_info in archive_infos:
             if recreater.is_temporary_archive(archive_info.name):
                 continue
diff --git a/src/borg/helpers/parseformat.py b/src/borg/helpers/parseformat.py
index 789232a5d7..12476c6036 100644
--- a/src/borg/helpers/parseformat.py
+++ b/src/borg/helpers/parseformat.py
@@ -295,7 +295,8 @@ def __str__(self):
 
 def ChunkerParams(s):
     if isinstance(s, (list, tuple)):
-        return tuple(s)
+        # e.g. the default value or a YAML list from a config file: validate it like the string form.
+        s = ",".join(str(p) for p in s)
     params = s.strip().split(",")
     count = len(params)
     if count == 0:
@@ -336,6 +337,8 @@ def ChunkerParams(s):
             raise ArgumentTypeError("min. chunk size exponent must not be less than 6 (2^6 = 64B min. chunk size)")
         if chunk_max > 23:
             raise ArgumentTypeError("max. chunk size exponent must not be more than 23 (2^23 = 8MiB max. chunk size)")
+        if window_size < 1:
+            raise ArgumentTypeError("window_size must be at least 1")
         # the chunker needs the hash window plus one byte beyond the min. chunk size to fit into
         # a max. chunk size sized buffer, see the assertion in ChunkerBuzHash64.__cinit__.
         if window_size + 2**chunk_min + 1 > 2**chunk_max:
@@ -409,8 +412,14 @@ def ChunkerParams(s):
             raise ArgumentTypeError("min. chunk size exponent must not be less than 6 (2^6 = 64B min. chunk size)")
         if chunk_max > 23:
             raise ArgumentTypeError("max. chunk size exponent must not be more than 23 (2^23 = 8MiB max. chunk size)")
+        if window_size < 1:
+            raise ArgumentTypeError("window_size must be at least 1")
         if window_size % 2 == 0:
             raise ArgumentTypeError("window_size must be an uneven (odd) number")
+        # the chunker needs the hash window plus one byte beyond the min. chunk size to fit into
+        # a max. chunk size sized buffer, see the assertion in Chunker.__cinit__ (buzhash.pyx).
+        if window_size + 2**chunk_min + 1 > 2**chunk_max:
+            raise ArgumentTypeError("required: window_size + 2^chunk_min + 1 <= 2^chunk_max")
         return CH_BUZHASH, chunk_min, chunk_max, chunk_mask, window_size
     raise ArgumentTypeError("invalid chunker params")
 
@@ -431,7 +440,7 @@ def DigestAlgos(s):
 
 def FilesCacheMode(s):
     ENTRIES_MAP = dict(ctime="c", mtime="m", size="s", inode="i", rechunk="r", disabled="d")
-    VALID_MODES = ("cis", "ims", "cs", "ms", "cr", "mr", "d", "s")  # letters in alpha order
+    VALID_MODES = ("cis", "ims", "cs", "ms", "cr", "mr", "d")  # letters in alpha order
     if s in VALID_MODES:
         return s
     entries = set(s.strip().split(","))
diff --git a/src/borg/testsuite/archiver/benchmark_cmd_test.py b/src/borg/testsuite/archiver/benchmark_cmd_test.py
index 088e188918..7bc50a6c7e 100644
--- a/src/borg/testsuite/archiver/benchmark_cmd_test.py
+++ b/src/borg/testsuite/archiver/benchmark_cmd_test.py
@@ -3,6 +3,7 @@
 
 import pytest
 
+from ...archiver import Archiver
 from ...constants import *  # NOQA
 from ...helpers import CommandError
 from . import cmd, RK_ENCRYPTION
@@ -18,6 +19,16 @@ def test_benchmark_crud(archiver, monkeypatch):
     assert "MB/s" in output
 
 
+def test_benchmark_crud_failing_command(archiver, monkeypatch):
+    # an inner borg command failing on the user's repository is an error, not a borg bug.
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
+    monkeypatch.setattr(Archiver, "do_extract", lambda self, args: EXIT_WARNING)
+    # fork=False, so the monkeypatched do_extract is the one the benchmark uses
+    with pytest.raises(CommandError, match="extract failed with rc 1"):
+        cmd(archiver, "benchmark", "crud", archiver.input_path, fork=False)
+
+
 def test_benchmark_crud_json_lines(archiver, monkeypatch):
     cmd(archiver, "repo-create", RK_ENCRYPTION)
     monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
diff --git a/src/borg/testsuite/archiver/create_cmd_test.py b/src/borg/testsuite/archiver/create_cmd_test.py
index 3fdcc6552a..f337b03a6f 100644
--- a/src/borg/testsuite/archiver/create_cmd_test.py
+++ b/src/borg/testsuite/archiver/create_cmd_test.py
@@ -14,6 +14,7 @@
 
 from ... import platform
 from ...constants import *  # NOQA
+from ...archiver.create_cmd import strip_prefix_and_path
 from ...constants import zeros
 from ...manifest import Manifest
 from ...platform import is_win32, is_cygwin
@@ -1390,6 +1391,21 @@ def test_create_dotslash_hack_root_metadata(archivers, request):
     assert "first" not in output
 
 
+@pytest.mark.parametrize(
+    "path, expected",
+    [
+        ("input/first/./second", ("input/first/", "input/first/second")),
+        ("/./", (None, "/")),
+        # the slashdot hack pointing to the root directory strips nothing, like /./ does.
+        ("//./", (None, "//")),  # normpath keeps a leading "//"
+        ("///./", (None, "/")),
+        ("//./etc", ("//", "//etc")),
+    ],
+)
+def test_strip_prefix_and_path(path, expected):
+    assert strip_prefix_and_path(path) == expected
+
+
 def test_log_json(archivers, request):
     archiver = request.getfixturevalue(archivers)
     create_test_files(archiver.input_path)
diff --git a/src/borg/testsuite/helpers/parseformat_test.py b/src/borg/testsuite/helpers/parseformat_test.py
index c1abc6f728..91ab45e0c6 100644
--- a/src/borg/testsuite/helpers/parseformat_test.py
+++ b/src/borg/testsuite/helpers/parseformat_test.py
@@ -34,6 +34,7 @@
     eval_escapes,
     ChunkerParams,
     DigestAlgos,
+    FilesCacheMode,
     files_cache_mode_no_ctime,
     get_size_units,
     normalize_local_path,
@@ -917,6 +918,11 @@ def test_valid_chunkerparams(chunker_params, expected_return):
         "buzhash,23,19,21,4095",  # violates min <= mask <= max
         "buzhash,19,23,21,4096",  # even window size
         "buzhash,19,23,21",  # missing window_size (must not fall into old-style compat mode)
+        "buzhash,19,23,21,-1",  # window_size < 1
+        "buzhash,20,20,20,4095",  # window_size + 2^chunk_min + 1 > 2^chunk_max
+        "20,20,20,4095",  # dito, old-style compat mode
+        "buzhash64,19,23,21,0,2",  # window_size < 1
+        "buzhash64,19,23,21,-1,2",  # dito
         "buzhash64,20,20,20,4095,2",  # window_size + 2^chunk_min + 1 > 2^chunk_max
         "buzhash64,19,19,19,4095,2",  # dito, chunk_min == chunk_max
         "buzhash64,19,23,21,4095",  # missing nc_level
@@ -932,6 +938,61 @@ def test_invalid_chunkerparams(invalid_chunker_params):
         ChunkerParams(invalid_chunker_params)
 
 
+@pytest.mark.parametrize(
+    "chunker_params, expected_return",
+    [
+        (DEFAULT_CHUNKER_PARAMS, DEFAULT_CHUNKER_PARAMS),  # the default value of --chunker-params
+        (["default"], DEFAULT_CHUNKER_PARAMS),
+        (["buzhash", 19, 23, 21, 4095], ("buzhash", 19, 23, 21, 4095)),
+        ([19, 23, 21, 4095], ("buzhash", 19, 23, 21, 4095)),
+        (("fastcdc", 19, 23, 21, 2), ("fastcdc", 19, 23, 21, 2)),
+        (["fixed", "4096"], ("fixed", 4096, 0)),
+    ],
+)
+def test_valid_chunkerparams_sequence(chunker_params, expected_return):
+    # e.g. a YAML list from a config file: it is validated like the string form.
+    assert ChunkerParams(chunker_params) == expected_return
+
+
+@pytest.mark.parametrize(
+    "invalid_chunker_params",
+    [
+        [],
+        ["crap", 1, 2, 3, 4],
+        ["buzhash", 5, 7, 6, 4095],  # too small min. size
+        ("buzhash", 19, 23, 21, 4096),  # even window size
+        ["fastcdc", 20, 20, 20, 2],  # chunk_min == chunk_max
+        ["fixed", 63],  # too small block size
+    ],
+)
+def test_invalid_chunkerparams_sequence(invalid_chunker_params):
+    with pytest.raises(ArgumentTypeError):
+        ChunkerParams(invalid_chunker_params)
+
+
+@pytest.mark.parametrize(
+    "mode, expected_mode",
+    [
+        ("ctime,size,inode", "cis"),
+        ("cis", "cis"),
+        ("mtime,size", "ms"),
+        ("rechunk,ctime", "cr"),
+        ("disabled", "d"),
+        ("d", "d"),
+    ],
+)
+def test_files_cache_mode(mode, expected_mode):
+    assert FilesCacheMode(mode) == expected_mode
+
+
+@pytest.mark.parametrize(
+    "mode", ["size", "s", "size,inode", "ctime,mtime", "crap"]  # neither ctime nor mtime nor disabled
+)
+def test_invalid_files_cache_mode(mode):
+    with pytest.raises(ArgumentTypeError):
+        FilesCacheMode(mode)
+
+
 @pytest.mark.parametrize(
     "mode, expected_mode, expected_changed",
     [