测试入站 Webhook 的麻烦程度与其实际问题并不成正比。

一个提供商——Stripe、GitHub、Shopify 等——会将 JSON 正文 POST 到您拥有的 URL。您需要准确看到他们发送的内容，并在修改处理器后再次发送。两者都不难，但通常的设置是：

启动隧道，以便提供商能够访问您的笔记本电脑
将生成的 URL 粘贴到提供商的仪表板中
在他们的 UI 中触发事件
阅读您的应用程序日志，并希望您记录得足够详细
修改您的处理器
返回步骤 3，因为您无法重放已到达的内容

步骤 6 是成本最高的一步。通过第三方的 UI 重新触发真实事件来测试一行代码的更改是一个缓慢的循环，而且有些事件根本不容易按需触发。

您真正需要的部分

剥离后，只有三个要求：

捕获到达的任何内容：方法、路径、标头、正文——不做任何修改
检查它，包括标头（签名通常位于此处）
根据需要多次将其重放到您的处理器

这些都不需要隧道。隧道解决的是不同的可达性问题——来自公共互联网的可达性——一旦捕获了有效负载，您就不再需要提供商再次参与。

使用标准库实现

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，它是一个付费产品。本文中的代码仅使用标准库，您可以随意使用——不购买任何东西，文章也能正常工作。