Bagaimana cara mendapatkan body request mentah di FastAPI?

Terima objek Request sebagai parameter dan panggil await request.body(). Jangan deklarasikan model Pydantic sebagai parameter body untuk route webhook; model itu mem-parsing JSON sebelum Anda bisa memverifikasinya.

Apakah paket Python standardwebhooks berfungsi dengan Ikiform?

Ya. Instal dengan pip install standardwebhooks dan panggil Webhook(secret).verify(body, headers) dengan secret whsec_ dari dashboard. Fungsi itu memunculkan WebhookVerificationError saat pemeriksaan gagal.

Mengapa memakai hmac.compare_digest alih-alih ==?

== bisa langsung mengembalikan hasil begitu satu karakter berbeda, yang membocorkan informasi waktu tentang seberapa banyak tebakan yang benar. compare_digest memakan waktu yang sama apa pun hasilnya.

Apakah saya butuh csrf_exempt untuk view webhook Django?

Ya. Middleware CSRF Django menolak post tanpa token CSRF, dan Ikiform tidak mengirimnya. Pemeriksaan tanda tangan menjadi perlindungan view itu sebagai gantinya.

Mengapa saya mendapat TypeError tentang bytes dan str saat menghitung HMAC?

hmac.new butuh bytes untuk key maupun pesan. Decode secret dengan base64.b64decode, yang mengembalikan bytes, dan susun pesan sebagai f"{msg_id}.{timestamp}.".encode() + body, dengan body berupa byte mentah dari request.

Verifikasi tanda tangan webhook

25 Oktober 2026Verifikasi tanda tangan webhook5 menit baca

Verifikasi tanda tangan webhook di Python

Verifikasi tanda tangan webhook Ikiform di Flask, FastAPI, dan Django, dengan paket standardwebhooks atau standard library, ditambah test pytest.

Preet SutharPendiri, Ikiform

Baca body request sebagai byte sebelum ada yang mem-parsing-nya: request.get_data() di Flask, await request.body() di FastAPI, request.body di Django. Berikan byte itu dan header ke Webhook(secret).verify() dari paket standardwebhooks, atau hitung HMAC-SHA256 sendiri dengan hmac dan bandingkan dengan hmac.compare_digest. Jawab 400 saat gagal dan parse JSON hanya setelah lolos.

Tiga framework, dua cara verifikasi. Flask dan Django memakai paket standardwebhooks; FastAPI memakai function standard library dengan suite pytest yang dibangun di atas test vector. Campur sesuka Anda: function dan paket itu bisa saling menggantikan.

Apa yang berbeda di Python

Framework Python memberi Anda body dalam bentuk yang tampak mentah tetapi terkadang tidak. FastAPI yang paling berisiko: deklarasikan model Pydantic sebagai parameter dan FastAPI membaca, mem-parsing, dan memvalidasi body sebelum function Anda berjalan. Anda mendapat model, dan byte yang ditandatangani Ikiform tidak pernah sampai ke tangan Anda. Jebakan yang sama ada di Flask dengan request.get_json() dan Django REST framework dengan request.data. Di Starlette dan FastAPI body adalah stream yang dibaca dan di-cache oleh await request.body(), jadi panggil itu sebelum ada yang mengiterasi request.stream().

Validasi dengan Pydantic setelah verifikasi, jika Anda menginginkan model: Event.model_validate_json(body).

Flask dengan standardwebhooks

Instal paketnya dengan 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() mengembalikan body sebagai byte. Flask meng-cache-nya, jadi memanggil get_json() sesudahnya tetap berfungsi. Jangan berikan as_text=True: men-decode lebih dulu adalah langkah yang nantinya harus Anda balikkan dengan persis.
Webhook(secret) menerima secret whsec_ lengkap dan menghapus awalannya sendiri. Buat sekali saat import, bukan per request.
verify mengubah nama header menjadi huruf kecil, jadi dict(request.headers) berfungsi apa pun huruf yang dipakai proxy. Fungsi ini memeriksa timestamp, setiap tanda tangan v1, dan mengembalikan JSON yang sudah di-parse.

FastAPI dengan 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)

