Merge pull request #10497 from ThomasWaldmann/encrypt-files-cache

seal the files cache with the repository key, fixes #5041

docs/faq.rst

@@ -680,13 +680,20 @@ Make sure that only you have access to the Borg data directory.
 Do I need to take security precautions regarding the cache?
 -----------------------------------------------------------
 
-The cache contains a lot of metadata information about the files in
-your repositories and it is not encrypted.
+The files cache contains a lot of metadata about the files in your backups
+(file sizes, timestamps, the chunk lists of the files and, if you use
+``borg create --digests``, digests of the file contents). It is protected with the
+repository key, like the data in the repository: it is encrypted and authenticated
+(authenticated only in the ``authenticated-*`` modes), so someone who can read the
+cache directory, but does not have the key, can not read it, and a modified files
+cache is detected (borg ignores it with a warning and rebuilds it). The pack cache
+(``BORG_STORE_CACHE``) only holds copies of the encrypted packs of the repository.
 
-However, the assumption is that the cache is being stored on the very
-same system which also contains the original files which are being
-backed up. So someone with access to the cache files would also have
-access the original files anyway.
+What is not protected: the names, sizes and timestamps of the files in the cache
+directory, the repository id (the cache directory is named after it) and the cache
+``config`` file (it only contains the cache version and the repository id). Someone
+who can write to the cache directory can delete files there, which only makes the
+next backup slower.
 
 The Internals section contains more details about :ref:`cache`. If you ever need to move the cache
 to a different location, this can be achieved by using the appropriate :ref:`env_vars`.

docs/internals/data-structures.rst

@@ -138,6 +138,35 @@ missing or fails the authentication, ``borg check --repair`` replaces it by empt
 (see :ref:`repo_defaults`). The ``chunkindex-invalid`` marker has no content and is
 stored as is.
 
