适用场景

支付回调、代码仓库事件、消息推送和第三方审批通知通常通过 Webhook 进入业务系统。很多服务已经实现了 HMAC 签名校验，却仍会遇到重复发货、重复记账或重复触发流水线的问题。

这类故障的关键在于：签名只能证明请求内容来自持有密钥的一方，不能证明这次请求从未被处理过。 一个合法请求被代理层、第三方平台或攻击者完整重放时，签名依然有效。

本文以 Python/FastAPI 接收端为例，建立一条完整的防护链路：校验原始请求体、限制时间窗口、使用常量时间比较、以事件 ID 实现原子幂等，并把密钥轮换和可观测性纳入设计。

现象描述

线上常见表现包括：

  • 同一个支付事件在几秒内触发两次，两个请求的签名都正确；
  • 应用日志显示接口均返回 200，但订单状态变更或下游任务执行了两次；
  • 第三方平台记录了一次主动重试，网关日志还出现了额外的相同请求；
  • 重放一段历史请求后，服务仍然接受，说明签名没有有效期；
  • 数据库虽然有唯一约束，但业务副作用发生在写入唯一记录之前。

不要先把问题归因于“第三方多发了一次”。Webhook 的投递语义通常是至少一次，超时、连接中断和响应丢失都可能触发重试。接收端必须把重复投递视为正常输入，而不是异常情况。

为什么验签成功仍不安全

假设发送方用下面的内容计算签名：

signature = HMAC-SHA256(secret, request_body)

攻击者即使不知道 secret，只要截获完整的请求体和签名，就可以原样再次发送。接收端重新计算出的结果完全相同，因此仍会通过验证。

可靠的签名载荷至少应绑定：

signed_payload = timestamp + "." + raw_request_body

接收端再同时验证：

  1. 时间戳与当前时间的偏差不超过允许窗口；
  2. 事件 ID 没有被成功处理过；
  3. 签名使用原始字节计算，且通过常量时间函数比较；
  4. 业务写入和幂等状态在同一个事务边界内完成。

时间窗口降低旧请求被重放的价值，事件 ID 则阻止窗口内的重复请求。两者不能互相替代。

排查思路

1. 先确认请求是否完全相同

记录经过脱敏的诊断字段，不要记录密钥、完整签名或原始业务请求体：

  • provider：事件来源；
  • event_id：发送方提供的唯一事件编号；
  • event_type：事件类型；
  • timestamp：签名时间戳；
  • body_sha256：原始请求体摘要；
  • delivery_attempt：发送方提供的投递次数；
  • trace_id：本次请求的链路标识；
  • result：accepted、duplicate、expired 或 invalid_signature。

如果两次请求的 event_id 和 body_sha256 都相同，优先按重复投递处理。如果事件 ID 相同但摘要不同，应拒绝请求并触发高优先级安全告警，因为这可能是发送方数据异常、事件 ID 冲突或攻击行为。

2. 检查验签是否使用原始请求体

以下做法容易导致验签逻辑失真：

  • 先把 JSON 解析成对象，再序列化后验签；
  • 对请求体执行字符集转换、去空格或字段排序；
  • 在中间件中读取请求体后没有正确传递原始字节；
  • 只签名业务字段，没有绑定时间戳和协议版本。

JSON 的空格、字段顺序和转义形式都可能变化。验签必须针对网络收到的原始字节进行，业务解析应放在验签成功之后。

3. 检查副作用发生顺序

常见错误流程是：先发货，再尝试写入 processed_webhook_events。当第二次投递到达时，即使唯一约束阻止了重复记录，发货动作已经发生。

正确流程应在数据库事务中先争抢事件处理权，再修改业务状态。邮件、消息或调用外部服务等无法放入同一数据库事务的副作用，应通过事务型 Outbox 在提交后异步发送。

可直接使用的接收端实现

下面示例使用 FastAPI，演示协议边界和处理顺序。生产环境中的幂等存储必须换成数据库或具备原子写能力的共享存储，不能使用进程内集合。

from __future__ import annotations

import hashlib
import hmac
import json
import time
from dataclasses import dataclass

from fastapi import FastAPI, Header, HTTPException, Request, status


app = FastAPI()
WEBHOOK_SECRET = b"replace-with-secret-manager-value"
MAX_CLOCK_SKEW_SECONDS = 300


@dataclass(frozen=True)
class VerifiedEvent:
    event_id: str
    event_type: str
    payload: dict[str, object]


