from __future__ import annotations import base64 import hashlib import json from typing import Any from cryptography.fernet import Fernet, InvalidToken from app.core.config import get_settings from app.core.exceptions import ValidationError SECRET_PREFIX = "fernet:v1:" class SecretConfigError(ValidationError): code = "secret_config_invalid" message = "Secret configuration cannot be decrypted." def encrypt_secret_config(secret_config: dict[str, Any]) -> str: payload = json.dumps(secret_config, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") token = _fernet().encrypt(payload).decode("utf-8") return SECRET_PREFIX + token def decrypt_secret_config(raw_value: str) -> dict[str, Any]: if not raw_value: return {} if not raw_value.startswith(SECRET_PREFIX): value = json.loads(raw_value) return value if isinstance(value, dict) else {} token = raw_value.removeprefix(SECRET_PREFIX).encode("utf-8") try: payload = _fernet().decrypt(token) except InvalidToken as exc: raise SecretConfigError() from exc value = json.loads(payload.decode("utf-8")) return value if isinstance(value, dict) else {} def list_secret_fields(raw_value: str) -> list[str]: return sorted(decrypt_secret_config(raw_value)) def _fernet() -> Fernet: digest = hashlib.sha256(get_settings().secret_key.encode("utf-8")).digest() key = base64.urlsafe_b64encode(digest) return Fernet(key)