replace asserts on external data by real checks, refs #8649

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__
+    )

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)

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,23 +77,23 @@ 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(
                     self.do_extract(self.parse_args([f"--repo={repo}", "extract", "borg-benchmark-crud1", "--dry-run"]))
                 )
                 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(
                     self.do_delete(self.parse_args([f"--repo={repo}", "delete", "-a", "borg-benchmark-crud1"]))
                 )
                 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)

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)

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:

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

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)

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)

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)
 

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]