52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""Encryption at rest for device credentials."""
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from sqlalchemy.types import Text, TypeDecorator
|
|
|
|
from app.core.config import settings
|
|
|
|
CREDENTIAL_PREFIX = "enc:v1:"
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
key = settings.CREDENTIAL_ENCRYPTION_KEY.strip()
|
|
if not key:
|
|
raise RuntimeError("Credential encryption key is not configured")
|
|
try:
|
|
return Fernet(key.encode())
|
|
except (TypeError, ValueError) as error:
|
|
raise RuntimeError("Credential encryption key is invalid") from error
|
|
|
|
|
|
def encrypt_credential(value: str) -> str:
|
|
"""Encrypt a plaintext credential with the deployment-provided key."""
|
|
if value.startswith(CREDENTIAL_PREFIX):
|
|
return value
|
|
return CREDENTIAL_PREFIX + _fernet().encrypt(value.encode()).decode()
|
|
|
|
|
|
def decrypt_credential(value: str) -> str:
|
|
"""Decrypt an encrypted credential; retain legacy plaintext only for migration."""
|
|
if not value.startswith(CREDENTIAL_PREFIX):
|
|
return value
|
|
try:
|
|
return _fernet().decrypt(value[len(CREDENTIAL_PREFIX):].encode()).decode()
|
|
except InvalidToken as error:
|
|
raise RuntimeError("Credential decryption failed") from error
|
|
|
|
|
|
class EncryptedCredential(TypeDecorator):
|
|
"""SQLAlchemy column type that stores credentials encrypted and reads plaintext."""
|
|
|
|
impl = Text
|
|
cache_ok = True
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
return encrypt_credential(value)
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is None:
|
|
return None
|
|
return decrypt_credential(value)
|