78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
 
|
"""Credential vault — envelope encryption (SPEC §10.2).
|
||
|
|
|
||
|
|
Mailbox secrets are encrypted with a fresh AES-256-GCM *data key*; the data key
|
||
|
|
is itself wrapped with the master key from ``MASTER_KEY`` (32 bytes, base64). A
|
||
|
|
key id is embedded to allow future master-key rotation.
|
||
|
|
|
||
|
|
Wire format of ``secret_ct`` (all lengths fixed unless noted)::
|
||
|
|
|
||
|
|
b"WMV1" # 4-byte magic + version
|
||
|
|
key_id # 1 byte (master key id, 0 for now)
|
||
|
|
wrap_nonce # 12 bytes (GCM nonce for the wrapped data key)
|
||
|
|
wrapped_key # 32 + 16 bytes (data key ciphertext + tag)
|
||
|
|
data_nonce # 12 bytes (GCM nonce for the payload)
|
||
|
|
payload_ciphertext # variable (payload + 16-byte tag)
|
||
|
|
|
||
|
|
The decryption path exists only in workers and ``accounts/test`` (SPEC §10.2);
|
||
|
|
this module enforces nothing about callers but is imported only from there.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import base64
|
||
|
|
import os
|
||
|
|
|
||
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
|
|
|
||
|
|
from app.config import get_settings
|
||
|
|
|
||
|
|
_MAGIC = b"WMV1"
|
||
|
|
_CURRENT_KEY_ID = 0
|
||
|
|
|
||
|
|
|
||
|
|
class VaultError(Exception):
|
||
|
|
"""Raised when a ciphertext is malformed or cannot be authenticated."""
|
||
|
|
|
||
|
|
|
||
|
|
def _master_key() -> bytes:
|
||
|
|
raw = base64.b64decode(get_settings().master_key)
|
||
|
|
if len(raw) != 32:
|
||
|
|
raise VaultError("MASTER_KEY must decode to exactly 32 bytes")
|
||
|
|
return raw
|
||
|
|
|
||
|
|
|
||
|
|
def encrypt_secret(plaintext: bytes) -> bytes:
|
||
|
|
"""Envelope-encrypt ``plaintext`` and return the ``secret_ct`` blob."""
|
||
|
|
data_key = AESGCM.generate_key(256)
|
||
|
|
wrap_nonce = os.urandom(12)
|
||
|
|
wrapped = AESGCM(_master_key()).encrypt(wrap_nonce, data_key, _MAGIC)
|
||
|
|
|
||
|
|
data_nonce = os.urandom(12)
|
||
|
|
ciphertext = AESGCM(data_key).encrypt(data_nonce, plaintext, _MAGIC)
|
||
|
|
|
||
|
|
return _MAGIC + bytes([_CURRENT_KEY_ID]) + wrap_nonce + wrapped + data_nonce + ciphertext
|
||
|
|
|
||
|
|
|
||
|
|
def decrypt_secret(blob: bytes) -> bytes:
|
||
|
|
"""Authenticate and decrypt a ``secret_ct`` blob back to plaintext bytes."""
|
||
|
|
if len(blob) < 4 + 1 + 12 + 48 + 12 or blob[:4] != _MAGIC:
|
||
|
|
raise VaultError("malformed secret ciphertext")
|
||
|
|
pos = 4
|
||
|
|
key_id = blob[pos]
|
||
|
|
pos += 1
|
||
|
|
if key_id != _CURRENT_KEY_ID:
|
||
|
|
raise VaultError(f"unknown master key id {key_id}")
|
||
|
|
wrap_nonce = blob[pos : pos + 12]
|
||
|
|
pos += 12
|
||
|
|
wrapped = blob[pos : pos + 48] # 32-byte key + 16-byte GCM tag
|
||
|
|
pos += 48
|
||
|
|
data_nonce = blob[pos : pos + 12]
|
||
|
|
pos += 12
|
||
|
|
ciphertext = blob[pos:]
|
||
|
|
|
||
|
|
try:
|
||
|
|
data_key = AESGCM(_master_key()).decrypt(wrap_nonce, wrapped, _MAGIC)
|
||
|
|
return AESGCM(data_key).decrypt(data_nonce, ciphertext, _MAGIC)
|
||
|
|
except Exception as exc: # cryptography raises InvalidTag etc.
|
||
|
|
raise VaultError("secret decryption failed") from exc
|