5 Commits
Author SHA1 Message Date
dsql 5de8b5d736 fix: OpenSSH private-key fingerprint fallback + clean error on missing password
get_rsa_key_fingerprint(is_private=True) only loaded PEM private keys, so an OpenSSH-format private key raised — unlike decrypt_aes_key_with_rsa, which already had the fallback. mirrored it: on a PEM load failure, an OPENSSH-marked key is loaded via load_ssh_private_key. also normalized the encrypted-key-without-password case: cryptography raises TypeError there, which now becomes a clear ValueError('private key is encrypted but no password was provided') in both methods instead of leaking the raw TypeError.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-29 01:39:21 -04:00
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 44 additions and 22 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.0 envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.2
``` ```
Direct: Direct:
```bash ```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.2"
``` ```
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(cfg.local_pub) crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
# verify the keypair actually round-trips BEFORE storing anything # 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 # 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(cfg.local_pub) fingerprint = crypto.get_rsa_key_fingerprint("public_key.pem")
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"], cfg.local_priv)) crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
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.0" version = "0.1.2"
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 = [
+37 -15
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(cfg.local_pub) crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
crypto.self_test(cfg.local_pub, cfg.local_priv) # raises if anything is wrong 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 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(cfg.local_pub) fp = crypto.get_rsa_key_fingerprint("public_key.pem")
record = caller_lookup(fp) 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): 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 os
import copy
import json import json
import base64 import base64
import hashlib import hashlib
@@ -147,9 +148,18 @@ 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. 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).
"""
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()
@@ -161,7 +171,18 @@ class EnvelopeCrypto:
) )
if is_private: if is_private:
private_key = serialization.load_pem_private_key(key_data, password=None) pw = password.encode() if password else None
try:
private_key = serialization.load_pem_private_key(key_data, password=pw)
except ValueError as error:
if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
private_key = serialization.load_ssh_private_key(key_data, password=pw)
else:
raise error
except TypeError as error:
raise ValueError(
"private key is encrypted but no password was provided"
) from error
public_key = private_key.public_key() public_key = private_key.public_key()
else: else:
try: try:
@@ -214,17 +235,18 @@ class EnvelopeCrypto:
"""unwrap an AES key with an RSA private key""" """unwrap an AES key with an RSA private key"""
with open(rsa_private_key_path, "rb") as key_file: with open(rsa_private_key_path, "rb") as key_file:
key_data = key_file.read() key_data = key_file.read()
pw = password.encode() if password else None
try: try:
private_key = serialization.load_pem_private_key( private_key = serialization.load_pem_private_key(key_data, password=pw)
key_data, password=password.encode() if password else None
)
except ValueError as error: except ValueError as error:
if b"BEGIN OPENSSH PRIVATE KEY" in key_data: if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
private_key = serialization.load_ssh_private_key( private_key = serialization.load_ssh_private_key(key_data, password=pw)
key_data, password=password.encode() if password else None
)
else: else:
raise error raise error
except TypeError as error:
raise ValueError(
"private key is encrypted but no password was provided"
) from error
wrapped = base64.b64decode(encrypted_key_base64) wrapped = base64.b64decode(encrypted_key_base64)
aes_key = private_key.decrypt( aes_key = private_key.decrypt(
@@ -305,7 +327,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 = record.copy() result = copy.deepcopy(record)
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))
@@ -353,7 +375,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 = record.copy() result = copy.deepcopy(record)
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: