# Python で Webhook の署名を検証する > Flask、FastAPI、Django で Ikiform の Webhook 署名を検証する方法。standardwebhooks パッケージか標準ライブラリを使い、pytest のテストも付けます。 公開日 2026-10-25 · https://ikiform.com/ja/guides/webhook-signature-verification/python 何かが解析する前に、リクエストのボディをバイト列として読みます。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 は標準ライブラリの関数と、[テストベクター](https://ikiform.com/ja/guides/webhook-signature-verification.md#test-vector)に基づく pytest のテストを使います。関数とパッケージは置き換え可能なので、好きなように組み合わせてください。 ## Python ならではの違い Python のフレームワークは、生に見えて実はそうでないこともある形でボディを渡してきます。特に注意が必要なのは FastAPI です。Pydantic モデルを引数に宣言すると、関数が動く前に FastAPI がボディを読み、解析し、検証します。手に入るのはモデルで、Ikiform が署名したバイト列には一切触れられません。同じ罠は Flask の `request.get_json()` や Django REST framework の `request.data` にもあります。Starlette と FastAPI ではボディはストリームで、`await request.body()` がそれを読んでキャッシュするので、何かが `request.stream()` を反復する前に呼んでください。 図: Pydantic のボディ引数を持つ FastAPI ハンドラーと、生のリクエストボディを読むハンドラーの比較 1. Pydantic モデルをボディ引数にすると、ハンドラーが動く前に FastAPI が JSON を解析して検証します。 2. モデルを再びシリアライズすると別のバイト列になるので、署名の検証は失敗します。 3. Request を受け取って request.body() を await すれば、送られたとおりのバイト列が手に入ります。 4. ハンドラーはそれを検証し、json.loads で解析して 204 を返し、時間のかかる処理を BackgroundTasks に渡します。 モデルが欲しいなら、検証の後で Pydantic を使います：Event.model_validate_json(body)。 ## Flask と standardwebhooks `pip install standardwebhooks` でパッケージをインストールします。 `app.py`: ```python 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`: ```python 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`: ```python 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`: ```python 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`: ```python 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`: ```python 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`: ```python 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 なら問題ありません。標準ライブラリ版はバイト列を直接ハッシュ化するので、どんなデコードにも依存しません。 ## 動作を確認する 1. `pytest` を実行します。4 つのテストがすべて通ります。 2. 公開された `https:` の URL でアプリを起動し、ダッシュボードの **Webhooks** に追加します。 3. `IKIFORM_WEBHOOK_SECRET` にエンドポイントの `whsec_` のシークレットを設定し、プロセスを再起動します。 4. **Send test** を選びます。配信ログに `204` が表示され、アプリケーションのログには `dlv_` の ID 付きで `endpoint.test` イベントが記録されます。 5. `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 で同期してください。署名自体は正しいかもしれません | ## よくある質問 ### 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 の署名を検証する](https://ikiform.com/ja/guides/webhook-signature-verification.md): フォームの Webhook が本当に Ikiform から来たかを確認する方法。Standard Webhooks の HMAC-SHA256 方式、テストベクター、生のボディ、タイムスタンプ、シークレットのローテーションを解説します。 - [Node.js](https://ikiform.com/ja/guides/webhook-signature-verification/nodejs.md): Node.js で Ikiform の Webhook 署名を検証する方法。Express と Fastify での生のボディ、SDK、standardwebhooks パッケージ、テスト済みの node:crypto 版。 - [Go の net/http](https://ikiform.com/ja/guides/webhook-signature-verification/go.md): Go の net/http で Ikiform の Webhook 署名を検証する方法。ボディを 1 回だけ読み、crypto/hmac で確認してテストするか、Standard Webhooks ライブラリを使います。 - [PHP](https://ikiform.com/ja/guides/webhook-signature-verification/php.md): Ikiform の Webhook 署名を、素の PHP では hash_hmac と hash_equals で、Laravel では Standard Webhooks ライブラリと CSRF の除外で検証します。 - [Ruby](https://ikiform.com/ja/guides/webhook-signature-verification/ruby.md): Ikiform の Webhook 署名を、Sinatra では OpenSSL と Minitest のテストで、Rails では standardwebhooks gem と request.raw_post で検証します。