3 Commits

Author SHA1 Message Date
markosbenitez
05b2f84eb8 security: vault v3 — PBKDF2 + envelope encryption (DEK/KEK)
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>
2026-05-26 11:04:51 -03:00
markosbenitez
5eb934d4a1 chore: release v0.3.1 — CHANGELOG + política de versiones en CLAUDE.md
- CHANGELOG.md: historial completo desde v0.1.0 hasta v0.3.1
- CLAUDE.md: sección 'Gestión de versiones' con reglas semver,
  tabla MAJOR/MINOR/PATCH, proceso de release copy-paste
- Scope y archivos del proyecto actualizados al estado real (v0.3.1)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-26 10:43:59 -03:00
markosbenitez
fa72ced5a6 security: hardening contra 7 vectores de ataque identificados
Análisis ofensivo → defensas implementadas:

1. /health/history sin auth (CRÍTICO)
   - Requiere Bearer token (HEALTH_AUTH_TOKEN en .env)
   - chat_ids enmascarados como hash SHA256[:8] opaco ("usr_a3b4c5d6")
   - Accesos denegados loggeados con IP + User-Agent

2. /health revela inteligencia operacional (MEDIO)
   - Sin auth: solo {"status", "timestamp"} — sin conteo de tenants ni timing
   - Con auth: respuesta completa con métricas internas
   - Misma URL, auth desbloquea detalle

3. Variables sensibles en proceso env (CRÍTICO)
   - Después de cargar Settings(), se limpian FERNET_KEY, TELEGRAM_BOT_TOKEN,
     PJ_USUARIO_ENC, PJ_PASSWORD_ENC, WEBHOOK_SECRET, HEALTH_AUTH_TOKEN, INVITE_CODE
   - Protege contra /proc/<pid>/environ y herencia por subprocesos

4. Race condition TOCTOU en invite codes (MEDIO)
   - SELECT+UPDATE reemplazado por UPDATE WHERE usado=0
   - rowcount=0 → False; imposible que dos requests simultáneos consuman el mismo código

5. Timing oracle en invite codes (BAJO-MEDIO)
   - Delay mínimo de 800ms en _paso_codigo() independientemente del resultado
   - secrets.compare_digest() para comparación del código del .env

6. Information disclosure en respuestas a no-registrados (BAJO)
   - _rechazar_no_registrado: "Para usar este bot escribí /start." (sin revelar comandos)
   - handle_texto_libre: silencio total para usuarios sin registro (no confirmar que el bot existe)

7. Nginx rate limiting (DEPLOY.md)
   - limit_req_zone pedrito_webhook: 30r/s, burst=50
   - limit_req_zone pedrito_health: 10r/m, burst=5
   - Headers de seguridad: X-Content-Type-Options, X-Frame-Options

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-26 08:06:36 -03:00
11 changed files with 732 additions and 90 deletions

163
CHANGELOG.md Normal file
View File

