You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
50 lines
1.5 KiB
50 lines
1.5 KiB
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)
|
|
|