4 Commits
Author SHA1 Message Date
dsql 16205e810a fix: deepcopy in reencrypt/decrypt_record so input is not mutated
both used record.copy() (shallow), leaving unencrypted mutable fields shared between the input and the returned dict, violating the documented 'input is not mutated' contract. switched to copy.deepcopy.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-28 17:18:28 -04:00
dsql 313b0c7d56 fix: forward password to private-key fingerprinting (v0.1.1)
get_rsa_key_fingerprint(is_private=True) called load_pem_private_key(password=None),
so an encrypted private key raised a raw TypeError. add an optional password param
forwarded to the load; unencrypted keys ignore it.

verified: encrypted private key fingerprints with its password and matches the
public key's fingerprint; missing password still raises.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-28 15:53:04 -04:00
dsql 659aa7849d add package: pyproject + src
EnvelopeCrypto: hybrid envelope encryption for dict records — a random
AES-256-GCM data key (DEK) encrypts the data, wrapped per-system via
RSA-OAEP (SHA-256) for distribution. config-free (DEK + key paths
injected), storage-agnostic, object-only. covers bootstrap/self_test,
authorize/deauthorize, rotate + reencrypt, and record-level decrypt.
src/ layout, hatchling build, cryptography backend.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-24 21:36:43 -04:00
dsql 0b708cdf9a init: envelope encryption (RSA-OAEP + AES-256-GCM) for dict records
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-24 21:25:27 -04:00
3 changed files with 25 additions and 16 deletions
+6 -6
View File
@@ -11,13 +11,13 @@ and storage-agnostic.
`requirements.txt`:
```
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.1
```
Direct:
```bash
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0"
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.1"
```
Requires `cryptography` (pulled transitively).
@@ -37,10 +37,10 @@ openssl rsa -in local_priv.pem -pubout -out local_pub.pem
from envelope_crypto import EnvelopeCrypto
# generate the DEK and wrap it for this system in one call
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap(cfg.local_pub)
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
# verify the keypair actually round-trips BEFORE storing anything
crypto.self_test(cfg.local_pub, cfg.local_priv) # raises if keys don't pair
crypto.self_test("public_key.pem", "private_key.pem") # raises if keys don't pair
# store the wrapped key — this is now the ONLY record of the DEK
await db.create_document("keys", {"_id": fingerprint, "key": wrapped})
@@ -53,11 +53,11 @@ re-derived each boot by unwrapping. **Never persist the plaintext key.**
```python
crypto = EnvelopeCrypto()
fingerprint = crypto.get_rsa_key_fingerprint(cfg.local_pub)
fingerprint = crypto.get_rsa_key_fingerprint("public_key.pem")
record = await db.get_document("keys", {"_id": fingerprint})
if not record:
raise RuntimeError("this system is not authorized")
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], cfg.local_priv))
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
bot.crypto = crypto
```
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "envelope_crypto"
version = "0.1.0"
version = "0.1.1"
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
requires-python = ">=3.10"
dependencies = [
+18 -9
View File
@@ -17,15 +17,15 @@ envelope-encryption pattern used by KMS-style systems.
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(cfg.local_pub)
crypto.self_test(cfg.local_pub, cfg.local_priv) # raises if anything is wrong
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(cfg.local_pub)
fp = crypto.get_rsa_key_fingerprint("public_key.pem")
record = caller_lookup(fp)
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], cfg.local_priv))
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
authorize another system (this instance must already hold the DEK):
@@ -54,6 +54,7 @@ function variants are the same functions — use whichever fits your storage.
"""
import os
import copy
import json
import base64
import hashlib
@@ -147,9 +148,15 @@ class EnvelopeCrypto:
return key
def get_rsa_key_fingerprint(
self, key_path_or_data: str, is_private: bool = False, is_file: bool = True
self, key_path_or_data: str, is_private: bool = False, is_file: bool = True,
password: Optional[str] = None,
) -> str:
"""return a base64 SHA-256 fingerprint of an RSA key for identification"""
"""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.
"""
if is_file:
with open(key_path_or_data, "rb") as key_file:
key_data = key_file.read()
@@ -161,7 +168,9 @@ class EnvelopeCrypto:
)
if is_private:
private_key = serialization.load_pem_private_key(key_data, password=None)
private_key = serialization.load_pem_private_key(
key_data, password=password.encode() if password else None
)
public_key = private_key.public_key()
else:
try:
@@ -305,7 +314,7 @@ class EnvelopeCrypto:
if not self.master_key:
raise ValueError("destination not initialized with data key")
result = record.copy()
result = copy.deepcopy(record)
for key, value in record.items():
if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
result[key] = self.encrypt_data(source_crypto.decrypt_data(value))
@@ -353,7 +362,7 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
if not isinstance(record, dict):
return record
result = record.copy()
result = copy.deepcopy(record)
for key, value in record.items():
if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
try: