Compare commits
13
Commits
v0.1.5
..
19528e89f8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
19528e89f8 | ||
|
|
1edd7ddfea | ||
|
|
4b9820b18e | ||
|
|
6d370a45e1 | ||
|
|
581ff712e0 | ||
|
|
8dffb6bc61 | ||
|
|
dc8f80c690 | ||
|
|
dfd794159e | ||
|
|
d0c3bcd7c8 | ||
|
|
857e9380c6 | ||
|
|
abf7491d26 | ||
|
|
7287a52947 | ||
|
|
d446f50942 |
@@ -11,18 +11,18 @@ and storage-agnostic.
|
||||
`requirements.txt`:
|
||||
|
||||
```
|
||||
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.5
|
||||
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10
|
||||
```
|
||||
|
||||
Direct:
|
||||
|
||||
```bash
|
||||
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.5"
|
||||
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10"
|
||||
```
|
||||
|
||||
Requires `cryptography` (pulled transitively).
|
||||
|
||||
Drop the `@v0.1.5` suffix from the line above to install the latest unpinned.
|
||||
Drop the `@v0.1.10` suffix from the line above to install the latest unpinned.
|
||||
|
||||
## First-time setup
|
||||
|
||||
@@ -80,10 +80,12 @@ enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ...,
|
||||
plain = crypto.decrypt_data(enc) # -> {"ssn": "..."}
|
||||
```
|
||||
|
||||
Dict keys must be `str`. `encrypt_data` raises `TypeError` on a non-str key (e.g. an
|
||||
int-keyed dict of Discord snowflakes) instead of silently stringifying it — the
|
||||
underlying JSON encoding has no other key type, so a coerced key would come back out
|
||||
of `decrypt_data` as a `str` and no longer match the original lookup key.
|
||||
Dict keys must be `str`, at any nesting depth (including a dict nested inside a list
|
||||
or tuple). `encrypt_data` raises `TypeError` on a non-str key anywhere in the payload
|
||||
(e.g. an int-keyed dict of Discord snowflakes, even nested a few levels down) instead
|
||||
of silently stringifying it — the underlying JSON encoding has no other key type, so a
|
||||
coerced key would come back out of `decrypt_data` as a `str` and no longer match the
|
||||
original lookup key.
|
||||
|
||||
For whole records: `decrypt_record(crypto, doc)` decrypts every `{secure, iv, data}`
|
||||
field (nested up to `traversal_level`, default 2); `is_encrypted_record(doc)` reports
|
||||
@@ -98,10 +100,11 @@ if is_encrypted_record(doc):
|
||||
doc = decrypt_record(crypto, doc)
|
||||
```
|
||||
|
||||
`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is
|
||||
exhausted, so it reliably reports `True` for a blob left behind by a shallower
|
||||
`is_encrypted_record` always falls back to an unbounded-depth scan whenever its bounded
|
||||
pass finds nothing, so it reliably reports `True` for a blob left behind by a shallower
|
||||
`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
|
||||
rotation, regardless of how deep the blob is nested.
|
||||
rotation, regardless of how deep the blob is nested (including inside a list or tuple
|
||||
at any depth) and regardless of which `traversal_level` you pass; it never false-negatives.
|
||||
|
||||
Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
|
||||
= `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
|
||||
@@ -177,6 +180,18 @@ The lib never touches a database; only the caller's storage layer differs.
|
||||
- The scheme is envelope/hybrid encryption (AES-256-GCM data key wrapped by RSA-OAEP).
|
||||
Using it does not by itself confer PCI-DSS or any other compliance — that is a
|
||||
whole-system property.
|
||||
- `initialize(master_key)` requires exactly 32 bytes (`bytes`, `len == 32`) — a
|
||||
16/24-byte key or a `str` raises `ValueError` instead of silently downgrading to
|
||||
AES-128/192 or failing late at first encrypt.
|
||||
- Wrap/unwrap (`encrypt_aes_key_with_rsa`, `decrypt_aes_key_with_rsa`,
|
||||
`authorize_system`, `bootstrap`, `rotate_master_key`) is **RSA-only**. A non-RSA
|
||||
key (e.g. Ed25519/EC) loads and fingerprints fine but raises a clear `ValueError`
|
||||
at wrap/unwrap rather than a raw `AttributeError`.
|
||||
- `fingerprint_data` serializes non-JSON-native values (datetime/date/time,
|
||||
bytes/bytearray, ObjectId-like objects) via a stable `default=` handler instead of
|
||||
raising `TypeError`, and type-tags dict keys internally so `{1: "a"}` and
|
||||
`{"1": "a"}` no longer collide to the same fingerprint. Never logs the data it
|
||||
fingerprints.
|
||||
|
||||
## Versioning
|
||||
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "envelope_crypto"
|
||||
version = "0.1.5"
|
||||
version = "1.0.0"
|
||||
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from importlib.metadata import version, PackageNotFoundError
|
||||
|
||||
from .envelope_crypto import (
|
||||
EnvelopeCrypto,
|
||||
DocumentCrypto,
|
||||
@@ -12,6 +14,11 @@ from .envelope_crypto import (
|
||||
fingerprint_data,
|
||||
)
|
||||
|
||||
try:
|
||||
__version__ = version("envelope_crypto")
|
||||
except PackageNotFoundError:
|
||||
__version__ = "0.0.0+unknown"
|
||||
|
||||
__all__ = [
|
||||
"EnvelopeCrypto",
|
||||
"DocumentCrypto",
|
||||
@@ -24,4 +31,5 @@ __all__ = [
|
||||
"decrypt_document",
|
||||
"decrypt_dict",
|
||||
"fingerprint_data",
|
||||
"__version__",
|
||||
]
|
||||
|
||||
@@ -1,68 +1,23 @@
|
||||
"""
|
||||
envelope encryption for dict records
|
||||
|
||||
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and
|
||||
that key is wrapped (RSA-OAEP) per authorized system's public key (KEK) for
|
||||
distribution. the wrapped key is stored by the caller, keyed by fingerprint;
|
||||
each system unwraps its own copy with its private key. this is the same
|
||||
envelope-encryption pattern used by KMS-style systems.
|
||||
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped
|
||||
(RSA-OAEP) per authorized system's public key (KEK) for distribution and stored by
|
||||
the caller, keyed by fingerprint. RSA-envelope only - a non-RSA key (e.g. Ed25519/EC)
|
||||
loads and fingerprints fine but raises ValueError at wrap/unwrap. never logs key
|
||||
material (DEK, PEM, wrapped key) - only fingerprints and counts. config-free and
|
||||
storage-agnostic; see README for bootstrap/boot/authorize/rotate flows.
|
||||
|
||||
from envelope_crypto import EnvelopeCrypto
|
||||
|
||||
crypto = EnvelopeCrypto()
|
||||
crypto.initialize(master_key) # 32-byte AES DEK
|
||||
crypto.initialize(master_key) # exactly 32 bytes (AES-256)
|
||||
enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data}
|
||||
plain = crypto.decrypt_data(enc) # -> original
|
||||
|
||||
first-time setup: generate the DEK and wrap it for the first system in one call,
|
||||
then verify the pipeline before storing anything:
|
||||
|
||||
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
|
||||
crypto.self_test("public_key.pem", "private_key.pem") # raises if anything is wrong
|
||||
caller_store({"_id": fingerprint, "key": wrapped}) # the only record of the DEK
|
||||
|
||||
boot (already set up): fingerprint own pubkey, fetch the wrapped DEK, unwrap:
|
||||
|
||||
fp = crypto.get_rsa_key_fingerprint("public_key.pem")
|
||||
record = caller_lookup(fp)
|
||||
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
|
||||
|
||||
authorize another system (this instance must already hold the DEK):
|
||||
|
||||
fp, wrapped = crypto.authorize_system(other_pub_path)
|
||||
caller_store({"_id": fp, "key": wrapped})
|
||||
|
||||
deauthorize: caller deletes that fingerprint's record. note this stops future
|
||||
unwraps but does not revoke a DEK already in a running system's memory — rotate
|
||||
if compromised.
|
||||
|
||||
rotate (new DEK + re-encrypt): generate a new DEK, wrap for the still-authorized
|
||||
set, then re-encrypt existing records old -> new:
|
||||
|
||||
new_key, wrapped = crypto.rotate_master_key([pub_a, pub_b])
|
||||
new_crypto = EnvelopeCrypto(); new_crypto.initialize(new_key)
|
||||
for record in caller_iter():
|
||||
caller_update(new_crypto.reencrypt(crypto, record))
|
||||
|
||||
reencrypt/is_encrypted_record/decrypt_record all detect a bare {secure, iv, data}
|
||||
blob used AS the whole record (the file-storage pattern), not just blobs nested
|
||||
under a key. reencrypt fails loud (raises) rather than silently leaving a field
|
||||
under the old key — including when a blob sits deeper than `traversal_level`;
|
||||
is_encrypted_record falls back to an unbounded-depth scan past traversal_level so
|
||||
it reliably catches leftovers as a post-rotation audit.
|
||||
|
||||
config-free: the host supplies the DEK and RSA key paths; this lib never imports
|
||||
config, configures logging, or touches a database. storage-agnostic — the
|
||||
encrypted blob is a plain dict; store it in mongo, a sql json column, or a file.
|
||||
|
||||
encrypt_data/decrypt_data round-trip a dict only when every key is a str —
|
||||
json (the wire format under the hood) has no other key type, so encrypt_data
|
||||
raises TypeError on a non-str key (e.g. an int-keyed dict of discord snowflakes)
|
||||
rather than silently stringifying it and losing the original key on decrypt.
|
||||
|
||||
naming: EnvelopeCrypto is canonical. PCICrypto / DocumentCrypto / RecordCrypto are
|
||||
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict
|
||||
function variants are the same functions — use whichever fits your storage.
|
||||
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function
|
||||
variants are the same functions - use whichever fits your storage.
|
||||
"""
|
||||
|
||||
import os
|
||||
@@ -75,20 +30,28 @@ from typing import Any, Dict, List, Optional, Tuple, Union
|
||||
|
||||
from cryptography.exceptions import UnsupportedAlgorithm
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
from cryptography.hazmat.primitives.serialization import load_ssh_public_key
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
_OAEP_PADDING = padding.OAEP(
|
||||
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
||||
algorithm=hashes.SHA256(),
|
||||
label=None,
|
||||
)
|
||||
|
||||
|
||||
def _password_mismatch_message(pw: Optional[bytes]) -> str:
|
||||
"""clear ValueError text for a TypeError raised by a password/encryption mismatch"""
|
||||
if pw is not None:
|
||||
return "password was given but private key is not encrypted"
|
||||
return "private key is encrypted but no password was provided"
|
||||
|
||||
|
||||
def _load_private_key(key_data: bytes, pw: Optional[bytes]):
|
||||
"""load a PEM or OpenSSH private key, normalizing the missing-password error
|
||||
|
||||
cryptography raises TypeError when a key is encrypted but no password was given
|
||||
(PEM raises it on the first call; OpenSSH raises it inside the openssh fallback).
|
||||
both are normalized to a clear ValueError so callers see one error type.
|
||||
"""
|
||||
"""load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error"""
|
||||
try:
|
||||
return serialization.load_pem_private_key(key_data, password=pw)
|
||||
except ValueError as error:
|
||||
@@ -96,26 +59,16 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
|
||||
try:
|
||||
return serialization.load_ssh_private_key(key_data, password=pw)
|
||||
except TypeError as ssh_error:
|
||||
raise ValueError(
|
||||
"private key is encrypted but no password was provided"
|
||||
) from ssh_error
|
||||
raise ValueError(_password_mismatch_message(pw)) from ssh_error
|
||||
if pw is not None:
|
||||
# a password was given but the PEM load still failed — most likely a wrong
|
||||
# password; give a clearer message than cryptography's raw "Bad decrypt"
|
||||
raise ValueError("could not load private key (wrong password or malformed key)") from error
|
||||
raise error
|
||||
except TypeError as error:
|
||||
raise ValueError(
|
||||
"private key is encrypted but no password was provided"
|
||||
) from error
|
||||
raise ValueError(_password_mismatch_message(pw)) from error
|
||||
|
||||
|
||||
def _fingerprint_of(public_key) -> str:
|
||||
"""base64 SHA-256 fingerprint of an already-loaded public key
|
||||
|
||||
factored so callers that already hold a loaded key (e.g. encrypt_aes_key_with_rsa)
|
||||
don't re-open and re-parse the key file just to fingerprint it.
|
||||
"""
|
||||
"""base64 SHA-256 fingerprint of an already-loaded public key"""
|
||||
key_bytes = public_key.public_bytes(
|
||||
encoding=serialization.Encoding.DER,
|
||||
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
@@ -131,25 +84,38 @@ def _is_blob(value: Any) -> bool:
|
||||
|
||||
|
||||
def _has_encrypted_field(record: Any) -> bool:
|
||||
"""unbounded-depth scan: does record (or anything nested under it) contain a blob
|
||||
"""unbounded-depth scan: does record (or anything nested under it, incl. list/tuple items) contain a blob"""
|
||||
if isinstance(record, dict):
|
||||
if _is_blob(record):
|
||||
return True
|
||||
return any(_has_encrypted_field(value) for value in record.values())
|
||||
if isinstance(record, (list, tuple)):
|
||||
return any(_has_encrypted_field(item) for item in record)
|
||||
return False
|
||||
|
||||
used to detect a blob left behind by a depth-limited traversal — no traversal_level
|
||||
cutoff here, since the whole point is to catch what a bounded pass would miss.
|
||||
"""
|
||||
if not isinstance(record, dict):
|
||||
return False
|
||||
if _is_blob(record):
|
||||
return True
|
||||
return any(_has_encrypted_field(value) for value in record.values())
|
||||
|
||||
def _non_str_key_types(data: Any) -> List[str]:
|
||||
"""unbounded-depth scan collecting type names of non-str dict keys, incl. dicts nested inside list/tuple items"""
|
||||
found: List[str] = []
|
||||
if isinstance(data, dict):
|
||||
for key, value in data.items():
|
||||
if not isinstance(key, str):
|
||||
found.append(type(key).__name__)
|
||||
found.extend(_non_str_key_types(value))
|
||||
elif isinstance(data, (list, tuple)):
|
||||
for item in data:
|
||||
found.extend(_non_str_key_types(item))
|
||||
return found
|
||||
|
||||
|
||||
def _require_rsa(key) -> None:
|
||||
"""raise ValueError unless key is an RSA public or private key"""
|
||||
if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)):
|
||||
raise ValueError(f"RSA key required for envelope wrap/unwrap, got {type(key).__name__}")
|
||||
|
||||
|
||||
def _load_public_key(key_data: bytes):
|
||||
"""load a PEM or OpenSSH public key, normalizing non-key input to ValueError
|
||||
|
||||
load_ssh_public_key raises UnsupportedAlgorithm (not ValueError) on non-SSH/garbage
|
||||
input; normalize it so a bad public key always surfaces as a clear ValueError,
|
||||
consistent with the private-key path.
|
||||
"""
|
||||
"""load a PEM or OpenSSH public key, normalizing non-key input to ValueError"""
|
||||
try:
|
||||
return serialization.load_pem_public_key(key_data)
|
||||
except ValueError:
|
||||
@@ -174,10 +140,20 @@ class EnvelopeCrypto:
|
||||
def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]:
|
||||
"""first-time setup: generate a DEK and wrap it for the first system
|
||||
|
||||
returns (crypto, fingerprint, wrapped_key) — an initialized instance plus
|
||||
the record to store as the first authorization. the plaintext DEK is never
|
||||
returned or persisted; it survives only as the wrapped copy. run self_test
|
||||
before storing to confirm the keypair round-trips.
|
||||
the plaintext DEK is never returned or persisted; it survives only as the
|
||||
wrapped copy. run self_test before storing to confirm the keypair round-trips.
|
||||
|
||||
Args:
|
||||
rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
|
||||
the first system's RSA public key.
|
||||
is_file: True (default) treats rsa_public_key as a path.
|
||||
|
||||
Returns:
|
||||
(crypto, fingerprint, wrapped_key): an initialized instance plus the
|
||||
record to store as the first authorization.
|
||||
|
||||
Raises:
|
||||
ValueError: rsa_public_key is not an RSA key, or is malformed.
|
||||
"""
|
||||
crypto = cls()
|
||||
crypto.initialize(crypto.create_aes_key())
|
||||
@@ -190,14 +166,24 @@ class EnvelopeCrypto:
|
||||
) -> bool:
|
||||
"""verify the full pipeline against a keypair; raises on any mismatch
|
||||
|
||||
round-trips sample data through this instance's DEK, then wraps the DEK
|
||||
with the public key and unwraps with the private key, confirming they
|
||||
match. run after bootstrap (or anytime as a health check) to catch a bad
|
||||
keypair or wrong key path before relying on it. is_file (default True) is
|
||||
threaded through to both the public and private key loads, so
|
||||
is_file=False correctly treats both rsa_public_key and rsa_private_key as
|
||||
in-memory PEM/OpenSSH data rather than opening either as a file path.
|
||||
returns True on success.
|
||||
round-trips sample data through this instance's DEK, then wraps and unwraps
|
||||
the DEK with the given keypair. run after bootstrap or as a health check.
|
||||
|
||||
Args:
|
||||
rsa_public_key: public half of the keypair to verify (path or, if
|
||||
is_file=False, raw PEM/OpenSSH data).
|
||||
rsa_private_key: private half of the keypair to verify.
|
||||
is_file: True (default) treats both keys as paths; False treats both
|
||||
as in-memory PEM/OpenSSH data.
|
||||
password: password for an encrypted private key, if any.
|
||||
|
||||
Returns:
|
||||
True on success.
|
||||
|
||||
Raises:
|
||||
ValueError: instance not initialized with a data key.
|
||||
RuntimeError: data round-trip fails, or the keypair does not pair
|
||||
(unwrap fails or the recovered key mismatches).
|
||||
"""
|
||||
if not self.master_key:
|
||||
raise ValueError("self_test: not initialized with a key")
|
||||
@@ -221,7 +207,15 @@ class EnvelopeCrypto:
|
||||
return True
|
||||
|
||||
def initialize(self, master_key: bytes) -> None:
|
||||
"""arm the instance with the AES data key (DEK)"""
|
||||
"""arm the instance with the AES data key (DEK)
|
||||
|
||||
requires exactly 32 bytes (AES-256): a shorter key would silently downgrade
|
||||
to AES-128/192 with no warning, and a non-bytes value would otherwise fail
|
||||
late and opaquely at first encrypt/decrypt - both rejected here instead.
|
||||
never logs the key material itself.
|
||||
"""
|
||||
if not isinstance(master_key, bytes) or len(master_key) != 32:
|
||||
raise ValueError("master_key must be exactly 32 bytes (AES-256)")
|
||||
self.master_key = master_key
|
||||
_log.info("crypto initialized with data key")
|
||||
|
||||
@@ -247,12 +241,23 @@ class EnvelopeCrypto:
|
||||
) -> str:
|
||||
"""return a base64 SHA-256 fingerprint of an RSA key for identification
|
||||
|
||||
for an encrypted private key (is_private=True), pass its `password`; an
|
||||
unencrypted key ignores it. fingerprinting always uses the public half, so a
|
||||
private and its public key produce the same fingerprint. PEM and OpenSSH
|
||||
private-key formats are both accepted (mirrors decrypt_aes_key_with_rsa). an
|
||||
encrypted key with no/wrong password raises ValueError with a clear message
|
||||
(cryptography raises TypeError for the missing-password case — normalized here).
|
||||
always fingerprints the public half, so a private key and its public key
|
||||
match. PEM and OpenSSH accepted (mirrors decrypt_aes_key_with_rsa).
|
||||
|
||||
Args:
|
||||
key_path_or_data: path to (or, if is_file=False, raw PEM/OpenSSH data
|
||||
of) the key.
|
||||
is_private: fingerprint the public half of a private key; pass its
|
||||
`password` if encrypted (an unencrypted key ignores it).
|
||||
is_file: True (default) treats key_path_or_data as a path.
|
||||
password: password for an encrypted private key, if is_private.
|
||||
|
||||
Returns:
|
||||
base64 SHA-256 fingerprint.
|
||||
|
||||
Raises:
|
||||
ValueError: key is malformed, or password/encryption mismatch
|
||||
(cryptography's raw TypeError normalized here).
|
||||
"""
|
||||
if is_file:
|
||||
with open(key_path_or_data, "rb") as key_file:
|
||||
@@ -278,7 +283,23 @@ class EnvelopeCrypto:
|
||||
def encrypt_aes_key_with_rsa(
|
||||
self, aes_key: bytes, rsa_key: str, is_file: bool = True
|
||||
) -> Tuple[str, str]:
|
||||
"""wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)"""
|
||||
"""wrap an AES key with an RSA public key
|
||||
|
||||
Args:
|
||||
aes_key: the AES data key to wrap.
|
||||
rsa_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
|
||||
the RSA public key to wrap with.
|
||||
is_file: True (default) treats rsa_key as a path.
|
||||
|
||||
Returns:
|
||||
(fingerprint, wrapped_b64): the wrapping key's fingerprint and the
|
||||
RSA-OAEP wrapped key, base64-encoded.
|
||||
|
||||
Raises:
|
||||
ValueError: rsa_key is not an RSA key (this lib is RSA-envelope
|
||||
only; e.g. an Ed25519/EC key loads and fingerprints fine but
|
||||
cannot wrap), or is malformed.
|
||||
"""
|
||||
if is_file:
|
||||
with open(rsa_key, "rb") as key_file:
|
||||
key_data = key_file.read()
|
||||
@@ -286,16 +307,9 @@ class EnvelopeCrypto:
|
||||
key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
|
||||
|
||||
public_key = _load_public_key(key_data)
|
||||
_require_rsa(public_key)
|
||||
|
||||
wrapped = public_key.encrypt(
|
||||
aes_key,
|
||||
padding.OAEP(
|
||||
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
||||
algorithm=hashes.SHA256(),
|
||||
label=None,
|
||||
),
|
||||
)
|
||||
# fingerprint from the already-loaded public_key — no second open/parse of the file
|
||||
wrapped = public_key.encrypt(aes_key, _OAEP_PADDING)
|
||||
fingerprint = _fingerprint_of(public_key)
|
||||
wrapped_b64 = base64.b64encode(wrapped).decode()
|
||||
_log.info("wrapped data key for fingerprint %s", fingerprint[:8])
|
||||
@@ -307,10 +321,20 @@ class EnvelopeCrypto:
|
||||
) -> bytes:
|
||||
"""unwrap an AES key with an RSA private key
|
||||
|
||||
is_file defaults to True (rsa_private_key is a path), matching
|
||||
encrypt_aes_key_with_rsa / get_rsa_key_fingerprint; pass is_file=False to
|
||||
supply the PEM/OpenSSH key data directly (e.g. an in-memory or vault-sourced
|
||||
key) instead of a file path.
|
||||
Args:
|
||||
encrypted_key_base64: the RSA-OAEP wrapped key, base64-encoded.
|
||||
rsa_private_key: path to (or, if is_file=False, raw PEM/OpenSSH data
|
||||
of) the private key to unwrap with.
|
||||
is_file: True (default) treats rsa_private_key as a path; pass False
|
||||
to supply PEM/OpenSSH key data directly (e.g. vault-sourced).
|
||||
password: password for an encrypted private key, if any.
|
||||
|
||||
Returns:
|
||||
the unwrapped AES data key.
|
||||
|
||||
Raises:
|
||||
ValueError: rsa_private_key is not an RSA key, is malformed, or a
|
||||
password/encryption mismatch.
|
||||
"""
|
||||
if is_file:
|
||||
with open(rsa_private_key, "rb") as key_file:
|
||||
@@ -319,25 +343,31 @@ class EnvelopeCrypto:
|
||||
key_data = rsa_private_key.encode() if isinstance(rsa_private_key, str) else rsa_private_key
|
||||
pw = password.encode() if password else None
|
||||
private_key = _load_private_key(key_data, pw)
|
||||
_require_rsa(private_key)
|
||||
|
||||
wrapped = base64.b64decode(encrypted_key_base64)
|
||||
aes_key = private_key.decrypt(
|
||||
wrapped,
|
||||
padding.OAEP(
|
||||
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
||||
algorithm=hashes.SHA256(),
|
||||
label=None,
|
||||
),
|
||||
)
|
||||
aes_key = private_key.decrypt(wrapped, _OAEP_PADDING)
|
||||
_log.info("unwrapped data key with RSA private key")
|
||||
return aes_key
|
||||
|
||||
def authorize_system(self, rsa_public_key: str, is_file: bool = True) -> Tuple[str, str]:
|
||||
"""wrap the current data key for another system's public key
|
||||
|
||||
returns (fingerprint, wrapped_b64) for the caller to store as that
|
||||
system's key-authorization record. requires this instance to already
|
||||
hold the data key — only an authorized system can authorize others.
|
||||
requires this instance to already hold the data key - only an
|
||||
authorized system can authorize others.
|
||||
|
||||
Args:
|
||||
rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data
|
||||
of) the other system's RSA public key.
|
||||
is_file: True (default) treats rsa_public_key as a path.
|
||||
|
||||
Returns:
|
||||
(fingerprint, wrapped_b64) for the caller to store as that system's
|
||||
key-authorization record.
|
||||
|
||||
Raises:
|
||||
ValueError: this instance is not initialized, or rsa_public_key is
|
||||
not an RSA key.
|
||||
"""
|
||||
if not self.master_key:
|
||||
raise ValueError("cannot authorize another system: not initialized")
|
||||
@@ -348,9 +378,20 @@ class EnvelopeCrypto:
|
||||
) -> Tuple[bytes, Dict[str, str]]:
|
||||
"""generate a NEW data key and wrap it for each authorized public key
|
||||
|
||||
returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing
|
||||
data — build a new instance with the new key and call reencrypt() on each
|
||||
record. systems not in the list get no wrapped copy (deauthorized).
|
||||
does NOT re-encrypt existing data - build a new instance with the new key
|
||||
and call reencrypt() on each record. systems not in the list get no
|
||||
wrapped copy (deauthorized).
|
||||
|
||||
Args:
|
||||
authorized_public_keys: RSA public keys (paths, or raw data if
|
||||
is_file=False) to wrap the new key for.
|
||||
is_file: True (default) treats each entry as a path.
|
||||
|
||||
Returns:
|
||||
(new_key, {fingerprint: wrapped_b64}).
|
||||
|
||||
Raises:
|
||||
ValueError: any entry in authorized_public_keys is not an RSA key.
|
||||
"""
|
||||
new_key = self.create_aes_key()
|
||||
wrapped = {}
|
||||
@@ -363,27 +404,23 @@ class EnvelopeCrypto:
|
||||
def encrypt_data(self, data: Union[Dict[str, Any], str]) -> Dict[str, str]:
|
||||
"""encrypt a dict or string under the data key with a unique IV
|
||||
|
||||
dict keys must be str: json.dumps silently stringifies int/float/bool/None
|
||||
keys (e.g. a snowflake-int-keyed dict), which would make decrypt_data return
|
||||
a dict that no longer matches the original by key type or identity — a
|
||||
dict keys must be str, at any nesting depth (including dicts nested inside
|
||||
lists/tuples): json.dumps silently stringifies int/float/bool/None keys
|
||||
(e.g. a snowflake-int-keyed dict), which would make decrypt_data return a
|
||||
dict that no longer matches the original by key type or identity - a
|
||||
non-str key is rejected here instead, so the failure is loud at encrypt
|
||||
time rather than a silent lookup miss after decrypt.
|
||||
"""
|
||||
if not self.master_key:
|
||||
raise ValueError("not initialized with data key")
|
||||
if not isinstance(data, (dict, str)):
|
||||
# a non-dict/non-str would .encode()-fail with an opaque AttributeError below;
|
||||
# reject it clearly (decrypt_data only round-trips dict or str anyway)
|
||||
raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}")
|
||||
if isinstance(data, dict):
|
||||
non_str_keys = [key for key in data if not isinstance(key, str)]
|
||||
if non_str_keys:
|
||||
# json.dumps would silently stringify these (int/float/bool/None keys),
|
||||
# corrupting the round-trip (decrypt_data would return a dict keyed by
|
||||
# the stringified value) — fail loud instead of coercing
|
||||
non_str_key_types = _non_str_key_types(data)
|
||||
if non_str_key_types:
|
||||
raise TypeError(
|
||||
"encrypt_data requires str dict keys, got non-str key(s): "
|
||||
f"{[type(key).__name__ for key in non_str_keys]}"
|
||||
"encrypt_data requires str dict keys (at any nesting depth), got non-str "
|
||||
f"key(s): {non_str_key_types}"
|
||||
)
|
||||
|
||||
data_str = json.dumps(data) if isinstance(data, dict) else data
|
||||
@@ -399,23 +436,25 @@ class EnvelopeCrypto:
|
||||
def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]:
|
||||
"""decrypt a {secure, iv, data} blob; returns the original dict or string
|
||||
|
||||
`encrypt_data` only json-encodes dicts (a string is stored verbatim), so decrypt
|
||||
only treats a json-OBJECT plaintext as a dict and returns everything else as the
|
||||
raw string. this keeps the type faithful for the common cases: a string that is
|
||||
json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]')
|
||||
round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a
|
||||
string whose exact value is a json OBJECT ('{"a":1}') decrypts to a dict, because
|
||||
without a type marker it is indistinguishable from a stored dict — don't store a
|
||||
bare string that is a json object if you need it back as a string. existing stored
|
||||
blobs are unaffected — a dict was stored as a json object and still parses to a dict.
|
||||
dict KEYS round-trip faithfully only because encrypt_data now requires str keys —
|
||||
json has no other key type, so this is the only shape decrypt_data ever sees.
|
||||
a json-shaped but non-object plaintext ('123', 'true', '[1,2]') round-trips
|
||||
as a STRING, not int/bool/list. one irreducible ambiguity: a string whose
|
||||
exact value is a json object ('{"a":1}') decrypts to a dict, indistinguishable
|
||||
from a stored dict - don't store a bare json-object string if you need it
|
||||
back as a string.
|
||||
|
||||
Args:
|
||||
encrypted_data: a {secure, iv, data} blob from encrypt_data.
|
||||
|
||||
Returns:
|
||||
the original dict or string.
|
||||
|
||||
Raises:
|
||||
ValueError: instance not initialized, or encrypted_data is not a
|
||||
{secure, iv, data} blob.
|
||||
"""
|
||||
if not self.master_key:
|
||||
raise ValueError("not initialized with data key")
|
||||
if not isinstance(encrypted_data, dict) or "iv" not in encrypted_data or "data" not in encrypted_data:
|
||||
# a structurally-malformed blob would raise a raw KeyError/TypeError; surface
|
||||
# a clear ValueError instead, matching the documented {secure, iv, data} shape
|
||||
raise ValueError("decrypt_data expects a {secure, iv, data} blob")
|
||||
|
||||
iv = base64.b64decode(encrypted_data["iv"])
|
||||
@@ -433,24 +472,30 @@ class EnvelopeCrypto:
|
||||
|
||||
self holds the destination (new) key; source_crypto holds the source (old)
|
||||
key. only {secure, iv, data} fields are touched; plaintext fields are left
|
||||
as-is. returns a new dict; the input is not mutated. used during rotation.
|
||||
as-is. used during rotation. if `record` itself is a bare blob (file-storage
|
||||
pattern) it is re-encrypted directly and returned in place of `record`.
|
||||
|
||||
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where
|
||||
the blob IS the whole document) it is re-encrypted directly and the result is
|
||||
returned in place of `record` — not nested under a key.
|
||||
traversal recurses into nested DICTS only - a blob nested inside a LIST is
|
||||
not re-encrypted. the depth-limit raise below IS reached for a list-nested
|
||||
blob when the cutoff dict scan finds it (it walks lists too), but a blob
|
||||
one level shallower - hit during normal traversal instead of the cutoff
|
||||
scan - is skipped silently; flatten list-nested blobs to dict fields before
|
||||
rotation or they can be silently left under the old key.
|
||||
|
||||
unlike decrypt_record (which logs a failed field and leaves it encrypted), a
|
||||
per-field decrypt failure here RAISES — rotation must fail loud, since silently
|
||||
keeping a field under the old key would lose it once the old key is retired. for
|
||||
the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead
|
||||
of being silently left under the old key: the caller either needs a higher
|
||||
`traversal_level` or must flatten the record before rotation.
|
||||
Args:
|
||||
source_crypto: instance holding the old (source) key.
|
||||
record: the record to re-encrypt. not mutated; a new dict is returned.
|
||||
traversal_level: max nesting depth to recurse into (default 2).
|
||||
|
||||
traversal recurses into nested DICTS only; a blob nested inside a LIST is NOT
|
||||
re-encrypted and is NOT covered by the depth-limit raise above. records in this
|
||||
scheme key blobs by field name, not inside arrays, so this doesn't arise in
|
||||
practice — but if you store list-nested blobs, flatten them to dict fields before
|
||||
rotation or they'll be silently left under the old key.
|
||||
Returns:
|
||||
a new dict with encrypted fields re-wrapped under this instance's key.
|
||||
|
||||
Raises:
|
||||
ValueError: destination not initialized with a data key; a per-field
|
||||
decrypt failure (unlike decrypt_record, this fails loud rather
|
||||
than silently stranding a field under the old key); or a blob
|
||||
nested deeper than `traversal_level` - raise traversal_level or
|
||||
flatten the record before rotation instead.
|
||||
"""
|
||||
if not self.master_key:
|
||||
raise ValueError("destination not initialized with data key")
|
||||
@@ -469,29 +514,56 @@ class EnvelopeCrypto:
|
||||
raise ValueError(
|
||||
f"reencrypt: field {key!r} contains an encrypted blob nested deeper "
|
||||
"than traversal_level; increase traversal_level or flatten the record "
|
||||
"before rotation — leaving it would strand the field under the old key"
|
||||
"before rotation - leaving it would strand the field under the old key"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
# naming aliases — same class
|
||||
# naming aliases - same class
|
||||
DocumentCrypto = EnvelopeCrypto
|
||||
RecordCrypto = EnvelopeCrypto
|
||||
PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate
|
||||
|
||||
|
||||
def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
|
||||
"""bounded-pass check of a single field value, walking list/tuple items too
|
||||
|
||||
falls back to the unbounded _has_encrypted_field scan once traversal_level is
|
||||
exhausted, so this can never return False for a value that still contains a
|
||||
blob at any depth
|
||||
"""
|
||||
if _is_blob(value):
|
||||
return True
|
||||
if isinstance(value, dict):
|
||||
if traversal_level > 0:
|
||||
return is_encrypted_record(value, traversal_level - 1)
|
||||
return _has_encrypted_field(value)
|
||||
if isinstance(value, (list, tuple)):
|
||||
return any(_is_encrypted_value(item, traversal_level) for item in value)
|
||||
return False
|
||||
|
||||
|
||||
def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
||||
"""return whether a record has any encrypted ({secure, iv, data}) fields
|
||||
|
||||
checks `record` itself (the file-storage pattern stores the blob AS the whole
|
||||
document, not nested under a key) as well as fields up to `traversal_level` deep.
|
||||
beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up —
|
||||
so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper
|
||||
than their traversal_level) is still reported as encrypted. this makes the function
|
||||
safe to use as a leftover-detecting post-rotation audit: it never returns False for
|
||||
a record that still contains a blob, at any depth.
|
||||
checks `record` itself (the file-storage pattern) as well as fields up to
|
||||
`traversal_level` deep, then always falls back to an unbounded-depth scan if
|
||||
the bounded pass found nothing - safe to use as a leftover-detecting
|
||||
post-rotation audit; never returns False for a record that still contains a
|
||||
blob, at any depth (including one nested inside a list or tuple), for any
|
||||
traversal_level value, odd or even (both passes walk list/tuple items, not
|
||||
just dict values).
|
||||
|
||||
aliases: is_encrypted_document, is_encrypted_dict — same function
|
||||
Args:
|
||||
record: the record (or bare blob) to check.
|
||||
traversal_level: bounded-pass nesting depth tried before the unbounded
|
||||
fallback scan runs (default 2); purely a fast-path - the fallback
|
||||
always makes the final call when the bounded pass finds nothing.
|
||||
|
||||
Returns:
|
||||
True if record or anything nested under it is an encrypted blob.
|
||||
|
||||
aliases: is_encrypted_document, is_encrypted_dict - same function
|
||||
"""
|
||||
if not isinstance(record, dict):
|
||||
return False
|
||||
@@ -505,9 +577,8 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
||||
|
||||
if traversal_level > 0:
|
||||
for value in record.values():
|
||||
if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
|
||||
if _is_encrypted_value(value, traversal_level - 1):
|
||||
return True
|
||||
return False
|
||||
|
||||
return any(_has_encrypted_field(value) for value in record.values())
|
||||
|
||||
@@ -515,16 +586,25 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
||||
def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> Union[dict, Any]:
|
||||
"""decrypt a record's encrypted fields into a new dict (up to traversal_level deep)
|
||||
|
||||
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where the
|
||||
blob IS the whole document) it is decrypted directly and the decrypted value
|
||||
(dict or string — see decrypt_data) is returned in place of `record`.
|
||||
if `record` itself is a bare {secure, iv, data} blob (file-storage pattern) it
|
||||
is decrypted directly and the value (dict or string, see decrypt_data) is
|
||||
returned in place of `record`. a failure on a single field (or on `record`
|
||||
itself) is logged and left encrypted, so a partial failure stays visible
|
||||
rather than silent.
|
||||
|
||||
failures on a single field are logged and that field is left encrypted, so a
|
||||
partial failure is visible (the {secure,...} blob remains) rather than silent. a
|
||||
failure decrypting `record` itself (the self-blob case above) is likewise logged
|
||||
and the still-encrypted blob is returned unchanged.
|
||||
Args:
|
||||
crypto: instance holding the data key to decrypt with.
|
||||
record: the record (or bare blob) to decrypt. not mutated.
|
||||
traversal_level: max nesting depth to recurse into (default 2).
|
||||
|
||||
aliases: decrypt_document, decrypt_dict — same function
|
||||
Returns:
|
||||
a new dict with encrypted fields decrypted (or record unchanged if not
|
||||
a dict).
|
||||
|
||||
Raises:
|
||||
ValueError: crypto is not initialized with a data key.
|
||||
|
||||
aliases: decrypt_document, decrypt_dict - same function
|
||||
"""
|
||||
if not crypto.master_key:
|
||||
raise ValueError("not initialized with data key")
|
||||
@@ -551,11 +631,24 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
|
||||
|
||||
|
||||
def fingerprint_data(data: dict) -> str:
|
||||
"""return a deterministic SHA-256 hex fingerprint of a dict"""
|
||||
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
|
||||
"""deterministic sha256 fingerprint of json-shaped dict data
|
||||
|
||||
the input is a record sent over the wire, so it is json-serializable by contract:
|
||||
keys are canonically sorted for an order-independent digest, separators are fixed for
|
||||
whitespace stability, and any non-json scalar (uuid/decimal/path/datetime/objectid) is
|
||||
stringified rather than crashing the encode. a callable value raises - its str() embeds a
|
||||
memory address, which would make the digest vary across processes. never logs the data.
|
||||
"""
|
||||
def _stringify(value: object) -> str:
|
||||
if callable(value):
|
||||
raise TypeError(f"fingerprint_data: cannot fingerprint a callable: {value!r}")
|
||||
return str(value)
|
||||
|
||||
encoded = json.dumps(data, sort_keys=True, separators=(",", ":"), default=_stringify)
|
||||
return hashlib.sha256(encoded.encode()).hexdigest()
|
||||
|
||||
|
||||
# function aliases — same functions, naming preference only
|
||||
# function aliases - same functions, naming preference only
|
||||
is_encrypted_document = is_encrypted_record
|
||||
is_encrypted_dict = is_encrypted_record
|
||||
decrypt_document = decrypt_record
|
||||
|
||||
Reference in New Issue
Block a user