Generated with: knock-webhooks skill

https://github.com/hookdeck/webhook-skills

─────────────────────────────────────────────────────────────────────────────

NutriFlow — Knock Webhook Handler (FastAPI)

Cliente: NutriFlow SaaS · Stack: Python 3.11 + FastAPI

Generado por CULTIVA IA · https://cultivaia.com

─────────────────────────────────────────────────────────────────────────────

Instalación:

pip install fastapi uvicorn python-dotenv

Variables de entorno (.env):

KNOCK_WEBHOOK_SECRET=whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxx # por endpoint

Arrancar en local con tunnel Hookdeck:

npx hookdeck-cli listen 8000 knock --path /webhooks/knock

uvicorn webhook_handler:app --reload --port 8000

─────────────────────────────────────────────────────────────────────────────

import base64 import hashlib import hmac import json import logging import os import time from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any

from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import JSONResponse

load_dotenv()

── Configuración ─────────────────────────────────────────────────────────────

logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s — %(message)s", ) logger = logging.getLogger("nutriflow.webhooks.knock")

KNOCK_WEBHOOK_SECRET: str = os.environ.get("KNOCK_WEBHOOK_SECRET", "") FIVE_MINUTES_MS: int = 5 * 60 * 1000

── Almacén de idempotencia (en producción → Redis o PostgreSQL) ──────────────

_processed_event_ids: set[str] = set()

app = FastAPI(title="NutriFlow — Knock Webhook Handler", version="1.0.0")

── Modelos de datos ──────────────────────────────────────────────────────────

@dataclass class NotificationEvent: """Registro normalizado de evento de notificación Knock para auditoría."""

event_id: str
event_type: str
message_id: str
channel: str
recipient_id: str
recipient_email: str
workflow: str
occurred_at: datetime
raw_payload: dict[str, Any] = field(default_factory=dict)
bounced: bool = False
bounce_reason: str | None = None

── Verificación de firma ─────────────────────────────────────────────────────

def verify_knock_signature( raw_body: bytes, header: str | None, secret: str, tolerance_ms: int = FIVE_MINUTES_MS, ) -> tuple[bool, str | None]: """ Verifica la cabecera x-knock-signature usando HMAC-SHA256 + timestamp.

Formato de cabecera:  t=<timestamp_ms>,s=<base64_signature>
Contenido firmado:    f"{timestamp_ms}.{raw_body_str}"
Algoritmo:            HMAC-SHA256 con secreto por endpoint, base64.

⚠️  GOTCHA CRÍTICO — Knock usa MILISEGUNDOS, Stripe usa segundos.
     Si portas un verificador de Stripe, cambia `time.time()` por
     `int(time.time() * 1000)` o tendrás fallos silenciosos al comparar
     timestamps (el drift simulado es ×1000, siempre fuera de tolerancia).

Args:
    raw_body:     Cuerpo crudo de la request (bytes, sin parsear).
    header:       Valor de la cabecera x-knock-signature.
    secret:       Secreto de firma del endpoint (del dashboard de Knock).
    tolerance_ms: Ventana de tiempo en ms (por defecto 5 minutos).

Returns:
    (True, None) si la firma es válida.
    (False, "<razón>") si no lo es.
"""
if not header:
    return False, "Falta la cabecera x-knock-signature"

# Parsear t=... y s=...
parts: dict[str, str] = {}
for segment in header.split(","):
    if "=" in segment:
        k, v = segment.split("=", 1)
        parts[k.strip()] = v.strip()

timestamp_ms_str = parts.get("t")
signature = parts.get("s")

if not timestamp_ms_str or not signature:
    return False, "Cabecera x-knock-signature malformada"

try:
    timestamp_ms = int(timestamp_ms_str)
except ValueError:
    return False, "Timestamp no es un entero válido"

# ── Validación de ventana temporal ──
# Knock: timestamp en MILISEGUNDOS (int(time.time() * 1000))
# Stripe: timestamp en segundos    (int(time.time()))
# ← Esta línea corrige el bug habitual al portar desde Stripe:
now_ms = int(time.time() * 1000)
if abs(now_ms - timestamp_ms) > tolerance_ms:
    return False, f"Timestamp fuera de tolerancia ({abs(now_ms - timestamp_ms)} ms)"

