FastAPI で生のリクエストボディを取得するには？

Request オブジェクトを引数に取り、await request.body() を呼びます。Webhook のルートでは Pydantic モデルをボディの引数として宣言しないでください。検証する前に JSON を解析してしまいます。

Python の standardwebhooks パッケージは Ikiform で使えますか？

はい。pip install standardwebhooks でインストールし、ダッシュボードの whsec_ のシークレットで Webhook(secret).verify(body, headers) を呼びます。確認に失敗すると WebhookVerificationError を送出します。

== ではなく hmac.compare_digest を使うのはなぜ？

== は 1 文字でも違えばすぐに返ることがあり、推測がどこまで合っていたかというタイミング情報が漏れます。compare_digest は常に同じ時間がかかります。

Django の Webhook ビューに csrf_exempt は必要ですか？

はい。Django の CSRF ミドルウェアは CSRF トークンのない送信を拒否し、Ikiform はトークンを送りません。代わりに署名のチェックがビューを守ります。

HMAC の計算で bytes と str に関する TypeError が出るのはなぜ？

hmac.new はキーとメッセージの両方に bytes を必要とします。シークレットは bytes を返す base64.b64decode でデコードし、メッセージは f"{msg_id}.{timestamp}.".encode() + body として組み立てます。body はリクエストの生のバイト列です。

Webhook の署名を検証する

2026年10月25日Webhook の署名を検証する2 分で読めます

Python で Webhook の署名を検証する

Flask、FastAPI、Django で Ikiform の Webhook 署名を検証する方法。standardwebhooks パッケージか標準ライブラリを使い、pytest のテストも付けます。

Preet SutharIkiform 創業者

何かが解析する前に、リクエストのボディをバイト列として読みます。Flask なら request.get_data()、FastAPI なら await request.body()、Django なら request.body です。そのバイト列とヘッダーを standardwebhooks パッケージの Webhook(secret).verify() に渡すか、hmac で HMAC-SHA256 を自分で計算して hmac.compare_digest で比較します。失敗したら 400 を返し、成功してから初めて JSON を解析します。

3 つのフレームワークと、2 つの検証方法です。Flask と Django は standardwebhooks パッケージを使い、FastAPI は標準ライブラリの関数と、テストベクターに基づく pytest のテストを使います。関数とパッケージは置き換え可能なので、好きなように組み合わせてください。

Python ならではの違い

Python のフレームワークは、生に見えて実はそうでないこともある形でボディを渡してきます。特に注意が必要なのは FastAPI です。Pydantic モデルを引数に宣言すると、関数が動く前に FastAPI がボディを読み、解析し、検証します。手に入るのはモデルで、Ikiform が署名したバイト列には一切触れられません。同じ罠は Flask の request.get_json() や Django REST framework の request.data にもあります。Starlette と FastAPI ではボディはストリームで、await request.body() がそれを読んでキャッシュするので、何かが request.stream() を反復する前に呼んでください。

モデルが欲しいなら、検証の後で Pydantic を使います：Event.model_validate_json(body)。

Flask と standardwebhooks

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() はボディをバイト列で返します。Flask はそれをキャッシュするので、後から get_json() を呼んでも動きます。as_text=True は渡さないでください。先にデコードすると、後でそれを正確に元に戻す手間が増えるだけです。
Webhook(secret) は whsec_ 付きのシークレット全体を受け取り、プレフィックスを自分で取り除きます。リクエストごとではなく、インポート時に 1 回だけ作ってください。
verify はヘッダー名を小文字にするので、プロキシがどの大文字小文字を使っても dict(request.headers) で動きます。タイムスタンプとすべての v1 署名を確認し、解析済みの JSON を返します。

FastAPI と標準ライブラリ

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)

