aeat.adapters.persistence.storage.envelope._envelope module

Schema-version envelope for file-backed persistence.

The envelope is the single contract every file-backed persistence consumer adheres to. It pins:

  • the on-disk schema version, which must match the consumer’s current schema exactly;

  • the timestamp of the write (timezone-aware datetime);

  • the sensitivity classification (so the substrate can refuse to load a record if a consumer accidentally bypasses its repository);

  • the payload itself (typed strict pydantic v2 model);

  • optional encryption metadata (when the payload is at-rest ciphertext).

The save_envelope() and load_envelope() helpers atomically write and read the envelope JSON via the project’s standard tempfile.NamedTemporaryFile + os.replace pattern. Encrypted envelopes require an explicit MasterKeyProvider and HKDF context; the helpers derive a per-consumer key via HKDF-SHA256 and do not resolve an ambient provider themselves.

The substrate refuses any payload whose schema_version differs from the consumer’s expected version, or which fails classification validation. Migrated sensitive repositories should use SecureBoundRepository, which stores the same envelope payload shape in encrypted SQL secure objects rather than plain files.

class AeadAlgorithm(*values)[source]

Bases: StrEnum

Closed catalogue of AEAD identifiers recognised by the substrate.

Members:
AES_256_GCM_V1: AES-256 Galois Counter Mode, version 1 wire

format (12-byte nonce, 16-byte tag). The only algorithm shipping today.

AES_256_GCM_V1
class EncryptionMetadata(**data)[source]

Bases: BaseModel

Encryption envelope describing how the payload was encrypted.

Variables:

  • algorithm – Stable identifier for the AEAD primitive used. Today only aes-256-gcm-v1 is defined; future primitives register their own identifier.

  • nonce_b64 – Base64-encoded 12-byte nonce.

  • ciphertext_b64 – Base64-encoded ciphertext_with_tag.

  • associated_data_b64 – Base64-encoded AAD bytes. The field is required so persisted metadata distinguishes an explicitly empty AAD from malformed metadata where the AAD member is missing.

Parameters:
algorithm: AeadAlgorithm
nonce_b64: str
ciphertext_b64: str
associated_data_b64: str
classmethod from_blob(blob, *, associated_data=b'')[source]

Build metadata from an encrypted blob.

Return type:

EncryptionMetadata

Returns:

EncryptionMetadata derived from an EncryptedBlob.

Parameters:
to_blob()[source]

Reconstruct the EncryptedBlob from encoded fields.

Return type:

EncryptedBlob

associated_data()[source]

Decode the associated-data bytes.

Return type:

bytes

class Envelope(**data)[source]

Bases: BaseModel, Generic

Frozen pydantic v2 envelope wrapping a typed file-backed payload.

Variables:

  • schema_version – Integer version that consumers compare to their expected version. Older and newer versions are refused.

  • written_at – Timezone-aware datetime captured at write time.

  • classification – The SensitivityClass declared by the writer. Mismatches at load time raise ClassificationError.

  • payload – The typed payload. Plaintext is stored when encryption is None; ciphertext lives in encryption.ciphertext_b64 when present, and payload is then a placeholder consumer-typed value.

  • encryption – Optional encryption metadata. None for plaintext envelopes.

Parameters:
schema_version: int
written_at: datetime
classification: SensitivityClass
payload: PayloadT
encryption: EncryptionMetadata | None
classmethod for_payload_type(payload_cls)[source]

Return the Envelope parameterised for payload_cls.

This typed factory avoids a bare cast(Any, Envelope).__class_getitem__(...) at call sites. The returned class is the concrete generic alias Pydantic needs at the JSON validation boundary. The cast to type[Envelope[PayloadT]] is safe because __class_getitem__ on a PEP-695 generic model returns exactly the parameterised subtype; Pydantic registers it as a model class whose payload field is constrained to payload_cls.

Return type:

type[Envelope[PayloadT]]

Parameters:

payload_cls (type[PayloadT])

save_envelope(envelope, path)[source]

Atomically persist envelope as JSON to path.

Parameters:

  • envelope (Envelope[TypeVar]) – The Envelope to write.

  • path (Path) – Destination file. Parent directory is created if absent.

Raises:

StorageValidationError – When the temporary file or atomic replace operation fails.

Return type:

None

load_envelope(path, envelope_type, *, expected_class, max_supported_version)[source]

Load and validate an envelope from disk.

Parameters:

  • path (Path) – Source file (must exist).

  • envelope_type (type[Envelope[TypeVar]]) – The parameterised envelope class (e.g. Envelope[MyPayloadV1]). Pydantic uses this to validate the JSON against the typed payload.

  • expected_class (SensitivityClass) – The SensitivityClass the consumer expects. Mismatch raises ClassificationError.

  • max_supported_version (int) – The current schema_version the consumer expects. Any different version raises EnvelopeVersionError.

Return type:

Envelope[TypeVar]

Returns:

The validated Envelope at the consumer’s expected version.

Raises:
class CipherEnvelope(**data)[source]

Bases: BaseModel

On-disk wire form for ciphertext-at-rest envelopes.

A CipherEnvelope is structurally distinct from Envelope — it carries no typed payload field, only the encryption metadata and the same classification gate. The plaintext Envelope (with payload) is JSON-serialised, encrypted with AES-256-GCM, and the ciphertext lives inside encryption.ciphertext_b64.

