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.
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().
- Con un modelo de Pydantic como parámetro del cuerpo, FastAPI analiza y valida el JSON antes de que se ejecute el handler.
- Volver a serializar el modelo produce bytes distintos, así que la firma falla.
- Recibir el Request y esperar request.body() te da los bytes exactos.
- El handler los verifica, los analiza con json.loads, responde 204 y pasa el trabajo lento a BackgroundTasks.
Flask con standardwebhooks
Instala el paquete con pip install standardwebhooks.
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 "", 204request.get_data()devuelve el cuerpo como bytes. Flask lo guarda en caché, así que llamar después aget_json()sigue funcionando. No pasesas_text=True: decodificar primero es un paso que luego tienes que deshacer exactamente.Webhook(secret)recibe el secretowhsec_completo y quita el prefijo por sí solo. Créalo una vez al importar, no en cada petición.verifypasa a minúsculas los nombres de las cabeceras, así quedict(request.headers)funciona sea cual sea el uso de mayúsculas del proxy. Comprueba la marca de tiempo y cada firmav1, y devuelve el JSON analizado.
FastAPI con la librería estándar
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")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 primeroawait request.body(). request.headersno distingue mayúsculas y minúsculas, así que.get("webhook-id")encuentra la cabecera como sea que se haya enviado.hmac.compare_digestcompara los bytes decodificados en tiempo constante.validate=Truehace que una firma mal formada no coincida, en lugar de decodificar en silencio un valor parcial.BackgroundTasksse ejecuta después de enviar el204, 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
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
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:
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:
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 TrueLlama 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.
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
- Ejecuta
pytest. Los cuatro tests pasan. - Arranca la app detrás de una URL
https:pública y añádela en Webhooks en el panel. - Pon en
IKIFORM_WEBHOOK_SECRETel secretowhsec_del endpoint y reinicia el proceso. - Elige Send test. El registro de entregas muestra
204, y el log de tu aplicación muestra el eventoendpoint.testcon su iddlv_. - Envía el formulario una vez con un nombre con acentos, como
Zoë, y comprueba que llega intacto.
Errores comunes
| Problema | Causa y solución |
|---|---|
| El handler de FastAPI siempre falla con un modelo de cuerpo | El modelo analizó el cuerpo primero. Recibe Request, verifica await request.body() y luego Model.model_validate_json(body) |
403 Forbidden de Django | El middleware CSRF. Añade @csrf_exempt solo a la vista del webhook |
missing_headers en la versión con la librería estándar | Un 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 arrancar | El secreto se copió sin su = final. Vuelve a copiar el valor whsec_ completo |
timestamp_too_old en todas las entregas | El reloj del servidor va desfasado más de 5 minutos. Sincronízalo con NTP; puede que la firma en sí esté bien |
Preguntas
Recibe el objeto Request como parámetro y llama a await request.body(). No declares un modelo de Pydantic como parámetro del cuerpo en una ruta de webhook; analiza el JSON antes de que puedas verificarlo.
Sí. Instálalo con pip install standardwebhooks y llama a Webhook(secret).verify(body, headers) con el secreto whsec_ del panel. Lanza WebhookVerificationError cuando falla una comprobación.
== puede terminar en cuanto un carácter difiere, lo que filtra información de tiempo sobre cuánto de un intento era correcto. compare_digest tarda lo mismo en cualquier caso.
Sí. El middleware CSRF de Django rechaza los envíos sin token CSRF, e Ikiform no envía ninguno. La comprobación de la firma es la protección de la vista en su lugar.
hmac.new necesita bytes tanto para la clave como para el mensaje. Decodifica el secreto con base64.b64decode, que devuelve bytes, y construye el mensaje como f"{msg_id}.{timestamp}.".encode() + body, donde body son los bytes sin procesar de la petición.
¿Te atascaste en un paso o quieres una guía para tu caso? Escribe a hello@ikiform.com.