配信の検証
`verify_webhook_signature`: 一定時間での比較、リプレイ許容ウィンドウ、そしてパース済みイベントの返却。
リクエストハンドラーでの使い方
Webhook の URL は公開されている。インターネット上の誰でも、正しい形の JSON をそこへ POST できる。したがって、署名を検証せずに payload['type'] を読むハンドラーは、誰でも書き込める API にほかならない。
import os from fastapi import FastAPI, Request, Responsefrom openemail import WebhookVerificationError, verify_webhook_signature app = FastAPI() @app.post('/webhooks/openemail')async def webhook(request: Request) -> Response: try: event = verify_webhook_signature( payload=await request.body(), headers=request.headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], tolerance_seconds=300, ) except WebhookVerificationError: return Response('bad signature', status_code=400) print(event['type'], event['data']) return Response(status_code=204)ボディは生のまま渡すこと。パースして再度シリアライズするとキーの順序や空白が変わり、署名が一致しなくなる。ボディは bytes でも str でもよい。FastAPI なら await request.body()、Django なら request.body、Flask なら request.get_data() である。headers は、これらの各フレームワークの request.headers やただの dict など、任意のマッピングを受け取り、X-OpenEmail-Signature は大文字小文字にかかわらず見つけられる。
自前の検証が取りこぼしがちな 2 点をこの関数は処理する。1 つは MAC を一定時間で比較することで、処理時間から正しい先頭部分を割り出せないようにする。もう 1 つは、前後いずれの方向でも tolerance_seconds(指定しなければ 5 分)を超えて古い配信を拒否することで、傍受されたリクエストが永久に再送可能にならないようにする。tolerance_seconds=0 にするとリプレイ検査は無効になる。どちらの不具合も表に出ない。いずれかを抱えたハンドラーでも、思いつくかぎりのテストはすべて通ってしまう。
失敗した場合は必ず WebhookVerificationError を送出する。secret が空である、X-OpenEmail-Signature ヘッダーがない、形式が t=<seconds>,v1=<hex> になっていない、タイムスタンプが許容範囲外である、署名が一致しない、ボディが JSON でない、のいずれの場合も同様である。成功した場合は、ボディを WebhookPayload としてパースした結果を返すため、間違えやすい 2 度目の json.loads は不要である。data に型を付けるには、WebhookPayload[EmailOpenedData] のように結果に注釈を付けること。
import os from openemail import verify_webhook_signaturefrom openemail.types import EmailOpenedData, WebhookPayload def on_open(body: bytes, headers: dict[str, str]) -> None: event: WebhookPayload[EmailOpenedData] = verify_webhook_signature( payload=body, headers=headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], ) print(event['data']['recipient'], event['data']['first'], event['data']['country'])WebhookVerificationError は OpenEmailError と ValueError の両方のサブクラスである。この関数はコルーチンではなく普通の関数なので、同期コードからも非同期コードからも await なしで呼び出す。標準ライブラリの hmac と hashlib 以外には何も必要としない。