Verifica las firmas de webhook

Verifica las firmas de webhook6 min de lectura

Verifica firmas de webhook en Python

Verifica las firmas de los webhooks de Ikiform en Flask, FastAPI y Django, con el paquete standardwebhooks o la librería estándar, más un test de pytest.

Preet SutharFundador, Ikiform

Lee el cuerpo de la petición como bytes antes de que nada lo analice: request.get_data() en Flask, await request.body() en FastAPI, request.body en Django. Pasa esos bytes y las cabeceras a Webhook(secret).verify() del paquete standardwebhooks, o calcula tú mismo el HMAC-SHA256 con hmac y compara con hmac.compare_digest. Responde 400 cuando falle y analiza el JSON solo después de que pase.

Tres frameworks, dos formas de verificar. Flask y Django usan el paquete standardwebhooks; FastAPI usa una función de la librería estándar con una suite de pytest basada en el vector de prueba. Combínalos como quieras: la función y el paquete son intercambiables.

Qué cambia en Python

Los frameworks de Python te dan el cuerpo de una forma que parece sin procesar pero a veces no lo es. FastAPI es el más delicado: declara un modelo de Pydantic como parámetro y FastAPI lee, analiza y valida el cuerpo antes de que se ejecute tu función. Recibes un modelo, y los bytes que firmó Ikiform nunca llegan a tus manos. La misma trampa existe en Flask con request.get_json() y en Django REST framework con request.data. En Starlette y FastAPI el cuerpo es un stream que await request.body() lee y guarda en caché, así que llámalo antes de que algo recorra request.stream().

Un handler de FastAPI con un parámetro de cuerpo de Pydantic, comparado con uno que lee el cuerpo sin procesar de la peticiónmodelo de Pydantic como parámetro del cuerpoawait request.body()bytes de lapeticiónanalizado yvalidadoantes de que seejecute tucódigomodel.model_dump_json()bytesdistintos400bytes de lapeticiónbytestal como seenviaronverifica yluegojson.loads204,trabajo ensegundoplanoUn handler de FastAPI con un parámetro de cuerpo de Pydantic, comparado con uno que lee el cuerpo sin procesar de la peticiónmodelo de Pydantic como parámetro del cuerpoawait request.body()bytes de la peticiónanalizado y validadoantes de que se ejecute tucódigomodel.model_dump_json()bytes distintos400bytes de la peticiónbytestal como se enviaronverifica y luegojson.loads204, trabajo en segundoplano
  1. Con un modelo de Pydantic como parámetro del cuerpo, FastAPI analiza y valida el JSON antes de que se ejecute el handler.
  2. Volver a serializar el modelo produce bytes distintos, así que la firma falla.
  3. Recibir el Request y esperar request.body() te da los bytes exactos.
  4. El handler los verifica, los analiza con json.loads, responde 204 y pasa el trabajo lento a BackgroundTasks.
Si quieres un modelo, valida con Pydantic después de verificar: Event.model_validate_json(body).

Flask con standardwebhooks

Instala el paquete con pip install standardwebhooks.

app.py
import os

from flask import Flask, request
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

app = Flask(__name__)
webhook = Webhook(os.environ["IKIFORM_WEBHOOK_SECRET"])


@app.post("/webhooks/ikiform")
def ikiform_webhook():
    try:
        event = webhook.verify(request.get_data(), dict(request.headers))
    except WebhookVerificationError as error:
        return str(error), 400
    if event["type"] == "submission.created":
        app.logger.info("submission %s %s", event["id"], event["data"]["fields"])
    return "", 204
  • request.get_data() devuelve el cuerpo como bytes. Flask lo guarda en caché, así que llamar después a get_json() sigue funcionando. No pases as_text=True: decodificar primero es un paso que luego tienes que deshacer exactamente.
  • Webhook(secret) recibe el secreto whsec_ completo y quita el prefijo por sí solo. Créalo una vez al importar, no en cada petición.
  • verify pasa a minúsculas los nombres de las cabeceras, así que dict(request.headers) funciona sea cual sea el uso de mayúsculas del proxy. Comprueba la marca de tiempo y cada firma v1, y devuelve el JSON analizado.

FastAPI con la librería estándar

signatures.py
import base64
import binascii
import hashlib
import hmac
import re
import time
from collections.abc import Mapping

TOLERANCE_SECONDS = 5 * 60


class InvalidSignature(Exception):
    pass


