77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""Symmetric encryption for secrets held at rest.
|
|
|
|
Module trust credentials — the HMAC secrets and static keys every module
|
|
integration is authenticated with — were stored as a plain JSON column. A
|
|
read-only database leak, a backup, or a support export handed over the signing
|
|
keys for every integration at once.
|
|
|
|
The key is derived from `ENCRYPTION_KEY` rather than used directly, so any
|
|
sufficiently long secret works as configuration without callers having to
|
|
produce a correctly-formatted Fernet key.
|
|
|
|
**This protects against database exposure, not repository exposure.** The
|
|
`ENCRYPTION_KEY` for this application currently lives in committed `.env.*`
|
|
files. Rotating those secrets out of version control is a separate task in the
|
|
same phase, and until it is done the honest description of this control is
|
|
"raises the cost of a database leak".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from app.config.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_PREFIX = "enc:v1:"
|
|
|
|
|
|
class EncryptionUnavailable(RuntimeError):
|
|
"""ENCRYPTION_KEY is not configured, so nothing can be encrypted or read."""
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
key = settings.ENCRYPTION_KEY
|
|
if not key:
|
|
raise EncryptionUnavailable(
|
|
"ENCRYPTION_KEY is not set. Module trust credentials cannot be "
|
|
"encrypted or decrypted without it."
|
|
)
|
|
digest = hashlib.sha256(key.encode("utf-8")).digest()
|
|
return Fernet(base64.urlsafe_b64encode(digest))
|
|
|
|
|
|
def encrypt(plaintext: str) -> str:
|
|
return _PREFIX + _fernet().encrypt(plaintext.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def decrypt(ciphertext: str) -> str:
|
|
if not is_encrypted(ciphertext):
|
|
raise ValueError("Value is not an encrypted payload")
|
|
try:
|
|
return _fernet().decrypt(ciphertext[len(_PREFIX) :].encode("ascii")).decode("utf-8")
|
|
except InvalidToken as e:
|
|
raise ValueError(
|
|
"Could not decrypt value — the ENCRYPTION_KEY does not match the one "
|
|
"it was encrypted with."
|
|
) from e
|
|
|
|
|
|
def is_encrypted(value: Optional[str]) -> bool:
|
|
return isinstance(value, str) and value.startswith(_PREFIX)
|
|
|
|
|
|
def encrypt_json(payload: dict[str, Any]) -> str:
|
|
return encrypt(json.dumps(payload, sort_keys=True, separators=(",", ":")))
|
|
|
|
|
|
def decrypt_json(ciphertext: str) -> dict[str, Any]:
|
|
return json.loads(decrypt(ciphertext))
|