Verify webhook signatures

Verify webhook signatures5 min read

Verify webhook signatures in Python

Verify Ikiform webhook signatures in Flask, FastAPI and Django, with the standardwebhooks package or the standard library, plus a pytest test.

Preet SutharFounder, Ikiform

Read the request body as bytes before anything parses it: request.get_data() in Flask, await request.body() in FastAPI, request.body in Django. Pass those bytes and the headers to Webhook(secret).verify() from the standardwebhooks package, or compute the HMAC-SHA256 yourself with hmac and compare with hmac.compare_digest. Answer 400 when it fails and parse the JSON only after it passes.

Three frameworks, two ways to verify. Flask and Django use the standardwebhooks package; FastAPI uses a standard-library function with a pytest suite built on the test vector. Mix them however you like: the function and the package are interchangeable.

What's different in Python

Python frameworks give you the body in a form that looks raw but sometimes isn't. FastAPI is the sharp one: declare a Pydantic model as a parameter and FastAPI reads, parses and validates the body before your function runs. You get a model, and the bytes Ikiform signed are never in your hands. The same trap exists in Flask with request.get_json() and Django REST framework with request.data. In Starlette and FastAPI the body is a stream that await request.body() reads and caches, so call it before anything iterates request.stream().

A FastAPI handler with a Pydantic body parameter compared with one that reads the raw request bodyPydantic model as the body parameterawait request.body()requestbytesparsed andvalidatedbefore yourcode runsmodel.model_dump_json()differentbytes400requestbytesbytesexactly as sentverify, thenjson.loads204, work inbackgroundA FastAPI handler with a Pydantic body parameter compared with one that reads the raw request bodyPydantic model as the body parameterawait request.body()request bytesparsed and validatedbefore your code runsmodel.model_dump_json()different bytes400request bytesbytesexactly as sentverify, then json.loads204, work in background
  1. With a Pydantic model as the body parameter, FastAPI parses and validates the JSON before the handler runs.
  2. Serializing the model again produces different bytes, so the signature fails.
  3. Taking the Request and awaiting request.body() gives the exact bytes.
  4. The handler verifies them, parses with json.loads, answers 204 and hands slow work to BackgroundTasks.
Validate with Pydantic after verifying, if you want a model: Event.model_validate_json(body).

Flask with standardwebhooks

Install the package with 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() returns the body as bytes. Flask caches it, so calling get_json() afterwards still works. Don't pass as_text=True: decoding first is a step you then have to undo exactly.
  • Webhook(secret) takes the full whsec_ secret and strips the prefix itself. Build it once at import time, not per request.
  • verify lowercases header names, so dict(request.headers) works whatever case the proxy used. It checks the timestamp, every v1 signature, and returns the parsed JSON.

FastAPI with the standard library

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)
  • The handler takes Request, not a model, and reads await request.body() first.
  • request.headers is case-insensitive, so .get("webhook-id") finds the header however it was sent.
  • hmac.compare_digest compares the decoded bytes in constant time. validate=True makes a malformed signature a non-match instead of a silently decoded partial value.
  • BackgroundTasks runs after the 204 is sent, so slow work doesn't count against Ikiform's 15-second wait.
  • The type hints need Python 3.10 or later.

A pytest test

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)

Run pytest. The fixed NOW keeps the vector's timestamp inside the 5-minute window. json.dumps(indent=2) is a realistic way a body gets changed in a Python codebase, so that case earns its place.

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 is the raw bytes. @csrf_exempt is required, because Ikiform has no CSRF token and Django would refuse the post with 403; the signature check replaces CSRF protection for this one view. Read request.body before touching request.POST, which only matters for form posts but can consume the stream.

Typed events after verifying

Pydantic is still useful, one step later. Verify the bytes, then validate the same bytes into a model. You get typed access to the payload and a clear error if a field you rely on is missing, without giving up the signature check:

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 ignores extra keys by default, so new fields Ikiform adds to the payload don't break the model. Call parse_submission(body) only for submission.created; an endpoint.test event has a different data shape.

Skipping repeats

A verified delivery can arrive twice when Ikiform doesn't get your answer within 15 seconds and retries with the same webhook-id. In Django, a unique column and an IntegrityError turn the second one into a no-op:

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

Call first_delivery(event["id"]) after verifying; if it returns False, answer 204 and stop. The atomic block keeps a failed insert from breaking the rest of the request's transaction. With SQLAlchemy, insert(...).on_conflict_do_nothing() does the same on PostgreSQL and SQLite.

The standardwebhooks package's verify decodes the body as UTF-8 before hashing, which is fine for Ikiform's JSON. The standard-library version hashes the bytes directly, so it doesn't depend on any decoding.

Check it works

  1. Run pytest. All four tests pass.
  2. Start the app behind a public https: URL and add it under Webhooks in the dashboard.
  3. Set IKIFORM_WEBHOOK_SECRET to the endpoint's whsec_ secret and restart the process.
  4. Choose Send test. The delivery log shows 204, and your application log shows the endpoint.test event with its dlv_ id.
  5. Submit the form once with an accented name, such as Zoë, and check it arrives intact.

Gotchas

ProblemCause and fix
FastAPI handler always fails with a body modelThe model parsed the body first. Take Request, verify await request.body(), then Model.model_validate_json(body)
403 Forbidden from DjangoCSRF middleware. Add @csrf_exempt to the webhook view only
missing_headers in the standard-library versionA plain dict of headers kept the server's capitalization. Use the framework's case-insensitive headers object, or lowercase the keys
binascii.Error: Incorrect padding at startupThe secret was copied without its trailing =. Copy the whole whsec_ value again
timestamp_too_old on every deliveryThe server clock is more than 5 minutes off. Sync it with NTP; the signature itself may be fine

Questions

Stuck on a step, or want a guide for your situation? Email hello@ikiform.com.