Guía de Seguridad de Webhooks

Esta guía explica cómo recibir de forma segura los eventos de webhook de Crypto Bot en producción.

Modelo de Amenazas

Un endpoint de webhook es una superficie HTTP pública. Sin verificación y controles, los atacantes pueden:

Falsificar notificaciones de pago
Reproducir actualizaciones antiguas
Agotar la capacidad de los workers con tráfico en ráfagas

Verificación de Firma

Crypto Bot firma cada solicitud con:

HMAC-SHA256(SHA256(api_token), raw_request_body)

La firma llega en el encabezado crypto-pay-api-signature.

Recomendado: Usar `Listener`

Listener verifica las firmas antes de ejecutar su callback.

::

callback puede ser síncrono (def) o asíncrono (async def).

Verificación Manual (Aplicación FastAPI Personalizada)

Si gestiona su propia ruta FastAPI, verifique contra el cuerpo crudo de la solicitud.

::

Protección contra Repetición

Listener soporta protección contra repetición extensible a través de replay_store. Para producción, prefiera un almacén compartido (por ejemplo, Redis) que implemente:

put_if_absent(key: str, ttl_seconds: int | None) -> bool
remove(key: str) -> None

Protección contra repetición en memoria integrada:

import os

from cryptobot.webhook import InMemoryReplayKeyStore, Listener

listener = Listener(
    host="127.0.0.1",
    callback=handle_webhook,
    api_token=os.environ["CRYPTOBOT_API_TOKEN"],
    replay_store=InMemoryReplayKeyStore(),
    replay_ttl_seconds=3600,
)

En producción, almacene esto en Redis o su base de datos en lugar de la memoria del proceso.

Claves de Repetición Personalizadas

Si su integración tiene un identificador de negocio estable, use replay_key_resolver:

import os

from cryptobot.webhook import InMemoryReplayKeyStore, Listener


def replay_key_resolver(data, raw_body, headers):
    payload = data.get("payload", {})
    invoice_id = payload.get("invoice_id")
    if invoice_id is not None:
        return f"invoice_paid:{invoice_id}"
    return None


listener = Listener(
    host="127.0.0.1",
    callback=handle_webhook,
    api_token=os.environ["CRYPTOBOT_API_TOKEN"],
    replay_store=InMemoryReplayKeyStore(),
    replay_ttl_seconds=3600,
    replay_key_resolver=replay_key_resolver,
)

Endurecimiento del Despliegue

Use HTTPS en el borde público.
Vincule la aplicación a 127.0.0.1 detrás de Nginx/Caddy/ingress cuando sea posible.
Aplique límites de solicitudes y timeouts del upstream.
Mantenga el manejador de webhook rápido; encole el trabajo costoso.
Nunca registre secretos o payloads crudos completos en producción.

Ejemplo de Proxy Reverso (Nginx)

server {
    listen 443 ssl http2;
    server_name pay.example.com;

    ssl_certificate /etc/letsencrypt/live/pay.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/pay.example.com/privkey.pem;

    location /webhook {
        proxy_pass http://127.0.0.1:2203/webhook;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Procesamiento Basado en Cola

Mueva la lógica de negocio no trivial fuera del hilo de la solicitud.

::

Pruebas Locales

1. Run listener locally

::

2. Expose local port with ngrok

ngrok http 2203

3. Set webhook URL in Crypto Bot

Use la URL HTTPS de ngrok más /webhook.

Prueba de Firma con Pytest

import hashlib
import hmac
import json

from fastapi.testclient import TestClient

from cryptobot.webhook import Listener


def sign(token: str, raw_body: str) -> str:
    secret = hashlib.sha256(token.encode()).digest()
    return hmac.new(secret, raw_body.encode("utf-8"), hashlib.sha256).hexdigest()


def test_valid_signature():
    received = {}

    def callback(headers, data):
        received.update(data)

    listener = Listener(host="127.0.0.1", callback=callback, api_token="test-token")
    client = TestClient(listener.app)

    body = {"update_id": 1, "update_type": "invoice_paid", "payload": {"invoice_id": 123}}
    raw = json.dumps(body)
    signature = sign("test-token", raw)

    resp = client.post("/webhook", content=raw, headers={"crypto-pay-api-signature": signature})

    assert resp.status_code == 200
    assert received["update_id"] == 1

Lista de Verificación de Seguridad

Verificación de firma habilitada (Listener o check_signature)
Terminación HTTPS configurada
El token proviene de variables de entorno/gestor de secretos
Protección contra repetición implementada
Operaciones lentas/en segundo plano movidas a workers
El registro de solicitudes oculta campos sensibles