done
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
Envelope encryption for HL API private keys.
|
||||
|
||||
Plaintext keys never touch disk: stored values are Fernet-encrypted with a KEK
|
||||
loaded from env (ENCRYPTION_KEY). Rotate KEK → re-encrypt all keys offline.
|
||||
"""
|
||||
import base64
|
||||
import hashlib
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from cryptography.fernet import Fernet, InvalidToken
|
||||
|
||||
from app.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _derive_fernet_key(raw: str) -> bytes:
|
||||
"""Accept any reasonably-long secret and derive a valid 32-byte Fernet key."""
|
||||
if not raw or len(raw) < 32:
|
||||
raise RuntimeError(
|
||||
"ENCRYPTION_KEY must be set to at least 32 random chars (e.g. `openssl rand -hex 32`)"
|
||||
)
|
||||
digest = hashlib.sha256(raw.encode("utf-8")).digest()
|
||||
return base64.urlsafe_b64encode(digest)
|
||||
|
||||
|
||||
_fernet: Optional[Fernet] = None
|
||||
|
||||
|
||||
def _cipher() -> Fernet:
|
||||
global _fernet
|
||||
if _fernet is None:
|
||||
_fernet = Fernet(_derive_fernet_key(settings.encryption_key))
|
||||
return _fernet
|
||||
|
||||
|
||||
# Prefix lets us distinguish encrypted blobs from any legacy plaintext rows during migration
|
||||
ENC_PREFIX = "enc:v1:"
|
||||
|
||||
|
||||
def encrypt_api_key(plaintext: str) -> str:
|
||||
token = _cipher().encrypt(plaintext.encode("utf-8")).decode("utf-8")
|
||||
return ENC_PREFIX + token
|
||||
|
||||
|
||||
def decrypt_api_key(stored: str) -> str:
|
||||
if not stored:
|
||||
raise ValueError("Empty api key")
|
||||
if not stored.startswith(ENC_PREFIX):
|
||||
# Legacy plaintext row (from before encryption was added). Refuse to use in prod.
|
||||
if settings.environment == "production":
|
||||
raise RuntimeError(
|
||||
"Found legacy-plaintext HL key; run migration script before production"
|
||||
)
|
||||
logger.warning("Reading LEGACY plaintext HL key — migrate ASAP")
|
||||
return stored
|
||||
try:
|
||||
return _cipher().decrypt(stored[len(ENC_PREFIX):].encode("utf-8")).decode("utf-8")
|
||||
except InvalidToken as exc:
|
||||
raise RuntimeError("HL key decryption failed — wrong ENCRYPTION_KEY?") from exc
|
||||
Reference in New Issue
Block a user