Como pego o corpo bruto da requisição no FastAPI?

Receba o objeto Request como parâmetro e chame await request.body(). Não declare um modelo Pydantic como parâmetro de corpo numa rota de webhook; ele faz o parse do JSON antes que você possa verificá-lo.

O pacote Python standardwebhooks funciona com o Ikiform?

Sim. Instale com pip install standardwebhooks e chame Webhook(secret).verify(body, headers) com o segredo whsec_ do painel. Ele lança WebhookVerificationError quando uma verificação falha.

Por que usar hmac.compare_digest em vez de ==?

O == pode retornar assim que um caractere for diferente, o que vaza informação de tempo sobre quanto de um palpite estava certo. O compare_digest leva o mesmo tempo em qualquer caso.

Preciso de csrf_exempt numa view de webhook do Django?

Sim. O middleware de CSRF do Django recusa envios sem token CSRF, e o Ikiform não envia um. A verificação da assinatura é a proteção da view no lugar dele.

Por que recebo um TypeError sobre bytes e str ao calcular o HMAC?

O hmac.new precisa de bytes tanto para a chave quanto para a mensagem. Decodifique o segredo com base64.b64decode, que devolve bytes, e monte a mensagem como f"{msg_id}.{timestamp}.".encode() + body, onde body são os bytes brutos da requisição.

Verificar assinaturas de webhook

25 de outubro de 2026Verificar assinaturas de webhook6 min de leitura

Verificar assinaturas de webhook em Python

Verifique assinaturas de webhook do Ikiform no Flask, FastAPI e Django, com o pacote standardwebhooks ou a biblioteca padrão, mais um teste com pytest.

Preet SutharFundador, Ikiform

Leia o corpo da requisição como bytes antes que algo faça o parse dele: request.get_data() no Flask, await request.body() no FastAPI, request.body no Django. Passe esses bytes e os headers para Webhook(secret).verify() do pacote standardwebhooks, ou calcule o HMAC-SHA256 você mesmo com hmac e compare com hmac.compare_digest. Responda 400 quando falhar e só faça o parse do JSON depois que passar.

Três frameworks, duas formas de verificar. Flask e Django usam o pacote standardwebhooks; o FastAPI usa uma função da biblioteca padrão com uma suíte pytest baseada no vetor de teste. Combine como quiser: a função e o pacote são intercambiáveis.

O que muda no Python

Os frameworks Python entregam o corpo de um jeito que parece bruto, mas às vezes não é. O FastAPI é o mais traiçoeiro: declare um modelo Pydantic como parâmetro e o FastAPI lê, faz o parse e valida o corpo antes de a sua função rodar. Você recebe um modelo, e os bytes que o Ikiform assinou nunca chegam às suas mãos. A mesma armadilha existe no Flask com request.get_json() e no Django REST framework com request.data. No Starlette e no FastAPI, o corpo é um stream que await request.body() lê e guarda em cache, então chame-o antes que qualquer coisa itere request.stream().

Valide com Pydantic depois de verificar, se quiser um modelo: Event.model_validate_json(body).

Flask com standardwebhooks

Instale o pacote com 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() devolve o corpo como bytes. O Flask guarda em cache, então chamar get_json() depois ainda funciona. Não passe as_text=True: decodificar antes é um passo que você depois precisa desfazer com exatidão.
Webhook(secret) recebe o segredo whsec_ completo e tira o prefixo sozinho. Crie-o uma vez na importação, não a cada requisição.
verify coloca os nomes dos headers em minúsculas, então dict(request.headers) funciona seja qual for a capitalização usada pelo proxy. Ele confere o timestamp, cada assinatura v1 e devolve o JSON parseado.

FastAPI com a biblioteca padrão

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)

O handler recebe Request, não um modelo, e lê await request.body() primeiro.
request.headers não diferencia maiúsculas de minúsculas, então .get("webhook-id") encontra o header seja como for que ele foi enviado.
hmac.compare_digest compara os bytes decodificados em tempo constante. validate=True faz uma assinatura malformada não bater, em vez de virar silenciosamente um valor parcial decodificado.
BackgroundTasks roda depois que o 204 é enviado, então o trabalho lento não conta para a espera de 15 segundos do Ikiform.
As anotações de tipo precisam do Python 3.10 ou posterior.

Um teste com 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)

Rode pytest. O NOW fixo mantém o timestamp do vetor dentro da janela de 5 minutos. json.dumps(indent=2) é um jeito realista de um corpo ser alterado numa base de código Python, então esse caso merece estar ali.

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 são os bytes brutos. @csrf_exempt é obrigatório, porque o Ikiform não tem token CSRF e o Django recusaria o envio com 403; a verificação da assinatura substitui a proteção CSRF nesta view. Leia request.body antes de mexer em request.POST, que só importa para envios de formulário mas pode consumir o stream.

Eventos tipados depois de verificar

O Pydantic continua útil, um passo depois. Verifique os bytes e então valide esses mesmos bytes num modelo. Você ganha acesso tipado ao payload e um erro claro se faltar um campo de que você depende, sem abrir mão da verificação da assinatura:

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)

Por padrão, o Pydantic ignora chaves extras, então campos novos que o Ikiform adicionar ao payload não quebram o modelo. Chame parse_submission(body) só para submission.created; um evento endpoint.test tem um formato de data diferente.

Ignorando repetições

Uma entrega verificada pode chegar duas vezes quando o Ikiform não recebe a sua resposta em 15 segundos e tenta de novo com o mesmo webhook-id. No Django, uma coluna única e um IntegrityError fazem a segunda não ter efeito:

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

Chame first_delivery(event["id"]) depois de verificar; se devolver False, responda 204 e pare. O bloco atomic impede que uma inserção que falhou quebre o resto da transação da requisição. Com SQLAlchemy, insert(...).on_conflict_do_nothing() faz o mesmo no PostgreSQL e no SQLite.

O verify do pacote standardwebhooks decodifica o corpo como UTF-8 antes de calcular o hash, o que não é problema para o JSON do Ikiform. A versão com a biblioteca padrão calcula o hash dos bytes diretamente, então não depende de nenhuma decodificação.

Confira se funciona

Rode pytest. Os quatro testes passam.
Suba o app atrás de uma URL https: pública e adicione-a em Webhooks no painel.
Defina IKIFORM_WEBHOOK_SECRET com o segredo whsec_ do endpoint e reinicie o processo.
Escolha Send test. O log de entregas mostra 204, e o log da sua aplicação mostra o evento endpoint.test com o id dlv_ dele.
Envie o formulário uma vez com um nome acentuado, como Zoë, e confira se ele chega intacto.

Pegadinhas

Problema	Causa e correção
O handler FastAPI sempre falha com um modelo de corpo	O modelo fez o parse do corpo primeiro. Receba `Request`, verifique `await request.body()` e depois `Model.model_validate_json(body)`
`403 Forbidden` do Django	Middleware de CSRF. Adicione `@csrf_exempt` só na view do webhook
`missing_headers` na versão com a biblioteca padrão	Um `dict` simples de headers manteve a capitalização do servidor. Use o objeto de headers do framework, que não diferencia maiúsculas, ou coloque as chaves em minúsculas
`binascii.Error: Incorrect padding` na inicialização	O segredo foi copiado sem o `=` final. Copie o valor `whsec_` inteiro de novo
`timestamp_too_old` em toda entrega	O relógio do servidor está mais de 5 minutos fora. Sincronize-o com NTP; a assinatura em si pode estar certa

Perguntas

Travou em algum passo ou quer um guia para o seu caso? Envie um e-mail para hello@ikiform.com.