KaiMail

PythonでKaiMailのWebhookをテストする方法

KaiMail Webhook Testing for Developers

KaiMail Webhook Testing for Developers

KaiMailは受信メールをHTTP POSTリクエストとしてアプリケーションに配信できる。メールボックスをポーリングする代わりに、メールが届いた瞬間にコードが呼び出される。サポートチケットシステム、請求書処理、通知パイプラインなど、メールに反応する何かを構築しているなら便利な機能である。

問題はテストだ。ローカルマシンはインターネットに公開されていないので、KaiMailからアクセスできない。このチュートリアルでは、PythonのWebhookレシーバー、ファイアウォールを突破するngrok、そして実際のメール送信まで、全体のセットアップを順を追って説明する。

今回はpipのライブラリーを使わない。ここで使うのはすべてPython標準ライブラリのみである。

前提条件

必要なものは3つ:

  1. Webhook対応するカスタムドメインが設定済みのKaiMailアカウントとメールボックス（例: [email protected]）
  2. Python 3.10以降
  3. ngrokアカウント（無料プランで十分）

ステップ1―Webhookレシーバーを書く

POSTリクエストを受け取り、KaiMailから送信された内容を表示する小さなHTTPサーバーを作る。

最小バージョン

まずは動く最小限のものから:

from http.server import BaseHTTPRequestHandler, HTTPServer
import json

class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length)

        payload = json.loads(body)
        print(json.dumps(payload, indent=2, ensure_ascii=False))

        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"OK\n")

server = HTTPServer(("0.0.0.0", 8000), WebhookHandler)
server.serve_forever()

これだけでWebhookを受信できる。python receiver.pyで保存して実行すればポート8000でリクエスト待ちにする。しかし、リクエストが本当にKaiMailから来たものか、ランダムなボットからではないか、検証したい。

署名検証を追加する

KaiMailはルートのWebhookシークレットを使い、HMAC-SHA256ですべてのWebhookペイロードに署名する。署名はX-KAI-Webhook-Signatureヘッダーにsha256=<hex_digest>形式で送信される。

検証手順:

  1. 共有シークレットを使い、リクエストボディのHMAC-SHA256を計算する
  2. ヘッダー値からsha256=プレフィックスを除去する
  3. 2つのhexダイジェストを定数時間で比較する
import hashlib
import hmac

def verify_signature(payload, secret, received_signature):
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    received_hex = received_signature.removeprefix("sha256=")
    return hmac.compare_digest(expected, received_hex)

removeprefixの呼び出しは重要である。ヘッダーの値はsha256=abc123...だが、hmac.new().hexdigest()はabc123...だけを返す。プレフィックスを除去せずに比較すると、検証は常に失敗する。（実際、自分たちのサンプルコードでまさにこのバグを見つけた。）

タイミング攻撃を防ぐため、==ではなくhmac.compare_digestを使う。

サンプルスクリプト

署名検証とCLI引数を含む完全なレシーバーがこちら。webhook_test_receiver.pyとして保存する:

#!/usr/bin/env python

import argparse
import hashlib
import hmac
import json
import sys
from datetime import UTC, datetime
from http.server import BaseHTTPRequestHandler, HTTPServer

KAI_HEADERS = [
    "X-KAI-Webhook-Signature",
    "X-KAI-Tracking-ID",
    "X-KAI-Event",
]


def verify_signature(payload, secret, received_signature):
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    received_hex = received_signature.removeprefix("sha256=")
    return hmac.compare_digest(expected, received_hex)


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else b""

        now = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC")
        print(f"\n{'=' * 70}")
        print(f"  POST {self.path}  [{now}]")
        print(f"{'=' * 70}")

        for header in KAI_HEADERS:
            value = self.headers.get(header)
            if value:
                print(f"  {header}: {value}")

        secret = self.server.webhook_secret
        signature = self.headers.get("X-KAI-Webhook-Signature", "")
        if secret:
            if signature:
                valid = verify_signature(body, secret, signature)
                status = "VALID" if valid else "INVALID"
            else:
                status = "MISSING"
            print(f"  Signature: {status}")

        print()

        if body:
            try:
                payload = json.loads(body)
                print(json.dumps(payload, indent=2, ensure_ascii=False))
            except (json.JSONDecodeError, UnicodeDecodeError):
                print(body.decode("utf-8", errors="replace"))

        print()

        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(b"OK\n")
        sys.stdout.flush()

    def log_message(self, format, *args):
        pass


def main():
    parser = argparse.ArgumentParser(
        description="KaiMail Webhookテスト用レシーバー"
    )
    parser.add_argument(
        "--port", type=int, default=8000, help="ポート (デフォルト: 8000)"
    )
    parser.add_argument(
        "--secret", type=str, default="", help="Webhook署名シークレット"
    )
    args = parser.parse_args()

    server = HTTPServer(("0.0.0.0", args.port), WebhookHandler)
    server.webhook_secret = args.secret

    print(f"Listening on http://0.0.0.0:{args.port}")
    if args.secret:
        print("Signature verification: ENABLED")
    else:
        print("Signature verification: DISABLED (use --secret to enable)")
    print("Waiting for POST requests... (Ctrl+C to stop)\n")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")
        server.server_close()
        sys.exit(0)


