測試入站 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},在 / 提供內建的網頁 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,它是一款付費產品。本文中的程式碼僅使用標準函式庫,您可以自行使用 — 不購買任何東西,本文也能正常運作。
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.