marcos/hechakuaa_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
hechakuaa_bot/onboarding.py
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

549 lines
24 KiB
Python
Raw Permalink Blame History

"""
onboarding.py — Máquina de estados del flujo de registro.
Estado en memoria por chat_id. Si el bot se reinicia, el usuario hace /start de nuevo
(aceptable en este sprint).
Estados:
INICIO → mostró bienvenida con botones de consentimiento
ESPERANDO_NOMBRE → preguntó nombre, esperando texto libre
ESPERANDO_TONO → mostró botones de tono, esperando selección
ESPERANDO_TZ → mostró botones de timezone, esperando selección
ESPERANDO_TZ_CUSTOM → eligió "Otra", esperando texto con nombre de zona
ESPERANDO_USUARIO → preguntó cédula del PJ, esperando texto
ESPERANDO_PASSWORD → preguntó password del PJ, esperando texto (borrar tras procesar)
ESPERANDO_SALIR → mostró confirmación de /salir, esperando botón
"""
from __future__ import annotations
import asyncio
import secrets
import time
from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import structlog
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
import audit
import vault as vault_mod
from config import Settings
from csj_client import CSJAuthError, CSJClient, CSJConnectionError, CSJPasswordTemporalError
from database import delete_tenant, get_tenant, save_tenant, verificar_y_consumir_invite_code
logger = structlog.get_logger(__name__)
# Estados (strings internos)
_ESPERANDO_CODIGO = "codigo"
_INICIO = "inicio"
_ESPERANDO_NOMBRE = "nombre"
_ESPERANDO_TONO = "tono"
_ESPERANDO_TZ = "tz"
_ESPERANDO_TZ_CUSTOM = "tz_custom"
_ESPERANDO_USUARIO = "usuario"
_ESPERANDO_PASSWORD = "password"
_ESPERANDO_SALIR = "salir"
_MAX_INTENTOS = 3
_MAX_INTENTOS_CODIGO = 3
_TZ_MAP = {
"py": "America/Asuncion",
"ar": "America/Argentina/Buenos_Aires",
"uy": "America/Montevideo",
}
def _esc(text: str) -> str:
"""Escapa MarkdownV2 para texto plano dentro de mensajes de onboarding."""
text = text.replace("\\", "\\\\")
for ch in "_*[]()~`>#+-=|{}.!":
text = text.replace(ch, f"\\{ch}")
return text
class OnboardingFlow:
"""Máquina de estados del flujo de registro multi-tenant."""
def __init__(self, settings: Settings) -> None:
self._settings = settings
self._states: dict[int, str] = {}
self._temp: dict[int, dict] = {} # datos parciales por chat_id
self._attempts: dict[int, int] = {} # intentos de credenciales fallidos
# ──────────────────────────────────────────────────────────────────────────
# Helpers de estado
# ──────────────────────────────────────────────────────────────────────────
def is_in_onboarding(self, chat_id: int) -> bool:
return chat_id in self._states
def _clear(self, chat_id: int) -> None:
self._states.pop(chat_id, None)
self._temp.pop(chat_id, None)
self._attempts.pop(chat_id, None)
# ──────────────────────────────────────────────────────────────────────────
# Puntos de entrada desde telegram_bot.py
# ──────────────────────────────────────────────────────────────────────────
async def iniciar_registro(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Llamado desde cmd_start cuando el usuario no está registrado."""
chat_id = update.effective_chat.id # type: ignore[union-attr]
self._temp[chat_id] = {}
self._attempts[chat_id] = 0
await audit.log("onboarding_inicio", chat_id=chat_id)
if self._settings.invite_code:
self._states[chat_id] = _ESPERANDO_CODIGO
await update.effective_message.reply_text( # type: ignore[union-attr]
"Hola\\! Para registrarte en este bot necesitás un código de acceso\\.\n\n"
"Escribí el código que te compartió el administrador:",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
await self._mostrar_consentimiento(chat_id, update)
async def _mostrar_consentimiento(self, chat_id: int, update: Update) -> None:
self._states[chat_id] = _INICIO
texto = (
"Hola\\! Soy Pedrito, el asistente judicial del estudio\\.\n\n"
"Antes de empezar, necesito que sepas:\n"
"• Voy a consultar el sistema del Poder Judicial en tu nombre\n"
"• Guardo tus credenciales cifradas — solo yo las puedo leer\n"
"• Solo hago consultas\\. Nunca acuso notificaciones por vos\\.\n"
"• Podés borrar tus datos cuando quieras con /salir\n\n"
"¿Aceptás estas condiciones?"
)
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("✅ Sí, acepto", callback_data="onb:consent:si"),
InlineKeyboardButton("❌ No, gracias", callback_data="onb:consent:no"),
]
])
await update.effective_message.reply_text( # type: ignore[union-attr]
texto, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=keyboard
)
async def iniciar_salir(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Llamado desde cmd_salir."""
chat_id = update.effective_chat.id # type: ignore[union-attr]
self._states[chat_id] = _ESPERANDO_SALIR
texto = (
"¿Estás seguro que querés borrar tus datos?\n"
"Esto eliminará tus credenciales y preferencias del sistema\\."
)
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("✅ Sí, borrar todo", callback_data="onb:salir:si"),
InlineKeyboardButton("❌ Cancelar", callback_data="onb:salir:no"),
]
])
await update.effective_message.reply_text( # type: ignore[union-attr]
texto, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=keyboard
)
# ──────────────────────────────────────────────────────────────────────────
# Handlers de entrada (llamados desde telegram_bot.py)
# ──────────────────────────────────────────────────────────────────────────
async def handle_mensaje(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Procesa texto libre durante el onboarding."""
chat_id = update.effective_chat.id # type: ignore[union-attr]
estado = self._states.get(chat_id)
if not estado:
return
texto_msg = (update.effective_message.text or "").strip() # type: ignore[union-attr]
if estado == _ESPERANDO_CODIGO:
await self._paso_codigo(chat_id, texto_msg, update)
elif estado == _ESPERANDO_NOMBRE:
await self._paso_nombre(chat_id, texto_msg, update)
elif estado == _ESPERANDO_TZ_CUSTOM:
await self._paso_tz_custom(chat_id, texto_msg, update)
elif estado == _ESPERANDO_USUARIO:
await self._paso_usuario(chat_id, texto_msg, update)
elif estado == _ESPERANDO_PASSWORD:
await self._paso_password(chat_id, texto_msg, update, context)
async def handle_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Procesa botones de onboarding (callback_data = 'onb:accion:valor')."""
query = update.callback_query
await query.answer() # type: ignore[union-attr]
chat_id = update.effective_chat.id # type: ignore[union-attr]
parts = (query.data or "").split(":") # type: ignore[union-attr]
if len(parts) < 3:
return
accion, valor = parts[1], parts[2]
if accion == "consent":
await self._paso_consentimiento(chat_id, valor, update)
elif accion == "tono":
await self._paso_tono(chat_id, valor, update)
elif accion == "tz":
await self._paso_tz(chat_id, valor, update)
elif accion == "reintentar":
await self._pedir_usuario(chat_id, update)
elif accion == "cancelar_onb":
self._clear(chat_id)
await update.effective_message.reply_text( # type: ignore[union-attr]
"Onboarding cancelado\\. Escribí /start cuando quieras intentarlo de nuevo\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
elif accion == "salir":
await self._procesar_salir(chat_id, valor, update, context)
# ──────────────────────────────────────────────────────────────────────────
# Pasos del flujo
# ──────────────────────────────────────────────────────────────────────────
async def _paso_codigo(self, chat_id: int, codigo: str, update: Update) -> None:
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)
valido = await verificar_y_consumir_invite_code(codigo, chat_id)
# 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:
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:
await audit.log("onboarding_codigo_correcto", chat_id=chat_id)
await self._mostrar_consentimiento(chat_id, update)
return
intentos = self._attempts.get(chat_id, 0) + 1
self._attempts[chat_id] = intentos
await audit.log("onboarding_codigo_incorrecto", chat_id=chat_id, resultado="error", detalle={"intento": intentos})
if intentos >= _MAX_INTENTOS_CODIGO:
self._clear(chat_id)
await update.effective_message.reply_text( # type: ignore[union-attr]
"Código incorrecto\\. Contactá al administrador para obtener acceso\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
await update.effective_message.reply_text( # type: ignore[union-attr]
"Código incorrecto\\. Intentá de nuevo:",
parse_mode=ParseMode.MARKDOWN_V2,
)
async def _paso_consentimiento(self, chat_id: int, valor: str, update: Update) -> None:
if valor == "no":
self._clear(chat_id)
await audit.log("onboarding_consentimiento_rechazado", chat_id=chat_id)
await update.effective_message.reply_text( # type: ignore[union-attr]
"Entendido\\. Escribí /start cuando quieras empezar\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
await audit.log("onboarding_consentimiento_aceptado", chat_id=chat_id)
self._states[chat_id] = _ESPERANDO_NOMBRE
await update.effective_message.reply_text( # type: ignore[union-attr]
"¿Cómo querés que te llame?\n\\(Escribí tu nombre o apodo\\)",
parse_mode=ParseMode.MARKDOWN_V2,
)
async def _paso_nombre(self, chat_id: int, nombre: str, update: Update) -> None:
if not nombre:
await update.effective_message.reply_text( # type: ignore[union-attr]
"Por favor escribí tu nombre o apodo\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
nombre = nombre[:50]
self._temp[chat_id]["nombre_preferido"] = nombre
self._states[chat_id] = _ESPERANDO_TONO
await audit.log("onboarding_nombre_registrado", chat_id=chat_id)
nombre_esc = _esc(nombre)
texto = f"¿Cómo preferís que me comunique con vos, {nombre_esc}?"
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("📜 Formal", callback_data="onb:tono:formal"),
InlineKeyboardButton("😊 Cotidiano", callback_data="onb:tono:cotidiano"),
InlineKeyboardButton("🤙 Informal", callback_data="onb:tono:informal"),
]
])
await update.effective_message.reply_text( # type: ignore[union-attr]
texto, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=keyboard
)
async def _paso_tono(self, chat_id: int, tono: str, update: Update) -> None:
if tono not in ("formal", "cotidiano", "informal"):
return
self._temp[chat_id]["tono"] = tono
self._states[chat_id] = _ESPERANDO_TZ
await audit.log("onboarding_tono_seleccionado", chat_id=chat_id, detalle={"tono": tono})
texto = (
"¿Desde dónde vas a usar el bot?\n"
"\\(Para mostrarte las horas correctas\\)"
)
keyboard = InlineKeyboardMarkup([
[InlineKeyboardButton("🇵🇾 Paraguay (Asunción)", callback_data="onb:tz:py")],
[InlineKeyboardButton("🇦🇷 Argentina (Buenos Aires)", callback_data="onb:tz:ar")],
[InlineKeyboardButton("🇺🇾 Uruguay (Montevideo)", callback_data="onb:tz:uy")],
[InlineKeyboardButton("🌐 Otra zona horaria", callback_data="onb:tz:otra")],
])
await update.effective_message.reply_text( # type: ignore[union-attr]
texto, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=keyboard
)
async def _paso_tz(self, chat_id: int, valor: str, update: Update) -> None:
if valor == "otra":
self._states[chat_id] = _ESPERANDO_TZ_CUSTOM
await update.effective_message.reply_text( # type: ignore[union-attr]
"Escribí el nombre de tu zona horaria\\.\n"
"Ejemplos: `America/Lima`, `America/Bogota`, `Europe/Madrid`",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
tz = _TZ_MAP.get(valor, "America/Asuncion")
self._temp[chat_id]["timezone"] = tz
await audit.log("onboarding_timezone_seleccionada", chat_id=chat_id, detalle={"tz": tz})
await self._pedir_usuario(chat_id, update)
async def _paso_tz_custom(self, chat_id: int, tz_str: str, update: Update) -> None:
try:
ZoneInfo(tz_str)
except (ZoneInfoNotFoundError, KeyError):
await update.effective_message.reply_text( # type: ignore[union-attr]
f"No reconozco la zona \"{_esc(tz_str)}\"\\.\n"
"Probá con algo como `America/Lima` o `America/Bogota`\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
self._temp[chat_id]["timezone"] = tz_str
await audit.log("onboarding_timezone_seleccionada", chat_id=chat_id, detalle={"tz": tz_str})
await self._pedir_usuario(chat_id, update)
async def _pedir_usuario(self, chat_id: int, update: Update) -> None:
self._states[chat_id] = _ESPERANDO_USUARIO
await update.effective_message.reply_text( # type: ignore[union-attr]
"Ahora necesito tus credenciales del sistema judicial "
"\\(apps\\.csj\\.gov\\.py\\)\\.\n\n"
"¿Cuál es tu número de cédula de identidad?\n"
"\\(Es el usuario que usás para entrar al sistema\\)",
parse_mode=ParseMode.MARKDOWN_V2,
)
async def _paso_usuario(self, chat_id: int, texto: str, update: Update) -> None:
cedula = texto.strip().replace(".", "").replace("-", "").replace(" ", "")
if not cedula.isdigit():
await update.effective_message.reply_text( # type: ignore[union-attr]
"La cédula debe contener solo dígitos\\. Intentá de nuevo\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
return
self._temp[chat_id]["usuario"] = cedula
self._states[chat_id] = _ESPERANDO_PASSWORD
await update.effective_message.reply_text( # type: ignore[union-attr]
"¿Y tu contraseña del sistema judicial?\n\n"
"⚠️ Voy a borrar este mensaje apenas lo procese\\. "
"No lo reenvíes a nadie\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
async def _paso_password(
self,
chat_id: int,
password: str,
update: Update,
context: ContextTypes.DEFAULT_TYPE,
) -> None:
# Borrar el mensaje del usuario inmediatamente
try:
await context.bot.delete_message(
chat_id=chat_id,
message_id=update.effective_message.message_id, # type: ignore[union-attr]
)
except Exception:
pass # puede fallar si el bot no tiene permisos de borrado en el chat
usuario = self._temp[chat_id].get("usuario", "")
if not usuario:
await context.bot.send_message(
chat_id=chat_id,
text="Error interno\\. Escribí /start para reiniciar\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
self._clear(chat_id)
return
await context.bot.send_message(chat_id=chat_id, text="⏳ Verificando tus credenciales...")
await audit.log("onboarding_credenciales_verificando", chat_id=chat_id)
credentials_enc: str | None = None
error_tipo: str | None = None
try:
async with CSJClient(self._settings, usuario=usuario, password=password, chat_id=chat_id) as client:
await client.login()
# Login exitoso — cifrar antes de borrar password
credentials_enc = vault_mod.cifrar_credenciales(
chat_id=chat_id,
usuario=usuario,
password=password,
fernet_key=self._settings.fernet_key,
)
except CSJPasswordTemporalError:
error_tipo = "temporal"
except CSJAuthError:
error_tipo = "credenciales"
except CSJConnectionError:
error_tipo = "conexion"
finally:
# Limpiar password de memoria lo antes posible
password = "" # noqa: F841
del password
if credentials_enc is not None:
await audit.log("onboarding_credenciales_ok", chat_id=chat_id)
await save_tenant(
chat_id,
nombre_preferido=self._temp[chat_id].get("nombre_preferido", ""),
tono=self._temp[chat_id].get("tono", "cotidiano"),
timezone=self._temp[chat_id].get("timezone", "America/Asuncion"),
credentials_enc=credentials_enc,
estado="activo",
consentimiento_aceptado=1,
)
await self._completado(chat_id, context)
else:
await audit.log("onboarding_credenciales_error", chat_id=chat_id, resultado="error", detalle={"tipo": error_tipo})
await self._error_credenciales(chat_id, error_tipo, context)
async def _completado(self, chat_id: int, context: ContextTypes.DEFAULT_TYPE) -> None:
nombre = self._temp[chat_id].get("nombre_preferido", "")
tz_str = self._temp[chat_id].get("timezone", "America/Asuncion")
ciudad = {
"America/Asuncion": "Paraguay",
"America/Argentina/Buenos_Aires": "Argentina",
"America/Montevideo": "Uruguay",
}.get(tz_str, tz_str)
self._clear(chat_id)
texto = (
f"✅ ¡Todo listo, {_esc(nombre)}\\!\n\n"
f"Estás conectado al sistema del Poder Judicial\\.\n"
f"Voy a revisar tus notificaciones cada hora en horario hábil\n"
f"\\(lunes a viernes, 7:00\\-18:00 hora de {_esc(ciudad)}\\)\\.\n\n"
f"Escribí /notif para revisar ahora mismo\\."
)
await context.bot.send_message(
chat_id=chat_id,
text=texto,
parse_mode=ParseMode.MARKDOWN_V2,
)
logger.info("onboarding_completado", chat_id=chat_id)
await audit.log("onboarding_completado", chat_id=chat_id)
async def _error_credenciales(
self, chat_id: int, error_tipo: str | None, context: ContextTypes.DEFAULT_TYPE
) -> None:
self._attempts[chat_id] = self._attempts.get(chat_id, 0) + 1
if error_tipo == "temporal":
self._clear(chat_id)
await context.bot.send_message(
chat_id=chat_id,
text=(
"⚠️ Tu contraseña del sistema judicial es temporal\\. "
"Cambiala en apps\\.csj\\.gov\\.py y luego escribí /start de nuevo\\."
),
parse_mode=ParseMode.MARKDOWN_V2,
)
return
if self._attempts[chat_id] >= _MAX_INTENTOS:
self._clear(chat_id)
await context.bot.send_message(
chat_id=chat_id,
text=(
"❌ Demasiados intentos fallidos\\. "
"Escribí /start cuando quieras intentarlo de nuevo\\."
),
parse_mode=ParseMode.MARKDOWN_V2,
)
return
if error_tipo == "conexion":
detalle = "El sistema del PJ parece estar temporalmente caído\\."
else:
detalle = "El usuario o la contraseña no son correctos\\."
texto = (
f"❌ No pude conectarme con esas credenciales\\.\n\n"
f"{detalle}\n\n"
"¿Querés intentar de nuevo?"
)
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("🔄 Intentar de nuevo", callback_data="onb:reintentar:si"),
InlineKeyboardButton("❌ Cancelar", callback_data="onb:cancelar_onb:si"),
]
])
await context.bot.send_message(
chat_id=chat_id,
text=texto,
parse_mode=ParseMode.MARKDOWN_V2,
reply_markup=keyboard,
)
async def _procesar_salir(
self,
chat_id: int,
valor: str,
update: Update,
context: ContextTypes.DEFAULT_TYPE,
) -> None:
if valor == "no":
self._clear(chat_id)
await update.effective_message.reply_text("Operación cancelada\\.", parse_mode=ParseMode.MARKDOWN_V2) # type: ignore[union-attr]
return
await delete_tenant(chat_id)
snapshot_path = Path(self._settings.data_dir) / f"snapshot_{chat_id}.json"
if snapshot_path.exists():
snapshot_path.unlink()
self._clear(chat_id)
await update.effective_message.reply_text( # type: ignore[union-attr]
"✅ Tus datos fueron eliminados\\. "
"Escribí /start si querés registrarte de nuevo\\.",
parse_mode=ParseMode.MARKDOWN_V2,
)
logger.info("tenant_dado_de_baja", chat_id=chat_id)
await audit.log("tenant_eliminado", chat_id=chat_id, detalle={"origen": "salir"})
Reference in New Issue View Git Blame Copy Permalink