gemstone_utils.sqlalchemy.key_storage

Persisted KDF params, KEK canaries, wrapped DEKs, and resolver helpers.

class gemstone_utils.sqlalchemy.key_storage.GemstoneKeyKdf(**kwargs)[source]

Bases: GemstoneDB

KEK slot: persisted KDF JSON, canary wire, and re-encrypt flag.

key_id is the KEK slot id (canonical UUID) referenced as segment 2 inside wrapped DEK and canary wire strings.

app_reencrypt_pending
canary_wrapped
created_at
key_id
params
updated_at
class gemstone_utils.sqlalchemy.key_storage.GemstoneKeyRecord(**kwargs)[source]

Bases: GemstoneDB

Wrapped DEK row (no KEK canary).

wrapped uses the five-part encrypted-field wire format; segment 2 is the KEK slot id (GemstoneKeyKdf.key_id), not this row’s primary key. data_alg is the algorithm for application field encryption.

created_at
data_alg
is_active
key_id
updated_at
wrapped
gemstone_utils.sqlalchemy.key_storage.get_kdf_params(session, key_id)[source]

Load persisted KDF params for a KEK slot.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_id (str) – KEK slot primary key.

Returns:

Parsed params dict.

Raises:

KeyError – If no row exists for key_id.

Return type:

dict

gemstone_utils.sqlalchemy.key_storage.get_wrapped(session, logical_key_id)[source]

Return the wrapped wire string for a DEK row.

Parameters:

  • session (Session) – SQLAlchemy session.

  • logical_key_id (str) – DEK primary key.

Returns:

GemstoneKeyRecord.wrapped text.

Raises:

KeyError – If no DEK row exists.

Return type:

str

gemstone_utils.sqlalchemy.key_storage.iter_kek_slots(session)[source]

Iterate KEK slot rows ordered by key_id.

Parameters:

session (Session) – SQLAlchemy session.

Yields:

GemstoneKeyKdf instances.

Return type:

Iterator[GemstoneKeyKdf]

gemstone_utils.sqlalchemy.key_storage.iter_wrapped_rows(session, key_ids=None)[source]

Iterate DEK rows ordered by key_id.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_ids (Iterable[str] | None) – Optional subset of DEK ids; all rows when omitted.

Yields:

GemstoneKeyRecord instances.

Return type:

Iterator[GemstoneKeyRecord]

gemstone_utils.sqlalchemy.key_storage.keyrecord_to_wire(record, wrap_key_id)[source]

Serialize a KeyRecord to wire form.

Parameters:

  • record (KeyRecord) – Wrapped key record.

  • wrap_key_id (str) – KEK slot id for wire segment 2.

Returns:

Five-part wire string.

Return type:

str

gemstone_utils.sqlalchemy.key_storage.make_keyctx_resolver(*, get_session=<function get_session>, load_passphrase=<function load_passphrase>, max_cache_size=0)[source]

Build a KeyContext resolver for EncryptedString.

Loads the DEK row, derives the KEK from persisted KDF params and passphrase, unwraps the DEK, and returns KeyContext with alg from data_alg.

Parameters:

  • get_session (Callable[[], Session]) – Callable returning a new session (default get_session).

  • load_passphrase (Callable[[], str]) – Callable returning the vault passphrase.

  • max_cache_size (int) – When > 0, cache resolved contexts in-process.

Returns:

Resolver callable (dek_keyid: str) -> KeyContext.

Return type:

Callable[[str], KeyContext]

gemstone_utils.sqlalchemy.key_storage.new_kdf_params(salt=None)[source]

Return recommended KDF params for new KEK slots.

Alias of gemstone_utils.key_mgmt.recommended_kdf_params().

Parameters:

salt (bytes | None) – Optional fixed salt.

Returns:

Params dict for derive_kek() and set_kdf_params().

Return type:

dict

