rethink-public/envelope_crypto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: rethink-public:main
Branches Tags
rethink-public:main
rethink-public:v0.1.3
rethink-public:v0.1.2
rethink-public:v0.1.1
rethink-public:v0.1.0
..
compare: rethink-public:v0.1.2
Branches Tags
rethink-public:main
rethink-public:v0.1.3
rethink-public:v0.1.2
rethink-public:v0.1.1
rethink-public:v0.1.0

No commits in common. "main" and "v0.1.2" have entirely different histories.
main ... v0.1.2

4 changed files with 45 additions and 106 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,5 +1,5 @@
# claude
.claude/
CLAUDE.md
# python
__pycache__/

8
README.md
View File

@ -11,19 +11,17 @@ and storage-agnostic.
`requirements.txt`:
```
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.2
```
Direct:
```bash
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3"
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.2"
```
Requires `cryptography` (pulled transitively).
Drop the `@v0.1.3` suffix from the line above to install the latest unpinned.
## First-time setup
Run once, ever, to create the data key and authorize the first system. You need an
@ -154,4 +152,4 @@ The lib never touches a database; only the caller's storage layer differs.
## Versioning
Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs.
Tagged `vX.Y.Z`. Pin the tag in `requirements.txt`.

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "envelope_crypto"
version = "0.1.3"
version = "0.1.2"
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
requires-python = ">=3.10"
dependencies = [

139
src/envelope_crypto/envelope_crypto.py
View File

@ -61,7 +61,6 @@ import hashlib
import logging
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.ciphers.aead import AESGCM
@ -70,65 +69,6 @@ from cryptography.hazmat.primitives.serialization import load_ssh_public_key
_log = logging.getLogger(__name__)
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.
"""
try:
return serialization.load_pem_private_key(key_data, password=pw)
except ValueError as error:
if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
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
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
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.
"""
key_bytes = public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
digest = hashes.Hash(hashes.SHA256())
digest.update(key_bytes)
return base64.b64encode(digest.finalize()).decode()
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.
"""
try:
return serialization.load_pem_public_key(key_data)
except ValueError:
try:
return load_ssh_public_key(key_data)
except (UnsupportedAlgorithm, ValueError) as error:
raise ValueError("not a valid PEM or OpenSSH public key") from error
class EnvelopeCrypto:
"""hybrid RSA/AES-256-GCM envelope encryption for dict records
@ -232,12 +172,31 @@ class EnvelopeCrypto:
if is_private:
pw = password.encode() if password else None
private_key = _load_private_key(key_data, pw)
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()
else:
public_key = _load_public_key(key_data)
try:
public_key = serialization.load_pem_public_key(key_data)
except ValueError:
public_key = load_ssh_public_key(key_data)
fingerprint = _fingerprint_of(public_key)
key_bytes = public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
digest = hashes.Hash(hashes.SHA256())
digest.update(key_bytes)
fingerprint = base64.b64encode(digest.finalize()).decode()
_log.info("generated %s key fingerprint", "private" if is_private else "public")
return fingerprint
@ -251,7 +210,10 @@ class EnvelopeCrypto:
else:
key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
public_key = _load_public_key(key_data)
try:
public_key = serialization.load_pem_public_key(key_data)
except ValueError:
public_key = load_ssh_public_key(key_data)
wrapped = public_key.encrypt(
aes_key,
@ -261,8 +223,7 @@ class EnvelopeCrypto:
label=None,
),
)
# fingerprint from the already-loaded public_key — no second open/parse of the file
fingerprint = _fingerprint_of(public_key)
fingerprint = self.get_rsa_key_fingerprint(rsa_key, is_private=False, is_file=is_file)
wrapped_b64 = base64.b64encode(wrapped).decode()
_log.info("wrapped data key for fingerprint %s", fingerprint[:8])
return fingerprint, wrapped_b64
@ -275,7 +236,17 @@ class EnvelopeCrypto:
with open(rsa_private_key_path, "rb") as key_file:
key_data = key_file.read()
pw = password.encode() if password else None
private_key = _load_private_key(key_data, pw)
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
wrapped = base64.b64decode(encrypted_key_base64)
aes_key = private_key.decrypt(
@ -321,10 +292,6 @@ class EnvelopeCrypto:
"""encrypt a dict or string under the data key with a unique IV"""
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__}")
data_str = json.dumps(data) if isinstance(data, dict) else data
iv = os.urandom(12)
@ -337,34 +304,18 @@ 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.
"""
"""decrypt a {secure, iv, data} blob; returns the original dict or string"""
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"])
ciphertext = base64.b64decode(encrypted_data["data"])
aesgcm = AESGCM(self.master_key)
plaintext = aesgcm.decrypt(iv, ciphertext, None).decode()
try:
parsed = json.loads(plaintext)
return json.loads(plaintext)
except json.JSONDecodeError:
return plaintext
return parsed if isinstance(parsed, dict) else plaintext
def reencrypt(self, source_crypto: "EnvelopeCrypto", record: dict, traversal_level: int = 2) -> dict:
"""re-encrypt a record's encrypted fields from source_crypto's key to this one's
@ -372,16 +323,6 @@ 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.
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.
traversal recurses into nested DICTS only (up to `traversal_level`); a blob nested
inside a LIST is NOT re-encrypted. 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 left
under the old key.
"""
if not self.master_key:
raise ValueError("destination not initialized with data key")