Webhook署名を検証する: Node、Python、GoでのHMAC-SHA256

Webhook署名を検証するには、送信者が署名したバイト列とまったく同じ内容に、共有しているシークレットを使ってHMACを再計算し、値と署名ヘッダーを定数時間で比較します。続いて、署名されたタイムスタンプが新しいことを確認します。ElidoのWebhookでは、全体の whsec_... シークレットを鍵にしたHMAC-SHA256を、{X-Webhook-Timestamp}.{raw body} に対して計算し、v1= を付けた16進数にして X-Elido-Signature と照合します。

アルゴリズム全体はこれだけです。失敗の原因は、その周辺の細部にあります。bodyパーサーが早すぎる段階で実行された、シークレットがデコードされた、16進数のダイジェストをbase64のものと比較した、といった問題です。この記事では、HMACによるWebhook署名検証の一般的な仕組みから、Node、Python、Go、n8n Codeノードで動作するElidoのコード、さらに多くのクイックスタートで省略されるリプレイウィンドウとシークレットローテーションまで説明します。

まだエンドポイントを設定していない場合は、すべてのイベント種別とペイロードのエンベロープを一覧にしたリンクイベント向けWebhookから始めてください。このページでは、署名付きリクエストがサーバーに届いた後の処理を扱います。

HMAC Webhook署名の仕組み

Webhookエンドポイントは公開URLです。見つけた人は誰でも本物のイベントに見えるJSON bodyをPOSTできるため、受信側には送信元の証明が必要です。HMACなら低コストで実現できます。送信者と受信者がシークレットを共有し、送信者が HMAC-SHA256(secret, message) を計算してヘッダーに入れ、受信者も同じ計算をして比較します。シークレットがなければ一致する値を生成できず、メッセージの1バイトを変更しただけでもダイジェスト全体が変わります。

プロバイダーによって異なる重要な点は3つあり、間違えるとそれぞれ検証に失敗します。

1. メッセージに含めるもの。 GitHubは生のbodyだけに署名します。StripeとElidoはタイムスタンプ、ドット、bodyに署名します。Standard Webhooks仕様はメッセージID、タイムスタンプ、bodyに署名します。
2. ダイジェストのエンコード方法。 v1= や sha256= のような方式プレフィックスを付けた16進数またはbase64です。
3. 鍵の内容。 プロバイダーによってはプレフィックスの後のシークレットをbase64デコードします。Elidoはデコードしません。鍵は完全な whsec_ 文字列をUTF-8バイト列にしたものです。

署名するメッセージにタイムスタンプを含めることが重要です。これにより、攻撃者が過去に有効な署名を得たbodyと新しいタイムスタンプヘッダーを組み合わせられなくなり、リプレイウィンドウを適用できるようになります。

Elidoが署名する内容と、それを運ぶヘッダー

event または siem エンドポイントへのすべての配信は、Content-Type: application/json を持ち、{"type", "workspace_id", "data", "timestamp"} のようなbodyを含むPOSTです。チャット型のエンドポイント種別（Discord、Telegram、Sentry）はURLで認証し、HMACヘッダーを持ちません。以下は最初の2種類にだけ適用されます。

シークレットはエンドポイント作成時に生成されます。whsec_ に64文字の16進数が続く形式です。作成レスポンスで一度だけ返され、その後は二度と返されないため、すぐにシークレットストアへ保存してください。

コードで実行できるテストベクトルを示します。シークレットが whsec_test_only_do_not_use、タイムスタンプが 1789000000、bodyが {"type":"link.created","workspace_id":42} の場合、正しいヘッダー値は次のとおりです。

v1=b9369aa411a8b7ce705bcd5bba112dea9d72d2e787aa88959ff62f33942d1a15

シェルから再現できます。受信側と送信側で結果が食い違うとき、私はいつも最初にこれを実行します。