def parse_signature_header(header: str) -> tuple[int, str]:
    """解析形如 t=时间戳,v1=十六进制摘要的签名头。"""
    parts = {}
    for item in header.split(","):
        key, separator, value = item.strip().partition("=")
        if not separator or not key or not value:
            raise ValueError("签名头格式错误")
        parts[key] = value

    if "t" not in parts or "v1" not in parts:
        raise ValueError("签名头缺少必要字段")

    timestamp = int(parts["t"])
    signature = parts["v1"].lower()
    if len(signature) != 64:
        raise ValueError("签名长度错误")
    return timestamp, signature


def verify_event(raw_body: bytes, signature_header: str, now: int) -> VerifiedEvent:
    """验证时间窗口与 HMAC，然后解析并校验业务事件。"""
    try:
        timestamp, provided_signature = parse_signature_header(signature_header)
    except (ValueError, TypeError) as exc:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "签名头无效") from exc

    if abs(now - timestamp) > MAX_CLOCK_SKEW_SECONDS:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "请求时间戳已过期")

    signed_payload = str(timestamp).encode("ascii") + b"." + raw_body
    expected_signature = hmac.new(
        WEBHOOK_SECRET,
        signed_payload,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected_signature, provided_signature):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "签名校验失败")

    try:
        data = json.loads(raw_body)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "请求体不是有效 JSON") from exc

    if not isinstance(data, dict):
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "请求体必须是对象")

    event_id = data.get("id")
    event_type = data.get("type")
    if not isinstance(event_id, str) or not 1 <= len(event_id) <= 128:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "事件 ID 无效")
    if not isinstance(event_type, str) or not 1 <= len(event_type) <= 128:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "事件类型无效")

    return VerifiedEvent(event_id=event_id, event_type=event_type, payload=data)


@app.post("/webhooks/provider")
async def receive_webhook(
    request: Request,
    x_webhook_signature: str = Header(alias="X-Webhook-Signature"),
) -> dict[str, str]:
    """接收并验证 Webhook，业务幂等操作应交给事务服务完成。"""
    raw_body = await request.body()
    if len(raw_body) > 1_048_576:
        raise HTTPException(status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, "请求体过大")

    event = verify_event(raw_body, x_webhook_signature, int(time.time()))
    result = process_event_in_transaction(event)
    return {"status": result}


def process_event_in_transaction(event: VerifiedEvent) -> str:
    """示意事务入口，生产环境应由数据库仓储实现。"""
    raise NotImplementedError("请接入数据库事务与幂等表")

关键点如下：

  • request.body() 返回的原始字节同时用于验签和 JSON 解析；
  • timestamp + b"." + raw_body 明确绑定时间与内容；
  • hmac.compare_digest() 避免普通字符串比较暴露时序差异；
  • 在验签前限制请求体大小，降低恶意大包带来的资源消耗；
  • abs(now - timestamp) 同时限制过旧和明显来自未来的请求；
  • 所有字段仍在服务端验证类型和长度，验签成功不代表业务数据可信。

示例中的固定密钥仅用于说明接口。正式环境应从密钥管理系统读取，并避免出现在源码、日志和异常信息中。

用数据库实现原子幂等

以 PostgreSQL 为例，先建立事件处理表：

CREATE TABLE processed_webhook_events (
    provider       text        NOT NULL,
    event_id       text        NOT NULL,
    body_sha256    char(64)    NOT NULL,
    event_type     text        NOT NULL,
    processed_at   timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (provider, event_id)
);

处理请求时开启事务，并先执行：

INSERT INTO processed_webhook_events (
    provider,
    event_id,
    body_sha256,
    event_type
)
VALUES ($1, $2, $3, $4)
ON CONFLICT (provider, event_id) DO NOTHING
RETURNING event_id;

根据结果分支：

  • 返回一行：当前事务取得处理权，继续更新订单、账务或任务状态；
  • 未返回行：事件已存在，读取原记录并比较 body_sha256；
  • 摘要一致：将本次请求记为 duplicate，返回 200，避免发送方持续重试；
  • 摘要不一致：回滚并返回错误，同时触发安全告警。

业务状态更新、幂等记录和 Outbox 消息必须在同一个事务内提交。这样即使应用在提交后、返回 HTTP 响应前崩溃，发送方重试时也只会得到“已处理”，不会再次产生业务副作用。

不要简单地对整个处理过程加分布式锁。锁到期、进程崩溃和网络分区仍可能造成重复执行，而数据库唯一约束可以作为最终一致的事实边界。

密钥轮换怎么做

