gemstone_utils.key_mgmt

KEK derivation, wrap/unwrap, passphrase loading, and rotation helpers.

exception gemstone_utils.key_mgmt.KEKVerificationError(secret_name, last_updated_iso)

Bases: ValueError

Raised when a derived KEK fails KEK-check decryption.

Variables:

• secret_name – Configured secret name from init().

• last_updated_iso – UTC ISO timestamp of last passphrase rotation, or "unknown".

Parameters:

• secret_name (str)

• last_updated_iso (str)

gemstone_utils.key_mgmt.derive_and_verify_kek(passphrase, kdf_params, kek_check_record, last_updated=None)

Derive a KEK and verify it against the KEK-check record.

Parameters:

• passphrase (str) – Vault passphrase.

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

• kek_check_record (KeyRecord) – KEK-check KeyRecord.

• last_updated (datetime | None) – Optional timestamp for KEKVerificationError.

Returns:

Verified KEK bytes.

Raises:

• KEKVerificationError – If the check blob does not decrypt correctly.

• ValueError – If KDF params are invalid.

Return type:

bytes

gemstone_utils.key_mgmt.init(secret_name, check_plaintext, env_allowed=False, env_var_name=None)

Configure module-level key management settings.

Must be called before make_kek_check_record, verify_kek, or load_passphrase.

Parameters:

• secret_name (str) – Name passed to resolve_secret("secret:...").

• check_plaintext (bytes) – Fixed plaintext bytes encrypted into the KEK-check blob.

• env_allowed (bool) – Whether to fall back to env: when the secret mount fails.

• env_var_name (str | None) – Environment variable name for fallback (defaults to secret_name).

Return type:

None

gemstone_utils.key_mgmt.load_keyctx(kek, record)

Build a KeyContext from a KEK and DEK KeyRecord.

Parameters:

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

• record (KeyRecord) – DEK record (keyid must be set).

Returns:

KeyContext with unwrapped key and metadata from record.

Raises:

ValueError – If record is a KEK-check record.

Return type:

KeyContext

gemstone_utils.key_mgmt.load_passphrase()

Load the vault passphrase via resolve_secret.

Tries secret:{SECRET_NAME}, then env:{ENV_VAR_NAME} when ENV_ALLOWED is true.

Returns:

Passphrase string.

Raises:

• RuntimeError – If init() was not called.

• ValueError – If neither configured source resolves.

Return type:

str

gemstone_utils.key_mgmt.make_kek_check_record(kek, alg='A256GCM')

Build a KEK-check KeyRecord (keyid is None).

Parameters:

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

• alg (str) – Symmetric wrap algorithm id.

Returns:

KeyRecord with encrypted CHECK_PLAINTEXT from init().

Raises:

RuntimeError – If init() was not called.

Return type:

KeyRecord

gemstone_utils.key_mgmt.recommended_kdf_params(salt=None)

Return params for the library’s current recommended KDF.

Today: PBKDF2-HMAC-SHA256 via kdf.pbkdf2.recommended_pbkdf2_params.

Parameters:

salt (bytes | None) – Optional fixed salt; random 16-byte salt when omitted.

Returns:

Params dict suitable for derive_kek and set_kdf_params.

Return type:

dict[str, Any]

gemstone_utils.key_mgmt.reencrypt_keys(old_kek, new_kek, records, new_alg=None)

Re-wrap DEK records under a new KEK.

Parameters:

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

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

• records (Iterable[KeyRecord]) – Iterable of DEK KeyRecord instances.

• new_alg (str | None) – Optional wrap algorithm; preserves each record’s alg when omitted.

Returns:

New KeyRecord list with updated ciphertext. KEK-check records (keyid is None) are skipped.

Return type:

list[KeyRecord]

gemstone_utils.key_mgmt.rotate_kek(old_kek, new_kek, records, new_alg=None)

Rotate KEK: rewrap DEKs and produce a new KEK-check record.

Parameters:

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

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

• records (Iterable[KeyRecord]) – Iterable of DEK KeyRecord instances.

• new_alg (str | None) – Optional wrap algorithm for all DEK records.

Returns:

Tuple (new_kek_check_record, updated_dek_records).

Return type:

tuple[KeyRecord, list[KeyRecord]]

gemstone_utils.key_mgmt.unwrap_key(kek, record)

Decrypt a DEK from a wrapped KeyRecord.

Parameters:

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

• record (KeyRecord) – DEK record (keyid must be set).

Returns:

Unwrapped key bytes.

Raises:

ValueError – If record is a KEK-check record.

Return type:

bytes

gemstone_utils.key_mgmt.verify_kek(kek, record, last_updated=None)

Verify that kek decrypts the KEK-check record.

Parameters:

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

• record (KeyRecord) – KEK-check record (keyid must be None).

• last_updated (datetime | None) – Optional timestamp included in KEKVerificationError.

Raises:

• ValueError – If record is not a KEK-check record.

• RuntimeError – If init() was not called.

• KEKVerificationError – If decryption does not yield CHECK_PLAINTEXT.

Return type:

None

gemstone_utils.key_mgmt.wrap_key(kek, key, alg='A256GCM')

Encrypt key material under a KEK.

Parameters:

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

• key (bytes) – Plaintext key bytes to wrap.

• alg (str) – Symmetric wrap algorithm id.

Returns:

KeyRecord with keyid=None; caller sets keyid when persisting.

Return type:

KeyRecord