crypto: add sealed streams (key-sealed frames over a byte stream)

A sealed stream cuts a byte stream into 1 MiB frames and puts every
frame into the key's envelope (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.

The AAD of a frame is a caller-given context plus the frame number and
a "last frame" flag, so the reader detects changed, reordered, dropped
or duplicated frames, truncation and trailing data. Every frame is
sealed with encrypt_oneshot(), so sealed streams are thread-safe.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Thomas Waldmann
2026-10-11 00:06:04 +02:00
co-authored by Claude Opus 5.5
parent ab0e8e9cef
commit 8cdc03b67e
4 changed files with 394 additions and 2 deletions
+29
View File
@@ -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
~~~~
+3 -2
View File
@@ -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
+149
View File
@@ -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")
@@ -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))