Webhook 密钥也需要轮换，但切换期间不能把新旧密钥拼接使用。推荐给每个密钥分配稳定的 key_id，发送方在签名头中带上版本：

X-Webhook-Signature: kid=2026-09,t=1790636400,v1=...

接收端根据 kid 精确选择候选密钥，只在明确的迁移期同时保留新旧两个版本。监控旧版本的最后使用时间，确认发送方全部切换且超过最大重试周期后，再撤销旧密钥。

若第三方协议不支持 kid，可以在短暂迁移期分别用新旧密钥计算摘要，但要限制候选数量，并记录命中的密钥版本。不要把失败签名或密钥内容写入日志。

测试防护是否有效

至少覆盖以下用例：

  1. 正确时间戳、正确请求体和正确签名，应处理一次；
  2. 同一事件再次投递，应返回成功但不重复执行业务；
  3. 时间戳超过五分钟，即使签名正确也应拒绝；
  4. 请求体修改一个字节，原签名必须失效；
  5. 事件 ID 相同但请求体摘要不同，应告警并拒绝；
  6. 两个并发请求携带相同事件 ID，只允许一个事务取得处理权；
  7. 业务事务回滚后，幂等记录也必须回滚，后续合法重试可以继续处理；
  8. 应用提交事务后模拟响应丢失，重试不得产生第二次副作用。

并发测试不能只用进程内假对象，因为它无法验证数据库唯一约束、事务隔离和连接池行为。应在隔离的测试数据库中同时发起多次相同事件，并断言业务表和 Outbox 表都只新增一条记录。

监控与告警

建议按 provider 和 result 统计计数器，但不要把 event_id 放进指标标签，以免产生高基数：

  • webhook_requests_total{provider,result}；
  • webhook_processing_duration_seconds{provider,event_type}；
  • webhook_timestamp_skew_seconds{provider}；
  • webhook_outbox_pending_total。

以下情况值得告警：

  • invalid_signature 突然升高，可能是密钥或代理配置错误，也可能是攻击；
  • expired 集中出现，可能是发送方时钟漂移或队列严重积压；
  • 同一事件 ID 出现不同摘要，这是比普通重复投递更强的异常信号；
  • Outbox 长时间未投递，说明主业务已提交但下游副作用尚未完成；
  • 重复率突然升高，可能是响应超时、网关断连或发送方重试策略变化。

日志、指标和追踪应共享 trace_id，但签名、密钥、Cookie、Authorization 和完整敏感请求体必须排除在观测数据之外。

常见误区

只校验来源 IP

IP 白名单可以减少暴露面，但云服务出口地址会变化，代理链也可能让源地址失真。它只能作为附加控制，不能替代加密签名。

验签成功后直接返回，再异步处理

如果请求入队失败却已返回 200，事件会永久丢失。应先把事件或 Outbox 记录可靠写入事务存储，再确认接收成功。

重复请求返回 409

对已经成功处理且内容一致的事件，通常应返回 2xx。返回 409 可能让发送方认为投递失败并持续重试，放大流量。

用 Redis 的先查后写判断重复

GET 后再 SET 存在竞态。若使用 Redis，至少需要 SET key value NX EX seconds 这样的原子操作，并评估数据丢失、过期时间和故障转移语义。涉及账务等关键业务时，仍建议以业务数据库的唯一约束作为最终边界。

预防措施清单

  • 在接入文档中固定签名算法、字段顺序、编码、分隔符和版本；
  • 只对原始请求体验签，成功后再解析业务数据；
  • 同时使用时间窗口和持久化事件 ID 防重放；
  • 使用常量时间函数比较摘要；
  • 密钥放入专用密钥管理系统，建立轮换和撤销流程；
  • 幂等记录、业务写入与 Outbox 在同一事务内提交；
  • 重复成功事件返回 2xx，不重复产生副作用；
  • 对事件 ID 相同但摘要不同的请求拒绝并告警；
  • 对请求体大小、字段类型、长度和事件类型建立白名单校验；
  • 定期执行并发重放、事务回滚和响应丢失测试。

总结

Webhook 的正确安全模型不是“签名通过就执行”，而是“来源可信、时间有效、内容完整、事件唯一、事务可恢复”五个条件同时成立。

HMAC 解决身份与完整性，时间戳压缩重放窗口，事件 ID 和数据库唯一约束保证业务幂等，事务型 Outbox 则把无法同步提交的外部副作用纳入可靠流程。把这几层组合起来，才能同时应对正常重试、网络故障、并发请求和恶意重放，而不是在重复扣款或重复发货之后依赖人工补偿。