ハンドラーはモデルではなく Request を受け取り、最初に await request.body() を読みます。
request.headers は大文字小文字を区別しないので、どう送られても .get("webhook-id") でヘッダーが見つかります。
hmac.compare_digest はデコードしたバイト列を定数時間で比較します。validate=True により、形式の崩れた署名は黙って一部だけデコードされるのではなく、一致しないものとして扱われます。
BackgroundTasks は 204 を送った後に動くので、時間のかかる処理が Ikiform の 15 秒の待ち時間に含まれません。
型ヒントには Python 3.10 以降が必要です。

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)

pytest を実行します。固定した NOW によって、テストベクターのタイムスタンプが 5 分の範囲内に収まります。json.dumps(indent=2) は Python のコードベースでボディが変わってしまう現実的な例なので、このケースには入れる価値があります。

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 は生のバイト列です。Ikiform は CSRF トークンを持たず、Django は送信を 403 で拒否するので、@csrf_exempt が必要です。このビューに限っては、署名の確認が CSRF 対策の代わりになります。request.POST に触れる前に request.body を読んでください。これはフォーム送信のときにしか関係しませんが、ストリームを消費してしまうことがあります。

検証後に型付きのイベントを扱う

Pydantic は 1 ステップ後でも役に立ちます。バイト列を検証してから、同じバイト列をモデルとして検証します。署名の確認をあきらめずに、ペイロードに型付きでアクセスでき、頼りにしているフィールドが欠けていれば明確なエラーが得られます。

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 はデフォルトで余分なキーを無視するので、Ikiform がペイロードに新しいフィールドを追加してもモデルは壊れません。parse_submission(body) は submission.created のときだけ呼んでください。endpoint.test イベントは data の形が違います。

重複をスキップする

Ikiform が 15 秒以内に応答を受け取れず、同じ webhook-id でリトライした場合、検証済みの配信が 2 回届くことがあります。Django では、一意の列と IntegrityError で 2 回目を何もしない処理にできます。

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

検証のあとに first_delivery(event["id"]) を呼び、False が返ったら 204 を返して終了します。atomic ブロックがあるので、挿入に失敗してもリクエストの残りのトランザクションは壊れません。SQLAlchemy では、PostgreSQL と SQLite で insert(...).on_conflict_do_nothing() が同じ働きをします。

standardwebhooks パッケージの verify はハッシュ化の前にボディを UTF-8 としてデコードしますが、Ikiform の JSON なら問題ありません。標準ライブラリ版はバイト列を直接ハッシュ化するので、どんなデコードにも依存しません。

動作を確認する

pytest を実行します。4 つのテストがすべて通ります。
公開された https: の URL でアプリを起動し、ダッシュボードの Webhooks に追加します。
IKIFORM_WEBHOOK_SECRET にエンドポイントの whsec_ のシークレットを設定し、プロセスを再起動します。
Send test を選びます。配信ログに 204 が表示され、アプリケーションのログには dlv_ の ID 付きで endpoint.test イベントが記録されます。
Zoë のようなアクセント付きの名前でフォームを 1 回送信し、そのまま届くことを確認します。

ハマりどころ

問題	原因と対処
ボディのモデルを使うと FastAPI のハンドラーが常に失敗する	モデルが先にボディを解析しています。`Request` を受け取り、`await request.body()` を検証してから `Model.model_validate_json(body)` を呼んでください
Django から `403 Forbidden`	CSRF ミドルウェアです。Webhook のビューにだけ `@csrf_exempt` を付けてください
標準ライブラリ版で `missing_headers`	ヘッダーを普通の `dict` にしたため、サーバーの大文字小文字がそのまま残っています。フレームワークの大文字小文字を区別しないヘッダーオブジェクトを使うか、キーを小文字にしてください
起動時に `binascii.Error: Incorrect padding`	シークレットが末尾の `=` なしでコピーされています。`whsec_` の値全体をコピーし直してください
すべての配信で `timestamp_too_old`	サーバーの時計が 5 分以上ずれています。NTP で同期してください。署名自体は正しいかもしれません

よくある質問

手順で詰まったときや、あなたの状況に合うガイドがほしいときはメールでどうぞ: hello@ikiform.com