symfonic.services.shadow.crypto

crypto

Encryption at rest for the recording store.

The store keeps governed tenant data, so §6.2 requires it encrypted at rest. The seam is the EncryptionPort protocol: an operator wires whatever KMS or AEAD their deployment mandates. What ships here is a stdlib-only default — an HMAC-SHA256 counter-mode keystream with encrypt-then-MAC — so the store has a working, testable, non-optional cipher instead of a TODO and a plaintext column.

RefusingCipher is the other half of the fail-closed rule: a store constructed without a cipher gets one that raises, so "we forgot to configure encryption" surfaces as a refusal to record rather than a silent plaintext write.

EncryptionPort

Bases: Protocol

Seal and unseal recording bytes.

HmacStreamCipher

HmacStreamCipher(key: bytes, *, key_id: str = 'shadow-recording-key')

HMAC-SHA256 CTR keystream with encrypt-then-MAC.

Encryption and authentication use separately derived subkeys, and the MAC covers key id, nonce, and ciphertext, so neither a swapped nonce nor a relabelled key can be passed off as a valid record.

Source code in src/symfonic/services/shadow/crypto.py

def __init__(self, key: bytes, *, key_id: str = "shadow-recording-key") -> None:
    if len(key) < 32:
        raise ValueError("recording-store key must be at least 32 bytes")
    self._key_id = key_id
    self._enc = hmac.new(key, b"symfonic/shadow/enc", hashlib.sha256).digest()
    self._mac = hmac.new(key, b"symfonic/shadow/mac", hashlib.sha256).digest()

RefusingCipher

The absence of a configured cipher, made explicit and loud.

SealedPayload dataclass

SealedPayload(key_id: str, nonce: bytes, ciphertext: bytes, tag: bytes)

Ciphertext plus everything needed to verify and open it. No plaintext.