Webhookを安全に受信する設計｜署名検証・再送・冪等性の実務ルール

2026-09-24

Webhookは、あるシステムで起きたイベントを別のシステムへHTTPで通知する仕組みです。定期的にAPIを巡回するポーリングを減らせますが、単にPOSTを受け取るだけでは安全な連携になりません。

受信側には、送信元の確認、改ざん検知、再送による二重処理の防止、順序が逆転した場合の処理、スキーマ変更への対応が必要です。この記事では、Webhookを「通知用HTTPリクエスト」ではなく「再送され得るイベントデータ」として設計します。

Webhookで解決すること

Webhookは、図面承認、部品更新、決済完了、計算終了などを送信側から通知します。ただし、完全な同期データではなく「変化が起きた」という合図として、受信後に送信元APIから最新状態を取得する構成もあります。

ネットワーク障害や再試行により、同じイベントが複数回来る可能性を前提にします。発生順に届くとも限りません。Stripeの公式資料も、重複への備えと配信順が保証されないことを明記しています。

イベントを共通エンベロープで表す

業務データだけを直接送るより、識別・検証・版管理に必要な情報を共通部分へ分けると扱いやすくなります。

{
  "event_id": "evt_01K5W8M3P4",
  "event_type": "drawing.released",
  "occurred_at": "2026-09-24T01:30:00Z",
  "schema_version": 1,
  "data": {
    "drawing_id": "dwg_000184",
    "revision": "C",
    "status": "released"
  }
}

このobjectでは、event_idなどがfield、右側がvalueです。schema_versionはnumber、dataは業務データを持つobjectです。

event_id：配信をまたいで変わらないイベント識別子
event_type：受信側が処理を選ぶための種類
occurred_at：送信元でイベントが発生した日時
schema_version：payload契約の版
data：対象IDとイベント固有の値

再送のたびにevent_idを作り直してはいけません。配信試行IDや対象のdrawing_idとも役割を分けます。

受信処理を短い同期部分と非同期処理に分ける

安定した受信経路は、次の順序で組み立てます。

HTTPSで受信
  ↓
本文サイズ・Content-Typeを確認
  ↓
生の本文と署名headerを検証
  ↓
JSONを解析し、スキーマとevent_typeを検証
  ↓
event_idを一意キーとして永続化
  ↓
速やかに2xxを返す
  ↓
キューまたはworkerで業務処理
  ↓
成功・再試行・隔離を記録

重要なのは、署名検証より前にJSONを整形しないことです。空白、改行、keyの順序が変わると、意味が同じJSONでもbyte列は変わります。GitHubとStripeの公式資料も、署名検証には変更前の本文を使うよう説明しています。

重い処理の完了までHTTP応答を待たせると、送信側がタイムアウトして再送する可能性が高まります。検証済みイベントを永続化した時点で成功応答し、業務処理は分離します。成功status codeや再送間隔は送信元の仕様を確認します。

HMAC署名で送信元と本文を検証する

RFC 2104のHMACは、共有秘密鍵とhash関数でメッセージ認証値を計算します。受信側は、本文が想定した送信元から届き、途中で変更されていないかを検証できます。

HMAC署名は暗号化ではありません。payload自体は読めるため、通信にはHTTPSを使い、機密情報を必要以上に含めない設計が必要です。

署名対象、header名、時刻の連結方法、hexかBase64かはサービスごとに異なります。次の例は、独自Webhook契約を説明するための例です。

X-Webhook-Timestamp: 1790213400
X-Webhook-Signature: sha256=<hex digest>

署名対象 = timestamp + "." + raw_body
署名方式 = HMAC-SHA256

受信側は、必ず契約どおりのbyte列で再計算します。JSON parse後のobjectを再度serializeした文字列では検証しません。

Pythonで署名と再送時刻を検証する

Python標準ライブラリのhmacとhashlibで、最小の検証関数を作れます。

import hashlib
import hmac
import json
import time


