260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""
|
|
health.py — Servidor HTTP para health check y webhook de Telegram.
|
|
|
|
Rutas:
|
|
GET /health → estado mínimo (público) | estado completo (con Bearer token)
|
|
GET /health/history → últimos N eventos del audit_log — REQUIERE Bearer token
|
|
POST /webhook → recibe updates de Telegram (solo en modo webhook)
|
|
|
|
Seguridad:
|
|
- /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
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
_PY_TZ = ZoneInfo("America/Asuncion")
|
|
from typing import TYPE_CHECKING
|
|
|
|
import aiohttp.web
|
|
import structlog
|
|
from telegram import Update
|
|
|
|
if TYPE_CHECKING:
|
|
from config import Settings
|
|
from telegram.ext import Application
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
# ── Estado global del poller ──────────────────────────────────────────────────
|
|
_last_poll_at: datetime | None = None
|
|
_last_poll_tenant_count: int = 0
|
|
_alert_sent: bool = False
|
|
|
|
|
|
def record_poll(tenant_count: int) -> None:
|
|
"""Llamar desde poller.py después de cada ronda automática."""
|
|
global _last_poll_at, _last_poll_tenant_count, _alert_sent
|
|
_last_poll_at = datetime.now(timezone.utc)
|
|
_last_poll_tenant_count = tenant_count
|
|
_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 ──────────────────────────────────────────────────────────────────
|
|
|
|
class HealthServer:
|
|
def __init__(self, settings: Settings, application: Application) -> None:
|
|
self._settings = settings
|
|
self._app_ptb = application
|
|
self._runner: aiohttp.web.AppRunner | None = None
|
|
self._watcher_task: asyncio.Task | None = None
|
|
|
|
async def start(self) -> None:
|
|
app = aiohttp.web.Application()
|
|
app.router.add_get("/health", self._handle_health)
|
|
app.router.add_get("/health/history", self._handle_history)
|
|
if self._settings.webhook_url:
|
|
app.router.add_post("/webhook", self._handle_webhook)
|
|
logger.info("webhook_endpoint_activo", url=f"{self._settings.webhook_url}/webhook")
|
|
|
|
self._runner = aiohttp.web.AppRunner(app)
|
|
await self._runner.setup()
|
|
site = aiohttp.web.TCPSite(self._runner, "0.0.0.0", self._settings.health_port)
|
|
await site.start()
|
|
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())
|
|
|
|
async def stop(self) -> None:
|
|
if self._watcher_task:
|
|
self._watcher_task.cancel()
|
|
try:
|
|
await self._watcher_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
if self._runner:
|
|
await self._runner.cleanup()
|
|
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 ─────────────────────────────────────────────────────────────
|
|
|
|
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
|
|
conn = get_conn()
|
|
async with conn.execute(
|
|
"SELECT COUNT(*) as n FROM tenants WHERE estado = 'activo'"
|
|
) as cur:
|
|
row = await cur.fetchone()
|
|
tenant_count = row["n"] if row else 0
|
|
|
|
seconds_since_poll: int | None = None
|
|
if _last_poll_at:
|
|
seconds_since_poll = int((now - _last_poll_at).total_seconds())
|
|
|
|
data.update({
|
|
"tenants_activos": tenant_count,
|
|
"last_poll": _last_poll_at.isoformat() if _last_poll_at else None,
|
|
"last_poll_tenants": _last_poll_tenant_count,
|
|
"seconds_since_last_poll": seconds_since_poll,
|
|
"poll_interval_minutes": self._settings.poll_interval_minutes,
|
|
"webhook_mode": bool(self._settings.webhook_url),
|
|
})
|
|
|
|
return aiohttp.web.Response(
|
|
text=json.dumps(data, indent=2),
|
|
content_type="application/json",
|
|
status=503 if stale else 200,
|
|
)
|
|
|
|
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
|
|
conn = get_conn()
|
|
|
|
try:
|
|
limit = min(int(request.rel_url.query.get("limit", "50")), 200)
|
|
except (ValueError, TypeError):
|
|
limit = 50
|
|
|
|
async with conn.execute(
|
|
"SELECT timestamp, timestamp_local, chat_id, evento, resultado, duracion_ms "
|
|
"FROM audit_log ORDER BY id DESC LIMIT ?",
|
|
(limit,),
|
|
) as cur:
|
|
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(
|
|
text=json.dumps(records, indent=2),
|
|
content_type="application/json",
|
|
)
|
|
|
|
async def _handle_webhook(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
|
|
provided = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "")
|
|
if self._settings.webhook_secret:
|
|
if not secrets.compare_digest(provided, self._settings.webhook_secret):
|
|
logger.warning("webhook_token_invalido", ip=request.remote)
|
|
# Respuesta genérica — no revelar si es auth vs otro error
|
|
return aiohttp.web.Response(status=404, text="Not Found")
|
|
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return aiohttp.web.Response(status=400, text="Bad Request")
|
|
|
|
update = Update.de_json(data=body, bot=self._app_ptb.bot)
|
|
await self._app_ptb.update_queue.put(update)
|
|
return aiohttp.web.Response(status=200, text="OK")
|
|
|
|
# ── Watcher ───────────────────────────────────────────────────────────────
|
|
|
|
async def _watcher_loop(self) -> None:
|
|
"""Alerta al admin si el poller lleva más de health_alert_hours sin actualizar."""
|
|
global _alert_sent
|
|
while True:
|
|
await asyncio.sleep(300)
|
|
if _last_poll_at is None or _alert_sent:
|
|
continue
|
|
ahora_py = datetime.now(_PY_TZ)
|
|
en_horario = (
|
|
ahora_py.weekday() < 5
|
|
and self._settings.poll_hora_inicio <= ahora_py.hour < self._settings.poll_hora_fin
|
|
)
|
|
if not en_horario:
|
|
continue
|
|
delta = datetime.now(timezone.utc) - _last_poll_at
|
|
if delta <= timedelta(hours=self._settings.health_alert_hours):
|
|
continue
|
|
minutos = int(delta.total_seconds() // 60)
|
|
try:
|
|
await self._app_ptb.bot.send_message(
|
|
chat_id=self._settings.telegram_chat_id,
|
|
text=(
|
|
f"⚠️ El poller no actualizó hace {minutos} minutos.\n"
|
|
"Puede haber un problema con el bot o la conexión al PJ."
|
|
),
|
|
)
|
|
_alert_sent = True
|
|
logger.warning("health_alerta_enviada", minutos=minutos)
|
|
except Exception as e:
|
|
logger.error("health_alerta_error", error=str(e))
|