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.
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().
- With a Pydantic model as the body parameter, FastAPI parses and validates the JSON before the handler runs.
- Serializing the model again produces different bytes, so the signature fails.
- Taking the Request and awaiting request.body() gives the exact bytes.
- The handler verifies them, parses with json.loads, answers 204 and hands slow work to BackgroundTasks.
Flask with standardwebhooks
Install the package with 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()returns the body as bytes. Flask caches it, so callingget_json()afterwards still works. Don't passas_text=True: decoding first is a step you then have to undo exactly.Webhook(secret)takes the fullwhsec_secret and strips the prefix itself. Build it once at import time, not per request.verifylowercases header names, sodict(request.headers)works whatever case the proxy used. It checks the timestamp, everyv1signature, and returns the parsed JSON.
FastAPI with the standard library
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)- The handler takes
Request, not a model, and readsawait request.body()first. request.headersis case-insensitive, so.get("webhook-id")finds the header however it was sent.hmac.compare_digestcompares the decoded bytes in constant time.validate=Truemakes a malformed signature a non-match instead of a silently decoded partial value.BackgroundTasksruns after the204is 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
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
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:
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:
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 TrueCall 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.
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
- Run
pytest. All four tests pass. - Start the app behind a public
https:URL and add it under Webhooks in the dashboard. - Set
IKIFORM_WEBHOOK_SECRETto the endpoint'swhsec_secret and restart the process. - Choose Send test. The delivery log shows
204, and your application log shows theendpoint.testevent with itsdlv_id. - Submit the form once with an accented name, such as
Zoë, and check it arrives intact.
Gotchas
| Problem | Cause and fix |
|---|---|
| FastAPI handler always fails with a body model | The model parsed the body first. Take Request, verify await request.body(), then Model.model_validate_json(body) |
403 Forbidden from Django | CSRF middleware. Add @csrf_exempt to the webhook view only |
missing_headers in the standard-library version | A 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 startup | The secret was copied without its trailing =. Copy the whole whsec_ value again |
timestamp_too_old on every delivery | The server clock is more than 5 minutes off. Sync it with NTP; the signature itself may be fine |
Questions
Take the Request object as a parameter and call await request.body(). Don't declare a Pydantic model as the body parameter for a webhook route; it parses the JSON before you can verify it.
Yes. Install it with pip install standardwebhooks and call Webhook(secret).verify(body, headers) with the whsec_ secret from the dashboard. It raises WebhookVerificationError when a check fails.
== can return as soon as one character differs, which leaks timing information about how much of a guess was right. compare_digest takes the same time regardless.
Yes. Django's CSRF middleware refuses posts without a CSRF token, and Ikiform doesn't send one. The signature check is the view's protection instead.
hmac.new needs bytes for both the key and the message. Decode the secret with base64.b64decode, which returns bytes, and build the message as f"{msg_id}.{timestamp}.".encode() + body, where body is the raw bytes from the request.
Stuck on a step, or want a guide for your situation? Email hello@ikiform.com.