printf '%s.%s' 1789000000 '{"type":"link.created","workspace_id":42}' \
  | openssl dgst -sha256 -hmac 'whsec_test_only_do_not_use' -r

ここには1つ落とし穴があります。body自身の timestamp フィールドはイベントが発生した時刻です。署名対象のタイムスタンプは、リクエスト送信時に設定される X-Webhook-Timestamp ヘッダーの値です。混同しないでください。

NodeでWebhook署名を検証する

Expressでは、ほとんどの失敗を解決するのは1行です。Webhookルートに express.raw() を取り付け、req.body が受信したバイト列そのままのBufferになるようにします。このルートはグローバルな app.use(express.json()) より前に登録してください。JSONパーサーがストリームを読み取った後では、rawパーサーは読むものが残っていません。

import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";

const SECRET = process.env.ELIDO_WEBHOOK_SECRET; // the full whsec_... string
const TOLERANCE_SEC = 300;

function matches(expected, got) {
  const a = Buffer.from(expected);
  const b = Buffer.from(got ?? "");
  return a.length === b.length && timingSafeEqual(a, b);
}

const app = express();

app.post(
  "/webhooks/elido",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const ts = req.get("X-Webhook-Timestamp") ?? "";
    if (
      !/^\d+$/.test(ts) ||
      Math.abs(Date.now() / 1000 - Number(ts)) > TOLERANCE_SEC
    ) {
      return res.status(400).send("stale or missing timestamp");
    }
    const expected =
      "v1=" +
      createHmac("sha256", SECRET)
        .update(`${ts}.`)
        .update(req.body)
        .digest("hex");
    const ok = [
      req.get("X-Elido-Signature"),
      req.get("X-Elido-Signature-Previous"),
    ].some((got) => matches(expected, got));
    if (!ok) return res.status(401).send("bad signature");

    const event = JSON.parse(req.body.toString("utf8"));
    // enqueue event, then acknowledge fast
    res.sendStatus(200);
  },
);

長さのチェックは飾りではありません。長さが異なるBufferを渡すと、timingSafeEqual はfalseを返さず例外を投げます。APIとSDKのクイックスタートのTypeScript SDKを使う場合、@elido/sdk の webhooks.verify() が同じHMAC計算とタイミングセーフな比較を行います。ただし { maxSkewSec: 300 } は明示的に渡してください。このオプションがないと、タイムスタンプの経過時間をまったく確認しません。

PythonとGoでWebhook HMAC SHA256を検証する

Pythonの標準ライブラリで対応できます。FastAPIでは await request.body() が生のバイト列を返します。Flaskでは request.json に何かが触れる前に request.get_data() を呼び出してください。

import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, HTTPException, Request

SECRET = os.environ["ELIDO_WEBHOOK_SECRET"].strip().encode()
TOLERANCE = 300
app = FastAPI()


def verify(raw: bytes, ts: str, candidates: list) -> bool:
    if not ts.isdigit() or abs(time.time() - int(ts)) > TOLERANCE:
        return False
    digest = hmac.new(SECRET, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    expected = "v1=" + digest
    return any(c and hmac.compare_digest(c, expected) for c in candidates)


@app.post("/webhooks/elido")
async def elido_webhook(request: Request):
    raw = await request.body()
    h = request.headers
    sigs = [h.get("x-elido-signature"), h.get("x-elido-signature-previous")]
    if not verify(raw, h.get("x-webhook-timestamp", ""), sigs):
        raise HTTPException(status_code=401, detail="bad signature")
    event = json.loads(raw)
    # enqueue event
    return {"ok": True}

Goではbodyを一度だけ読み取り、サイズに上限を設け、定数時間で比較する crypto/hmac の hmac.Equal を使います。解析した整数を再フォーマットするのではなく、受信したヘッダー文字列をそのまま使ってメッセージを組み立てます。

package webhook

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"io"
	"net/http"
	"strconv"
	"time"
)

