marcos/hechakuaa_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

security: vault v3 — PBKDF2 + envelope encryption (DEK/KEK)

Browse Source 
Reemplaza derivación SHA256 por PBKDF2-HMAC-SHA256 (600k iter) con salt
aleatorio por tenant + DEK de 32 bytes aleatoria por cifrado. Robar la DB
sin FERNET_KEY o FERNET_KEY sin la DB ya no alcanza para comprometer credenciales.
Compatibilidad v1 preservada para migración. Agrega migrar_vault_v3.py con
dry-run, pre-flight check y backup atómico.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
markosbenitez
2026-05-26 11:04:51 -03:00
parent 5eb934d4a1
commit 05b2f84eb8
3 changed files with 333 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@@ -11,6 +11,28 @@ Versiones: [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
---
## [0.4.0] — 2026-05-26
### Security
- **Envelope encryption v3 en `vault.py`** — credenciales cifradas en dos capas independientes:
- DEK (Data Encryption Key): 32 bytes aleatorios, única por operación de cifrado.
- Wrapping Key: derivada de `FERNET_KEY` con PBKDF2-HMAC-SHA256, 600k iteraciones, salt de 16 bytes aleatorio por tenant. ~200ms por derivación — fuerza bruta inviable incluso con GPU.
- Formato en DB: `{"v":3,"s":"<salt_b64>","k":"<dek_enc>","p":"<payload_enc>"}`
- Propiedades: robar la DB sin `FERNET_KEY` es inútil; robar `FERNET_KEY` sin la DB también.
- **Cache de wrapping keys** (`lru_cache`) — evita ejecutar PBKDF2 600k iter por cada tenant en cada ciclo del poller. Cache key incluye fingerprint del KEK para invalidar si cambia `FERNET_KEY`.
- **Compatibilidad v1 preservada** — `descifrar_credenciales` detecta automáticamente formato v1 (Fernet token puro) vs v3 (JSON). No se generan nuevas v1.
### Added
- **`scripts/migrar_vault_v3.py`** — migración segura de credenciales v1 → v3:
- Verifica que todos los tenants v1 sean descifrables ANTES de tocar la DB (aborta si alguno falla).
- Backup automático de la DB antes de cualquier escritura.
- Flag `--dry-run` para ver qué haría sin modificar nada.
- Commit atómico: o migra todos o no migra ninguno.
---
## [0.3.1] — 2026-05-26
### Security

163
scripts/migrar_vault_v3.py Normal file
View File

@@ -0,0 +1,163 @@
#!/usr/bin/env python3
"""
scripts/migrar_vault_v3.py — Migración de credenciales v1 → v3 (PBKDF2 + envelope encryption).
Qué hace:
1. Lee FERNET_KEY del .env
2. Conecta a la DB y lista todos los tenants con credentials_enc
3. Saltea tenants ya en v3 (JSON, empieza con '{')
4. Para cada tenant v1: descifra con el algoritmo legacy, re-cifra con v3
5. Si algún tenant falla al descifrar: aborta SIN escribir nada
6. Si todo OK: hace backup de la DB, aplica todos los UPDATE en una transacción atómica
Uso:
python scripts/migrar_vault_v3.py # migra de verdad
python scripts/migrar_vault_v3.py --dry-run # solo muestra qué haría, no escribe
Seguridad:
- Las credenciales descifradas NUNCA se escriben en disco ni en logs
- El backup de la DB se hace ANTES de cualquier modificación
- Si el proceso se interrumpe a mitad, el backup permite restaurar
"""
from __future__ import annotations
import argparse
import asyncio
import shutil
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
async def main() -> None:
parser = argparse.ArgumentParser(
description="Migra credenciales de vault v1 a v3 (PBKDF2 + envelope encryption)"
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Muestra qué haría sin escribir nada en la DB",
)
args = parser.parse_args()
env_path = Path(".env")
if not env_path.exists():
print("ERROR: No encontré .env. Ejecutá desde el directorio raíz del proyecto.")
sys.exit(1)
# Leer FERNET_KEY del .env
import re
env_contenido = env_path.read_text(encoding="utf-8")
m = re.search(r"^FERNET_KEY=(.+)$", env_contenido, re.MULTILINE)
if not m or not m.group(1).strip():
print("ERROR: FERNET_KEY no encontrada o vacía en .env")
sys.exit(1)
fernet_key = m.group(1).strip()
from config import get_settings
from database import run_migrations, get_conn
import vault
settings = get_settings()
await run_migrations(settings.db_path)
conn = get_conn()
# Traer todos los tenants con credentials_enc
async with conn.execute(
"SELECT chat_id, credentials_enc FROM tenants WHERE credentials_enc IS NOT NULL"
) as cur:
tenants = list(await cur.fetchall())
if not tenants:
print("No hay tenants con credenciales en la DB. Nada que migrar.")
return
total = len(tenants)
ya_v3 = 0
a_migrar: list[tuple[int, str]] = [] # (chat_id, credentials_enc_v1)
for row in tenants:
if vault.es_formato_v3(row["credentials_enc"]):
ya_v3 += 1
else:
a_migrar.append((row["chat_id"], row["credentials_enc"]))
print()
print("=== Migración vault v1 → v3 ===")
print()
print(f" Tenants totales: {total}")
print(f" Ya en v3: {ya_v3} (se saltean)")
print(f" A migrar (v1): {len(a_migrar)}")
print()
if not a_migrar:
print(" Todos los tenants ya están en v3. Nada que hacer.")
return
if args.dry_run:
print(" [DRY RUN] Los siguientes tenants serían migrados:")
for chat_id, _ in a_migrar:
print(f" chat_id={chat_id}")
print()
print(" [DRY RUN] No se escribió nada.")
return
# Descifrar todos ANTES de tocar la DB — abortar si alguno falla
actualizaciones: list[tuple[str, int]] = [] # (nuevo_enc, chat_id)
errores: list[tuple[int, str]] = []
print(" Verificando que todos los v1 sean descifrables...")
for chat_id, credentials_enc in a_migrar:
try:
usuario, password = vault.descifrar_credenciales(chat_id, credentials_enc, fernet_key)
nuevo_enc = vault.cifrar_credenciales(chat_id, usuario, password, fernet_key)
actualizaciones.append((nuevo_enc, chat_id))
del usuario, password # no dejar en memoria más de lo necesario
except ValueError as e:
errores.append((chat_id, str(e)))
print(f" ERROR tenant {chat_id}: {e}")
if errores:
print()
print(f" {len(errores)} tenant(s) no pudieron descifrarse.")
print(" La DB NO fue modificada. Revisá los errores de arriba.")
sys.exit(1)
print(f" OK — {len(actualizaciones)} tenant(s) listos para migrar.")
print()
# Confirmar
confirmar = input(" Escribí 'migrar' para confirmar y escribir en la DB: ").strip()
if confirmar != "migrar":
print(" Operación cancelada.")
sys.exit(0)
# Backup de la DB ANTES de cualquier escritura
db_path = settings.db_path
bak_path = db_path.with_suffix(".db.bak")
shutil.copy2(db_path, bak_path)
print(f"\n Backup guardado en {bak_path}")
# Aplicar todos los UPDATE en una sola transacción
for nuevo_enc, chat_id in actualizaciones:
await conn.execute(
"UPDATE tenants SET credentials_enc = ? WHERE chat_id = ?",
(nuevo_enc, chat_id),
)
await conn.commit()
print(f" {len(actualizaciones)} tenant(s) migrados a v3.")
print()
print(" ✅ Migración completada.")
print(f" Backup de la DB anterior: {bak_path}")
print()
print(" Próximo paso: reiniciar el bot.")
print(" systemctl restart pedrito (en VPS con systemd)")
print(" docker compose restart (en VPS con Docker)")
print(" o Ctrl+C y uv run python main.py (en local)")
print()
if __name__ == "__main__":
asyncio.run(main())

169
vault.py
View File

@@ -1,55 +1,182 @@
"""
vault.py — Cifrado de credenciales por tenant.
Cada tenant tiene sus credenciales cifradas con una clave derivada de:
FERNET_KEY (global, del .env) + str(chat_id) (único por tenant)
## Arquitectura v3: Envelope Encryption con PBKDF2 + DEK por tenant
Esto asegura que las credenciales de un tenant no se pueden descifrar
con solo la FERNET_KEY — hace falta también el chat_id.
Cada credencial se cifra en dos capas independientes:
1. DEK (Data Encryption Key): 32 bytes aleatorios, única por operación de cifrado.
2. KEK→Wrapping Key: derivada de FERNET_KEY usando PBKDF2-HMAC-SHA256 con salt único
por tenant. El salt se genera al cifrar y se guarda junto al ciphertext.
La KEK (FERNET_KEY del .env) nunca cifra datos directamente — solo envuelve la DEK.
Formato credentials_enc v3:
JSON compacto: {"v":3,"s":"<salt_b64>","k":"<dek_enc_b64>","p":"<payload_enc_b64>"}
s = salt de 16 bytes (base64url) para derivar la wrapping key via PBKDF2
k = DEK cifrada con Fernet(wrapping_key)
p = {"usuario": ..., "password": ...} cifrado con Fernet(DEK)
Ventajas sobre v1 (SHA256):
- PBKDF2 (600k iter): derivar la wrapping key cuesta ~200ms — fuerza bruta inviable
- Salt único por tenant: mismo FERNET_KEY + mismo password → ciphertexts distintos
- DEK aleatoria: comprometer una credencial no ayuda con las demás
- Dos capas: robar la DB sin FERNET_KEY es inútil; robar FERNET_KEY sin la DB también
## Formato v1 (legacy, solo lectura)
Fernet(SHA256(FERNET_KEY + str(chat_id))) — mantenido para migración desde el script
`scripts/migrar_vault_v3.py`. No se generan nuevos v1.
"""
from __future__ import annotations
import base64
import hashlib
import json
import os
from functools import lru_cache
import structlog
from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from utils.sanitize import scrub
logger = structlog.get_logger(__name__)
# PBKDF2: NIST SP 800-132 (2023) recomienda mínimo 600k iteraciones para SHA-256
_PBKDF2_ITERATIONS = 600_000
_SALT_BYTES = 16
_DEK_BYTES = 32
def _derive_key(fernet_key: str, chat_id: int) -> bytes:
"""Deriva una clave Fernet única por tenant usando SHA-256(fernet_key + chat_id)."""
raw = (fernet_key + str(chat_id)).encode()
digest = hashlib.sha256(raw).digest()
return base64.urlsafe_b64encode(digest)
def _derive_wrapping_key(kek: str, salt: bytes) -> bytes:
"""
Deriva una wrapping key de 32 bytes usando PBKDF2-HMAC-SHA256.
~200ms intencionalmente — hace fuerza bruta inviable incluso con GPU.
"""
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=_DEK_BYTES,
salt=salt,
iterations=_PBKDF2_ITERATIONS,
)
return base64.urlsafe_b64encode(kdf.derive(kek.encode("utf-8")))
@lru_cache(maxsize=256)
def _wrapping_key_cached(kek_fingerprint: str, salt_hex: str, kek: str) -> bytes:
"""
Cache de wrapping keys por (salt, kek) para no ejecutar PBKDF2 600k veces
en el mismo proceso cuando el poller revisa múltiples tenants.
kek_fingerprint es el SHA256[:8] de kek — sirve para invalidar entradas
si el FERNET_KEY cambia (nueva instancia del proceso).
"""
return _derive_wrapping_key(kek, bytes.fromhex(salt_hex))
def _get_wrapping_key(kek: str, salt: bytes) -> bytes:
"""Obtiene wrapping key del cache o la deriva si no está."""
fingerprint = hashlib.sha256(kek.encode()).hexdigest()[:8]
return _wrapping_key_cached(fingerprint, salt.hex(), kek)
# ── API pública ───────────────────────────────────────────────────────────────
def cifrar_credenciales(chat_id: int, usuario: str, password: str, fernet_key: str) -> str:
"""Retorna JSON cifrado con Fernet listo para guardar en DB."""
key = _derive_key(fernet_key, chat_id)
f = Fernet(key)
payload = json.dumps({"usuario": usuario, "password": password}).encode()
encrypted = f.encrypt(payload).decode()
logger.info("credenciales_cifradas", chat_id=chat_id)
return encrypted
"""
Cifra credenciales con envelope encryption v3.
Retorna JSON compacto listo para guardar en DB.
"""
# Capa 1: generar salt (para PBKDF2) y DEK aleatoria
salt = os.urandom(_SALT_BYTES)
dek_raw = os.urandom(_DEK_BYTES)
dek_b64 = base64.urlsafe_b64encode(dek_raw)
# Capa 2: cifrar DEK con wrapping key derivada de FERNET_KEY + salt
wrapping_key = _get_wrapping_key(fernet_key, salt)
f_wrap = Fernet(wrapping_key)
dek_enc = f_wrap.encrypt(dek_b64)
# Capa 3: cifrar payload con DEK
f_dek = Fernet(dek_b64)
payload_bytes = json.dumps({"usuario": usuario, "password": password}).encode()
payload_enc = f_dek.encrypt(payload_bytes)
envelope = {
"v": 3,
"s": base64.urlsafe_b64encode(salt).decode(),
"k": dek_enc.decode(),
"p": payload_enc.decode(),
}
logger.info("credenciales_cifradas_v3", chat_id=chat_id)
return json.dumps(envelope, separators=(",", ":"))
def descifrar_credenciales(chat_id: int, credentials_enc: str, fernet_key: str) -> tuple[str, str]:
"""
Retorna (usuario, password).
Levanta ValueError si no se puede descifrar (clave incorrecta o datos corruptos).
Descifra credenciales. Soporta v1 (legacy) y v3 (actual).
Levanta ValueError si no se puede descifrar.
"""
key = _derive_key(fernet_key, chat_id)
stripped = credentials_enc.strip()
if stripped.startswith("{"):
return _descifrar_v3(chat_id, stripped, fernet_key)
return _descifrar_v1(chat_id, stripped, fernet_key)
def es_formato_v3(credentials_enc: str) -> bool:
"""Retorna True si las credenciales ya están en formato v3."""
return credentials_enc.strip().startswith("{")
# ── Implementaciones internas ─────────────────────────────────────────────────
def _descifrar_v3(chat_id: int, credentials_enc: str, fernet_key: str) -> tuple[str, str]:
try:
envelope = json.loads(credentials_enc)
if envelope.get("v") != 3:
raise ValueError(f"Versión desconocida: {envelope.get('v')}")
salt = base64.urlsafe_b64decode(envelope["s"])
dek_enc = envelope["k"].encode()
payload_enc = envelope["p"].encode()
# Descifrar DEK con wrapping key derivada de PBKDF2
wrapping_key = _get_wrapping_key(fernet_key, salt)
f_wrap = Fernet(wrapping_key)
dek_b64 = f_wrap.decrypt(dek_enc)
# Descifrar payload con DEK
f_dek = Fernet(dek_b64)
payload = json.loads(f_dek.decrypt(payload_enc).decode())
return payload["usuario"], payload["password"]
except (InvalidToken, KeyError, ValueError, json.JSONDecodeError) as e:
logger.error("error_descifrar_v3", chat_id=chat_id, error=scrub(str(e)))
raise ValueError(
f"No se pudieron descifrar credenciales (v3) para chat_id={chat_id}"
) from e
def _descifrar_v1(chat_id: int, credentials_enc: str, fernet_key: str) -> tuple[str, str]:
"""
Legacy: clave derivada con SHA256(fernet_key + chat_id).
Disponible para migración — no se generan nuevas v1.
"""
raw = (fernet_key + str(chat_id)).encode()
key = base64.urlsafe_b64encode(hashlib.sha256(raw).digest())
f = Fernet(key)
try:
payload = json.loads(f.decrypt(credentials_enc.encode()).decode())
logger.warning("credenciales_v1_descifradas_migracion_pendiente", chat_id=chat_id)
return payload["usuario"], payload["password"]
except (InvalidToken, KeyError, ValueError) as e:
logger.error("error_descifrar_credenciales", chat_id=chat_id, error=scrub(str(e)))
logger.error("error_descifrar_v1", chat_id=chat_id, error=scrub(str(e)))
raise ValueError(
f"No se pudieron descifrar credenciales para chat_id={chat_id}"
f"No se pudieron descifrar credenciales (v1) para chat_id={chat_id}"
) from e