2 Commits
Author SHA1 Message Date
dsqlandClaude Opus 4.8 8a220a8810 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.

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