Variables:

  • cipher_schema_version – Wire-format version of the cipher envelope itself (independent of the inner plaintext envelope’s Envelope.schema_version).

  • written_at – Timezone-aware datetime captured at write time.

  • classification – The SensitivityClass of the inner payload. Replicated at the cipher layer so a load can reject foreign-class ciphertext before the master key is consulted (defense in depth).

  • encryption – Required encryption metadata.

Parameters:
cipher_schema_version: int
written_at: datetime
classification: SensitivityClass
encryption: EncryptionMetadata
build_aad(classification, hkdf_context)[source]

Build the AEAD associated-data binding for a cipher envelope.

The AAD authenticates both the SensitivityClass classification and the consumer’s HKDF context, so an attacker cannot relabel ciphertext as a different sensitivity class or graft a payload from one consumer onto another.

Return type:

bytes

Parameters:
derive_envelope_key(*, master_key, hkdf_context)[source]

Derive a per-consumer 32-byte key from the master key via HKDF-SHA256.

Return type:

bytes

Parameters:
save_encrypted_envelope(envelope, path, *, master_key_provider, hkdf_context)[source]

Atomically persist envelope as an AES-256-GCM ciphertext on disk.

The plaintext Envelope is JSON-serialised, encrypted with AES-256-GCM under a per-consumer key derived from the master key via HKDF-SHA256, and written to path as a CipherEnvelope wire form. The classification and HKDF context are bound to the ciphertext via AAD so an attacker cannot relabel or cross-consumer-graft.

The caller supplies master_key_provider explicitly. Tests can pass EphemeralMasterKeyProvider; production callers pass the provider selected by the custody flow. This helper does not resolve settings, active sessions, or default key providers on its own.

Parameters:

  • envelope (Envelope[TypeVar]) – The plaintext envelope to encrypt and persist.

  • path (Path) – Destination file. Parent directory is created if absent.

  • master_key_provider (MasterKeyProvider) – MasterKeyProvider supplying the master key used to derive the per-consumer encryption key via HKDF-SHA256.

  • hkdf_context (bytes) – Per-consumer context bytes (e.g. b"aeat.domain.transactions.v1"). Different consumers MUST use distinct contexts so cross-consumer ciphertext substitution fails.

Raises:

StorageValidationError – When the temporary file or atomic replace operation fails.

Return type:

None

load_encrypted_envelope(path, envelope_type, *, expected_class, master_key_provider, hkdf_context, max_supported_version)[source]

Load and decrypt an at-rest-ciphertext envelope.

The on-disk shape MUST be a CipherEnvelope. The classification gate is enforced before the master key is consulted — a foreign-class ciphertext is rejected without any crypto attempt (defense in depth). After decryption, the inner plaintext is parsed back into the typed Envelope, classification-checked again, and version-checked.

Parameters:

  • path (Path) – Source file (must exist).

  • envelope_type (type[Envelope[TypeVar]]) – The parameterised envelope class.

  • expected_class (SensitivityClass) – The SensitivityClass the consumer expects. Mismatch raises ClassificationError before any crypto attempt.

  • master_key_provider (MasterKeyProvider) – MasterKeyProvider supplying the master key used to derive the per-consumer decryption key via HKDF-SHA256.

  • hkdf_context (bytes) – Per-consumer context bytes; MUST match the value supplied at save time.

  • max_supported_version (int) – Current inner-envelope schema version the consumer expects.

Return type:

Envelope[TypeVar]

Returns:

The decrypted and version-checked inner Envelope.

Raises:

  • ClassificationError – If the cipher envelope’s class differs from expected_class, or if the inner plaintext envelope’s class drifts from the cipher layer (which would indicate tampering since the AAD binds them).

  • DecryptionError – If the AEAD tag fails to verify.

  • EnvelopeVersionError – If the inner plaintext envelope’s schema version differs from max_supported_version.

reencrypt_envelope_file(path, envelope_type, *, expected_class, master_key_provider, hkdf_context, max_supported_version)[source]

Re-encrypt a single plaintext envelope file in place.

Read once: if path is already a CipherEnvelope, return False (already ciphertext, nothing to do). Otherwise parse as a plaintext Envelope and re-write through save_encrypted_envelope().

Returns True iff the file was re-encrypted, False if the file was already ciphertext or did not exist. The atomic-replace pattern from save_encrypted_envelope() governs the on-disk rewrite: a crash mid-rewrite leaves either the plaintext OR the ciphertext on disk, never a torn write.

Repository load paths are strict ciphertext-only; this function is the only sanctioned path that touches plaintext envelopes.

Parameters:

  • path (Path) – Target file to re-encrypt in place.

  • envelope_type (type[Envelope[TypeVar]]) – The parameterised envelope class.

  • expected_class (SensitivityClass) – The SensitivityClass the consumer expects.

  • master_key_provider (MasterKeyProvider) – MasterKeyProvider supplying the master key used to derive the per-consumer encryption key via HKDF-SHA256.

  • hkdf_context (bytes) – Per-consumer context bytes; MUST match those used for subsequent load calls.

  • max_supported_version (int) – Current inner-envelope schema version the consumer expects.

Return type:

bool