@@ -0,0 +1,163 @@
# Changelog — Pedrito Hechakuaa
Todos los cambios notables se documentan acá.
Formato: [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
Versiones: [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
---
## [Unreleased]
---
## [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
- **`/health/history` requiere auth** — `Authorization: Bearer <HEALTH_AUTH_TOKEN>`; sin token
devuelve 401 con header `WWW-Authenticate`. Sin auth configurada, retorna 401 igualmente.
- **chat_ids enmascarados en historial** — reemplazados por hash SHA256[:8] opaco (`usr_a3b4c5d6`);
preserva correlación de eventos sin exponer Telegram IDs (PII).
- **`/health` público mínimo** — sin auth devuelve solo `{status, timestamp}`. Con auth, respuesta
completa con métricas internas. Elimina exposición de conteo de tenants y timing del poller.
- **Env vars borradas del proceso** — después de cargar `Settings()`, se eliminan de `os.environ`:
`FERNET_KEY`, `TELEGRAM_BOT_TOKEN`, `PJ_USUARIO_ENC`, `PJ_PASSWORD_ENC`, `WEBHOOK_SECRET`,
`HEALTH_AUTH_TOKEN`, `INVITE_CODE`. Protege contra `/proc/<pid>/environ`.
- **Race condition TOCTOU en invite codes eliminada** — `SELECT + UPDATE` reemplazado por
`UPDATE WHERE codigo=? AND usado=0`; `rowcount=0` indica código no válido o ya usado.
Imposible que dos requests simultáneos consuman el mismo código.
- **Timing oracle en validación de código mitigado** — delay mínimo garantizado de 800ms en
`_paso_codigo()` independientemente del resultado; `secrets.compare_digest()` para comparación
contra código del `.env`.
- **Respuestas a usuarios no registrados reducidas al mínimo** — `_rechazar_no_registrado` ya no
revela la arquitectura de comandos; texto libre de usuarios sin registro recibe silencio total.
- **Webhook inválido devuelve 404** — en vez de 403, no confirma la existencia del endpoint.
- **Nginx rate limiting** documentado en `DEPLOY.md`: `limit_req_zone` 30r/s para webhook,
10r/min para health; headers `X-Content-Type-Options`, `X-Frame-Options`.
- **`HEALTH_AUTH_TOKEN`** agregado a `config.py` y a la guía de `.env` en `DEPLOY.md`.
---
## [0.3.0] — 2026-05-25
### Added
- **Docker** — `Dockerfile` (imagen `ghcr.io/astral-sh/uv:python3.12-bookworm-slim`),
`docker-compose.yml` con `restart: unless-stopped` y healthcheck, `.dockerignore`.
- **Health check HTTP** (`health.py`) — servidor aiohttp en puerto 8080:
- `GET /health` — estado del bot (ok/degraded)
- `GET /health/history` — últimos N eventos del audit log
- Watcher loop que alerta al admin via Telegram si el poller no actualiza en N horas
- **Schema migrations** — directorio `migrations/001_initial_schema.sql`, función
`run_migrations()` en `database.py`; idempotente con `CREATE TABLE IF NOT EXISTS`.
Tabla `schema_migrations` lleva registro de versiones aplicadas.
- **Webhook opt-in** — `POST /webhook` en aiohttp valida `X-Telegram-Bot-Api-Secret-Token` con
`compare_digest` y pone el Update en `application.update_queue` de PTB. Se activa con
`WEBHOOK_URL` en `.env`; vacío = long-polling.
- `aiohttp>=3.9` como nueva dependencia.
- `WEBHOOK_URL`, `WEBHOOK_SECRET`, `HEALTH_PORT`, `HEALTH_ALERT_HOURS` en `config.py`.
- Poller notifica a health vía callback `on_poll_done(tenant_count)`.
- `revisar_ahora()` retorna `int` (cantidad de tenants revisados).
### Changed
- `init_db()` reemplazado por `run_migrations()` en `main.py`; `init_db()` queda como alias
para compatibilidad con scripts.
- `main.py` arranca `HealthServer` antes del poller; elige polling vs webhook según config;
registra webhook con Telegram en modo webhook; elimina webhook al detener.
- `DEPLOY.md` actualizado con guía completa de 20 secciones: DNS Bluehost/cPanel (con mensaje
exacto en inglés para delegar), Nginx, Certbot/SSL, Docker, webhook, health check, rate limiting.
---
## [0.2.1] — 2026-05-25
### Fixed
- **Campos del API del PJ corregidos** — la API real usa `fechaNotificacion` (no `fecha`),
`descripcionDespacho` (no `juzgado`), `descripcionCircunscripcion` (no `circunscripcion`).
`NotificacionPendiente.__init__` ahora usa fallback `or` para ambos nombres.
- **PDF callback** — usaba `cod_notificacion_origen` como ID de actuación; corregido a
`cod_actuacion_caso` (campo `codActuacionCaso` de la API).
- **403 en `listar_notificaciones`** — manejado explícitamente como `CSJAuthError` en vez de
caer en `raise_for_status()` genérico.
---
## [0.2.0] — 2026-05-25
### Added
- **Multi-tenant** — `database.py` con SQLite vía aiosqlite; tabla `tenants` con credenciales
cifradas por chat_id; `list_active_tenants()`, `save_tenant()`, `delete_tenant()`.
- **Onboarding flow** (`onboarding.py`) — máquina de estados de 8 pasos:
código de invitación → consentimiento → nombre → tono → timezone → cédula → password → guardado.
- **Códigos de invitación** — tabla `invite_codes`, validación de un solo uso atómica,
fallback al código maestro del `.env`; `scripts/generar_codigo.py`.
- **Audit log** — tabla `audit_log`, función `audit.log()` fire-and-forget; cobertura de todos
los eventos de onboarding, polling, comandos y errores.
- **Rate limiting** (`utils/rate_limiter.py`) — ventana deslizante por chat_id, 20 req/60s;
reset en `/resetear`.
- **FERNET_KEY rotation** (`scripts/rotar_fernet_key.py`) — re-cifra todos los tenants de forma
atómica; hace backup de `.env` antes de modificar; aborta si algún tenant falla.
- Comando `/exp` — búsqueda de expedientes por número o carátula.
- Comando `/pdf` — descarga PDF de la actuación más reciente via `permisoFirmado`.
- Tono personalizable por abogado: formal / cotidiano / informal.
- Zona horaria configurable por abogado.
- `TELEGRAM_EXTRA_IDS` para autorizar IDs adicionales sin onboarding.
- `DEPLOY.md` — primera versión de la guía de despliegue en VPS Ubuntu con systemd.
### Changed
- `poller.py` itera sobre todos los tenants activos (era single-tenant).
- Snapshots por tenant: `data/snapshot_{chat_id}.json` (era `data/snapshot.json` global).
- `vault.py` usa derivación de clave Fernet por chat_id: `base64(SHA256(FERNET_KEY + str(chat_id)))`.
### Security
- Contraseña del PJ borrada de memoria inmediatamente tras cifrar (`del password` en onboarding).
- Mensaje de Telegram con la contraseña eliminado del chat vía `delete_message` antes de procesar.
---
## [0.1.0] — 2026-05-25
### Added
- Login al sistema judicial `apps.csj.gov.py`:
`GET /login` → cookie `cookiesession1``POST /autenticador/login` → bearer token (TTL 1h).
- Polling de `GET /Notificaciones/PorRecibir` cada N minutos en horario hábil (lun-vie, 7-18hs).
- Diff vs snapshot JSON local — solo notifica notificaciones con `codNotificacionOrigen` nuevo.
- Relogin automático transparente al expirar el token.
- Bot de Telegram con comandos: `/start`, `/notif`, `/estado`, `/ayuda`, `/salir`.
- Credenciales del PJ cifradas con Fernet en `.env` (`scripts/cifrar_credenciales.py`).
- `scrub()` en `utils/sanitize.py` — sanitiza bearer tokens, JWTs, cookies y passwords
antes de pasar cualquier dato externo a los logs.
- Logging estructurado con `structlog` (JSON en prod, ConsoleRenderer en dev vía `isatty`).
- Configuración vía `pydantic-settings` — único punto de config de la app (`config.py`).
- Stack base: `httpx`, `python-telegram-bot[ext] v21`, `aiosqlite`, `cryptography`, `structlog`,
`pydantic-settings`, `apscheduler`.
- Manejo de señales `SIGINT`/`SIGTERM` para shutdown limpio.

View File

@@ -6,19 +6,21 @@ Pedrito Hechakuaa: un bot de Telegram que monitorea notificaciones pendientes
en el sistema judicial de Paraguay (`apps.csj.gov.py`) y avisa cuando aparece en el sistema judicial de Paraguay (`apps.csj.gov.py`) y avisa cuando aparece
algo nuevo. Uso propio de un solo abogado (single-tenant). algo nuevo. Uso propio de un solo abogado (single-tenant).
**Scope estricto de la v0:** **Versión actual: v0.3.1** — ver historial completo en `CHANGELOG.md`.
- Login al PJ + polling de notificaciones pendientes
- Diff vs snapshot anterior → solo notifica lo nuevo
- Telegram como canal de salida
- Correr en local o VPS con docker-compose
**Fuera del scope de v0 (no agregar sin consulta explícita):** **Scope implementado:**
- Login al PJ + polling de notificaciones pendientes (multi-tenant)
- Diff vs snapshot por tenant → solo notifica lo nuevo
- Telegram como canal de salida con onboarding, tono personalizable y timezone
- Búsqueda de expedientes y descarga de PDFs
- Infraestructura: Docker, health check HTTP, schema migrations, webhook opt-in
**Fuera del scope actual (no agregar sin consulta explícita):**
- Motor de reglas de plazos - Motor de reglas de plazos
- LLM / conversación libre - LLM / conversación libre
- Multi-tenant
- Web de onboarding - Web de onboarding
- Actuaciones de expedientes (es v1) - Actuaciones de expedientes (es v1)
- Acuse de notificaciones (nunca en MVP) - Acuse de notificaciones (nunca — efecto procesal irreversible)
--- ---
@@ -253,9 +255,54 @@ Ante cambio de schema del PJ: actualizar `csj_client.py` y documentar en
| Archivo | Propósito | | Archivo | Propósito |
|---------|-----------| |---------|-----------|
| `CLAUDE.md` | Este archivo — reglas para Claude Code | | `CLAUDE.md` | Este archivo — reglas para Claude Code |
| `OBSERVATIONS.md` | Hallazgos durante el desarrollo (crear si no existe) | | `CHANGELOG.md` | Historial de versiones — actualizar antes de cada release |
| `TECH_DEBT.md` | Deuda técnica anotada (crear si no existe) | | `DEPLOY.md` | Guía de despliegue en VPS: Docker, Nginx, SSL, DNS |
| `data/snapshot.json` | Estado persistido del poller | | `OBSERVATIONS.md` | Hallazgos durante el desarrollo (schema real del PJ, etc.) |
| `TECH_DEBT.md` | Deuda técnica anotada |
| `migrations/` | Archivos `.sql` de schema — nunca editar los ya aplicados |
| `data/pasante.db` | Base de datos SQLite (tenants, audit_log, invite_codes) |
| `data/snapshot_*.json` | Estado del diff de notificaciones por tenant |
---
## Gestión de versiones
### Esquema: Semantic Versioning (semver 2.0)
`MAJOR.MINOR.PATCH` — tags git con prefijo `v` (ej: `v0.3.1`).
| Número | Incrementar cuando... | Ejemplos |
|--------|----------------------|---------|
| `MAJOR` | Cambio incompatible con versiones anteriores: migración de DB no retrocompatible, cambio de protocolo del PJ, rediseño de arquitectura | `v1.0.0` (migración a Postgres) |
| `MINOR` | Feature nuevo sin romper lo existente: comando nuevo, integración nueva, mejora visible al usuario | `/recordatorio`, modo silencioso |
| `PATCH` | Bug fix, security patch, mejora interna, documentación operacional | corrección de campo del PJ, hardening |
### Reglas que Claude Code sigue siempre
1. **Cada release requiere entrada en `CHANGELOG.md`** con fecha y secciones
`Added / Changed / Fixed / Security` según corresponda.
2. **El tag se crea después del commit de changelog**, no antes.
3. **`[Unreleased]`** acumula todos los cambios hasta que Marcos decide liberar.
4. **Claude propone el número de versión** basándose en las reglas de arriba;
Marcos lo aprueba antes de tagear.
5. **`git tag` y `git push --tags` los ejecuta Claude solo cuando Marcos lo confirma** —
son acciones externas (modifican el remoto) que requieren aprobación explícita.
6. **Tags retroactivos** son válidos para dar historia a commits anteriores;
se crean con `git tag -a vX.Y.Z <hash>`.
### Proceso de release (copy-paste)
```bash
# 1. CHANGELOG.md ya actualizado (mover [Unreleased] → [X.Y.Z] con fecha)
git add CHANGELOG.md
git commit -m "chore: release vX.Y.Z"
# 2. Tag anotado
git tag -a vX.Y.Z -m "vX.Y.Z — resumen en una línea"
# 3. Push incluyendo el tag
git push origin main --tags
```
--- ---

View File

@@ -429,6 +429,9 @@ WEBHOOK_SECRET= # ← dejar vacío por ahora, completar en el pas
# ── HEALTH CHECK ───────────────────────────────────────────────────────────── # ── HEALTH CHECK ─────────────────────────────────────────────────────────────
HEALTH_PORT=8080 # dejar así HEALTH_PORT=8080 # dejar así
HEALTH_ALERT_HOURS=2 # alerta al admin si el poller no actualizó en 2 horas HEALTH_ALERT_HOURS=2 # alerta al admin si el poller no actualizó en 2 horas
# Token para /health (detallado) y /health/history — generar con:
# python3 -c "import secrets; print(secrets.token_hex(32))"
HEALTH_AUTH_TOKEN= # ← COMPLETAR (recomendado)
# ── POLLING ────────────────────────────────────────────────────────────────── # ── POLLING ──────────────────────────────────────────────────────────────────
POLL_INTERVAL_MINUTES=60 # revisar el PJ cada 60 minutos POLL_INTERVAL_MINUTES=60 # revisar el PJ cada 60 minutos
@@ -577,6 +580,14 @@ Pegar **exactamente** este contenido (reemplazá el nombre de dominio si usás u
# /etc/nginx/sites-available/pedrito # /etc/nginx/sites-available/pedrito
# Reverse proxy para Pedrito Hechakuaa # Reverse proxy para Pedrito Hechakuaa
# ── Rate limiting ─────────────────────────────────────────────────────────────
# Limita requests por IP para mitigar escaneo automatizado y ataques de fuerza bruta.
# La zona se define fuera del bloque server (en el contexto http).
# Si tenés otros sites en este Nginx, mover estas líneas a /etc/nginx/nginx.conf
# dentro del bloque http { }.
limit_req_zone $binary_remote_addr zone=pedrito_webhook:10m rate=30r/s;
limit_req_zone $binary_remote_addr zone=pedrito_health:10m rate=10r/m;
server { server {
listen 80; listen 80;
server_name pedrito.inqualityhq.com; server_name pedrito.inqualityhq.com;
@@ -594,14 +605,19 @@ server {
include /etc/letsencrypt/options-ssl-nginx.conf; include /etc/letsencrypt/options-ssl-nginx.conf;
ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem;
# Seguridad: no exponer versión de Nginx # Seguridad: no exponer versión de Nginx ni headers informativos
server_tokens off; server_tokens off;
add_header X-Content-Type-Options nosniff;
add_header X-Frame-Options DENY;
# Tamaño máximo de payload (Telegram puede enviar archivos) # Tamaño máximo de payload (Telegram puede enviar archivos)
client_max_body_size 20M; client_max_body_size 20M;
# ── Health check (acceso público para monitoreo) ────────────────────────── # ── Health check ──────────────────────────────────────────────────────────
# /health es público (respuesta mínima sin datos sensibles)
# /health/history requiere Bearer token — el bot lo rechaza sin él
location /health { location /health {
limit_req zone=pedrito_health burst=5 nodelay;
proxy_pass http://127.0.0.1:8080/health; proxy_pass http://127.0.0.1:8080/health;
proxy_set_header Host $host; proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
@@ -609,25 +625,31 @@ server {
} }
location /health/history { location /health/history {
limit_req zone=pedrito_health burst=3 nodelay;
proxy_pass http://127.0.0.1:8080/health/history; proxy_pass http://127.0.0.1:8080/health/history;
proxy_set_header Host $host; proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
# Pasar Authorization header para que el bot pueda validar el Bearer token
proxy_set_header Authorization $http_authorization;
proxy_read_timeout 10s; proxy_read_timeout 10s;
} }
# ── Webhook de Telegram ─────────────────────────────────────────────────── # ── Webhook de Telegram ───────────────────────────────────────────────────
# Solo Telegram envía aquí — los IPs de Telegram son bien conocidos (149.154.x, 91.108.x)
# El bot valida el X-Telegram-Bot-Api-Secret-Token internamente
location /webhook { location /webhook {
limit_req zone=pedrito_webhook burst=50 nodelay;
proxy_pass http://127.0.0.1:8080/webhook; proxy_pass http://127.0.0.1:8080/webhook;
proxy_set_header Host $host; proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme; proxy_set_header X-Forwarded-Proto $scheme;
# Pasar el header de autenticación del webhook de Telegram
proxy_set_header X-Telegram-Bot-Api-Secret-Token $http_x_telegram_bot_api_secret_token; proxy_set_header X-Telegram-Bot-Api-Secret-Token $http_x_telegram_bot_api_secret_token;
proxy_read_timeout 30s; proxy_read_timeout 30s;
} }
# ── Denegar cualquier otra ruta ─────────────────────────────────────────── # ── Denegar cualquier otra ruta ───────────────────────────────────────────
# Respuesta genérica — no revelar que hay un bot ni su propósito
location / { location / {
return 404; return 404;
} }

View File

@@ -57,6 +57,10 @@ class Settings(BaseSettings):
# Código de invitación para registro (vacío = registro abierto) # Código de invitación para registro (vacío = registro abierto)
invite_code: str = "" invite_code: str = ""
# Token para los endpoints HTTP protegidos (/health detallado, /health/history)
# Sin auth configurada los endpoints devuelven respuesta mínima (status only)
health_auth_token: str = ""
# Webhook (vacío = long-polling) # Webhook (vacío = long-polling)
webhook_url: str = "" # ej: https://bot.midominio.com webhook_url: str = "" # ej: https://bot.midominio.com
webhook_secret: str = "" # token secreto para X-Telegram-Bot-Api-Secret-Token webhook_secret: str = "" # token secreto para X-Telegram-Bot-Api-Secret-Token

View File

@@ -162,23 +162,20 @@ async def crear_invite_code(codigo: str) -> None:
async def verificar_y_consumir_invite_code(codigo: str, chat_id: int) -> bool: async def verificar_y_consumir_invite_code(codigo: str, chat_id: int) -> bool:
""" """
Verifica que el código existe y no fue usado, lo marca como consumido y Verifica y consume un código de invitación en una sola operación atómica.
retorna True. Retorna False si no existe o ya fue usado.
Operación atómica en una sola transacción SQLite. Usa UPDATE con condición AND usado=0 en vez de SELECT+UPDATE para eliminar la
race condition TOCTOU: dos requests concurrentes con el mismo código solo
pueden actualizar la fila una vez — el segundo UPDATE afecta 0 rows y retorna False.
""" """
conn = _assert_conn() conn = _assert_conn()
async with conn.execute( cur = await conn.execute(
"SELECT usado FROM invite_codes WHERE codigo = ?", (codigo,) "UPDATE invite_codes SET usado = 1, usado_por = ?, usado_at = ? "
) as cur: "WHERE codigo = ? AND usado = 0",
row = await cur.fetchone()
if row is None or row["usado"]:
return False
await conn.execute(
"UPDATE invite_codes SET usado = 1, usado_por = ?, usado_at = ? WHERE codigo = ?",
(chat_id, _now_iso(), codigo), (chat_id, _now_iso(), codigo),
) )
await conn.commit() await conn.commit()
if cur.rowcount > 0:
logger.info("invite_code_consumido", chat_id=chat_id) logger.info("invite_code_consumido", chat_id=chat_id)
return True return True
return False

119
health.py
View File

@@ -2,20 +2,31 @@
health.py — Servidor HTTP para health check y webhook de Telegram. health.py — Servidor HTTP para health check y webhook de Telegram.
Rutas: Rutas:
GET /health → JSON con estado actual del bot GET /health → estado mínimo (público) | estado completo (con Bearer token)
GET /health/history → últimos N eventos del audit_log (?limit=50) GET /health/history → últimos N eventos del audit_log — REQUIERE Bearer token
POST /webhook → recibe updates de Telegram (solo en modo webhook) POST /webhook → recibe updates de Telegram (solo en modo webhook)
Modo webhook opt-in: solo se activa si settings.webhook_url está configurado. Seguridad:
En modo long-polling, /webhook no se registra. - /health sin auth: solo devuelve {"status": "ok"|"degraded", "timestamp": "..."}
Sin conteos de tenants ni timing que revele inteligencia operacional.
- /health con auth: respuesta completa con métricas internas.
- /health/history siempre requiere auth.
- chat_ids en el historial se enmascaran como hashes opacos (no reversibles desde HTTP).
- Webhook valida X-Telegram-Bot-Api-Secret-Token con compare_digest.
Configurar en .env:
HEALTH_AUTH_TOKEN=<token largo aleatorio>
(generar con: python -c "import secrets; print(secrets.token_hex(32))")
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import hashlib
import json import json
import secrets
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import asyncio
import aiohttp.web import aiohttp.web
import structlog import structlog
from telegram import Update from telegram import Update
@@ -29,7 +40,7 @@ logger = structlog.get_logger(__name__)
# ── Estado global del poller ────────────────────────────────────────────────── # ── Estado global del poller ──────────────────────────────────────────────────
_last_poll_at: datetime | None = None _last_poll_at: datetime | None = None
_last_poll_tenant_count: int = 0 _last_poll_tenant_count: int = 0
_alert_sent: bool = False # evitar spam de alertas consecutivas _alert_sent: bool = False
def record_poll(tenant_count: int) -> None: def record_poll(tenant_count: int) -> None:
@@ -40,6 +51,17 @@ def record_poll(tenant_count: int) -> None:
_alert_sent = False _alert_sent = False
def _mask_chat_id(chat_id: int | None) -> str | None:
"""
Convierte un chat_id en un identificador opaco no reversible.
Preserva la capacidad de correlacionar eventos del mismo usuario
sin exponer el Telegram ID real.
"""
if chat_id is None:
return None
return "usr_" + hashlib.sha256(str(chat_id).encode()).hexdigest()[:8]
# ── Servidor ────────────────────────────────────────────────────────────────── # ── Servidor ──────────────────────────────────────────────────────────────────
class HealthServer: class HealthServer:
@@ -61,7 +83,11 @@ class HealthServer:
await self._runner.setup() await self._runner.setup()
site = aiohttp.web.TCPSite(self._runner, "0.0.0.0", self._settings.health_port) site = aiohttp.web.TCPSite(self._runner, "0.0.0.0", self._settings.health_port)
await site.start() await site.start()
logger.info("health_server_iniciado", port=self._settings.health_port) logger.info(
"health_server_iniciado",
port=self._settings.health_port,
auth_configurada=bool(self._settings.health_auth_token),
)
self._watcher_task = asyncio.create_task(self._watcher_loop()) self._watcher_task = asyncio.create_task(self._watcher_loop())
@@ -76,37 +102,60 @@ class HealthServer:
await self._runner.cleanup() await self._runner.cleanup()
logger.info("health_server_detenido") logger.info("health_server_detenido")
# ── Auth ──────────────────────────────────────────────────────────────────
def _is_authenticated(self, request: aiohttp.web.Request) -> bool:
"""
Valida el header Authorization: Bearer <token>.
Si health_auth_token no está configurado, siempre retorna False
(los endpoints protegidos devuelven respuesta mínima, no error).
"""
if not self._settings.health_auth_token:
return False
auth = request.headers.get("Authorization", "")
if not auth.startswith("Bearer "):
return False
provided = auth[7:]
# compare_digest previene timing attacks sobre el token de admin
return secrets.compare_digest(provided, self._settings.health_auth_token)
# ── Handlers ───────────────────────────────────────────────────────────── # ── Handlers ─────────────────────────────────────────────────────────────
async def _handle_health(self, request: aiohttp.web.Request) -> aiohttp.web.Response: async def _handle_health(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
now = datetime.now(timezone.utc)
stale = False
if _last_poll_at:
stale = (now - _last_poll_at) > timedelta(hours=self._settings.health_alert_hours)
# Respuesta pública mínima — no revela conteos ni timing operacional
data: dict = {
"status": "degraded" if stale else "ok",
"timestamp": now.isoformat(),
}
# Respuesta extendida solo con auth válida
if self._is_authenticated(request):
from database import get_conn from database import get_conn
conn = get_conn() conn = get_conn()
async with conn.execute( async with conn.execute(
"SELECT COUNT(*) as n FROM tenants WHERE estado = 'activo'" "SELECT COUNT(*) as n FROM tenants WHERE estado = 'activo'"
) as cur: ) as cur:
row = await cur.fetchone() row = await cur.fetchone()
tenant_count = row["n"] if row else 0 tenant_count = row["n"] if row else 0
now = datetime.now(timezone.utc)
seconds_since_poll: int | None = None seconds_since_poll: int | None = None
stale = False
if _last_poll_at: if _last_poll_at:
delta = now - _last_poll_at seconds_since_poll = int((now - _last_poll_at).total_seconds())
seconds_since_poll = int(delta.total_seconds())
stale = delta > timedelta(hours=self._settings.health_alert_hours)
data = { data.update({
"status": "degraded" if stale else "ok",
"timestamp": now.isoformat(),
"tenants_activos": tenant_count, "tenants_activos": tenant_count,
"last_poll": _last_poll_at.isoformat() if _last_poll_at else None, "last_poll": _last_poll_at.isoformat() if _last_poll_at else None,
"last_poll_tenants": _last_poll_tenant_count, "last_poll_tenants": _last_poll_tenant_count,
"seconds_since_last_poll": seconds_since_poll, "seconds_since_last_poll": seconds_since_poll,
"poll_interval_minutes": self._settings.poll_interval_minutes, "poll_interval_minutes": self._settings.poll_interval_minutes,
"webhook_mode": bool(self._settings.webhook_url), "webhook_mode": bool(self._settings.webhook_url),
} })
return aiohttp.web.Response( return aiohttp.web.Response(
text=json.dumps(data, indent=2), text=json.dumps(data, indent=2),
content_type="application/json", content_type="application/json",
@@ -114,6 +163,20 @@ class HealthServer:
) )
async def _handle_history(self, request: aiohttp.web.Request) -> aiohttp.web.Response: async def _handle_history(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
# Este endpoint siempre requiere autenticación — contiene PII (chat_ids)
if not self._is_authenticated(request):
logger.warning(
"health_history_acceso_denegado",
ip=request.remote,
ua=request.headers.get("User-Agent", "")[:80],
)
return aiohttp.web.Response(
status=401,
text=json.dumps({"error": "Authentication required"}),
content_type="application/json",
headers={"WWW-Authenticate": 'Bearer realm="pedrito-health"'},
)
from database import get_conn from database import get_conn
conn = get_conn() conn = get_conn()
@@ -129,16 +192,26 @@ class HealthServer:
) as cur: ) as cur:
rows = await cur.fetchall() rows = await cur.fetchall()
# Enmascarar chat_ids: el consumidor puede correlacionar eventos del mismo usuario
# pero no puede determinar la identidad real sin acceso a la DB.
records = []
for r in rows:
row_dict = dict(r)
row_dict["chat_id"] = _mask_chat_id(row_dict.get("chat_id"))
records.append(row_dict)
return aiohttp.web.Response( return aiohttp.web.Response(
text=json.dumps([dict(r) for r in rows], indent=2), text=json.dumps(records, indent=2),
content_type="application/json", content_type="application/json",
) )
async def _handle_webhook(self, request: aiohttp.web.Request) -> aiohttp.web.Response: async def _handle_webhook(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") provided = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
if self._settings.webhook_secret and secret != self._settings.webhook_secret: if self._settings.webhook_secret:
if not secrets.compare_digest(provided, self._settings.webhook_secret):
logger.warning("webhook_token_invalido", ip=request.remote) logger.warning("webhook_token_invalido", ip=request.remote)
return aiohttp.web.Response(status=403, text="Forbidden") # Respuesta genérica — no revelar si es auth vs otro error
return aiohttp.web.Response(status=404, text="Not Found")
try: try:
body = await request.json() body = await request.json()
@@ -155,7 +228,7 @@ class HealthServer:
"""Alerta al admin si el poller lleva más de health_alert_hours sin actualizar.""" """Alerta al admin si el poller lleva más de health_alert_hours sin actualizar."""
global _alert_sent global _alert_sent
while True: while True:
await asyncio.sleep(300) # revisar cada 5 minutos await asyncio.sleep(300)
if _last_poll_at is None or _alert_sent: if _last_poll_at is None or _alert_sent:
continue continue
delta = datetime.now(timezone.utc) - _last_poll_at delta = datetime.now(timezone.utc) - _last_poll_at

27
main.py
View File

@@ -7,6 +7,7 @@ Al arrancar: aplica migraciones y migra el tenant principal del .env si no exist
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import os
import signal import signal
import sys import sys
@@ -71,8 +72,34 @@ async def _migrar_tenant_principal(settings) -> None:
logger.info("tenant_principal_migrado", chat_id=chat_id) logger.info("tenant_principal_migrado", chat_id=chat_id)
def _limpiar_env_sensibles() -> None:
"""
Elimina variables sensibles del entorno del proceso después de cargar Settings.
Protege contra:
- cat /proc/<pid>/environ (cualquier usuario con acceso al PID)
- herencia accidental de env vars por subprocesos
- stack traces que vuelquen os.environ
NO protege contra 'docker inspect' (que lee la config del contenedor, no el env live).
Para eso usar Docker secrets con Swarm mode.
"""
_sensibles = (
"FERNET_KEY",
"TELEGRAM_BOT_TOKEN",
"PJ_USUARIO_ENC",
"PJ_PASSWORD_ENC",
"WEBHOOK_SECRET",
"HEALTH_AUTH_TOKEN",
"INVITE_CODE",
)
for var in _sensibles:
os.environ.pop(var, None)
async def main() -> None: async def main() -> None:
settings = get_settings() settings = get_settings()
_limpiar_env_sensibles()
logger.info( logger.info(
"pedrito_hechakuaa_iniciando", "pedrito_hechakuaa_iniciando",

View File

@@ -16,6 +16,9 @@ Estados:
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import secrets
import time
from pathlib import Path from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
@@ -201,13 +204,28 @@ class OnboardingFlow:
async def _paso_codigo(self, chat_id: int, codigo: str, update: Update) -> None: async def _paso_codigo(self, chat_id: int, codigo: str, update: Update) -> None:
codigo = codigo.strip() codigo = codigo.strip()
# Registrar tiempo de inicio para garantizar respuesta de duración constante.
# Objetivo: ocultar la diferencia de tiempo entre "código no existe en DB"
# (~0.001ms) y "código existe pero ya fue usado" (~3-5ms de DB lookup).
t0 = time.monotonic()
# Primero: verificar contra la tabla de códigos de un solo uso (DB) # Primero: verificar contra la tabla de códigos de un solo uso (DB)
valido = await verificar_y_consumir_invite_code(codigo, chat_id) valido = await verificar_y_consumir_invite_code(codigo, chat_id)
# Fallback: código global del .env (ilimitado, para uso interno/pruebas) # Fallback: código global del .env (ilimitado, para uso interno/pruebas)
# secrets.compare_digest previene timing attacks sobre el valor del .env
if not valido and self._settings.invite_code: if not valido and self._settings.invite_code:
valido = (codigo == self._settings.invite_code) valido = secrets.compare_digest(
codigo.encode("utf-8"),
self._settings.invite_code.encode("utf-8"),
)
# Garantizar un tiempo mínimo de 800ms independientemente del resultado.
# Esto hace que medir tiempos de respuesta no revele información sobre los códigos.
elapsed = time.monotonic() - t0
_MIN_RESP_SECS = 0.8
if elapsed < _MIN_RESP_SECS:
await asyncio.sleep(_MIN_RESP_SECS - elapsed)
if valido: if valido:
await audit.log("onboarding_codigo_correcto", chat_id=chat_id) await audit.log("onboarding_codigo_correcto", chat_id=chat_id)

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())

View File

@@ -258,8 +258,10 @@ class PedritoBot:
return None return None
async def _rechazar_no_registrado(self, update: Update) -> None: async def _rechazar_no_registrado(self, update: Update) -> None:
# Mensaje mínimo: no revela qué hace el bot ni su arquitectura interna.
# Solo dice que hay un comando /start, lo cual es estándar en todos los bots de Telegram.
await update.effective_message.reply_text( # type: ignore[union-attr] await update.effective_message.reply_text( # type: ignore[union-attr]
"Primero tenés que registrarte\\. Escribí /start\\.", "Para usar este bot escribí /start\\.",
parse_mode=ParseMode.MARKDOWN_V2, parse_mode=ParseMode.MARKDOWN_V2,
) )
@@ -635,10 +637,9 @@ class PedritoBot:
tenant = await get_tenant(chat_id) tenant = await get_tenant(chat_id)
if tenant and tenant.get("estado") == "activo": if tenant and tenant.get("estado") == "activo":
texto = _escape_md(_texto_tono("texto_no_reconocido", tenant)) texto = _escape_md(_texto_tono("texto_no_reconocido", tenant))
else:
texto = _escape_md("Solo respondo comandos específicos. Escribí /ayuda para ver la lista.")
await update.effective_message.reply_text(texto, parse_mode=ParseMode.MARKDOWN_V2) # type: ignore[union-attr] await update.effective_message.reply_text(texto, parse_mode=ParseMode.MARKDOWN_V2) # type: ignore[union-attr]
# Usuarios sin registro: silencio total.
# No confirmar que el bot existe ni qué hace — quien lo debe usar ya sabe cómo.
# ────────────────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────────────────
# Handlers de callbacks # Handlers de callbacks

169
vault.py
View File

@@ -1,55 +1,182 @@
""" """
vault.py — Cifrado de credenciales por tenant. vault.py — Cifrado de credenciales por tenant.
Cada tenant tiene sus credenciales cifradas con una clave derivada de: ## Arquitectura v3: Envelope Encryption con PBKDF2 + DEK por tenant
FERNET_KEY (global, del .env) + str(chat_id) (único por tenant)
Esto asegura que las credenciales de un tenant no se pueden descifrar Cada credencial se cifra en dos capas independientes:
con solo la FERNET_KEY — hace falta también el chat_id.
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 from __future__ import annotations
import base64 import base64
import hashlib import hashlib
import json import json
import os
from functools import lru_cache
import structlog import structlog
from cryptography.fernet import Fernet, InvalidToken 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 from utils.sanitize import scrub
logger = structlog.get_logger(__name__) 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: 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) Cifra credenciales con envelope encryption v3.
f = Fernet(key) Retorna JSON compacto listo para guardar en DB.
payload = json.dumps({"usuario": usuario, "password": password}).encode() """
encrypted = f.encrypt(payload).decode() # Capa 1: generar salt (para PBKDF2) y DEK aleatoria
logger.info("credenciales_cifradas", chat_id=chat_id) salt = os.urandom(_SALT_BYTES)
return encrypted 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]: def descifrar_credenciales(chat_id: int, credentials_enc: str, fernet_key: str) -> tuple[str, str]:
""" """
Retorna (usuario, password). Descifra credenciales. Soporta v1 (legacy) y v3 (actual).
Levanta ValueError si no se puede descifrar (clave incorrecta o datos corruptos). 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) f = Fernet(key)
try: try:
payload = json.loads(f.decrypt(credentials_enc.encode()).decode()) 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"] return payload["usuario"], payload["password"]
except (InvalidToken, KeyError, ValueError) as e: 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( 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 ) from e