106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from functools import lru_cache
|
|
from typing import Any, Mapping, Protocol, runtime_checkable
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from govoplan_core.settings import settings
|
|
|
|
|
|
class SecretConfigurationError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class SecretDecryptionError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class TransientPayloadError(RuntimeError):
|
|
pass
|
|
|
|
|
|
CAPABILITY_SECURITY_SECRET_PROVIDER = "security.secretProvider" # noqa: S105 # nosec B105 - capability identifier.
|
|
|
|
|
|
@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
|
|
|
|
|
|
def seal_transient_payload(payload: Mapping[str, Any]) -> str:
|
|
"""Seal a short-lived JSON object without persisting server-side state."""
|
|
|
|
encoded = json.dumps(
|
|
dict(payload),
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
return _fernet().encrypt(encoded).decode("utf-8")
|
|
|
|
|
|
def open_transient_payload(token: str, *, ttl_seconds: int) -> dict[str, Any]:
|
|
"""Open a sealed JSON object and enforce its maximum age."""
|
|
|
|
if ttl_seconds <= 0:
|
|
raise ValueError("Transient payload TTL must be positive")
|
|
try:
|
|
encoded = _fernet().decrypt(token.encode("utf-8"), ttl=ttl_seconds)
|
|
payload = json.loads(encoded.decode("utf-8"))
|
|
except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
|
|
raise TransientPayloadError("Transient payload is invalid or expired") from exc
|
|
if not isinstance(payload, dict):
|
|
raise TransientPayloadError("Transient payload must contain a JSON object")
|
|
return payload
|