def verify_signature(body: bytes, headers: Mapping[str, str], secret: str, now: float | None = None) -> None:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures:
        raise InvalidSignature("missing_headers")
    if not re.fullmatch(r"[0-9]{1,12}", timestamp):
        raise InvalidSignature("invalid_timestamp")

    age = int(time.time() if now is None else now) - int(timestamp)
    if age > TOLERANCE_SECONDS:
        raise InvalidSignature("timestamp_too_old")
    if age < -TOLERANCE_SECONDS:
        raise InvalidSignature("timestamp_too_new")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    expected = hmac.new(key, f"{msg_id}.{timestamp}.".encode() + body, hashlib.sha256).digest()

    for entry in signatures.split(" "):
        version, _, value = entry.partition(",")
        if version != "v1":
            continue
        try:
            received = base64.b64decode(value, validate=True)
        except binascii.Error:
            continue
        if hmac.compare_digest(received, expected):
            return
    raise InvalidSignature("no_matching_signature")
main.py
import json
import logging
import os

from fastapi import BackgroundTasks, FastAPI, Request, Response

from signatures import InvalidSignature, verify_signature

app = FastAPI()
log = logging.getLogger("webhooks")
SECRET = os.environ["IKIFORM_WEBHOOK_SECRET"]


def handle_submission(delivery_id: str, fields: dict) -> None:
    log.info("submission %s %s", delivery_id, fields)


@app.post("/webhooks/ikiform")
async def ikiform_webhook(request: Request, background: BackgroundTasks) -> Response:
    body = await request.body()
    try:
        verify_signature(body, request.headers, SECRET)
    except InvalidSignature as error:
        return Response(str(error), status_code=400)
    event = json.loads(body)
    if event["type"] == "submission.created":
        background.add_task(handle_submission, event["id"], event["data"]["fields"])
    return Response(status_code=204)
  • El handler recibe Request, no un modelo, y lee primero await request.body().
  • request.headers no distingue mayúsculas y minúsculas, así que .get("webhook-id") encuentra la cabecera como sea que se haya enviado.
  • hmac.compare_digest compara los bytes decodificados en tiempo constante. validate=True hace que una firma mal formada no coincida, en lugar de decodificar en silencio un valor parcial.
  • BackgroundTasks se ejecuta después de enviar el 204, así que el trabajo lento no cuenta contra la espera de 15 segundos de Ikiform.
  • Las anotaciones de tipo necesitan Python 3.10 o posterior.

Un test de pytest

test_signatures.py
import json

import pytest

from signatures import InvalidSignature, verify_signature

SECRET = "whsec_Fg8c6hm1Gt41v5RSPKm72LggDx8S1mlWkbBYpHPAnpw="
ROLLED = "whsec_4YWiO+F83kohh/SGNr2zcgChkpVJ65ySVeV23Qabr2A="
BODY = b'{"id":"dlv_01k7v2d4n8r2p6s0w3y5b7c9e1","type":"endpoint.test","timestamp":"2026-10-17T12:00:00.000Z","api_version":"2026-10-17","data":{"endpoint_id":"whk_01k7v2e1b3d5f7h9k1m3n5q7s9","message":"This is a test event from Ikiform."}}'
HEADERS = {
    "webhook-id": "dlv_01k7v2d4n8r2p6s0w3y5b7c9e1",
    "webhook-timestamp": "1792300000",
    "webhook-signature": "v1,Duq+Iuh0xbRZyS+C/TO5W1hcGLGwg+ifBiWiYkq15qA=",
}
NOW = 1792300000 + 30


def test_accepts_the_vector():
    verify_signature(BODY, HEADERS, SECRET, now=NOW)


def test_accepts_either_secret_during_a_roll():
    both = {**HEADERS, "webhook-signature": HEADERS["webhook-signature"] + " v1,gfHP3s7ANdIpnM7gewtWtSvTrHgmsbEFgCaeRRR0Eak="}
    verify_signature(BODY, both, SECRET, now=NOW)
    verify_signature(BODY, both, ROLLED, now=NOW)


def test_rejects_a_reserialized_body():
    pretty = json.dumps(json.loads(BODY), indent=2).encode()
    with pytest.raises(InvalidSignature, match="no_matching_signature"):
        verify_signature(pretty, HEADERS, SECRET, now=NOW)


def test_rejects_a_stale_timestamp():
    with pytest.raises(InvalidSignature, match="timestamp_too_old"):
        verify_signature(BODY, HEADERS, SECRET, now=NOW + 600)

Ejecuta pytest. El NOW fijo mantiene la marca de tiempo del vector dentro de la ventana de 5 minutos. json.dumps(indent=2) es una forma realista en que un cuerpo cambia en un código Python, así que ese caso se gana su sitio.