def verify_webhook(
    raw_body: bytes,
    timestamp_header: str,
    signature_header: str,
    secret: bytes,
    *,
    now: int | None = None,
    tolerance_seconds: int = 300,
) -> dict:
    try:
        sent_at = int(timestamp_header)
    except (TypeError, ValueError) as exc:
        raise ValueError("timestamp headerが不正です") from exc

    current = int(time.time()) if now is None else now
    if abs(current - sent_at) > tolerance_seconds:
        raise ValueError("許容時間外の配信です")

    prefix = "sha256="
    if not signature_header.startswith(prefix):
        raise ValueError("署名方式が不正です")

    signed = timestamp_header.encode("ascii") + b"." + raw_body
    expected = hmac.new(secret, signed, hashlib.sha256).hexdigest()
    received = signature_header[len(prefix):]

    if not hmac.compare_digest(expected, received):
        raise ValueError("署名が一致しません")

    try:
        return json.loads(raw_body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("UTF-8のJSONとして解析できません") from exc

hmac.compare_digest()は、通常の==よりtiming attackを受けにくい比較に使います。Python公式資料とGitHubの例でも案内されています。

時刻の許容幅はreplay attackの範囲を狭めます。ただし、正規の再送に新しい署名時刻が付く場合もあるため、event_idでも重複排除します。サーバー時計を同期し、許容秒数は送信元仕様に合わせます。

SQLiteで受信と処理を分離する

受信したイベントは、業務処理の前に一意制約のあるtableへ保存します。

CREATE TABLE webhook_events (
    event_id          TEXT PRIMARY KEY NOT NULL,
    event_type        TEXT NOT NULL,
    schema_version    INTEGER NOT NULL,
    occurred_at       TEXT NOT NULL,
    payload_json      TEXT NOT NULL,
    processing_status TEXT NOT NULL
        CHECK (processing_status IN ('pending', 'processing', 'succeeded', 'failed')),
    received_at       TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    processed_at      TEXT
);

登録時はevent_idの重複をDB制約へ任せます。

import json
import sqlite3


def store_event(conn: sqlite3.Connection, event: dict) -> bool:
    cursor = conn.execute(
        """
        INSERT OR IGNORE INTO webhook_events (
            event_id, event_type, schema_version,
            occurred_at, payload_json, processing_status
        ) VALUES (?, ?, ?, ?, ?, 'pending')
        """,
        (
            event["event_id"],
            event["event_type"],
            event["schema_version"],
            event["occurred_at"],
            json.dumps(event, ensure_ascii=False, separators=(",", ":")),
        ),
    )
    conn.commit()
    return cursor.rowcount == 1

初回はTrue、同じevent_idの再送はFalseになります。重複でも署名が正しく、既に保存済みなら成功応答を返せます。そうしないと送信側が再送を続ける場合があります。

検索後のINSERTだけでは同時受信で競合します。PRIMARY KEYまたはUNIQUE制約を最終防衛線にします。

スキーマ検証は署名検証の後に行う

署名が正しくても、必要fieldが欠けたイベントを処理してはいけません。少なくとも次を検証します。

payload全体がobjectである
event_id、event_type、occurred_at、schema_version、dataが存在する
各fieldの型と文字数が契約内である
event_typeがallowlistに含まれる
schema_versionを受信側が処理できる
data内の対象IDや状態値が業務ルールに合う

署名欠落、本文改変、期限切れ、未知のtype、未対応version、同一IDの再送、順序逆転もテストします。

順序逆転と差分更新に備える

drawing.releasedの後に古いdrawing.reviewedが届くと、単純な上書きでは状態が巻き戻ります。発生日時だけで順序を決めるのも安全ではありません。複数イベントが同時刻になり得るうえ、時計差もあるためです。

順序が重要なら、送信側が対象ごとの単調増加versionまたはsequenceを発行し、受信側は現在versionより新しい場合だけ反映します。送信元APIが正本なら、Webhookを更新命令ではなく取得の契機として扱い、対象IDから最新状態を再取得する方法もあります。

再処理に備えてschema_versionと元payloadを保持します。変換処理にもversionを付ければ、DBへ反映した規則を追跡できます。

失敗時の扱いを決める

エラーを一つの再試行ループへ入れると、永久に成功しないデータが滞留します。

一時的なDB障害・外部API障害：回数と間隔を制限して再試行
未対応version・必須field欠落：隔離して通知
署名不一致・期限切れ：処理せず監査ログへ記録
重複event ID：業務処理をせず受信済みとして扱う
workerの処理失敗：failedと試行回数を記録し、手動再開可能にする

2xxは「最終処理が完了した」ではなく、「安全に受け付けた」を表す設計にできます。

秘密情報と公開データを分離する

Webhook secretをpayload、ソースコード、通常ログへ含めてはいけません。環境変数やsecret管理機能から取得し、送信元・環境・endpointごとに分けます。

鍵の移行期間だけ旧鍵と新鍵を検証する手順やkey_idを用意します。ログにはsecretや本文全体を出さず、event IDや判定理由だけを残します。

IP allowlistだけに依存せず、署名検証、HTTPS、request size上限、rate limit、認可を組み合わせます。Webhook secretは業務APIのcredentialと分離します。

CAD・DB・自動化へ再利用する設計

設計データでは巨大なCAD fileや部品表を直接含めず、対象ID、revision、変更種別を渡し、権限のあるAPIから正本を取得する方法があります。

共通エンベロープ、署名契約、受信table、処理statusを再利用すれば、drawing.released、part.updated、calculation.completedなどを同じ基盤へ載せられます。業務ごとの差はevent_typeとdataのschemaへ閉じ込めます。