Wie bekomme ich in FastAPI den rohen Request-Body?

Nimm das Request-Objekt als Parameter und ruf await request.body() auf. Deklariere für eine Webhook-Route kein Pydantic-Modell als Body-Parameter; es parst das JSON, bevor du es verifizieren kannst.

Funktioniert das Python-Paket standardwebhooks mit Ikiform?

Ja. Installiere es mit pip install standardwebhooks und ruf Webhook(secret).verify(body, headers) mit dem whsec_-Secret aus dem Dashboard auf. Es wirft WebhookVerificationError, wenn eine Prüfung scheitert.

Warum hmac.compare_digest statt == nutzen?

== kann zurückkehren, sobald ein Zeichen abweicht, was über die Laufzeit verrät, wie viel eines Rateversuchs stimmte. compare_digest braucht immer gleich lange.

Brauche ich csrf_exempt für eine Django-Webhook-View?

Ja. Die CSRF-Middleware von Django lehnt Posts ohne CSRF-Token ab, und Ikiform sendet keins. Die Signaturprüfung ist stattdessen der Schutz der View.

Warum bekomme ich beim Berechnen des HMAC einen TypeError zu bytes und str?

hmac.new braucht Bytes für Key und Nachricht. Dekodiere das Secret mit base64.b64decode, das Bytes zurückgibt, und baue die Nachricht als f"{msg_id}.{timestamp}.".encode() + body, wobei body die rohen Bytes aus dem Request sind.

Webhook-Signaturen prüfen

25. Oktober 2026Webhook-Signaturen prüfen6 Min. Lesezeit

Webhook-Signaturen in Python prüfen

Prüfe Webhook-Signaturen von Ikiform in Flask, FastAPI und Django, mit dem Paket standardwebhooks oder der Standardbibliothek, plus einem Test mit pytest.

Preet SutharGründer, Ikiform

Lies den Request-Body als Bytes, bevor ihn etwas parst: request.get_data() in Flask, await request.body() in FastAPI, request.body in Django. Übergib diese Bytes und die Header an Webhook(secret).verify() aus dem Paket standardwebhooks, oder berechne den HMAC-SHA256 selbst mit hmac und vergleiche mit hmac.compare_digest. Antworte mit 400, wenn die Prüfung scheitert, und parse das JSON erst, wenn sie besteht.

Drei Frameworks, zwei Wege zur Prüfung. Flask und Django nutzen das Paket standardwebhooks; FastAPI nutzt eine Funktion aus der Standardbibliothek mit einer pytest-Suite auf Basis des Testvektors. Kombiniere sie, wie du willst: Funktion und Paket sind austauschbar.

Was in Python anders ist

Python-Frameworks geben dir den Body in einer Form, die roh aussieht, es aber manchmal nicht ist. FastAPI ist der heikle Fall: Deklarierst du ein Pydantic-Modell als Parameter, liest, parst und validiert FastAPI den Body, bevor deine Funktion läuft. Du bekommst ein Modell, und die Bytes, die Ikiform signiert hat, hast du nie in der Hand. Dieselbe Falle gibt es in Flask mit request.get_json() und im Django REST Framework mit request.data. In Starlette und FastAPI ist der Body ein Stream, den await request.body() liest und cacht, also ruf es auf, bevor irgendetwas über request.stream() iteriert.

Validiere nach der Prüfung mit Pydantic, wenn du ein Modell willst: Event.model_validate_json(body).

Flask mit standardwebhooks

Installiere das Paket mit 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() gibt den Body als Bytes zurück. Flask cacht ihn, daher funktioniert get_json() danach weiterhin. Übergib nicht as_text=True: Vorher zu dekodieren ist ein Schritt, den du dann exakt rückgängig machen müsstest.
Webhook(secret) nimmt das vollständige whsec_-Secret und entfernt das Präfix selbst. Erzeuge es einmal beim Import, nicht pro Request.
verify schreibt Header-Namen klein, daher funktioniert dict(request.headers) unabhängig davon, welche Schreibweise der Proxy verwendet hat. Es prüft den Zeitstempel und jede v1-Signatur und gibt das geparste JSON zurück.

FastAPI mit der Standardbibliothek

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)