+.. _sealed_stream:
+
+Sealed streams
+~~~~~~~~~~~~~~
+
+Data that may be too big to be kept in memory twice is protected as a **sealed
+stream** (``borg.crypto.sealed_stream``): the byte stream is cut into frames of 1 MiB
+(the last frame holds the rest, it may be empty), and every frame is put into the
+key's envelope, so it is encrypted and authenticated in the encrypting modes and
+authenticated only in the ``authenticated-*`` modes::
+
+    stream := frame+
+    frame  := flag (1 byte: 0 = more frames follow, 1 = last frame)
+              length (4 bytes, unsigned, little endian: the length of the envelope)
+              envelope (the key's encrypt_oneshot() of the payload, empty id)
+
+The AAD of a frame's envelope is a *context* given by the user of the sealed stream,
+followed by the frame number (8 bytes, big endian, counting from 0) and the flag (1
+byte). The context starts with a domain prefix of its own and contains what the stream
+belongs to (e.g. the repository id and a name). So a frame can not be changed, moved
+to another position or into another stream, and a stream that ends before its last
+frame, claims an earlier frame to be the last one or has data after its last frame is
+rejected. What a sealed stream can not detect is the replacement of the whole stream
+by an older stream with the same context.
+
+Every frame is sealed in a one-off session (a new random session id, IV 0), so
+sealed streams can be used from any thread. Each frame costs a session key derivation
+(a few microseconds), which is negligible for 1 MiB of payload.
+
 
 Keys
 ~~~~
@@ -729,11 +758,27 @@ Borg can also work without using the files cache (saves memory if you have a
 lot of files or not much RAM free), then all files are assumed to have changed.
 This is usually much slower than with files cache.
 
-The on-disk format of the files cache is a stream of msgpacked tuples (key, value).
-There, the chunks list is stored in its uncompressed form (chunk id and size), as
-the chunks index indexes are only valid for one specific in-memory chunks index.
-Loading the files cache involves reading the file, one msgpack object at a time,
-unpacking it, and compressing the entry as described above.
+On disk, the files cache is a stream of msgpacked tuples (key, value). There, the
+chunks list is stored in its uncompressed form (chunk id and size), as the chunks
+index indexes are only valid for one specific in-memory chunks index. This stream
+is stored as a :ref:`sealed stream <sealed_stream>`, so it is encrypted and
+authenticated with the repository key (authenticated only in the ``authenticated-*``
+modes), like the objects in the repository. The context of the sealed stream is
+``b"borg-files-cache\0"``, followed by the repository id and the file name of the files
+cache (``files.<SUFFIX>``): a files cache copied from another repository (using the
+same key material) or from another archive series is rejected. Loading the files
+cache involves reading and verifying the file frame by frame, unpacking one msgpack
+object at a time and compressing the entry as described above.
+
+Without the key, one can only see the names, sizes and timestamps of the files
+cache files. A files cache that fails the verification (it is corrupted, was
+modified, or was not written with the key of this repository) is ignored with a
+warning: borg then rebuilds the files cache from the most recent archive of the
+series in the repository, or, failing that, starts with an empty files cache. The
+same happens, without a warning, if there is no files cache file (yet). Replacing a
+files cache file by an older version of it is not detected, but harmless: its
+entries were correct when it was written, and entries that refer to chunks that are
+no longer in the repository are dropped when loading it.
 
 .. _index:
 
@@ -1274,19 +1319,14 @@ just retry later.
 Checksumming data structures
 ----------------------------
 
-As detailed in the previous sections, Borg generates and stores files
-containing important meta data, currently the files cache.
-
-Data corruption in the files cache could create incorrect archives, e.g. due
-to wrong object IDs or sizes in the files cache.
-
-Therefore, Borg calculates checksums when writing these files and tests checksums
-when reading them. Checksums are generally 256-bit sha256 hashes.
+Borg 2 protects its own local files with the repository key (see :ref:`the files
+cache <cache>`). Checksums are only used for reading borg 1.x repositories (see
+``borg transfer``): borg 1.x stores the repository index and hints files of such a
+repository (``index.N``, ``hints.N``) with checksums, so that data corruption in them
+is detected. Checksums are generally 256-bit sha256 hashes.
 Checksums are stored as hexadecimal ASCII strings.
 
 For compatibility, checksums are not required and absent checksums do not trigger errors.
-The mechanisms have been designed to avoid false-positives when various Borg
-versions are used alternately on the same repositories.
 
 Checksums are a data safety mechanism. They are not a security mechanism.
 
@@ -1315,9 +1355,8 @@ mixed into the checksum state first (encoded as an ASCII string via `%10d`
 printf format), then the name of the part is mixed in as an UTF-8 string.
 Lastly, the current position (length) in the file is mixed in as well.
 
-Borg 2 uses parts only when reading a borg 1.x repository (see ``borg
-transfer``): its index and hints files have a ``HashHeader`` part. The files
-cache is written and read as a single part.
+The index file of a borg 1.x repository has a ``HashHeader`` part, its hints file
+is a single part.
 
 The checksum state is not reset at part boundaries.
 
@@ -1354,35 +1393,13 @@ DetachedIntegrityCheckedFile, which automatically writes and reads it from a
 Upper layer
 ~~~~~~~~~~~
 
-.. rubric:: The files cache
+The integrity data of ``index.N`` and ``hints.N`` is stored in ``integrity.N`` next
+to them, a msgpacked dict with the integrity data version (2) and the integrity data
+of both files (keys ``index`` and ``hints``). Without an ``integrity.N`` file, or if
+its version is unknown, the files are read without checking them.
 
-The files cache is the only file borg 2 protects this way. Its integrity data
-is stored in the ``[integrity]`` section of the cache ``config`` file, keyed by
-the file's name (see :ref:`the files cache <cache>` about that name):
-
-.. code-block:: none
-
-    [cache]
-    version = 1
-    repository = 3c4...e59
-
-    [integrity]
-    files.9f8...a08 = {"algorithm": "SHA256", "digests": {"final": "e2a...b24"}}
-
-The chunks index is not in this list: it is not a local file, but lives in the
-repository below ``index/`` and has its own integrity mechanism, see
-:ref:`pack-index-namespace`.
-
-The cache config file is read in its entirety (using the Python ConfigParser),
-modified and written back, so sections and values a Borg version does not
-understand are preserved. There is no guard against an older Borg version
-updating the files cache without updating its integrity data: every Borg
-version that can open a version 5 repository knows the ``[integrity]`` section.
-
-A files cache that fails its integrity check (or can not be read at all) is
-discarded, not used: borg then rebuilds the files cache from the most recent
-archive of the series in the repository, or, failing that, starts with an empty
-files cache.
+If the index or hints file fails its integrity check, borg rebuilds the index
+and hints from the segments, as for any other error reading them.
 
 
 HardLinkManager and the hlid concept

docs/internals/frontends.rst

@@ -214,7 +214,6 @@ Saving the local cache at the end of :ref:`borg_create`::
 
     {"message": "Saving files cache", "operation": 1, "msgid": "cache.close", "type": "progress_message", "finished": false, "time": 1787900398.719723}
     {"message": "Saving index", "operation": 1, "msgid": "cache.close", "type": "progress_message", "finished": false, "time": 1787900398.728792}
-    {"message": "Saving cache config", "operation": 1, "msgid": "cache.close", "type": "progress_message", "finished": false, "time": 1787900398.7294679}
     {"message": "", "operation": 1, "msgid": "cache.close", "type": "progress_message", "finished": true, "time": 1787900398.739775}
 
 A debug log message::
@@ -1028,7 +1027,7 @@ Warnings
 Operations
     - cache.close
 
-      Saving the local cache (files cache, chunks index, cache config) at the end of a command.
+      Saving the local cache (files cache, chunks index) at the end of a command.
     - cache.merge_chunkindex_fragments
 
       Loading the chunk index by merging the index fragments stored in the repository.

docs/internals/security.rst

@@ -42,6 +42,14 @@ includes the store objects borg keeps next to the packs: the chunk index fragmen
 and the per-archive reference caches are encrypted with the key, too, see
 :ref:`store object envelope <store_object_envelope>`.
 
+The client-local files cache (see :ref:`the files cache <cache>`) is protected with
+the key as well, as a :ref:`sealed stream <sealed_stream>`: someone who can read the
+borg cache directory, but does not have the key, can not read the file sizes,
+timestamps, chunk lists and content digests stored in it (in the encrypting modes),
+and a files cache that was modified is detected in all modes (it is ignored with a
+warning and rebuilt). The names, sizes and timestamps of the files cache files are
+not protected.
+
 The attacker can always impose a denial of service by definition (they could
 block connections to the repository, or delete it partly or entirely).
 

src/borg/cache.py

@@ -37,8 +37,8 @@ from .helpers import sig_int
 from .helpers import msgpack
 from .helpers.msgpack import int_to_timestamp, timestamp_to_int
 from .item import ChunkListEntry
-from .crypto.file_integrity import IntegrityCheckedFile, FileIntegrityError
 from .crypto.key import store_hash
+from .crypto.sealed_stream import SealedStreamWriter, SealedStreamError, read_sealed_stream
 from .manifest import Manifest
 from .platform import SaveFile
 from .repository import Repository, StoreObjectNotFound, PackReader
@@ -98,6 +98,9 @@ def discover_files_cache_names(path, files_cache_name="files"):
 # digests has a default, so that a files cache written by a borg without it still loads.
 FileCacheEntry = namedtuple("FileCacheEntry", "age inode size ctime mtime chunks digests", defaults=(None,))
 
+# domain prefix of the sealed stream context of a files cache, see FilesCacheMixin._files_cache_context.
+FILES_CACHE_AAD = b"borg-files-cache\0"
+
 
 def cache_dir(repository, path=None):
     return Path(path) if path else Path(get_cache_dir()) / repository.id_str
@@ -126,7 +129,6 @@ class CacheConfig:
         config.add_section("cache")
         config.set("cache", "version", "1")
         config.set("cache", "repository", self.repository.id_str)
-        config.add_section("integrity")
         with SaveFile(self.config_path) as fd:
             config.write(fd)
 
@@ -139,20 +141,6 @@ class CacheConfig:
             self._config.read_file(fd)
         self._check_upgrade(self.config_path)
         self.id = self._config.get("cache", "repository")
-        try:
-            self.integrity = dict(self._config.items("integrity"))
-        except configparser.NoSectionError:
-            logger.debug("Cache integrity: no [integrity] section in the cache config, no integrity data.")
-            self.integrity = {}
-
-    def save(self, with_integrity=False):
-        if with_integrity:
-            if not self._config.has_section("integrity"):
-                self._config.add_section("integrity")
-            for file, integrity_data in self.integrity.items():
-                self._config.set("integrity", file, integrity_data)
-        with SaveFile(self.config_path) as fd:
-            self._config.write(fd)
 
     def close(self):
         pass
@@ -367,6 +355,15 @@ class FilesCacheMixin:
     def discover_files_cache_names(self, path):
         return discover_files_cache_names(path, self.FILES_CACHE_NAME)
 
+    def _files_cache_context(self):
+        """
+        Return the sealed stream context of the files cache (see borg.crypto.sealed_stream).
+
+        It binds the files cache to the repository and to its file name, so a files cache copied from
+        another repository (using the same key material) or from another archive series is rejected.
+        """
+        return FILES_CACHE_AAD + self.repository.id + self.files_cache_name().encode()
+
     def _read_files_cache(self):
         """read files cache from cache directory"""
         if "d" in self.cache_mode:  # d(isabled)
@@ -375,18 +372,14 @@ class FilesCacheMixin:
         files = {}
         logger.debug("Reading files cache ...")
         files_cache_logger.debug("FILES-CACHE-LOAD: starting...")
-        msg = None
+        name = self.files_cache_name()
+        msg = warning = None
         try:
-            with IntegrityCheckedFile(
-                path=str(self.path / self.files_cache_name()),
-                write=False,
-                integrity_data=self.cache_config.integrity.get(self.files_cache_name()),
-            ) as fd:
+            with open(self.path / name, "rb") as fd:
                 u = msgpack.Unpacker(use_list=True)
-                while True:
-                    data = fd.read(64 * 1024)
-                    if not data:
-                        break
+                # the payloads are only trustworthy once the whole stream is read, so `files` is
+                # discarded below if reading the sealed stream fails at any frame.
+                for data in read_sealed_stream(fd, self.key, self._files_cache_context()):
                     u.feed(data)
                     try:
                         for path_hash, entry in u:
@@ -398,12 +391,17 @@ class FilesCacheMixin:
                                 # repo is missing a chunk referenced from entry
                                 logger.debug(f"compress_entry failed for {entry}, skipping.")
                     except (TypeError, ValueError) as exc:
-                        msg = "The files cache seems invalid. [%s]" % str(exc)
+                        # authenticated, but not what we expect - only possible through a bug.
+                        warning = f"Ignoring invalid files cache {name} [{exc}], it will be rebuilt."
                         break
         except OSError as exc:
             msg = "The files cache can't be read. [%s]" % str(exc)
-        except FileIntegrityError as fie:
-            msg = "The files cache is corrupted. [%s]" % str(fie)
+        except SealedStreamError as exc:
+            # corrupted, tampered with, or not written with this repository's key.
+            warning = f"Ignoring corrupted files cache {name} [{exc}], it will be rebuilt."
+        if warning is not None:
+            logger.warning(warning)
+            files = None
         if msg is not None:
             logger.debug(msg)
             files = None
@@ -422,49 +420,48 @@ class FilesCacheMixin:
         files_cache_logger.debug("FILES-CACHE-SAVE: starting...")
         cache_path = str(self.path / self.files_cache_name())
         with SaveFile(cache_path, binary=True) as sync_file:
-            with IntegrityCheckedFile(path=cache_path, write=True, override_fd=sync_file) as fd:
-                entries = 0
-                age_discarded = 0
-                race_discarded = 0
-                broken_discarded = 0
-                for path_hash, entry in files.items():
-                    try:
-                        entry = self.decompress_entry(entry)
-                    except KeyError:
-                        # the entry references a chunk that is no longer in the chunks index (e.g.
-                        # after an aborted / out-of-space backup rolled back pending chunks). such an
-                        # entry is unusable; drop it rather than crash while saving the cache — the
-                        # file will simply be re-chunked in a future backup. this mirrors the
-                        # compress_entry KeyError handling in _read_files_cache.
-                        broken_discarded += 1
-                        continue
-                    if entry.age == 0:  # current entries
-                        if max(timestamp_to_int(entry.ctime), timestamp_to_int(entry.mtime)) < discard_after:
-                            # Only keep files seen in this backup that old enough not to suffer race conditions
-                            # relating to filesystem snapshots and ctime/mtime granularity or being modified
-                            # while we read them.
-                            keep = True
-                        else:
-                            keep = False
-                            race_discarded += 1
-                    else:  # old entries
-                        if entry.age < ttl:
-                            # Also keep files from older backups that have not reached BORG_FILES_CACHE_TTL yet.
-                            keep = True
-                        else:
-                            keep = False
-                            age_discarded += 1
-                    if keep:
-                        msgpack.pack((path_hash, entry), fd)
-                        entries += 1
-            integrity_data = fd.integrity_data
+            writer = SealedStreamWriter(sync_file, self.key, self._files_cache_context())
+            entries = 0
+            age_discarded = 0
+            race_discarded = 0
+            broken_discarded = 0
+            for path_hash, entry in files.items():
+                try:
+                    entry = self.decompress_entry(entry)
+                except KeyError:
+                    # the entry references a chunk that is no longer in the chunks index (e.g.
+                    # after an aborted / out-of-space backup rolled back pending chunks). such an
+                    # entry is unusable; drop it rather than crash while saving the cache — the
+                    # file will simply be re-chunked in a future backup. this mirrors the
+                    # compress_entry KeyError handling in _read_files_cache.
+                    broken_discarded += 1
+                    continue
+                if entry.age == 0:  # current entries
+                    if max(timestamp_to_int(entry.ctime), timestamp_to_int(entry.mtime)) < discard_after:
+                        # Only keep files seen in this backup that old enough not to suffer race conditions
+                        # relating to filesystem snapshots and ctime/mtime granularity or being modified
+                        # while we read them.
+                        keep = True
+                    else:
+                        keep = False
+                        race_discarded += 1
+                else:  # old entries
+                    if entry.age < ttl:
+                        # Also keep files from older backups that have not reached BORG_FILES_CACHE_TTL yet.
+                        keep = True
+                    else:
+                        keep = False
+                        age_discarded += 1
+                if keep:
+                    msgpack.pack((path_hash, entry), writer)
+                    entries += 1
+            writer.finish()
         files_cache_logger.debug(f"FILES-CACHE-KILL: removed {age_discarded} entries with age >= TTL [{ttl}]")
         if broken_discarded:
             files_cache_logger.debug(f"FILES-CACHE-KILL: removed {broken_discarded} entries referencing missing chunks")
         t_str = datetime.fromtimestamp(discard_after / 1e9, UTC).isoformat()
         files_cache_logger.debug(f"FILES-CACHE-KILL: removed {race_discarded} entries with ctime/mtime >= {t_str}")
         files_cache_logger.debug(f"FILES-CACHE-SAVE: finished, {entries} remaining entries saved.")
-        return integrity_data
 
     def file_known_and_unchanged(self, hashed_path, path_hash, st):
         """
@@ -1378,14 +1375,12 @@ class AdHocWithFilesCache(FilesCacheMixin, ChunksMixin):
         pi = ProgressIndicatorMessage(msgid="cache.close")
         if self._files is not None:
             pi.output("Saving files cache")
-            integrity_data = self._write_files_cache(self._files)
-            self.cache_config.integrity[self.files_cache_name()] = integrity_data
+            self._write_files_cache(self._files)
             self._files = None  # release the (potentially large) files cache dict, like _chunks below
         if self._chunks is not None:
             for key, value in sorted(self._chunks.stats.items()):
                 logger.debug(f"Chunks index stats: {key}: {value}")
             pi.output("Saving index")
-            # note: index/* in repo has a different integrity mechanism
             now = datetime.now(UTC)
             self._maybe_write_chunks_index(now, force=True, clear=True)
             self._chunks = None  # nothing there (cleared!)
@@ -1403,8 +1398,6 @@ class AdHocWithFilesCache(FilesCacheMixin, ChunksMixin):
             # repository's reference too, so a later .chunks access rebuilds it from the repo instead
             # of seeing a valid-looking but empty index (and so is_chunk_index_loaded reports False).
             self.repository.invalidate_chunk_index()
-        pi.output("Saving cache config")
-        self.cache_config.save(with_integrity=True)
         self.cache_config.close()
         pi.finish()
         self.cache_config = None

src/borg/crypto/key.py

@@ -1404,8 +1404,9 @@ class AEADKeyBase(KeyBase):
         This does not use or change the key's current session (self.sessionid, self.cipher, the IV counter),
         so it is safe to call from any thread, also concurrently with encrypt(): two threads sharing the
         session's IV counter could otherwise encrypt two messages with the same IV. decrypt() reads the
-        result like any other envelope. Each call derives a new session key, so this is meant for rare,
-        small objects (e.g. the repository lock objects), not for chunks.
+        result like any other envelope. Each call derives a new session key (a few microseconds), so this
+        is meant for objects that are few compared to their size (e.g. the repository lock objects or the
+        ~1 MiB frames of a sealed stream, see sealed_stream.py), not for millions of chunks.
         """
         sessionid = os.urandom(24)
         iv = 0

src/borg/crypto/sealed_stream.py

@@ -0,0 +1,149 @@
+"""
+Sealed streams: a byte stream protected by a key's envelope, frame by frame.
+
+The writer cuts the stream into frames of FRAME_SIZE bytes (the last frame holds the rest, it may be
+empty) and puts every frame into the key's envelope (see KeyBase.encrypt): encrypted and authenticated
+in the encrypting modes, authenticated only in the authenticated-* modes. So neither the writer nor the
+reader needs memory for more than a frame, whatever the size of the stream.
+
+Format::
+
+    stream := frame+
+    frame  := flag (1 byte: 0 = more frames follow, 1 = last frame)
+              length (4 bytes, unsigned, little endian: the length of the envelope)
+              envelope (key.encrypt_oneshot(b"", payload, aad=context + frame number + flag))
+
+The frame number counts from 0 and is 8 bytes, big endian; the flag in the AAD is the same byte as in
+the frame header. The context is chosen by the caller: it must start with a caller-specific domain
+prefix (so streams of different users, and the other envelopes made with the key, can not be mistaken
+for each other) and should contain everything the stream belongs to (e.g. the repository id and a
+name). The fixed-size fields follow the context, so the AAD is unambiguous for contexts of any length.
+
+Given the context, the reader detects any change of a frame and any reordering, dropping or duplicating
+of frames, a stream that ends before its last frame, a "last" flag on an earlier frame and data after
+the last frame. It can not detect that a whole stream was replaced by an older stream sealed with the
+same context - each user has to decide whether that matters.
+
+Every frame is sealed in a one-off session (see encrypt_oneshot), so writing and reading a sealed stream
+is safe in any thread, also while other threads encrypt with the same key.
+"""
+
+import struct
+
+from ..helpers import IntegrityError
+
+FRAME_SIZE = 1024 * 1024  # the payload size of a frame (the last frame may be smaller)
+
+FLAG_MORE = 0
+FLAG_LAST = 1
+
+_HEADER = struct.Struct("<BI")  # flag, envelope length
+
+# the MAC keys length-prefix the AAD with 16 bits, leave room for the frame number and the flag.
+MAX_CONTEXT_SIZE = 0xFFFF - 8 - 1
+
+
+class SealedStreamError(IntegrityError):
+    """Sealed stream error: {}"""
+
+
+def _frame_aad(context, frame_no, last):
+    return context + frame_no.to_bytes(8, "big") + bytes([FLAG_LAST if last else FLAG_MORE])
+
+
+class SealedStreamWriter:
+    """
+    Write a sealed stream to the binary file *fd* (a file-like object with a write() method).
+
+    Use it like a file opened for writing (only write() is supported), then call finish() exactly once
+    to write the last frame. A stream without its last frame is rejected by the reader.
+    """
+
+    def __init__(self, fd, key, context, *, frame_size=None):
+        assert len(context) <= MAX_CONTEXT_SIZE
+        self.fd = fd
+        self.key = key
+        self.context = bytes(context)
+        self.frame_size = FRAME_SIZE if frame_size is None else frame_size
+        assert self.frame_size > 0
+        self.buffer = bytearray()
+        self.frame_no = 0
+        self.finished = False
+
+    def _write_frame(self, payload, last):
+        envelope = self.key.encrypt_oneshot(b"", payload, aad=_frame_aad(self.context, self.frame_no, last))
+        self.fd.write(_HEADER.pack(FLAG_LAST if last else FLAG_MORE, len(envelope)))
+        self.fd.write(envelope)
+        self.frame_no += 1
+
+    def write(self, data):
+        assert not self.finished, "write() after finish()"
+        self.buffer += data
+        while len(self.buffer) >= self.frame_size:
+            self._write_frame(self.buffer[: self.frame_size], last=False)
+            del self.buffer[: self.frame_size]
+        return len(data)
+
+    def finish(self):
+        assert not self.finished, "finish() called twice"
+        self._write_frame(self.buffer, last=True)
+        self.buffer = bytearray()
+        self.finished = True
+
+
+def _read_exactly(fd, size):
+    """
+    Read *size* bytes from the binary file *fd*, fewer only at EOF.
+
+    A single fd.read(size) may return fewer bytes before EOF, e.g. for raw (unbuffered) files, pipes,
+    sockets or other file-like objects, so read again until we have *size* bytes or EOF.
+    """
+    data = fd.read(size)
+    if len(data) == size or not data:
+        return data
+    buffer = bytearray(data)
+    while len(buffer) < size:
+        data = fd.read(size - len(buffer))
+        if not data:
+            break
+        buffer += data
+    return bytes(buffer)
+
+
+def read_sealed_stream(fd, key, context, *, max_frame_size=None):
+    """
+    Read the sealed stream from the binary file *fd*, yield the payloads of its frames in order.
+
+    Every frame is verified before its payload is yielded. Raises SealedStreamError if the stream is
+    not intact (see the module docstring) - this may happen after some payloads were yielded already,
+    so the caller must not use what it got before the generator is exhausted.
+
+    *max_frame_size* (default: FRAME_SIZE) is the largest frame payload accepted.
+    """
+    assert len(context) <= MAX_CONTEXT_SIZE
+    context = bytes(context)
+    max_envelope_size = (FRAME_SIZE if max_frame_size is None else max_frame_size) + key.PAYLOAD_OVERHEAD
+    frame_no = 0
+    while True:
+        header = _read_exactly(fd, _HEADER.size)
+        if len(header) != _HEADER.size:
+            raise SealedStreamError(f"frame {frame_no}: stream ends before the last frame")
+        flag, length = _HEADER.unpack(header)
+        if flag not in (FLAG_MORE, FLAG_LAST):
+            raise SealedStreamError(f"frame {frame_no}: invalid flag {flag}")
+        if length > max_envelope_size:
+            raise SealedStreamError(f"frame {frame_no}: envelope too large ({length} bytes)")
+        envelope = _read_exactly(fd, length)
+        if len(envelope) != length:
+            raise SealedStreamError(f"frame {frame_no}: stream ends inside the frame")
+        last = flag == FLAG_LAST
+        try:
+            payload = key.decrypt(b"", envelope, aad=_frame_aad(context, frame_no, last))
+        except IntegrityError as err:
+            raise SealedStreamError(f"frame {frame_no}: authentication failed") from err
+        yield payload
+        if last:
+            break
+        frame_no += 1
+    if fd.read(1):
+        raise SealedStreamError("data after the last frame")

src/borg/testsuite/archiver/create_cmd_test.py

@@ -2061,14 +2061,35 @@ def test_chunkindex_covers_committed_archive(archiver, monkeypatch):
     assert_dirs_equal("input", "output/input")
 
 
-def _remove_files_cache(archiver, archive_name):
-    """Remove the local files cache of an archive series, forcing a rebuild from the repository."""
+def _files_cache_path(archiver, archive_name):
+    """Return the path of the local files cache of an archive series."""
     from ...cache import files_cache_name
     from ...helpers import get_cache_dir
 
     repo_id = json.loads(cmd(archiver, "repo-info", "--json"))["repository"]["id"]
-    cache_file = Path(get_cache_dir(repo_id, create=False)) / files_cache_name(archive_name)
-    cache_file.unlink()
+    return Path(get_cache_dir(repo_id, create=False)) / files_cache_name(archive_name)
+
+
+def _remove_files_cache(archiver, archive_name):
+    """Remove the local files cache of an archive series, forcing a rebuild from the repository."""
+    _files_cache_path(archiver, archive_name).unlink()
+
+
+def test_files_cache_corrupted(archivers, request):
+    """A corrupted files cache is ignored with a warning and rebuilt from the previous archive."""
+    archiver = request.getfixturevalue(archivers)
+    create_regular_file(archiver.input_path, "file1", size=1024 * 80)
+    granularity_sleep()  # file2 must have newer timestamps than file1
+    create_regular_file(archiver.input_path, "file2", size=1024 * 80)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    cmd(archiver, "create", "home", "input")
+    cache_file = _files_cache_path(archiver, "home")
+    data = bytearray(cache_file.read_bytes())
+    data[len(data) // 2] ^= 0x01
+    cache_file.write_bytes(bytes(data))
+    output = cmd(archiver, "create", "--list", "home", "input")  # rc 0: the warning does not change it
+    assert f"Ignoring corrupted files cache {cache_file.name}" in output
+    assert "U input/file1" in output  # known from the files cache rebuilt from the previous archive
 
 
 def test_files_cache_rebuild_ignores_other_hosts(archivers, request, monkeypatch):

src/borg/testsuite/cache_test.py

@@ -32,7 +32,10 @@ from ..cache import (
 )
 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 msgpack
 from ..helpers.msgpack import int_to_timestamp
 from ..manifest import Manifest
 from ..repository import PackReader, Repository
@@ -120,13 +123,111 @@ class TestAdHocWithFilesCache:
         path_hash = H(42)
         files = {path_hash: cache.compress_entry(entry)}
         assert cache._newest_cmtime is None  # nothing was chunked in this backup
-        integrity_data = cache._write_files_cache(files)
-        cache.cache_config.integrity[cache.files_cache_name()] = integrity_data
+        cache._write_files_cache(files)
         # with the bug (initial value 0 instead of None) the cutoff would be the epoch and the
         # entry would be dropped; after the fix the entry must still be there:
         loaded = cache._read_files_cache()
         assert path_hash in loaded
 
+    def write_files_cache(self, manifest, *, archive_name="test", count=3):
+        """Write a files cache with *count* entries, return the cache and the entries (path_hash -> entry)."""
+        cache = AdHocWithFilesCache(manifest, cache_mode="cis", archive_name=archive_name)
+        # the chunk that the cached files reference (needed so the entries can be (de)compressed):
+        cache.add_chunk(H(5), {}, b"5678", stats=Statistics())
+        ts = int_to_timestamp(safe_ns(time.time_ns()))
+        entries = {
+            H(1000 + i): FileCacheEntry(
+                age=0, inode=i, size=4, ctime=ts, mtime=ts, chunks=[(H(5), 4)], digests={"sha256": H(2000 + i)}
+            )
+            for i in range(count)
+        }
+        cache._write_files_cache({path_hash: cache.compress_entry(entry) for path_hash, entry in entries.items()})
+        return cache, entries
+
+    def files_cache_path(self, cache):
+        return cache.path / cache.files_cache_name()
+
+    def test_files_cache_roundtrip(self, manifest, monkeypatch):
+        monkeypatch.setattr(sealed_stream, "FRAME_SIZE", 100)  # many frames, entries straddle frames
+        cache, entries = self.write_files_cache(manifest, count=50)
+        with open(self.files_cache_path(cache), "rb") as fd:
+            assert len(list(read_sealed_stream(fd, cache.key, cache._files_cache_context()))) > 10
+        loaded = cache._read_files_cache()
+        assert {path_hash: cache.decompress_entry(entry) for path_hash, entry in loaded.items()} == {
+            path_hash: entry._replace(age=1) for path_hash, entry in entries.items()
+        }
+
+    def test_files_cache_empty(self, manifest):
+        cache, _ = self.write_files_cache(manifest, count=0)
+        assert cache._read_files_cache() == {}
+
+    def test_files_cache_missing(self, manifest, caplog):
+        cache = AdHocWithFilesCache(manifest, cache_mode="cis", archive_name="test")
+        with caplog.at_level(logging.WARNING, logger="borg.cache"):
+            assert cache._read_files_cache() is None
+        assert caplog.text == ""  # no files cache yet is normal
+
+    def test_files_cache_encrypted(self, manifest):
+        cache, entries = self.write_files_cache(manifest)
+        data = self.files_cache_path(cache).read_bytes()
+        assert H(5) not in data
+        for path_hash, entry in entries.items():
+            assert path_hash not in data
+            assert entry.digests["sha256"] not in data
+
+    def check_ignored(self, cache, caplog, reason="corrupted"):
+        with caplog.at_level(logging.WARNING, logger="borg.cache"):
+            assert cache._read_files_cache() is None
+        assert f"Ignoring {reason} files cache {cache.files_cache_name()}" in caplog.text
+
+    def test_files_cache_tampered(self, manifest, caplog):
+        cache, _ = self.write_files_cache(manifest)
+        path = self.files_cache_path(cache)
+        data = bytearray(path.read_bytes())
+        data[len(data) // 2] ^= 0x01
+        path.write_bytes(bytes(data))
+        self.check_ignored(cache, caplog)
+
+    def test_files_cache_other_series(self, manifest, caplog):
+        cache_a, _ = self.write_files_cache(manifest, archive_name="series-a")
+        cache_b = AdHocWithFilesCache(manifest, cache_mode="cis", archive_name="series-b")
+        self.files_cache_path(cache_b).write_bytes(self.files_cache_path(cache_a).read_bytes())
+        self.check_ignored(cache_b, caplog)
+
+    def test_files_cache_other_repository(self, manifest, monkeypatch, caplog):
+        cache, _ = self.write_files_cache(manifest)
+        # a files cache written for another repository using the same key material:
+        monkeypatch.setattr(cache.repository, "id", bytes(32))
+        self.check_ignored(cache, caplog)
+
+    def test_files_cache_plaintext(self, manifest, caplog):
+        cache, entries = self.write_files_cache(manifest)
+        # a files cache in the format of borg versions without sealed files caches: a plain msgpack stream.
+        with open(self.files_cache_path(cache), "wb") as fd:
+            for path_hash, entry in entries.items():
+                msgpack.pack((path_hash, entry), fd)
+        self.check_ignored(cache, caplog)
+
+    def test_files_cache_invalid_content(self, manifest, caplog):
+        cache = AdHocWithFilesCache(manifest, cache_mode="cis", archive_name="test")
+        # authentic, but not a stream of (path_hash, entry) tuples:
+        with open(self.files_cache_path(cache), "wb") as fd:
+            writer = SealedStreamWriter(fd, cache.key, cache._files_cache_context())
+            msgpack.pack([1, 2, 3], writer)
+            writer.finish()
+        self.check_ignored(cache, caplog, reason="invalid")
+
+    def test_cache_config(self, manifest):
+        cache, _ = self.write_files_cache(manifest)
+        config_path = cache.path / "config"
+        config = config_path.read_text()
+        assert "[cache]" in config and "[integrity]" not in config
+        st = config_path.stat()
+        cache.close()
+        # nothing in the config changes after its creation, so close() does not rewrite it:
+        assert config_path.stat().st_ino == st.st_ino and config_path.stat().st_mtime_ns == st.st_mtime_ns
+        assert config_path.read_text() == config
+
 
 def test_delete_chunkindex_from_repo_missing(tmp_path):
     """delete_chunkindex_from_repo handles StoreObjectNotFound when index entries do not exist."""

src/borg/testsuite/crypto/sealed_stream_test.py

@@ -0,0 +1,213 @@
+import io
+import os
+import random
+import struct
+
+import pytest
+
+from ...crypto.key import AESOCBKey, CHPOKey, AuthenticatedKey, Blake3AuthenticatedKey
+from ...crypto import sealed_stream
+from ...crypto.sealed_stream import SealedStreamWriter, SealedStreamError, read_sealed_stream
+from ...crypto.sealed_stream import FLAG_LAST, FLAG_MORE
+from ...helpers import IntegrityError
+
+CONTEXT = b"borg-test-stream\0" + bytes(range(32)) + b"some-name"
+HEADER = struct.Struct("<BI")
+ENCRYPTING_CLASSES = (AESOCBKey, CHPOKey)
+AUTHENTICATING_CLASSES = (AuthenticatedKey, Blake3AuthenticatedKey)
+
+
+def make_key(cls):
+    key = cls(None)
+    key.init_from_given_data(crypt_key=bytes(range(64)), id_key=bytes(range(100, 132)), chunk_seed=0)
+    key.init_ciphers()
+    return key
+
+
+@pytest.fixture(params=ENCRYPTING_CLASSES + AUTHENTICATING_CLASSES, ids=lambda cls: cls.__name__)
+def key(request):
+    return make_key(request.param)
+
+
+def seal(key, data, *, frame_size=None, context=CONTEXT, write_sizes=None):
+    """Return the sealed stream of *data*, written in pieces of *write_sizes* (default: one write)."""
+    fd = io.BytesIO()
+    writer = SealedStreamWriter(fd, key, context, frame_size=frame_size)
+    pos = 0
+    for size in write_sizes or [len(data)]:
+        writer.write(data[pos : pos + size])
+        pos += size
+    writer.write(data[pos:])
+    writer.finish()
+    return fd.getvalue()
+
+
+def unseal(key, stream, *, max_frame_size=None, context=CONTEXT):
+    return b"".join(read_sealed_stream(io.BytesIO(stream), key, context, max_frame_size=max_frame_size))
+
+
+def split_frames(stream):
+    """Split a sealed stream into a list of (flag, envelope)."""
+    frames, pos = [], 0
+    while pos < len(stream):
+        flag, length = HEADER.unpack_from(stream, pos)
+        pos += HEADER.size
+        frames.append((flag, stream[pos : pos + length]))
+        pos += length
+    return frames
+
+
+def join_frames(frames):
+    return b"".join(HEADER.pack(flag, len(envelope)) + envelope for flag, envelope in frames)
+
+
+@pytest.mark.parametrize("size", [0, 1, 99, 100, 101, 1000])
+def test_roundtrip(key, size):
+    data = os.urandom(size)
+    rng = random.Random(size)
+    write_sizes = [rng.randint(0, 150) for _ in range(20)]
+    stream = seal(key, data, frame_size=100, write_sizes=write_sizes)
+    assert unseal(key, stream, max_frame_size=100) == data
+    frames = split_frames(stream)
+    # full frames, then the last one with the rest - empty if size is a multiple of the frame size:
+    assert len(frames) == size // 100 + 1
+    assert [flag for flag, _ in frames] == [FLAG_MORE] * (len(frames) - 1) + [FLAG_LAST]
+    assert len(frames[-1][1]) == size % 100 + key.PAYLOAD_OVERHEAD
+
+
+def test_roundtrip_default_frame_size(key):
+    data = os.urandom(3 * sealed_stream.FRAME_SIZE + 12345)
+    stream = seal(key, data)
+    assert len(split_frames(stream)) == 4
+    assert unseal(key, stream) == data
+
+
+@pytest.mark.parametrize("cls", ENCRYPTING_CLASSES, ids=lambda cls: cls.__name__)
+def test_encrypted(cls):
+    key = make_key(cls)
+    data = b"secret payload " * 100
+    stream = seal(key, data, frame_size=100)
+    assert b"secret payload" not in stream
+    frames = split_frames(stream)
+    # every frame is sealed in its own one-off session: a fresh session id and IV 0 (see AEADKeyBase).
+    session_ids = {envelope[8:32] for _, envelope in frames}
+    assert len(session_ids) == len(frames)
+    assert all(envelope[2:8] == bytes(6) for _, envelope in frames)
+    assert unseal(key, stream, max_frame_size=100) == data
+
+
+@pytest.mark.parametrize("cls", AUTHENTICATING_CLASSES, ids=lambda cls: cls.__name__)
+def test_authenticated_only(cls):
+    key = make_key(cls)
+    data = b"visible payload " * 10
+    stream = seal(key, data)
+    assert data in stream  # authenticated-* modes do not encrypt
+    assert unseal(key, stream) == data
+    tampered = stream.replace(b"visible", b"VISIBLE", 1)
+    with pytest.raises(SealedStreamError):
+        unseal(key, tampered)
+
+
+def test_tampering(key):
+    stream = seal(key, os.urandom(250), frame_size=100)
+    frames = split_frames(stream)
+    frame_len = HEADER.size + len(frames[0][1])
+    # offsets in the first frame: flag, length, envelope header, payload, and the last byte of the stream.
+    for offset in (0, 1, HEADER.size, HEADER.size + key.PAYLOAD_OVERHEAD, frame_len - 1, len(stream) - 1):
+        tampered = bytearray(stream)
+        tampered[offset] ^= 0x01
+        with pytest.raises(SealedStreamError):
+            unseal(key, bytes(tampered), max_frame_size=100)
+
+
+def test_truncation(key):
+    stream = seal(key, os.urandom(250), frame_size=100)
+    frames = split_frames(stream)
+    without_last = join_frames(frames[:-1])
+    with pytest.raises(SealedStreamError, match="ends before the last frame"):
+        unseal(key, without_last, max_frame_size=100)
+    with pytest.raises(SealedStreamError, match="ends before the last frame"):
+        unseal(key, without_last + stream[len(without_last) : len(without_last) + 2], max_frame_size=100)
+    with pytest.raises(SealedStreamError, match="ends inside the frame"):
+        unseal(key, stream[:-1], max_frame_size=100)
+    with pytest.raises(SealedStreamError, match="ends before the last frame"):
+        unseal(key, b"", max_frame_size=100)
+
+
+def test_trailing_data(key):
+    stream = seal(key, os.urandom(250), frame_size=100)
+    with pytest.raises(SealedStreamError, match="data after the last frame"):
+        unseal(key, stream + b"\0", max_frame_size=100)
+    with pytest.raises(SealedStreamError, match="data after the last frame"):
+        unseal(key, stream + stream, max_frame_size=100)
+
+
+def test_reordered_and_duplicated_frames(key):
+    frames = split_frames(seal(key, os.urandom(350), frame_size=100))
+    assert len(frames) == 4
+    for bad in (
+        [frames[1], frames[0], frames[2], frames[3]],  # swapped
+        [frames[0], frames[0], frames[1], frames[2], frames[3]],  # duplicated
+        [frames[0], frames[2], frames[3]],  # one dropped
+    ):
+        with pytest.raises(SealedStreamError, match="authentication failed"):
+            unseal(key, join_frames(bad), max_frame_size=100)
+
+
+def test_last_flag_on_earlier_frame(key):
+    frames = split_frames(seal(key, os.urandom(250), frame_size=100))
+    # claim that the first frame is the last one and cut the stream there:
+    with pytest.raises(SealedStreamError, match="frame 0: authentication failed"):
+        unseal(key, join_frames([(FLAG_LAST, frames[0][1])]), max_frame_size=100)
+    with pytest.raises(SealedStreamError, match="frame 0: invalid flag"):
+        unseal(key, join_frames([(2, frames[0][1])] + frames[1:]), max_frame_size=100)
+
+
+@pytest.mark.parametrize("other_context", [CONTEXT[:-1], CONTEXT + b"x", b"X" + CONTEXT[1:], b""])
+def test_context_mismatch(key, other_context):
+    stream = seal(key, os.urandom(50))
+    with pytest.raises(SealedStreamError, match="frame 0: authentication failed") as excinfo:
+        unseal(key, stream, context=other_context)
+    assert isinstance(excinfo.value, IntegrityError)
+    assert isinstance(excinfo.value.__cause__, IntegrityError)
+
+
+def test_oversize_frame(key):
+    stream = seal(key, os.urandom(250), frame_size=200)
+    with pytest.raises(SealedStreamError, match="frame 0: envelope too large"):
+        unseal(key, stream, max_frame_size=100)
+    # a huge length field is rejected before trying to read the envelope:
+    with pytest.raises(SealedStreamError, match="frame 0: envelope too large"):
+        unseal(key, HEADER.pack(FLAG_LAST, 2**32 - 1))
+
+
+def test_writer_misuse(key):
+    writer = SealedStreamWriter(io.BytesIO(), key, CONTEXT)
+    writer.write(b"data")
+    writer.finish()
+    with pytest.raises(AssertionError):
+        writer.write(b"more")
+    with pytest.raises(AssertionError):
+        writer.finish()
+
+
+class ShortReader:
+    """A binary file whose read() returns at most a few bytes per call, like a pipe or a raw socket."""
+
+    def __init__(self, data, max_read=3):
+        self.fd = io.BytesIO(data)
+        self.max_read = max_read
+
+    def read(self, size=-1):
+        return self.fd.read(min(size, self.max_read) if size >= 0 else self.max_read)
+
+
+def test_short_reads(key):
+    data = os.urandom(250)
+    stream = seal(key, data, frame_size=100)
+    assert b"".join(read_sealed_stream(ShortReader(stream), key, CONTEXT, max_frame_size=100)) == data
+    # short reads do not hide a truncated stream or trailing data:
+    with pytest.raises(SealedStreamError, match="ends inside the frame"):
+        b"".join(read_sealed_stream(ShortReader(stream[:-1]), key, CONTEXT, max_frame_size=100))
+    with pytest.raises(SealedStreamError, match="data after the last frame"):
+        b"".join(read_sealed_stream(ShortReader(stream + b"\0"), key, CONTEXT, max_frame_size=100))