Webhook 详解：签名负载与重试机制

你刚刚集成了支付提供商的 webhook。文档说“支付成功时，我们会向你的端点发送 POST 请求”。听起来很简单，对吧？然后你注意到了重复事件、遗漏的通知，以及关于伪造请求的安全警报。Webhook 概念简单，但在生产环境中却很棘手。本文解释了如何使用签名负载和重试机制正确实现 webhook，以便构建可靠且安全的集成。

什么是 Webhook？

Webhook 是一种反向 API：你的应用程序不再轮询服务以获取更新，而是由服务通过 HTTP POST 将事件推送到你的端点。这种方式高效且近乎实时。常见用途包括支付通知、CI/CD 触发器和聊天集成。

但 webhook 也带来了挑战：你如何知道请求确实来自提供商？如果你的服务器宕机了怎么办？如果同一个事件到达两次怎么办？让我们逐一解决这些问题。

使用签名负载保护 Webhook

任何人都可以向你的端点发送 HTTP 请求。如果没有验证，攻击者可以伪造事件，导致欺诈或数据损坏。标准的解决方案是使用共享密钥对负载进行签名。

HMAC 签名的工作原理

提供商和你共享一个密钥。对于每个事件，提供商使用该密钥对请求体计算 HMAC（基于哈希的消息认证码），并在头部（例如 X-Signature）中发送签名。你在自己这边重新计算 HMAC 并进行比较。如果匹配，则请求是真实的且未被篡改。

以下是一个使用 Flask 的 Python 最小示例：

import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = b'your_shared_secret'

@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Signature')
    if not signature:
        abort(400, 'Missing signature')
    body = request.get_data()
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        abort(401, 'Invalid signature')
    # Process the event
    return '', 200

始终使用 hmac.compare_digest 进行恒定时间比较，以防止时序攻击。此外，在解析 JSON 正文之前验证签名，以避免不必要的处理。

签名的最佳实践

• 使用强密钥：生成至少 32 字节的随机密钥，并安全存储（例如环境变量）。
• 包含时间戳：为了防止重放攻击，提供商通常会在签名负载中包含时间戳。拒绝超过几分钟的请求。
• 定期轮换密钥：在轮换期间支持多个活动密钥，以避免停机。
• 记录签名算法：指定哈希函数（SHA-256 是标准）和头部名称。

通过重试处理失败

从提供商的角度来看，webhook 是“发射后不管”的，但你的端点可能暂时不可用。为了确保交付，提供商实现了重试策略。作为接收方，你必须为重试和重复事件做好准备。

提供商使用的重试策略

大多数提供商使用指数退避进行重试：例如，立即重试，然后在 1 分钟、5 分钟、30 分钟后重试，直到达到最大尝试次数（通常为 5-10 次）。有些还允许你配置重试计划。具体行为各不相同，因此请阅读提供商的文档。

为重试设计端点

你的端点应该：

• 快速响应：在几秒钟内以 2xx 状态码确认接收。异步执行繁重的处理（例如，将事件放入队列）。
• 具有幂等性：同一个事件可能会被多次传递。使用幂等键（通常在事件负载或头部中提供）进行去重。将已处理的事件 ID 存储一段合理的时间。
• 返回适当的状态码：成功返回 2xx，客户端错误返回 4xx（提供商可能不会重试），服务器错误返回 5xx（提供商将重试）。

Python 中幂等性检查的示例：

processed_events = set()  # In production, use a database or cache

@app.route('/webhook', methods=['POST'])
def webhook():
    # ... signature verification ...
    event = request.json
    event_id = event['id']
    if event_id in processed_events:
        return '', 200  # Already processed
    # Process event
    processed_events.add(event_id)
    return '', 200

测试你的 Webhook 实现

在上线之前，彻底测试你的端点：

1. 模拟签名请求：编写单元测试，生成有效和无效的签名。
2. 测试重试场景：模拟失败并确保你的端点处理重复事件。
3. 使用提供商工具：许多提供商提供 CLI 或仪表板来发送测试事件。
4. 监控和记录日志：记录传入事件（不含敏感数据），并为签名失败或高错误率设置警报。

比较：轮询 vs Webhook

常见问题

如何验证 webhook 签名？

使用共享密钥计算原始请求体的 HMAC，并使用恒定时间比较函数将其与签名头部进行比较。始终在处理负载之前进行验证。

我的端点应返回什么以确认 webhook？

在收到事件并将其排队处理后，尽快返回 2xx 状态码（例如 200 OK）。避免在请求处理程序中进行长时间处理，以防止超时。

如何处理重复的 webhook 传递？

使用事件负载中的幂等键（或唯一事件 ID）来跟踪已处理的事件。将这些 ID 存储在数据库或缓存中，如果 ID 已存在则跳过处理。

准备好测试你的 webhook 负载了吗？使用我们的 JSON 格式化工具 来美化打印和验证传入 webhook 的 JSON 正文，使调试更容易。