TryQuickToolBox

Webhook解説：署名付きペイロードとリトライ

Backend2026-11-08TryQuickToolBox

決済プロバイダーのWebhookを統合したばかりだとします。ドキュメントには「決済が成功したらエンドポイントにPOSTします」と書いてあります。簡単ですよね？ しかし、重複イベント、通知の欠落、偽造リクエストに関するセキュリティアラートに気づくでしょう。Webhookは概念としてはシンプルですが、本番環境では厄介です。この記事では、署名付きペイロードとリトライを使用してWebhookを正しく実装する方法を説明し、信頼性が高く安全な統合を構築できるようにします。

Webhookとは？

Webhookは逆APIです。アプリケーションがサービスに更新をポーリングする代わりに、サービスがHTTP POSTを介してイベントをエンドポイントにプッシュします。効率的でほぼリアルタイムです。一般的な用途には、決済通知、CI/CDトリガー、チャット統合などがあります。

しかし、Webhookには課題があります。リクエストが本当にプロバイダーから来たものかどうかをどうやって確認しますか？ サーバーがダウンしていたらどうしますか？ 同じイベントが2回届いたらどうしますか？ これらを1つずつ解決していきましょう。

署名付きペイロードでWebhookを保護する

誰でもエンドポイントにHTTPリクエストを送信できます。検証がなければ、攻撃者がイベントを偽造し、詐欺やデータ破損を引き起こす可能性があります。標準的な解決策は、共有シークレットでペイロードに署名することです。

HMAC署名の仕組み

プロバイダーとあなたは共有の秘密鍵を持ちます。各イベントについて、プロバイダーはその秘密鍵を使用してリクエストボディのHMAC（Hash-based Message Authentication Code）を計算し、署名をヘッダー（例：X-Signature）で送信します。あなたは自分の側でHMACを再計算して比較します。一致すれば、リクエストは本物で改ざんされていません。

Flaskを使用したPythonの最小限の例を次に示します。

import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = b'your_shared_secret'

@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Signature')
    if not signature:
        abort(400, 'Missing signature')
    body = request.get_data()
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        abort(401, 'Invalid signature')
    # Process the event
    return '', 200

タイミング攻撃を防ぐために、常にhmac.compare_digestを使用して定数時間比較を行ってください。また、不要な処理を避けるために、JSONボディを解析する前に署名を検証してください。

署名のベストプラクティス

リトライで障害に対処する

プロバイダーの観点から見ると、Webhookはファイア・アンド・フォーゲットです。しかし、あなたのエンドポイントが一時的に利用できない場合があります。配信を確実にするために、プロバイダーはリトライポリシーを実装しています。受信者として、リトライと重複に備える必要があります。

プロバイダーが使用するリトライ戦略

ほとんどのプロバイダーは指数バックオフでリトライします。たとえば、即時、その後1分後、5分後、30分後など、最大試行回数（多くの場合5〜10回）までです。リトライスケジュールを設定できるプロバイダーもあります。正確な動作は異なるため、プロバイダーのドキュメントを読んでください。

リトライ用にエンドポイントを設計する

エンドポイントは次のようにする必要があります。

Pythonでの冪等性チェックの例：

processed_events = set()  # In production, use a database or cache

@app.route('/webhook', methods=['POST'])
def webhook():
    # ... signature verification ...
    event = request.json
    event_id = event['id']
    if event_id in processed_events:
        return '', 200  # Already processed
    # Process event
    processed_events.add(event_id)
    return '', 200

Webhook実装のテスト

本番環境に移行する前に、エンドポイントを徹底的にテストしてください。

  1. 署名付きリクエストをシミュレートする：有効な署名と無効な署名を生成する単体テストを書きます。
  2. リトライシナリオをテストする：障害をシミュレートし、エンドポイントが重複を処理できることを確認します。
  3. プロバイダーのツールを使用する：多くのプロバイダーは、テストイベントを送信するためのCLIまたはダッシュボードを提供しています。
  4. 監視とログ：受信イベントをログに記録し（機密データは除く）、署名の失敗や高いエラー率に対するアラートを設定します。

比較：ポーリング vs Webhook

側面ポーリングWebhook
レイテンシ間隔に依存（例：分単位）ほぼリアルタイム
サーバー負荷高い（頻繁なリクエスト）低い（イベント時のみ）
複雑さ実装が簡単署名検証とリトライ処理が必要
信頼性リトライを自分で制御プロバイダーがリトライを制御。重複を処理する必要あり

FAQ

Webhook署名を検証するにはどうすればよいですか？

共有秘密鍵を使用して生のリクエストボディのHMACを計算し、定数時間比較関数を使用して署名ヘッダーと比較します。ペイロードを処理する前に必ず検証してください。

Webhookを確認応答するためにエンドポイントは何を返すべきですか？

イベントを受信して処理のためにキューに入れたらすぐに、2xxステータスコード（例：200 OK）を返します。タイムアウトを防ぐために、リクエストハンドラー内での長時間の処理は避けてください。

重複したWebhook配信を処理するにはどうすればよいですか？

イベントペイロードの冪等性キー（または一意のイベントID）を使用して、処理済みイベントを追跡します。これらのIDをデータベースまたはキャッシュに保存し、IDが以前に確認された場合は処理をスキップします。

Webhookペイロードをテストする準備はできましたか？ JSON Formatterを使用して、受信するWebhookのJSONボディを整形して検証し、デバッグを容易にしましょう。