Webhooks 詳解：簽名負載與重試機制

你已經建立了一個 Webhook 端點來接收第三方服務的事件。在測試環境中一切正常，但在生產環境中，你注意到重複的事件、遺漏的通知，以及偶爾對偽造請求的安全疑慮。這些都是在沒有適當防護措施下實作 Webhook 時常見的陷阱。

Webhook 是 HTTP 回呼，讓服務在發生某事時通知你的應用程式。與輪詢不同，它們提供即時更新。但要可靠地使用它們，你需要解決兩個關鍵方面：真實性（確保請求來自預期的發送者）和傳遞（處理失敗與重試）。本文將透過實用程式碼範例說明這兩者。

Webhook 的運作方式

Webhook 是使用者定義的 HTTP POST 端點。當事件發生時，來源服務會向該 URL 發送一個包含事件描述負載的請求。例如，支付處理器可能會在扣款成功時發送 Webhook。

流程很簡單：

1. 你向服務註冊一個 Webhook URL。
2. 當事件觸發時，服務會向該 URL 發送 HTTP POST。
3. 你的端點處理負載並返回 2xx 狀態以確認收到。

如果沒有適當的驗證，任何人都可以向你的端點發送請求。如果沒有重試處理，暫時性故障可能導致事件遺失。讓我們逐一解決。

使用簽名負載保護 Webhook

為了確保 Webhook 來自預期的發送者，服務會使用共享密鑰對負載進行簽名。簽名通常是請求主體的 HMAC（基於雜湊的訊息驗證碼），並在類似 X-Signature 的標頭中發送。

HMAC 簽名如何運作

發送者使用密鑰和原始請求主體計算 HMAC，然後將十六進位編碼的簽名包含在標頭中。你的端點使用相同的密鑰重新計算 HMAC，並與收到的簽名進行比較。如果匹配，則請求是真實的。

以下是使用 Express 的 Node.js 最小範例：

const crypto = require('crypto');
const express = require('express');
const app = express();

// Use raw body for signature verification
app.use(express.raw({ type: 'application/json' }));

app.post('/webhook', (req, res) => {
  const signature = req.headers['x-signature'];
  const secret = process.env.WEBHOOK_SECRET;
  const payload = req.body;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');

  if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
    return res.status(401).send('Invalid signature');
  }

  // Process the event
  const event = JSON.parse(payload);
  // ... handle event
  res.status(200).send('OK');
});

重點：

• 務必使用原始請求主體進行驗證，而不是解析後的物件。解析可能會改變負載（例如空白字元）並破壞簽名。
• 使用時間安全的比較來防止時序攻擊。
• 安全地儲存密鑰（環境變數、密鑰管理器）並定期輪換。

額外的安全措施

• 時間戳驗證：在簽名中包含時間戳（例如 t=timestamp,v1=signature），並拒絕超過幾分鐘的請求以防止重放攻擊。
• IP 允許清單：如果供應商公佈了 IP 範圍，則將你的端點限制為這些 IP。
• HTTPS：始終使用 HTTPS 以保護傳輸中的負載。

實作可靠的重試

Webhook 傳遞並非保證。網路可能故障、你的伺服器可能宕機，或者你可能返回非 2xx 狀態。為了處理這種情況，發送者通常會使用指數退避進行重試。作為接收者，你必須設計端點以具備冪等性並快速回應。

發送者端（如果你發送 Webhook）

如果你正在建立發送 Webhook 的服務，請實作帶有退避的重試。常見策略：

• 在網路錯誤或 5xx 回應時重試。
• 不要在 4xx 回應時重試（除了 429 Too Many Requests）。
• 使用帶有抖動的指數退避：等待 1 秒、2 秒、4 秒、8 秒等，直到最大值。
• 設定最大重試次數（例如 5 次），並考慮為失敗使用死信佇列。

Python 中的重試邏輯範例：

import time
import requests
from requests.exceptions import RequestException

def send_webhook(url, payload, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = requests.post(url, json=payload, timeout=5)
            if 200 <= response.status_code < 300:
                return True
            elif 400 <= response.status_code < 500 and response.status_code != 429:
                # Client error, don't retry
                return False
        except RequestException:
            pass
        # Exponential backoff with jitter
        delay = (2 ** attempt) + random.uniform(0, 1)
        time.sleep(delay)
    return False

接收者端

你的端點應該：

• 快速確認：在幾秒內返回 2xx。如果處理時間較長，將事件加入佇列並非同步處理。
• 具備冪等性：使用負載中的事件 ID 進行去重。儲存已處理的 ID 並忽略重複。
• 處理併發：使用資料庫交易或鎖定來避免同時處理同一事件時的競爭條件。

使用資料庫唯一約束的冪等性範例：

CREATE TABLE processed_events (
  event_id VARCHAR(255) PRIMARY KEY,
  processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

在處理之前，嘗試插入事件 ID。如果違反主鍵約束，則跳過處理。

在本機測試 Webhook

在開發過程中，你需要一個公共 URL 讓服務能夠存取你的本機伺服器。像 ngrok 或 localtunnel 這樣的工具可以公開你的 localhost。在開發中始終使用測試密鑰並驗證簽名，以便及早發現問題。

比較：輪詢 vs Webhook

常見問題

什麼是簽名的 Webhook 負載？

簽名負載包含使用共享密鑰生成的加密簽名（通常是 HMAC）。它允許接收者驗證請求來自預期的發送者且未被篡改。

如何處理重複的 Webhook 傳遞？

透過使用負載中的唯一事件 ID 使你的端點具備冪等性。儲存已處理的 ID 並忽略重複。這確保重試或網路問題不會導致重複處理。

我應該重試失敗的 Webhook 傳遞嗎？

是的，但僅限於暫時性錯誤（網路問題、5xx）。使用帶有抖動的指數退避並設定最大重試限制。避免在 4xx 錯誤（除了 429）時重試，因為它們表示請求本身有問題。

在除錯 Webhook 負載時，你經常需要檢查 JSON 資料。JSON Formatter 可以幫助你快速美化並驗證 JSON 負載。