encryption

Field-level encryption for sensitive cache columns.

This module provides SQLAlchemy TypeDecorators (EncryptedString, EncryptedJSON) that transparently encrypt values on write and decrypt on read. Stores and ORM consumers do not need to know that encryption is happening — they continue to read and write plain Python str / list / dict values.

Encryption uses the existing _AESEncryption (AES-GCM, 256-bit) helper from bitfount.encryption.encryption. The encryption key is derived from the pod's existing RSA private key (the same pod_rsa.pem the pod uses for federated messaging) via _FernetEncryption.derive_key_from_path. Callers don't need to configure anything: open_background_cache accepts the pod's key path (pod.pod_key_path) and binds it into this module; the key is then derived lazily on first use.

Stored format

Encrypted columns are stored as TEXT. When encryption is enabled, the stored string is:

enc:v1:<base64(nonce || ciphertext)>

When encryption is disabled, the column stores the original plaintext (strings) or its JSON-serialised form (EncryptedJSON). Both shapes round-trip cleanly: the decorator detects the enc:v1: prefix on read and chooses to decrypt or pass through, so encrypted and plaintext rows can coexist (useful if encryption is toggled on/off during development).

Toggling

Encryption is enabled when config.settings.cache_encryption_enabled is True and a key is resolvable. For tests, prefer the context manager:

with CacheEncryption.disabled():
    ...

or, to round-trip with a known key, CacheEncryption.set_key(raw_key).

Threat model note

The PK columns task_hash and bitfount_patient_id are intentionally NOT encrypted: PKs need to support equality lookups and uniqueness checks, and bitfount_patient_id is already opaque per the EHR/MRN it came from.

Module

Functions

compute_blind_index

def compute_blind_index(plaintext: str) ‑> str | None:

Return a keyed-HMAC blind index for plaintext, or None if disabled.

Keyed by the cache encryption key (derived from the pod RSA key), the digest supports indexed equality lookups over an otherwise-encrypted value without decrypting rows, and is not reversible. Returns None when encryption is disabled, since no key is available and the value would otherwise be an unkeyed hash of PHI.

Arguments

• plaintext: The value to index (e.g. a canonical file path).

Returns The hex HMAC-SHA256 digest, or None when encryption is disabled.

Classes

CacheEncryption

class CacheEncryption():

Singleton holding the cache encryption key and on/off state.

The class-level state is intentionally global because SQLAlchemy TypeDecorator instances live in the ORM module and have no natural way to receive per-cache state. In practice a process talks to one cache at a time, so a singleton is fine.

Tests should prefer the disabled() context manager rather than mutating state directly, so the original state is restored on exit.

Static methods

---

disabled

def disabled(cls) ‑> collections.abc.Iterator[None]:

Temporarily force encryption off (e.g. inside a test).

enabled

def enabled(cls, key: bytes | None = None) ‑> collections.abc.Iterator[None]:

Temporarily force encryption on, optionally with a specific key.

get_key

def get_key() ‑> bytes:

Return the raw 32-byte AES key, deriving from the pod RSA key.

Raises

• RuntimeError: if the pod context is not bound or key derivation fails.

is_enabled

def is_enabled() ‑> bool:

Return True when encryption should be applied on read/write.

reset

def reset() ‑> None:

Forget all overrides, cached key, and bound pod. Primarily for tests.

set_key

def set_key(key: bytes | None) ‑> None:

Override the resolved key.

Pass None to clear the cached key so the next call re-derives from the bound pod's RSA key. Primarily for tests.

set_pod_key_path

def set_pod_key_path(pod_key_path: Path | None) ‑> None:

Bind the absolute path to the pod's RSA private key.

Arguments

• pod_key_path: Path to pod_rsa.pem for the pod that owns the cache being opened. None clears the binding.

Called from open_background_cache so callers don't have to thread the pod identity through to the cache layer manually. Changing the bound path clears the cached derived key.

EncryptedJSON

class EncryptedJSON(*args: Any, **kwargs: Any):

A nullable JSON column that transparently encrypts its contents.

The Python-side value is the deserialised JSON object (list / dict / etc.). The DB-side storage is TEXT — either an encrypted blob (when enabled) or the raw json.dumps output (when disabled).

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Ancestors

• typing.Generic

Variables

• static cache_ok

• static impl - A variably sized string type.

  In SQL, usually corresponds to CLOB or TEXT. In general, TEXT objects do not have a length; while some databases will accept a length argument here, it will be rejected by others.

Methods

---

process_bind_param

def process_bind_param(self, value: Any, dialect: Any) ‑> str | None:

Serialise and encrypt the JSON value before binding to the SQL statement.

process_result_value

def process_result_value(self, value: Any, dialect: Any) ‑> Any:

Decrypt and deserialise the JSON value when reading from the database.

EncryptedString

class EncryptedString(*args: Any, **kwargs: Any):

A nullable string column that transparently encrypts its contents.

On write: if encryption is enabled, the plaintext is encrypted and stored with the enc:v1: prefix. If disabled, the plaintext is stored as-is.

On read: if the stored value carries the prefix, it is decrypted; otherwise it is returned verbatim. This lets encrypted and plaintext rows coexist (e.g. across a toggle event).

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Ancestors

• typing.Generic

Variables

• static cache_ok

• static impl - A variably sized string type.

  In SQL, usually corresponds to CLOB or TEXT. In general, TEXT objects do not have a length; while some databases will accept a length argument here, it will be rejected by others.

Methods

---

process_bind_param

def process_bind_param(self, value: Any, dialect: Any) ‑> str | None:

Encrypt the value before binding it to the SQL statement.

process_result_value

def process_result_value(self, value: Any, dialect: Any) ‑> str | None:

Decrypt the value when reading from the database.

MigratingJSON

class MigratingJSON(*args: Any, **kwargs: Any):

A JSON column migrating from encrypted to plaintext at rest.

The JSON counterpart of MigratingString: stores the json.dumps output as plaintext on write — never encrypts, even when cache encryption is enabled — and on read decrypts a legacy enc:v1: blob or parses plaintext directly. Lets a column that used to be EncryptedJSON be read correctly while pre-existing ciphertext rows are rewritten to plaintext (on the next write of each row, or via a backfill).

Construct a :class:.TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute 'as-is', ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

Ancestors

• typing.Generic

Variables

• static cache_ok

• static impl - A variably sized string type.

  In SQL, usually corresponds to CLOB or TEXT. In general, TEXT objects do not have a length; while some databases will accept a length argument here, it will be rejected by others.

Methods

---

process_bind_param

def process_bind_param(self, value: Any, dialect: Any) ‑> str | None:

Serialise the JSON value and store it as plaintext (no encryption).

process_result_value

def process_result_value(self, value: Any, dialect: Any) ‑> Any:

Decrypt a legacy encrypted blob; parse plaintext directly.

MigratingString

class MigratingString(*args: Any, **kwargs: Any):

A string column migrating from encrypted to plaintext at rest.

For a column that used to be EncryptedString but is now stored plaintext so it can be substring-searched and ordered in

Ancestors

• typing.Generic

Variables

• static cache_ok

• static impl - A variably sized string type.

  In SQL, usually corresponds to CLOB or TEXT. In general, TEXT objects do not have a length; while some databases will accept a length argument here, it will be rejected by others.

Methods

---

process_bind_param

def process_bind_param(self, value: Any, dialect: Any) ‑> str | None:

Store the value as plaintext (no encryption).

process_result_value

def process_result_value(self, value: Any, dialect: Any) ‑> str | None:

Decrypt a legacy encrypted value; pass plaintext through.

---