A R R R Associates
42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from app.core.settings import get_settings
|
|
|
|
|
|
class VaultCryptoError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _master_key() -> bytes:
|
|
settings = get_settings()
|
|
raw = (getattr(settings, "VAULT_MASTER_KEY", "") or settings.SECRET_KEY or "").strip()
|
|
if not raw or raw == "change-me-to-a-long-random-string":
|
|
raise VaultCryptoError("VAULT_MASTER_KEY must be configured with a strong production secret before using the credential vault.")
|
|
return raw.encode("utf-8")
|
|
|
|
|
|
def _fernet(tenant_id: int) -> Fernet:
|
|
digest = hmac.new(_master_key(), f"audit-firm-vault:tenant:{tenant_id}:v1".encode(), hashlib.sha256).digest()
|
|
return Fernet(base64.urlsafe_b64encode(digest))
|
|
|
|
|
|
def encrypt_value(tenant_id: int, value: str | None) -> str | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return _fernet(tenant_id).encrypt(value.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def decrypt_value(tenant_id: int, token: str | None) -> str:
|
|
if not token:
|
|
return ""
|
|
try:
|
|
return _fernet(tenant_id).decrypt(token.encode("ascii")).decode("utf-8")
|
|
except InvalidToken as exc:
|
|
raise VaultCryptoError("The credential could not be decrypted with the configured vault key.") from exc
|