Handler menerima Request, bukan model, dan membaca await request.body() lebih dulu.
request.headers tidak peka huruf besar-kecil, jadi .get("webhook-id") menemukan header bagaimanapun ia dikirim.
hmac.compare_digest membandingkan byte hasil decode secara constant-time. validate=True membuat tanda tangan yang cacat menjadi tidak cocok, alih-alih nilai parsial yang di-decode diam-diam.
BackgroundTasks berjalan setelah 204 terkirim, jadi pekerjaan lambat tidak dihitung dalam waktu tunggu 15 detik Ikiform.
Type hint butuh Python 3.10 atau lebih baru.

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

Jalankan pytest. NOW yang tetap menjaga timestamp vector di dalam jendela 5 menit. json.dumps(indent=2) adalah cara realistis body berubah di codebase Python, jadi kasus itu layak dipertahankan.

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 adalah byte mentah. @csrf_exempt wajib, karena Ikiform tidak punya token CSRF dan Django akan menolak post dengan 403; pemeriksaan tanda tangan menggantikan perlindungan CSRF untuk view yang satu ini. Baca request.body sebelum menyentuh request.POST, yang hanya relevan untuk form post tetapi bisa menghabiskan stream.

Event bertipe setelah verifikasi

Pydantic tetap berguna, satu langkah kemudian. Verifikasi byte-nya, lalu validasi byte yang sama ke dalam model. Anda mendapat akses bertipe ke payload dan error yang jelas jika field yang Anda andalkan hilang, tanpa melepaskan pemeriksaan tanda tangan:

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 mengabaikan key tambahan secara default, jadi field baru yang ditambahkan Ikiform ke payload tidak merusak model. Panggil parse_submission(body) hanya untuk submission.created; event endpoint.test punya bentuk data yang berbeda.

Melewati pengulangan

Pengiriman yang terverifikasi bisa tiba dua kali saat Ikiform tidak mendapat jawaban Anda dalam 15 detik dan mencoba ulang dengan webhook-id yang sama. Di Django, kolom unik dan IntegrityError membuat yang kedua menjadi 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

Panggil first_delivery(event["id"]) setelah verifikasi; jika mengembalikan False, jawab 204 dan berhenti. Blok atomic mencegah insert yang gagal merusak sisa transaksi request. Dengan SQLAlchemy, insert(...).on_conflict_do_nothing() melakukan hal yang sama di PostgreSQL dan SQLite.

verify dari paket standardwebhooks men-decode body sebagai UTF-8 sebelum hashing, yang tidak masalah untuk JSON Ikiform. Versi standard library meng-hash byte secara langsung, jadi tidak bergantung pada decoding apa pun.

Pastikan berfungsi

Jalankan pytest. Keempat test lolos.
Jalankan aplikasi di balik URL https: publik dan tambahkan di Webhooks di dashboard.
Atur IKIFORM_WEBHOOK_SECRET ke secret whsec_ milik endpoint dan restart proses.
Pilih Send test. Log pengiriman menampilkan 204, dan log aplikasi Anda menampilkan event endpoint.test dengan id dlv_-nya.
Kirim formulir sekali dengan nama beraksen, seperti Zoë, dan pastikan nama itu tiba utuh.

Hal yang sering menjebak

Masalah	Penyebab dan solusi
Handler FastAPI selalu gagal dengan model body	Model mem-parsing body lebih dulu. Terima `Request`, verifikasi `await request.body()`, lalu `Model.model_validate_json(body)`
`403 Forbidden` dari Django	Middleware CSRF. Tambahkan `@csrf_exempt` hanya ke view webhook
`missing_headers` di versi standard library	`dict` header biasa mempertahankan kapitalisasi dari server. Gunakan objek header framework yang tidak peka huruf besar-kecil, atau ubah key menjadi huruf kecil
`binascii.Error: Incorrect padding` saat startup	Secret disalin tanpa `=` di akhirnya. Salin lagi seluruh nilai `whsec_`
`timestamp_too_old` di setiap pengiriman	Jam server meleset lebih dari 5 menit. Sinkronkan dengan NTP; tanda tangannya sendiri mungkin baik-baik saja

Pertanyaan

Mentok di salah satu langkah, atau ingin panduan untuk situasi Anda? Email hello@ikiform.com.