const toleranceSec = 300

// Verify returns the raw body when the request carries a valid Elido signature.
func Verify(w http.ResponseWriter, r *http.Request, secret []byte) ([]byte, bool) {
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
	if err != nil {
		return nil, false
	}
	tsHeader := r.Header.Get("X-Webhook-Timestamp")
	ts, err := strconv.ParseInt(tsHeader, 10, 64)
	if err != nil {
		return nil, false
	}
	if age := time.Now().Unix() - ts; age > toleranceSec || age < -toleranceSec {
		return nil, false
	}

	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(tsHeader + "."))
	mac.Write(body)
	expected := []byte("v1=" + hex.EncodeToString(mac.Sum(nil)))

	for _, name := range []string{"X-Elido-Signature", "X-Elido-Signature-Previous"} {
		if got := r.Header.Get(name); got != "" && hmac.Equal([]byte(got), expected) {
			return body, true
		}
	}
	return nil, false
}

3つの実装はいずれも、署名が不正なら 401 を返し、チェックに合格してからJSONを解析します。Elidoでは2xx以外を失敗した試行として扱うため、検証処理が間違っていると、本物の配信がエンドポイントの配信ログに401として積み上がります。デプロイ後はまずここを確認してください。

実際のトラフィックで試してみませんか？Webhook機能ページからワークスペースにエンドポイントを作成し、ローカルトンネルに向けてください。配信ログには各試行で検証処理が返したステータスコードが表示されます。

n8n Codeノード内でWebhook署名を検証する

n8nなら、追加サービスなしで同じチェックを実行できます。WebhookノードでRaw Bodyオプションを有効にすると、変更されていないリクエストがバイナリデータとして保存されます。その直後にCodeノードを追加してください。組み込みの crypto モジュールは n8n Codeノード で利用できるため、次のコードをそのまま実行できます。

const crypto = require("crypto");
const h = $input.first().json.headers;
const raw = await this.helpers.getBinaryDataBuffer(0, "data");

const ts = h["x-webhook-timestamp"] ?? "";
if (!/^\d+$/.test(ts) || Math.abs(Date.now() / 1000 - Number(ts)) > 300) {
  throw new Error("stale delivery");
}
const expected = Buffer.from(
  "v1=" +
    crypto
      .createHmac("sha256", $env.ELIDO_WEBHOOK_SECRET)
      .update(`${ts}.`)
      .update(raw)
      .digest("hex"),
);
const ok = ["x-elido-signature", "x-elido-signature-previous"].some((name) => {
  const got = Buffer.from(h[name] ?? "");
  return (
    got.length === expected.length && crypto.timingSafeEqual(got, expected)
  );
});
if (!ok) throw new Error("bad signature");
return [{ json: JSON.parse(raw.toString("utf8")) }];

エラーがスローされると実行が停止するため、後続の処理は偽造されたイベントに対して実行されません。セルフホスティングで注意が必要なのは、インスタンスに N8N_BLOCK_ENV_ACCESS_IN_NODE=true が設定されている場合です。このときCodeノードは $env を読み取れず、シークレットが空になってすべての配信がチェックに失敗します。セルフホストn8nガイドではリバースプロキシ側を、n8n URL短縮記事ではイベントが届いた後に構築できるものを説明しています。

リプレイウィンドウとシークレットローテーション

有効な署名で証明できるのは、誰がリクエストを送ったかであり、いつ送ったかではありません。ログ行や設定ミスのあるプロキシから署名済み配信を1件取得した人は、それを1週間後に再送でき、HMACはそれでも一致します。タイムスタンプのチェックがこの隙間を埋めます。これはあなたの役割です。Elidoの配信ワーカーはタイムスタンプに署名しますが、受信側でウィンドウを強制しません。私は300秒を使います。数分ずれるサーバーは正しいトラフィックを拒否し始めるため、受信側の時計をNTPと同期してください。