Der Handler nimmt Request statt eines Modells und liest zuerst await request.body().
request.headers unterscheidet nicht zwischen Groß- und Kleinschreibung, daher findet .get("webhook-id") den Header, egal wie er gesendet wurde.
hmac.compare_digest vergleicht die dekodierten Bytes in konstanter Zeit. validate=True macht eine fehlerhafte Signatur zu einem Nicht-Treffer statt zu einem still dekodierten Teilwert.
BackgroundTasks läuft, nachdem das 204 gesendet wurde, sodass langsame Arbeit nicht gegen die 15 Sekunden Wartezeit von Ikiform zählt.
Die Type Hints brauchen Python 3.10 oder neuer.

Ein 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)

Führe pytest aus. Das feste NOW hält den Zeitstempel des Testvektors im 5-Minuten-Fenster. json.dumps(indent=2) ist ein realistischer Weg, auf dem ein Body in einer Python-Codebasis verändert wird, also hat dieser Fall seinen Platz verdient.

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 sind die rohen Bytes. @csrf_exempt ist nötig, weil Ikiform kein CSRF-Token hat und Django den Post mit 403 ablehnen würde; die Signaturprüfung ersetzt den CSRF-Schutz für diese eine View. Lies request.body, bevor du request.POST anfasst, was nur bei Formular-Posts relevant ist, aber den Stream verbrauchen kann.

Typisierte Events nach dem Verifizieren

Pydantic ist weiterhin nützlich, einen Schritt später. Verifiziere die Bytes und validiere dann dieselben Bytes in ein Modell. So bekommst du typisierten Zugriff auf den Payload und einen klaren Fehler, wenn ein Feld fehlt, auf das du dich verlässt, ohne die Signaturprüfung aufzugeben:

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 ignoriert zusätzliche Keys standardmäßig, sodass neue Felder, die Ikiform zum Payload hinzufügt, das Modell nicht brechen. Ruf parse_submission(body) nur für submission.created auf; ein endpoint.test-Event hat eine andere Form von data.

Wiederholungen überspringen

Eine verifizierte Zustellung kann zweimal ankommen, wenn Ikiform deine Antwort nicht innerhalb von 15 Sekunden bekommt und mit derselben webhook-id wiederholt. In Django machen eine eindeutige Spalte und ein IntegrityError die zweite zu einem 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

Ruf first_delivery(event["id"]) nach dem Verifizieren auf; gibt es False zurück, antworte mit 204 und hör auf. Der atomic-Block verhindert, dass ein gescheiterter Insert den Rest der Transaktion des Requests kaputt macht. Mit SQLAlchemy tut insert(...).on_conflict_do_nothing() dasselbe auf PostgreSQL und SQLite.

verify aus dem Paket standardwebhooks dekodiert den Body vor dem Hashen als UTF-8, was für das JSON von Ikiform in Ordnung ist. Die Version mit der Standardbibliothek hasht die Bytes direkt, hängt also von keiner Dekodierung ab.

Prüfen, ob es funktioniert

Führe pytest aus. Alle vier Tests bestehen.
Starte die App hinter einer öffentlichen https:-URL und füge sie im Dashboard unter Webhooks hinzu.
Setze IKIFORM_WEBHOOK_SECRET auf das whsec_-Secret des Endpunkts und starte den Prozess neu.
Wähle Send test. Das Zustellungsprotokoll zeigt 204, und dein Anwendungslog zeigt das endpoint.test-Event mit seiner dlv_-ID.
Sende das Formular einmal mit einem Namen mit Akzent ab, etwa Zoë, und prüfe, dass er unversehrt ankommt.

Stolperfallen

Problem	Ursache und Lösung
FastAPI-Handler scheitert mit einem Body-Modell immer	Das Modell hat den Body zuerst geparst. Nimm `Request`, verifiziere `await request.body()` und dann `Model.model_validate_json(body)`
`403 Forbidden` von Django	CSRF-Middleware. Füge `@csrf_exempt` nur zur Webhook-View hinzu
`missing_headers` in der Version mit der Standardbibliothek	Ein einfaches `dict` mit Headern hat die Großschreibung des Servers behalten. Nutze das Header-Objekt des Frameworks, das Groß- und Kleinschreibung ignoriert, oder schreib die Keys klein
`binascii.Error: Incorrect padding` beim Start	Das Secret wurde ohne sein abschließendes `=` kopiert. Kopiere den ganzen `whsec_`-Wert erneut
`timestamp_too_old` bei jeder Zustellung	Die Serveruhr geht mehr als 5 Minuten falsch. Synchronisiere sie per NTP; die Signatur selbst ist vielleicht in Ordnung

Fragen

Hängst du bei einem Schritt fest oder willst eine Anleitung für deinen Fall? Schreib an hello@ikiform.com.