79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
|
"""Encryption and presentation helpers for credentials stored in the database."""
|
||
|
|
import re
|
||
|
|
|
||
|
|
from cryptography.fernet import Fernet, InvalidToken
|
||
|
|
|
||
|
|
from app.utils.Secrets import get_runtime_secret
|
||
|
|
|
||
|
|
|
||
|
|
PREFIX = "enc:v1:"
|
||
|
|
SENSITIVE_KEYS = re.compile(
|
||
|
|
r"(?:api[_-]?key|password|passwd|secret|token|authorization|credential)", re.I
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _fernet():
|
||
|
|
return Fernet(get_runtime_secret("credential_encryption_key").encode("ascii"))
|
||
|
|
|
||
|
|
|
||
|
|
def is_encrypted(value):
|
||
|
|
return isinstance(value, str) and value.startswith(PREFIX)
|
||
|
|
|
||
|
|
|
||
|
|
def encrypt_credential(value):
|
||
|
|
value = "" if value is None else str(value)
|
||
|
|
if not value or is_encrypted(value):
|
||
|
|
return value
|
||
|
|
return PREFIX + _fernet().encrypt(value.encode("utf-8")).decode("ascii")
|
||
|
|
|
||
|
|
|
||
|
|
def decrypt_credential(value):
|
||
|
|
value = "" if value is None else str(value)
|
||
|
|
if not is_encrypted(value):
|
||
|
|
return value
|
||
|
|
try:
|
||
|
|
return _fernet().decrypt(value[len(PREFIX):].encode("ascii")).decode("utf-8")
|
||
|
|
except (InvalidToken, ValueError) as exc:
|
||
|
|
raise RuntimeError("credential decryption failed; check the configured encryption key") from exc
|
||
|
|
|
||
|
|
|
||
|
|
def mask_credential(value):
|
||
|
|
plain = decrypt_credential(value)
|
||
|
|
if not plain:
|
||
|
|
return ""
|
||
|
|
if len(plain) <= 4:
|
||
|
|
return "****"
|
||
|
|
return plain[:2] + "****" + plain[-2:]
|
||
|
|
|
||
|
|
|
||
|
|
def redact_mapping(value):
|
||
|
|
if not isinstance(value, dict):
|
||
|
|
return value
|
||
|
|
return {key: ("[REDACTED]" if SENSITIVE_KEYS.search(str(key)) else val)
|
||
|
|
for key, val in value.items()}
|
||
|
|
|
||
|
|
|
||
|
|
def migrate_existing_credentials():
|
||
|
|
"""Idempotently encrypt legacy plaintext credentials in existing rows."""
|
||
|
|
from app.models import LLMModel, StreamModel
|
||
|
|
|
||
|
|
changed = 0
|
||
|
|
for item in LLMModel.objects.only("id", "api_key").iterator():
|
||
|
|
encrypted = encrypt_credential(item.api_key)
|
||
|
|
if encrypted != item.api_key:
|
||
|
|
LLMModel.objects.filter(pk=item.pk).update(api_key=encrypted)
|
||
|
|
changed += 1
|
||
|
|
for item in StreamModel.objects.only(
|
||
|
|
"id", "pull_stream_username", "pull_stream_password"
|
||
|
|
).iterator():
|
||
|
|
updates = {}
|
||
|
|
for field in ("pull_stream_username", "pull_stream_password"):
|
||
|
|
old_value = getattr(item, field)
|
||
|
|
new_value = encrypt_credential(old_value)
|
||
|
|
if new_value != old_value:
|
||
|
|
updates[field] = new_value
|
||
|
|
if updates:
|
||
|
|
StreamModel.objects.filter(pk=item.pk).update(**updates)
|
||
|
|
changed += 1
|
||
|
|
return changed
|