Compare commits
2
Commits
v0.1.2
..
8a220a8810
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8a220a8810 | ||
|
|
1864612d64 |
@@ -11,13 +11,13 @@ and storage-agnostic.
|
||||
`requirements.txt`:
|
||||
|
||||
```
|
||||
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.2
|
||||
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0
|
||||
```
|
||||
|
||||
Direct:
|
||||
|
||||
```bash
|
||||
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.2"
|
||||
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0"
|
||||
```
|
||||
|
||||
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("public_key.pem")
|
||||
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap(cfg.local_pub)
|
||||
|
||||
# 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
|
||||
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("public_key.pem")
|
||||
fingerprint = crypto.get_rsa_key_fingerprint(cfg.local_pub)
|
||||
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"], "private_key.pem"))
|
||||
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], cfg.local_priv))
|
||||
bot.crypto = crypto
|
||||
```
|
||||
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "envelope_crypto"
|
||||
version = "0.1.2"
|
||||
version = "0.1.0"
|
||||
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
|
||||
@@ -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("public_key.pem")
|
||||
crypto.self_test("public_key.pem", "private_key.pem") # raises if anything is wrong
|
||||
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap(cfg.local_pub)
|
||||
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
|
||||
|
||||
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)
|
||||
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):
|
||||
|
||||
@@ -54,7 +54,6 @@ function variants are the same functions — use whichever fits your storage.
|
||||
"""
|
||||
|
||||
import os
|
||||
import copy
|
||||
import json
|
||||
import base64
|
||||
import hashlib
|
||||
@@ -148,18 +147,9 @@ class EnvelopeCrypto:
|
||||
return key
|
||||
|
||||
def get_rsa_key_fingerprint(
|
||||
self, key_path_or_data: str, is_private: bool = False, is_file: bool = True,
|
||||
password: Optional[str] = None,
|
||||
self, key_path_or_data: str, is_private: bool = False, is_file: bool = True
|
||||
) -> 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).
|
||||
"""
|
||||
"""return a base64 SHA-256 fingerprint of an RSA key for identification"""
|
||||
if is_file:
|
||||
with open(key_path_or_data, "rb") as key_file:
|
||||
key_data = key_file.read()
|
||||
@@ -171,18 +161,7 @@ class EnvelopeCrypto:
|
||||
)
|
||||
|
||||
if is_private:
|
||||
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
|
||||
private_key = serialization.load_pem_private_key(key_data, password=None)
|
||||
public_key = private_key.public_key()
|
||||
else:
|
||||
try:
|
||||
@@ -235,18 +214,17 @@ class EnvelopeCrypto:
|
||||
"""unwrap an AES key with an RSA private key"""
|
||||
with open(rsa_private_key_path, "rb") as key_file:
|
||||
key_data = key_file.read()
|
||||
pw = password.encode() if password else None
|
||||
try:
|
||||
private_key = serialization.load_pem_private_key(key_data, password=pw)
|
||||
private_key = serialization.load_pem_private_key(
|
||||
key_data, password=password.encode() if password else None
|
||||
)
|
||||
except ValueError as error:
|
||||
if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
|
||||
private_key = serialization.load_ssh_private_key(key_data, password=pw)
|
||||
private_key = serialization.load_ssh_private_key(
|
||||
key_data, password=password.encode() if password else None
|
||||
)
|
||||
else:
|
||||
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)
|
||||
aes_key = private_key.decrypt(
|
||||
@@ -327,7 +305,7 @@ class EnvelopeCrypto:
|
||||
if not self.master_key:
|
||||
raise ValueError("destination not initialized with data key")
|
||||
|
||||
result = copy.deepcopy(record)
|
||||
result = record.copy()
|
||||
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))
|
||||
@@ -375,7 +353,7 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
|
||||
if not isinstance(record, dict):
|
||||
return record
|
||||
|
||||
result = copy.deepcopy(record)
|
||||
result = record.copy()
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user