"""Fernet 对称加密工具:用于加密存储敏感字段(MySQL 密码、S3 SecretKey)。""" from cryptography.fernet import Fernet, InvalidToken from app.config import get_settings _fernet: Fernet | None = None def _get_fernet() -> Fernet: global _fernet if _fernet is None: _fernet = Fernet(get_settings().FERNET_KEY.encode()) return _fernet def encrypt_str(plain: str) -> str: """加密字符串,返回 base64-urlsafe 字符串(可安全存数据库 Text 字段)。""" if plain is None: return "" return _get_fernet().encrypt(plain.encode()).decode() def decrypt_str(token: str) -> str: """解密字符串。空字符串返回空字符串。""" if not token: return "" try: return _get_fernet().decrypt(token.encode()).decode() except InvalidToken as e: raise ValueError(f"Failed to decrypt token (FERNET_KEY 可能已被更改): {e}") from e def encrypt_dict(data: dict) -> str: """加密字典:序列化为 JSON 再加密。""" import json return encrypt_str(json.dumps(data, ensure_ascii=False)) def decrypt_dict(token: str) -> dict: """解密字典。""" import json if not token: return {} return json.loads(decrypt_str(token))