再試行でウィンドウに引っかかることはありません。各試行には新しい X-Webhook-Timestamp と新しい署名が付く一方、X-Webhook-Delivery は同じままです。この分担により、2つの防御を両立できます。タイムスタンプが取得されたリクエストの有効期間を制限し、配信IDへの一意なインデックスが正当な再試行の二重処理を防ぎます。レート制限と冪等性の記事では、受信API側で同じパターンを説明しています。

ローテーションは POST /v1/workspaces/{workspace_id}/webhooks/{id}/rotate-secret、またはエンドポイントページのRotateボタンから実行します。レスポンスには新しいシークレットが一度だけ含まれ、grace_window_days (7) と previous_expires_at も返されます。この7日間、各配信には2つの署名が付きます。

• X-Elido-Signature、新しいシークレットで作成
• X-Elido-Signature-Previous、古いシークレットで作成

そのため、上のすべてのコード例では、保持している1つのシークレットに対して両方のヘッダーをチェックしています。古いシークレットで動作している受信側は2つ目のヘッダーに一致し、新しいシークレットをデプロイした後は1つ目に一致します。その間に失敗は起きません。ただし、前の鍵のヘッダーは保証ではなく橋渡しとして扱い、新しいシークレットを週の早い段階で配布してください。

Webhook署名の検証が失敗する理由

誰かのデバッグを手伝うとき、ほぼ毎回同じ短い一覧になります。次の順番で確認してください。

1. JSONを再シリアライズしている。 JSON.stringify(req.body) や json.dumps(payload) が生成するバイト列は、送信者がハッシュしたものと異なります。キーの順序、空白、エスケープされたスラッシュ、Unicodeエスケープなどが違うためです。生のbodyをハッシュしてください。フレームワークがすでに解析している場合は、文字列を再構築せずミドルウェアの順序を直します。
2. 鍵のバイト列が違う。 Elidoは完全な whsec_... 文字列をHMAC鍵として使います。プレフィックスを削る、残りを16進数デコードする、base64デコードする（Standard Webhooksライブラリが行う処理）と、別の鍵になります。echo からシークレットファイルに入った末尾の改行でも同じことが起きるため、Python例では .strip() を呼び出しています。
3. エンコーディングが一致していない。 v1= と小文字の16進数をヘッダーと比較してください。base64ダイジェスト、大文字の16進数、プレフィックスの欠落は一致しません。
4. タイムスタンプが違う。 bodyの timestamp フィールドではなく、X-Webhook-Timestamp ヘッダーの文字列を使います。解析して再フォーマットした数値も使わないでください。

細かい点が2つあります。== による比較は動作しますがタイミングが漏れるため、言語が提供する定数時間比較の関数を使ってください。また、プロキシがbodyを展開したり再エンコードしたりしても検証に失敗します。ただし、通常のJSON POSTではまれです。

それでも両者の結果が一致しない場合は、タイムスタンプ、bodyの長さ、ダイジェストの先頭数文字をログに記録し、先ほどのopenssl行を同じ入力で実行します。opensslと一致した側が正しい側です。各プロバイダーで署名が他の確認すべき対策の中でどの位置にあるかは、URL短縮サービスのセキュリティチェックリストを参照してください。

基礎記事も読んでください。リンクイベント向けWebhook

ブログの関連記事

• リンクイベント向けWebhook - イベント種別、ペイロードのエンベロープ、再試行ポリシー。
• クリック計測におけるWebhookとポーリングの比較 - pushがpullに勝つ場合と、そうでない場合。
• n8nによるセルフホストのリンク自動化 - リバースプロキシ、キューモード、署名チェックを1つの構成で実現します。
• URL短縮API: レート制限、再試行、冪等性 - 逆方向から見た重複排除パターン。
• URL短縮サービスのセキュリティチェックリスト - あらゆるプロバイダーで確認すべき9つの対策。