# ── Calcular firma esperada ──
signed_content = f"{timestamp_ms_str}.{raw_body.decode('utf-8')}"
expected_bytes = hmac.new(
    secret.encode("utf-8"),
    signed_content.encode("utf-8"),
    hashlib.sha256,
).digest()
expected = base64.b64encode(expected_bytes).decode("utf-8")

# Comparación en tiempo constante para evitar timing attacks
if not hmac.compare_digest(signature, expected):
    return False, "Firma inválida"

return True, None

── Lógica de negocio por evento ──────────────────────────────────────────────

def _extract_event(payload: dict[str, Any]) -> NotificationEvent: """Normaliza el payload crudo de Knock a NotificationEvent.""" data = payload.get("data") or {} recipient = data.get("recipient") or {} channel_data = data.get("channel_data") or {}

return NotificationEvent(
    event_id=payload.get("id", ""),
    event_type=payload.get("type", ""),
    message_id=data.get("id", ""),
    channel=data.get("channel_id", ""),
    recipient_id=recipient.get("id", ""),
    recipient_email=recipient.get("email", ""),
    workflow=data.get("source") or data.get("workflow") or "",
    occurred_at=datetime.now(timezone.utc),
    raw_payload=payload,
)

def handle_message_sent(evt: NotificationEvent) -> None: """Registra el envío en el log de auditoría.""" logger.info( "📤 [SENT] msg=%s | canal=%s | destinatario=%s | workflow=%s", evt.message_id, evt.channel, evt.recipient_email, evt.workflow, ) # En producción → INSERT INTO notification_events (...)

def handle_message_delivered(evt: NotificationEvent) -> None: """Marca el mensaje como entregado en la BD.""" logger.info( "✅ [DELIVERED] msg=%s | destinatario=%s", evt.message_id, evt.recipient_email, ) # En producción → UPDATE notification_events SET delivered_at=NOW()

def handle_message_bounced(evt: NotificationEvent) -> None: """ Alerta interna cuando un informe de paciente rebota.

Si el workflow es 'informe-paciente', abre una tarea urgente
para que el dietista actualice el email del paciente.
"""
bounce_reason = (evt.raw_payload.get("data") or {}).get("bounce_reason", "desconocido")
logger.warning(
    "🚨 [BOUNCED] msg=%s | destinatario=%s | razón=%s | workflow=%s",
    evt.message_id,
    evt.recipient_email,
    bounce_reason,
    evt.workflow,
)

if evt.workflow == "informe-paciente":
    logger.critical(
        "⚠️  INFORME DE PACIENTE NO ENTREGADO — "
        "Abrir tarea urgente para dietista. recipient_id=%s",
        evt.recipient_id,
    )
    # En producción:
    # → INSERT INTO internal_alerts (type='bounced_patient_report', ...)
    # → Notificar al dietista asignado vía canal in-app de Knock

def handle_message_read(evt: NotificationEvent) -> None: """Actualiza métricas de engagement.""" logger.info( "👁️ [READ] msg=%s | destinatario=%s", evt.message_id, evt.recipient_email, ) # En producción → UPDATE engagement_stats SET read_count = read_count + 1

def handle_link_clicked(evt: NotificationEvent) -> None: """Registra el clic en el tracking de conversiones.""" url = (evt.raw_payload.get("event_data") or {}).get("url", "") logger.info( "🔗 [LINK_CLICKED] msg=%s | destinatario=%s | url=%s", evt.message_id, evt.recipient_email, url, ) # En producción → INSERT INTO link_clicks (message_id, url, clicked_at, ...)

── Tabla de dispatch ─────────────────────────────────────────────────────────

EVENT_HANDLERS = { "message.sent": handle_message_sent, "message.delivered": handle_message_delivered, "message.bounced": handle_message_bounced, "message.read": handle_message_read, "message.link_clicked": handle_link_clicked, }

── Endpoint principal ────────────────────────────────────────────────────────