Django

webhooks/views.py
import os

from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

webhook = Webhook(os.environ["IKIFORM_WEBHOOK_SECRET"])


@csrf_exempt
@require_POST
def ikiform_webhook(request):
    try:
        event = webhook.verify(request.body, dict(request.headers))
    except WebhookVerificationError as error:
        return HttpResponseBadRequest(str(error))
    if event["type"] == "submission.created":
        print("submission", event["id"])
    return HttpResponse(status=204)

request.body son los bytes sin procesar. @csrf_exempt es obligatorio, porque Ikiform no tiene token CSRF y Django rechazaría el envío con 403; la comprobación de la firma sustituye a la protección CSRF en esta vista concreta. Lee request.body antes de tocar request.POST, que solo importa en envíos de formulario pero puede consumir el stream.

Eventos tipados después de verificar

Pydantic sigue siendo útil, un paso más tarde. Verifica los bytes y luego valida esos mismos bytes en un modelo. Obtienes acceso tipado al payload y un error claro si falta un campo del que dependes, sin renunciar a la comprobación de la firma:

events.py
from pydantic import BaseModel


class Form(BaseModel):
    id: str
    name: str


class SubmissionData(BaseModel):
    submission_id: str
    form: Form
    fields: dict[str, object]
    created_at: str
    truncated: bool


class SubmissionCreated(BaseModel):
    id: str
    type: str
    data: SubmissionData


def parse_submission(body: bytes) -> SubmissionCreated:
    return SubmissionCreated.model_validate_json(body)

Pydantic ignora por defecto las claves extra, así que los campos nuevos que Ikiform añada al payload no rompen el modelo. Llama a parse_submission(body) solo para submission.created; un evento endpoint.test tiene otra forma de data.

Saltar las repeticiones

Una entrega verificada puede llegar dos veces cuando Ikiform no recibe tu respuesta en 15 segundos y reintenta con el mismo webhook-id. En Django, una columna única y un IntegrityError hacen que la segunda no haga nada:

webhooks/models.py
from django.db import IntegrityError, models, transaction


class Delivery(models.Model):
    delivery_id = models.CharField(max_length=64, unique=True)
    received_at = models.DateTimeField(auto_now_add=True)


def first_delivery(delivery_id: str) -> bool:
    try:
        with transaction.atomic():
            Delivery.objects.create(delivery_id=delivery_id)
    except IntegrityError:
        return False
    return True

Llama a first_delivery(event["id"]) después de verificar; si devuelve False, responde 204 y para. El bloque atomic evita que una inserción fallida rompa el resto de la transacción de la petición. Con SQLAlchemy, insert(...).on_conflict_do_nothing() hace lo mismo en PostgreSQL y SQLite.

El verify del paquete standardwebhooks decodifica el cuerpo como UTF-8 antes de calcular el hash, lo cual va bien con el JSON de Ikiform. La versión con la librería estándar calcula el hash de los bytes directamente, así que no depende de ninguna decodificación.

Comprueba que funciona

  1. Ejecuta pytest. Los cuatro tests pasan.
  2. Arranca la app detrás de una URL https: pública y añádela en Webhooks en el panel.
  3. Pon en IKIFORM_WEBHOOK_SECRET el secreto whsec_ del endpoint y reinicia el proceso.
  4. Elige Send test. El registro de entregas muestra 204, y el log de tu aplicación muestra el evento endpoint.test con su id dlv_.
  5. Envía el formulario una vez con un nombre con acentos, como Zoë, y comprueba que llega intacto.

Errores comunes

ProblemaCausa y solución
El handler de FastAPI siempre falla con un modelo de cuerpoEl modelo analizó el cuerpo primero. Recibe Request, verifica await request.body() y luego Model.model_validate_json(body)
403 Forbidden de DjangoEl middleware CSRF. Añade @csrf_exempt solo a la vista del webhook
missing_headers en la versión con la librería estándarUn dict normal de cabeceras conservó las mayúsculas del servidor. Usa el objeto de cabeceras del framework, que no distingue mayúsculas, o pasa las claves a minúsculas
binascii.Error: Incorrect padding al arrancarEl secreto se copió sin su = final. Vuelve a copiar el valor whsec_ completo
timestamp_too_old en todas las entregasEl reloj del servidor va desfasado más de 5 minutos. Sincronízalo con NTP; puede que la firma en sí esté bien

Preguntas

¿Te atascaste en un paso o quieres una guía para tu caso? Escribe a hello@ikiform.com.