gemstone_utils.sqlalchemy.key_storage.put_keyrecord(session, *, key_id, wrapped, data_alg='A256GCM', is_active=False)[source]

Insert a new DEK row.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_id (str) – New DEK primary key (must not exist).

  • wrapped (str) – Five-part wire string.

  • data_alg (str) – Field encryption algorithm id for KeyContext.alg.

  • is_active (bool) – When True, clears is_active on other DEK rows.

Raises:

ValueError – If key_id exists or data_alg is unsupported.

Return type:

None

gemstone_utils.sqlalchemy.key_storage.rewrap_key_records(session, *, old_kek, new_kek, old_wrap_key_id, new_wrap_key_id, key_ids=None, new_alg=None)[source]

Rewrap DEK and canary wires in place under a new KEK.

Run inside an explicit transaction, for example:

with session.begin():
    rewrap_key_records(session, ...)

Does not commit the session.

Parameters:

  • session (Session) – SQLAlchemy session.

  • old_kek (bytes) – Current key-encryption key.

  • new_kek (bytes) – New key-encryption key.

  • old_wrap_key_id (str) – KEK slot id in existing wire segment 2.

  • new_wrap_key_id (str) – KEK slot id for updated wires.

  • key_ids (Iterable[str] | None) – Optional DEK subset; all DEK rows when omitted.

  • new_alg (str | None) – Optional new wrap algorithm for DEK blobs.

Raises:

ValueError – If required rows are missing or wire segment 2 mismatches.

Return type:

None

gemstone_utils.sqlalchemy.key_storage.set_app_reencrypt_pending(session, key_id, pending)[source]

Set the application re-encrypt flag on a KEK slot.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_id (str) – KEK slot primary key.

  • pending (bool) – New flag value.

Raises:

KeyError – If the KEK slot row does not exist.

Return type:

None

gemstone_utils.sqlalchemy.key_storage.set_kdf_params(session, key_id, params)[source]

Insert or update KDF params for a KEK slot.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_id (str) – KEK slot primary key.

  • params (dict) – KDF params dict (must include supported "kdf").

Raises:

ValueError – If params["kdf"] is missing or unsupported.

Return type:

None

gemstone_utils.sqlalchemy.key_storage.set_kek_canary(session, key_id, canary_wrapped)[source]

Set canary_wrapped on a KEK slot row.

Call set_kdf_params() first to create the row.

Parameters:

  • session (Session) – SQLAlchemy session.

  • key_id (str) – KEK slot primary key.

  • canary_wrapped (str) – KEK-check wire string.

Raises:

KeyError – If the KEK slot row does not exist.

Return type:

None

gemstone_utils.sqlalchemy.key_storage.unwrap_stored_key(kek, logical_key_id, wire)[source]

Unwrap DEK bytes from a stored wire string.

Parameters:

  • kek (bytes) – Key-encryption key bytes.

  • logical_key_id (str) – DEK row primary key (must match persisted id).

  • wire (str) – Wrapped DEK wire string.

Returns:

Unwrapped key bytes.

Return type:

bytes

gemstone_utils.sqlalchemy.key_storage.wire_to_keyrecord(logical_key_id, wire)[source]

Parse a wire string into a KeyRecord.

Parameters:

  • logical_key_id (str | None) – DEK id to store on the record (may be None for canary).

  • wire (str) – Five-part encrypted-field wire string.

Returns:

KeyRecord with ciphertext and params from the wire.

Return type:

KeyRecord

gemstone_utils.sqlalchemy.key_storage.wire_wrap(wrap_key_id, kek, plaintext_key_material, alg='A256GCM')[source]

Wrap key material as an encrypted-field wire string.

Parameters:

  • wrap_key_id (str) – KEK slot id (wire segment 2).

  • kek (bytes) – Key-encryption key bytes.

  • plaintext_key_material (bytes) – Key bytes to wrap.

  • alg (str) – Symmetric wrap algorithm id.

Returns:

Five-part wire string.

Return type:

str