@app.post("/webhooks/knock", status_code=status.HTTP_200_OK) async def knock_webhook(request: Request) -> JSONResponse: """ Recibe y procesa webhooks salientes de Knock.

Flujo:
  1. Verificar firma HMAC-SHA256 (primero, antes de parsear nada)
  2. Parsear JSON
  3. Idempotencia — ignorar eventos ya procesados (Knock reintenta ×8)
  4. Despachar al handler correspondiente
  5. Responder 200 OK (cualquier no-2xx fuerza reintento en Knock)
"""
# ── 1. Verificar firma (leer el body crudo ANTES de parsear) ──
raw_body = await request.body()
sig_header = request.headers.get("x-knock-signature")

valid, error = verify_knock_signature(raw_body, sig_header, KNOCK_WEBHOOK_SECRET)
if not valid:
    logger.warning("Webhook rechazado: %s | header=%s", error, sig_header)
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail=f"Webhook inválido: {error}",
    )

# ── 2. Parsear JSON ──
try:
    payload: dict[str, Any] = json.loads(raw_body)
except json.JSONDecodeError as exc:
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail=f"JSON malformado: {exc}",
    ) from exc

# ── 3. Idempotencia — Knock hace at-least-once delivery, reintenta ×8 ──
event_id: str = payload.get("id", "")
if event_id in _processed_event_ids:
    logger.info("Evento duplicado ignorado: id=%s", event_id)
    return JSONResponse({"received": True, "duplicate": True})

_processed_event_ids.add(event_id)
# En producción: usar Redis SETNX o INSERT ... ON CONFLICT DO NOTHING

# ── 4. Despachar ──
event_type: str = payload.get("type", "")
evt = _extract_event(payload)
handler = EVENT_HANDLERS.get(event_type)

if handler:
    handler(evt)
else:
    logger.debug("Evento no manejado: type=%s id=%s", event_type, event_id)

# ── 5. Responder 200 siempre (evitar reintento innecesario) ──
return JSONResponse({"received": True})

── Health check ──────────────────────────────────────────────────────────────

@app.get("/health") async def health() -> dict[str, str]: return {"status": "ok", "service": "nutriflow-knock-webhooks"}

── Tests con pytest ──────────────────────────────────────────────────────────

Ejecutar: pytest test_knock_webhook.py -v

Ejemplo de fixture (archivo separado test_knock_webhook.py):

import base64, hashlib, hmac, json, time

from fastapi.testclient import TestClient

from webhook_handler import app

SECRET = "test_secret_nutriflow"

os.environ["KNOCK_WEBHOOK_SECRET"] = SECRET

def make_signature(body: bytes, secret: str = SECRET) -> str:

ts = str(int(time.time() * 1000)) # ← milisegundos, no segundos

content = f"{ts}.{body.decode()}"

sig = base64.b64encode(

hmac.new(secret.encode(), content.encode(), hashlib.sha256).digest()

).decode()

return f"t={ts},s={sig}"

def test_message_bounced_triggers_alert(caplog):

client = TestClient(app)

payload = {

"id": "evt_bounce_001",

"type": "message.bounced",

"data": {

"id": "msg_abc123",

"channel_id": "email",

"source": "informe-paciente",

"recipient": {"id": "usr_42", "email": "paciente@clinica.es"},

"bounce_reason": "550 user unknown",

},

}

body = json.dumps(payload).encode()

resp = client.post(

"/webhooks/knock",

content=body,

headers={"x-knock-signature": make_signature(body)},

)

assert resp.status_code == 200

assert "INFORME DE PACIENTE NO ENTREGADO" in caplog.text

def test_duplicate_event_ignored():

# Segunda petición con el mismo event_id → duplicate=True

...

def test_invalid_signature_rejected():

client = TestClient(app)

body = b'{"id":"x","type":"message.sent"}'

resp = client.post(

"/webhooks/knock",

content=body,

headers={"x-knock-signature": "t=0,s=invalida"},

)

assert resp.status_code == 400

─────────────────────────────────────────────────────────────────────────────

if name == "main": import uvicorn

uvicorn.run(app, host="0.0.0.0", port=8000)