"""Fernet symmetric encryption for connector credentials at rest.
Reads the encryption key from the CONNECTOR_ENCRYPTION_KEY environment variable.
If not set, falls back to generating an ephemeral key (NOT secure for production
— data encrypted with the ephemeral key is lost on process restart).
Set in production:
export CONNECTOR_ENCRYPTION_KEY=$(python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
The key must be a 32 url-safe base64-encoded bytes (44 characters).
"""
from __future__ import annotations
import os
import base64
import json
import logging
from cryptography.fernet import Fernet, InvalidToken
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Key management
# ---------------------------------------------------------------------------
_KEY = os.environ.get("CONNECTOR_ENCRYPTION_KEY")
_fernet: Fernet
if _KEY:
# A valid Fernet key is exactly 44 url-safe base64 characters.
if len(_KEY) == 44:
_fernet = Fernet(_KEY.encode())
else:
# Assume raw 32-byte key — encode to the expected base64 format.
_fernet = Fernet(base64.urlsafe_b64encode(_KEY.encode()))
logger.info("Connector encryption initialised with env-provided key")
else:
# Check if we're in production mode
if os.environ.get('FLASK_ENV') == 'production':
raise RuntimeError(
"CONNECTOR_ENCRYPTION_KEY not set in production. "
"Connector credentials will be corrupted on restart. "
"Generate a key: python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'"
)
else:
# Fallback: generate ephemeral key.
# WARNING: encrypted data is lost on restart; this is ONLY for
# development / testing where no real secrets are stored.
_fernet = Fernet(Fernet.generate_key())
logger.warning(
"CONNECTOR_ENCRYPTION_KEY not set — using ephemeral key. "
"Encrypted data will be lost on restart. Set CONNECTOR_ENCRYPTION_KEY "
"for production use."
)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def encrypt_config(config: dict) -> str:
"""Encrypt a config dict to a Fernet token string.
Args:
config: Plain-text configuration dict (e.g. API keys, OAuth tokens).
Returns:
A Fernet-encrypted token as a UTF-8 string.
"""
plaintext = json.dumps(config, default=str).encode()
return _fernet.encrypt(plaintext).decode()
def decrypt_config(token: str) -> dict:
"""Decrypt a Fernet token string back to a config dict.
Args:
token: A Fernet-encrypted token string.
Returns:
The decrypted configuration dict.
Raises:
cryptography.fernet.InvalidToken: if the token is corrupted or the
key doesn't match.
"""
if not token:
return {}
try:
plaintext = _fernet.decrypt(token.encode())
return json.loads(plaintext.decode())
except InvalidToken:
logger.error("Failed to decrypt connector config — wrong key or corrupted token")
raise