インバウンドWebhookのテストは、問題の割に面倒です。

プロバイダー — Stripe、GitHub、Shopifyなど — は、あなたが所有するURLにJSONボディをPOSTします。送信内容を正確に確認し、ハンドラーを変更した後に再度送信する必要があります。これらはどれも難しいことではありません。しかし、通常のセットアップは次のようになります。

プロバイダーがあなたのノートPCに到達できるようにトンネルを開始する
生成されたURLをプロバイダーのダッシュボードに貼り付ける
UIでイベントをトリガーする
アプリケーションログを確認し、十分にログを記録したことを期待する
ハンドラーを変更する
すでに到着したものをリプレイできないため、手順3に戻る

手順6がコストのかかる部分です。サードパーティのUIを通じて実際のイベントを再トリガーして1行の変更をテストするのは遅いループであり、一部のイベントはオンデマンドでトリガーすることがそもそも簡単ではありません。

実際に必要な部分

整理すると、要件は3つだけです。

到着したものをすべてキャプチャする：メソッド、パス、ヘッダー、ボディ — 変更なし
ヘッダーを含めて検査する（通常、署名がここに存在する）
何度でもハンドラーでリプレイする

これらにはトンネルは必要ありません。トンネルは異なる問題 — パブリックインターネットからの到達可能性 — を解決するもので、一度ペイロードをキャプチャすれば、プロバイダーを再び関与させる必要はありません。

標準ライブラリで実装する

Pythonのhttp.serverで十分です。リクエストを記録し200を返すキャッチオールハンドラー：

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

CAPTURED = []

class Handler(BaseHTTPRequestHandler):
    def _capture(self):
        length = int(self.headers.get("Content-Length", 0))
        CAPTURED.append({
            "method": self.command,
            "path": self.path,
            "headers": dict(self.headers),
            "body": self.rfile.read(length).decode("utf-8", "replace"),
        })
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"status":"captured"}')

    do_POST = do_PUT = do_PATCH = do_GET = do_DELETE = _capture

    def log_message(self, *args):
        pass

ThreadingHTTPServer(("127.0.0.1", 8000), Handler).serve_forever()

Enter fullscreen mode Exit fullscreen mode

http://127.0.0.1:8000/hook/test に何かを向けると記録されます。

リプレイも同様に単純です — キャプチャしたリクエストを取得して別の場所に送信します：

import json, urllib.request

def replay(captured, target_url):
    req = urllib.request.Request(
        target_url,
        data=captured["body"].encode(),
        headers={k: v for k, v in captured["headers"].items()
                 if k.lower() not in ("host", "content-length")},
        method=captured["method"],
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return resp.status, resp.read().decode()

Enter fullscreen mode Exit fullscreen mode

HostとContent-Lengthを削除することが重要です — どちらも元のリクエストを参照しており、正しくないか、urllibが自身で設定するものと競合します。

これが基本的なアイデアです。これ以降はすべて利便性です：複数のリクエストを保存する、リクエストを読み取ってリプレイを発行するUI、エンドポイントによるマッチング、再起動後も保持する永続化。

注意すべき点

リプレイでは署名検証が失敗します。プロバイダーは生のボディに署名し、しばしばこの種のリプレイを防ぐためにタイムスタンプを使用します。ハンドラーが署名を検証する場合、開発時に検証をスキップする方法が必要になります。これは署名方式の特性であり、キャプチャツールで修正できるものではありません。

ボディは正確に一度だけ読み取ります。self.rfile.read() はストリームを消費します。まず変数に読み込んでから作業します。

盲目的にデコードしないでください。すべてのWebhookボディがUTF-8 JSONとは限りません。errors="replace" は、バイナリペイロードがハンドラーをダウンさせるのを防ぎます。

127.0.0.1 にバインドします。多くのサンプルコードのデフォルトは 0.0.0.0 で、これは認証されていないエンドポイントをマシンのすべてのインターフェースに公開し、認証ヘッダーを含む完全なリクエストボディをログに記録します。

自分でメンテナンスしたくない場合

これをプロジェクトに貼り付けるのに疲れたので、適切に構築してHookTrapとして販売しています — 49ドル、ワンタイム、ソースコード込み、サブスクリプションなし。

これは同じアイデアの角を丸めたものです：/hook/{endpoint_id} への任意のメソッドをキャプチャ、リクエストの読み取りとリプレイの発行のための組み込みWeb UI（/）、JSON API（GET /api/requests、POST /api/replay/{id}、DELETE /api/requests）、--file storage.json によるオプションの永続化。依然としてPython 3.9+標準ライブラリのみ — pip install不要、アカウント不要、何もマシンの外に出ません。

意図的に行わないこと：独自のHTTPSなし（TLSが必要な場合はプロキシの後ろで実行）、UIに認証なし、リプレイは同期的に実行されるため遅いターゲットが呼び出しをブロックする。これらは購入前ではなくREADMEに記載されています。

→ HookTrap — $49、セルフホスト

開示：私はHookTrapを構築し、有料製品です。この記事のコードは標準ライブラリのみで、自由に使用できます — 何も購入しなくても記事は問題なく機能します。