Spaces:
Paused
Paused
File size: 5,426 Bytes
4223796 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """
MFA Encryption Utilities — CrowData.
Cifra/descifra secretos TOTP y códigos de respaldo usando Fernet (AES-128-GCM).
La clave se deriva de MFA_ENCRYPTION_KEY (base64, 32 bytes) en settings.
Soporte de versiones:
- v2: Fernet con MFA_ENCRYPTION_KEY (actual, recomendado)
- v1: PBKDF2 derivado de SECRET_KEY (legacy, solo lectura para migración)
Formato v1: `v1:<fernet_token>`
Formato v2: `v2:<fernet_token>`
"""
from cryptography.fernet import Fernet, InvalidToken
from app.config import get_settings
import base64
import json
import logging
logger = logging.getLogger(__name__)
_settings = None
# ─── v2: Fernet con MFA_ENCRYPTION_KEY ───
def _get_fernet_v2() -> Fernet:
global _settings
if _settings is None:
_settings = get_settings()
key = _settings.mfa_encryption_key
if not key:
raise RuntimeError("MFA_ENCRYPTION_KEY no configurado en .env")
try:
decoded = base64.urlsafe_b64decode(key)
if len(decoded) != 32:
raise ValueError
except Exception:
raise RuntimeError("MFA_ENCRYPTION_KEY debe ser base64 url-safe de 32 bytes (generar con: python -c \"import base64, os; print(base64.urlsafe_b64encode(os.urandom(32)).decode())\")")
return Fernet(key.encode())
# ─── v1: PBKDF2 derivado de SECRET_KEY (legacy, solo lectura) ───
_v1_fernet = None
def _get_fernet_v1() -> Fernet:
global _v1_fernet
if _v1_fernet is None:
s = get_settings()
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b'crowdata-mfa-encryption',
iterations=100000,
)
key = base64.urlsafe_b64encode(kdf.derive(s.secret_key.encode()))
_v1_fernet = Fernet(key)
return _v1_fernet
# ─── API pública versionada ───
VERSION_PREFIX_V1 = b"v1:"
VERSION_PREFIX_V2 = b"v2:"
def _encrypt_with_version(data: str, version: int = 2) -> str:
"""Cifra con prefijo de versión."""
if version == 1:
f = _get_fernet_v1()
prefix = VERSION_PREFIX_V1
else:
f = _get_fernet_v2()
prefix = VERSION_PREFIX_V2
return (prefix + f.encrypt(data.encode())).decode()
def _decrypt_with_version(encrypted: str) -> str:
"""Descifra detectando automáticamente la versión por prefijo."""
if not encrypted:
raise ValueError("Empty encrypted data")
data_bytes = encrypted.encode()
if data_bytes.startswith(VERSION_PREFIX_V1):
f = _get_fernet_v1()
return f.decrypt(data_bytes[len(VERSION_PREFIX_V1):]).decode()
elif data_bytes.startswith(VERSION_PREFIX_V2):
f = _get_fernet_v2()
return f.decrypt(data_bytes[len(VERSION_PREFIX_V2):]).decode()
else:
# Sin prefijo = legacy sin versión, intentar v1 (formato anterior)
logger.warning("Encrypted data without version prefix, trying v1 (legacy)")
try:
return _get_fernet_v1().decrypt(data_bytes).decode()
except InvalidToken:
# Último intento con v2
return _get_fernet_v2().decrypt(data_bytes).decode()
# ─── API pública (usa v2 para escribir, ambas para leer) ───
def encrypt_mfa(data: str) -> str:
"""Cifra un string (secreto TOTP) usando v2."""
return _encrypt_with_version(data, version=2)
def decrypt_mfa(encrypted: str) -> str:
"""Descifra detectando versión automáticamente (v1 o v2)."""
return _decrypt_with_version(encrypted)
def encrypt_backup_codes(codes: list[str]) -> str:
"""Cifra una lista de códigos de respaldo como JSON usando v2."""
import json
return encrypt_mfa(json.dumps(codes))
def decrypt_backup_codes(encrypted: str) -> list[str]:
"""Descifra y parsea la lista de códigos de respaldo."""
import json
return json.loads(decrypt_mfa(encrypted))
# ─── Migración / compatibilidad ───
def is_encrypted(data: str) -> bool:
"""Detecta si un string parece estar cifrado (tiene prefijo v1: o v2: o parece Fernet)."""
try:
if data.startswith(("v1:", "v2:")):
return True
# Legacy sin prefijo
return len(data) > 40 and data.endswith("=")
except Exception:
return False
def safe_decrypt_mfa(data: str) -> str:
"""Descifra si está cifrado, sino retorna tal cual (para migración gradual)."""
if not data:
return data
if is_encrypted(data):
try:
return decrypt_mfa(data)
except Exception as e:
logger.warning(f"Failed to decrypt MFA data: {e}")
return data # Fallback
return data
def safe_decrypt_backup_codes(data: str) -> list[str]:
"""Descifra códigos de respaldo si están cifrados, sino parsea JSON o lista vacía."""
if not data:
return []
if is_encrypted(data):
try:
return decrypt_backup_codes(data)
except Exception:
pass
# Fallback: intentar parsear como JSON
import json
try:
return json.loads(data)
except Exception:
return []
def migrate_to_v2(encrypted_v1: str) -> str:
"""Re-cifra dato v1 a v2. Para migración batch."""
plaintext = _get_fernet_v1().decrypt(encrypted_v1.encode()).decode()
return encrypt_mfa(plaintext) |