75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
from functools import lru_cache
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from govoplan_core.settings import settings
|
|
|
|
|
|
class SecretConfigurationError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class SecretDecryptionError(RuntimeError):
|
|
pass
|
|
|
|
|
|
CAPABILITY_SECURITY_SECRET_PROVIDER = "security.secretProvider"
|
|
|
|
|
|
@runtime_checkable
|
|
class SecretProvider(Protocol):
|
|
def store_secret(self, *, scope: str, name: str, value: str) -> str:
|
|
...
|
|
|
|
def read_secret(self, secret_ref: str) -> str | None:
|
|
...
|
|
|
|
def delete_secret(self, secret_ref: str) -> None:
|
|
...
|
|
|
|
|
|
def _normalize_fernet_key(value: str) -> bytes:
|
|
candidate = value.strip().encode("utf-8")
|
|
try:
|
|
Fernet(candidate)
|
|
return candidate
|
|
except (TypeError, ValueError):
|
|
raw = None
|
|
try:
|
|
raw = base64.b64decode(candidate)
|
|
except Exception as exc:
|
|
raise SecretConfigurationError("MASTER_KEY_B64 must be a Fernet key or base64-encoded 32-byte key") from exc
|
|
if len(raw) != 32:
|
|
raise SecretConfigurationError("MASTER_KEY_B64 must decode to exactly 32 bytes")
|
|
return base64.urlsafe_b64encode(raw)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _fernet() -> Fernet:
|
|
if settings.master_key_b64:
|
|
return Fernet(_normalize_fernet_key(settings.master_key_b64))
|
|
if settings.app_env.lower() in {"dev", "test", "local"}:
|
|
key = base64.urlsafe_b64encode(hashlib.sha256(b"multi-seal-mail-development-master-key").digest())
|
|
return Fernet(key)
|
|
raise SecretConfigurationError("MASTER_KEY_B64 is required outside dev/test/local environments")
|
|
|
|
|
|
def encrypt_secret(value: str | None) -> str | None:
|
|
if value in (None, ""):
|
|
return None
|
|
return _fernet().encrypt(str(value).encode("utf-8")).decode("utf-8")
|
|
|
|
|
|
def decrypt_secret(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return _fernet().decrypt(value.encode("utf-8")).decode("utf-8")
|
|
except InvalidToken as exc:
|
|
raise SecretDecryptionError("Stored secret cannot be decrypted with the configured master key") from exc
|