57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""Encryption for Cloud-managed LLM Provider credentials."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
|
|
ENCRYPTION_KEY_ENV = "CLOUD_LLM_PROVIDER_ENCRYPTION_KEY"
|
|
|
|
|
|
class ProviderSecretConfigurationError(ValueError):
|
|
"""Raised when deployment configuration cannot protect a Provider secret."""
|
|
|
|
|
|
class ProviderSecretDecryptionError(ValueError):
|
|
"""Raised when a persisted Provider secret cannot be decrypted."""
|
|
|
|
|
|
class ProviderSecretBox:
|
|
"""Small, testable boundary around Fernet encryption."""
|
|
|
|
def __init__(self, key: str) -> None:
|
|
try:
|
|
self._fernet = Fernet(key.encode("ascii"))
|
|
except (UnicodeEncodeError, ValueError) as exc:
|
|
raise ProviderSecretConfigurationError(
|
|
f"{ENCRYPTION_KEY_ENV} must be a valid Fernet key"
|
|
) from exc
|
|
|
|
def encrypt(self, secret: str) -> str:
|
|
if not secret:
|
|
raise ValueError("Provider API key must not be empty")
|
|
return self._fernet.encrypt(secret.encode("utf-8")).decode("ascii")
|
|
|
|
def decrypt(self, ciphertext: str) -> str:
|
|
try:
|
|
return self._fernet.decrypt(ciphertext.encode("ascii")).decode("utf-8")
|
|
except (InvalidToken, UnicodeDecodeError, UnicodeEncodeError) as exc:
|
|
raise ProviderSecretDecryptionError(
|
|
"stored Provider API key cannot be decrypted"
|
|
) from exc
|
|
|
|
|
|
def load_provider_secret_box(
|
|
env: Mapping[str, str] | None = None,
|
|
) -> ProviderSecretBox:
|
|
values = os.environ if env is None else env
|
|
key = values.get(ENCRYPTION_KEY_ENV, "")
|
|
if not key:
|
|
raise ProviderSecretConfigurationError(
|
|
f"{ENCRYPTION_KEY_ENV} must be configured for database Provider management"
|
|
)
|
|
return ProviderSecretBox(key)
|