if __name__ == "__main__":
    main()

ステップ2―ngrokのインストールと設定

ngrokはローカルマシンへのトラフィックをトンネリングする公開HTTPS URLを作成する。インストール方法:

# Ubuntu/Debian
sudo snap install ngrok

# macOS
brew install ngrok

# または https://ngrok.com/download からダウンロード

ngrok.comで無料アカウントを作成し、認証トークンを追加する:

ngrok config add-authtoken YOUR_AUTH_TOKEN

レシーバーのポートに向けてトンネルを起動する:

ngrok http 8000

ngrokは以下のような出力を表示する:

Forwarding  https://a1b2c3d4.ngrok-free.app -> http://localhost:8000

この ngrok-free.app HTTPS URLは、KaiMailがPOSTする先になるのでコピーする。

注意: 無料プランではngrokを再起動するたびにランダムなURLが変わる。テスト用途なら問題ない。

ステップ3―KaiMailの設定

KaiMailのルートページにアクセスし、テストしたいメールボックス（例: [email protected]）を編集して、配信方法をWebhookに設定する。WebhookのURLフィールドにngrokのHTTPS URLを貼り付ける。

署名検証に必要になるので、ルートに表示される署名シークレットをメモしておくと便利かもしれない。

ステップ4―テストメールを送信する

署名シークレットを指定してレシーバーを起動する:

python webhook_test_receiver.py --port 8000 --secret YOUR_SIGNING_SECRET

あとは任意のメールクライアントからKaiMailに登録してる[email protected]にメールを送信してKaiMailで受信することでレシーバーにWebhookとして受け取るはず。

レシーバーのターミナルを確認する。数秒以内に、完全なJSONペイロードが表示され、Signature: VALIDと出力されるはずである。

ペイロードの構造

典型的なWebhookペイロードは以下のようになる:

{
  "version": "1.0",
  "timestamp": "2025-01-31T10:30:00Z",
  "headers": {
    "From": "[email protected]",
    "To": "[email protected]",
    "Subject": "Quick question",
    "Date": "Fri, 31 Jan 2025 10:30:00 +0000",
    "Message-ID": "<[email protected]>"
  },
  "body_text": "Hi,\n\nDo you offer annual discounts?\n\nThanks,\nAlice",
  "body_html": null,
  "attachments": [],
  "envelope": {
    "sender": "[email protected]",
    "recipient": "[email protected]"
  },
  "account": {
    "email": "[email protected]",
    "domain": "yourdomain.com",
    "mailbox": "[email protected]"
  },
  "metadata": {
    "authentication_results": {}
  }
}

各フィールドの説明:

  • version: ペイロードスキーマのバージョン。現在は"1.0"。
  • timestamp: KaiMailがメールを処理した日時（ISO 8601、UTC）。
  • headers: 元のメールヘッダーすべて。複数値を持つヘッダー（Receivedなど）は配列として格納される。
  • body_text: プレーンテキスト本文（存在する場合）。HTMLのみのメールではnull。
  • body_html: HTML本文（存在する場合）。テキストのみのメールではnull。マルチパートメールでは両方が存在しうる。
  • attachments: 添付ファイルオブジェクトの配列。各オブジェクトにはfilename、content_type、size（バイト）、url（24時間後に期限切れとなるプリサインURL）がある。添付ファイルはJSONに埋め込まれず外部に保存されるため、ペイロードが小さく保たれる。
  • envelope: SMTPエンベロープの送信者と受信者（From/Toヘッダーとは異なる場合がある）。
  • account: KaiMailアカウントのメールアドレス、ドメイン、メールを受信した特定のメールボックス。
  • metadata: メール認証結果（SPF、DKIM、ARC）を含む追加データ。

Webhookへの応答

サーバーは2xx HTTPステータスコードを返す必要がある。配信が成功したとみなされるためである。

サーバーが5xxエラー、408（リクエストタイムアウト）、または429（リクエスト過多）を返した場合、KaiMailは指数バックオフ（5秒、10秒、20秒）で最大3回リトライする。

その他の4xxレスポンスは永続的な失敗として扱われる。KaiMailはリトライしない。

サーバーがタイムアウト時間内に応答しない場合、5xxと同様に扱われる。

次のステップ

このセットアップは開発とテスト用である。本番環境では:

  1. HTTPSを使う。 本番のWebhookエンドポイントはTLSの背後に配置すべきである。KaiMailはデフォルトでSSL証明書を検証する。
  2. 処理前にデータベースに永続化する。 Webhookを受信したらストアし、即座に200を返してから非同期で処理する。遅い処理でのタイムアウトを回避できる。
  3. トラッキングIDで重複排除する。 X-KAI-Tracking-IDヘッダーは配信ごとにユニークである。サーバーが最初の試行を既に処理した後にリトライが到着した場合、再処理前にトラッキングIDを確認する。
  4. 常に署名を検証する。 本番では、無効または欠落したX-KAI-Webhook-Signatureを持つリクエストは拒否すべきである。テストレシーバーは結果を表示するだけだが、本番サーバーでは403を返すべきである。
← 戻る： 外部連携