适用场景
支付、订单、代码托管或消息平台通过 Webhook 主动回调业务系统。接口已经校验 HMAC 签名，但偶尔仍出现同一事件被重复执行，甚至攻击者截获一条合法请求后，可以在数小时后原样重放。
本文以 Python 3.11、FastAPI 和 Redis 为例，实现一套可直接落地的接收端：对原始请求体进行 HMAC-SHA256 验签，限制时间窗口，以事件 ID 做原子去重，并把“重复投递”和“恶意重放”纳入可观测范围。示例协议可按第三方平台的字段名调整。
现象描述
常见现场表现包括：
- Webhook 日志显示每次请求的签名都合法，但同一订单被更新多次；
- 上游因超时重试，短时间内重复发送同一事件；
- 攻击者复制完整请求后，能够在另一个时间再次调用接口；
- 代码先执行业务，再写“已处理”标记，并发请求同时穿透；
- 验签时重新序列化 JSON，导致合法请求偶发验签失败；
- 为了排查问题记录了签名密钥、完整请求体或用户隐私数据，形成新的安全风险。
关键认知是：签名只能证明“请求内容由持有密钥的一方生成且未被篡改”，不能天然证明“这是第一次收到该请求”。防重放必须另外校验时间戳、唯一事件 ID 和处理状态。
威胁模型与协议约定
接收端约定上游发送以下请求头：
X-Webhook-Timestamp: 1788062400
X-Webhook-Event-Id: evt_01HXYZ...
X-Webhook-Signature: v1=2f4c...
签名原文严格定义为：
<timestamp>.<event_id>.<原始请求体字节>
服务端使用共享密钥计算：
HMAC-SHA256(secret, signed_payload)
把时间戳和事件 ID 纳入签名非常重要。否则攻击者可以替换未签名的时间戳，绕过时间窗口；也可以替换事件 ID，绕过去重键。
本方案防护以下风险：
- 请求体或关键请求头被篡改；
- 合法请求在允许时间窗口之外被重放；
- 同一事件在多实例、并发场景下被重复消费；
- 密钥轮换期间新旧发送方切换造成大面积验签失败。
它不替代 HTTPS、业务权限校验和下游幂等。若攻击者已经获得共享密钥，就能自行构造合法签名，应立即轮换密钥并调查泄露范围。
实现思路
安全处理顺序应固定为：
- 限制请求体大小并读取原始字节；
- 校验请求头是否存在、格式是否合法；
- 检查时间戳与服务器当前时间的偏差；
- 使用原始字节计算 HMAC，并做常量时间比较；
- 使用事件 ID 在 Redis 中原子占位；
- 解析 JSON 并校验事件类型、对象 ID 等业务字段；
- 执行业务逻辑，并依靠数据库唯一约束或状态机实现最终幂等；
- 快速返回 2xx，耗时任务交给可靠队列。
不要在验签前解析并重新序列化 JSON。空格、换行、键顺序或 Unicode 转义方式变化都会改变字节序列，使相同 JSON 语义产生不同摘要。
可直接使用的验签代码
创建 webhook_security.py：
from __future__ import annotations
import hashlib
import hmac
import time
from dataclasses import dataclass
class WebhookVerificationError(ValueError):
"""表示 Webhook 请求未通过安全校验。"""
@dataclass(frozen=True)
class VerifiedWebhook:
"""保存通过验签的 Webhook 最小上下文。"""
event_id: str
timestamp: int
def verify_webhook(
*,
body: bytes,
timestamp_text: str,
event_id: str,
signature_header: str,
secrets: tuple[bytes, ...],
now: int | None = None,
tolerance_seconds: int = 300,
) -> VerifiedWebhook:
"""校验时间窗口、事件标识和 HMAC-SHA256 签名。"""
if not event_id or len(event_id) > 128:
raise WebhookVerificationError("事件 ID 缺失或长度非法")
try:
timestamp = int(timestamp_text)
except ValueError as exc:
raise WebhookVerificationError("时间戳格式非法") from exc
current_time = int(time.time()) if now is None else now
if abs(current_time - timestamp) > tolerance_seconds:
raise WebhookVerificationError("请求时间戳超出允许窗口")
if not signature_header.startswith("v1="):
raise WebhookVerificationError("签名版本不受支持")
supplied_digest = signature_header.removeprefix("v1=")
if len(supplied_digest) != 64:
raise WebhookVerificationError("签名长度非法")
signed_payload = (
timestamp_text.encode("ascii")
+ b"."
+ event_id.encode("utf-8")
+ b"."
+ body
)
# 轮换期同时接受新旧密钥，但日志中不得输出密钥或完整签名。
is_valid = any(
hmac.compare_digest(
hmac.new(secret, signed_payload, hashlib.sha256).hexdigest(),
supplied_digest,
)
for secret in secrets
)
if not is_valid:
raise WebhookVerificationError("签名不匹配")
return VerifiedWebhook(event_id=event_id, timestamp=timestamp)
关键点如下：
body必须是网络读取到的原始字节；hmac.compare_digest()避免普通字符串比较引入可利用的时间差；abs(current_time - timestamp)同时拒绝过旧和明显来自未来的请求；secrets支持轮换期同时验证新旧密钥，稳定后应移除旧密钥；- 错误文案只描述失败类型，不返回期望签名或内部密钥信息。
FastAPI 接收端与 Redis 原子去重
安装依赖后，可把下面的路由接入现有应用。生产环境应从密钥管理系统注入 WEBHOOK_SECRETS，不要把密钥提交到仓库。
import json
import os
from fastapi import APIRouter, HTTPException, Request, Response, status
from redis.asyncio import Redis
from webhook_security import WebhookVerificationError, verify_webhook
router = APIRouter()
redis_client = Redis.from_url(
os.environ["REDIS_URL"],
encoding="utf-8",
decode_responses=True,
)
webhook_secrets = tuple(
item.encode("utf-8")
for item in os.environ["WEBHOOK_SECRETS"].split(",")
if item
)
MAX_BODY_BYTES = 1024 * 1024
DEDUPLICATION_TTL_SECONDS = 24 * 60 * 60
@router.post("/webhooks/payment")
async def receive_payment_webhook(request: Request) -> Response:
"""接收支付事件，并在安全校验后执行幂等处理。"""
content_length = request.headers.get("content-length")
if content_length and int(content_length) > MAX_BODY_BYTES:
raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
body = await request.body()
if len(body) > MAX_BODY_BYTES:
raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
try:
verified = verify_webhook(
body=body,
timestamp_text=request.headers["X-Webhook-Timestamp"],
event_id=request.headers["X-Webhook-Event-Id"],
signature_header=request.headers["X-Webhook-Signature"],
secrets=webhook_secrets,
)
except (KeyError, WebhookVerificationError, UnicodeError) as exc:
# 对外统一返回，详细原因只在脱敏后的内部日志中记录。
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Webhook 校验失败",
) from exc
deduplication_key = f"webhook:payment:{verified.event_id}"
is_first_delivery = await redis_client.set(
deduplication_key,
"processing",
ex=DEDUPLICATION_TTL_SECONDS,
nx=True,
)
if not is_first_delivery:
# 对已验签的重复投递返回 200，避免上游持续重试。
return Response(status_code=status.HTTP_200_OK)
try:
event = json.loads(body)
await process_payment_event(event, verified.event_id)
except (json.JSONDecodeError, ValueError) as exc:
# 无效载荷可删除占位，让上游修正后使用相同事件 ID 重试。
await redis_client.delete(deduplication_key)
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Webhook 载荷非法",
) from exc
except Exception:
# 临时故障释放占位，使上游重试能够重新处理；此处必须记录错误链。
await redis_client.delete(deduplication_key)
raise
await redis_client.set(
deduplication_key,
"completed",
ex=DEDUPLICATION_TTL_SECONDS,
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
SET key value EX ttl NX 是一个原子操作：只有第一个请求能获得处理权。不要把 EXISTS 和 SET 拆成两条命令，否则两个并发实例可能同时看到键不存在并执行业务。
Redis 去重用于快速挡住重复投递，但不能成为唯一防线。Redis 故障、键过期或运维误删后，事件仍可能再次进入业务。因此，扣款、发货、记账等关键写入必须以 provider + event_id 建立数据库唯一约束，并在同一个事务内写入事件记录和业务状态。
CREATE TABLE processed_webhook_event (
provider VARCHAR(32) NOT NULL,
event_id VARCHAR(128) NOT NULL,
processed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (provider, event_id)
);
测试正常、边界与失败路径
验签逻辑应使用注入的 now，避免测试依赖真实时钟。以下 pytest 用例覆盖合法请求、篡改和过期重放：
import hashlib
import hmac
import pytest
from webhook_security import WebhookVerificationError, verify_webhook
def build_signature(secret: bytes, timestamp: str, event_id: str, body: bytes) -> str:
"""构造与发送方协议一致的测试签名。"""
payload = timestamp.encode() + b"." + event_id.encode() + b"." + body
return "v1=" + hmac.new(secret, payload, hashlib.sha256).hexdigest()
def test_verify_webhook_accepts_valid_request() -> None:
"""合法请求应返回已验证上下文。"""
body = b'{"type":"payment.succeeded"}'
signature = build_signature(b"secret", "1700000000", "evt_1", body)
result = verify_webhook(
body=body,
timestamp_text="1700000000",
event_id="evt_1",
signature_header=signature,
secrets=(b"secret",),
now=1700000060,
)
assert result.event_id == "evt_1"
def test_verify_webhook_rejects_modified_body() -> None:
"""请求体被篡改后必须验签失败。"""
signature = build_signature(b"secret", "1700000000", "evt_1", b"{}")
with pytest.raises(WebhookVerificationError, match="签名不匹配"):
verify_webhook(
body=b'{"amount":9999}',
timestamp_text="1700000000",
event_id="evt_1",
signature_header=signature,
secrets=(b"secret",),
now=1700000060,
)
def test_verify_webhook_rejects_expired_replay() -> None:
"""超过时间窗口的合法签名也应被拒绝。"""
body = b"{}"
signature = build_signature(b"secret", "1700000000", "evt_1", body)
with pytest.raises(WebhookVerificationError, match="超出允许窗口"):
verify_webhook(
body=body,
timestamp_text="1700000000",
event_id="evt_1",
signature_header=signature,
secrets=(b"secret",),
now=1700000601,
tolerance_seconds=300,
)
接口层还应补充以下测试：缺失请求头返回 401、请求体超过上限返回 413、两个并发请求只有一个进入业务函数、业务临时失败后允许重试、数据库唯一约束阻止过期后的重复处理。
定位与观测实践
建议记录结构化字段，但不要记录密钥、完整签名、Cookie、Authorization 或原始敏感请求体：
message="Webhook 校验失败" provider="payment" event_id_hash="..." reason="timestamp_expired" source_ip="..."
message="Webhook 重复投递" provider="payment" event_id_hash="..." deduplication_state="completed"
message="Webhook 处理完成" provider="payment" event_type="payment.succeeded" duration_ms=37
event_id_hash 可使用服务端专用盐计算摘要，既能关联重复事件，又避免直接暴露外部标识。重点监控：
webhook_verification_failed_total{reason}：按失败原因统计验签拒绝；webhook_duplicate_total{provider}：观察上游重试或重放异常；webhook_processing_duration_seconds：处理耗时分布；webhook_processing_failed_total{event_type}：业务处理失败；- 队列积压、Redis 错误率和数据库唯一键冲突数。
某来源短时间出现大量 signature_mismatch 或 timestamp_expired 时，应触发告警并结合网关限流。不要仅依赖来源 IP 白名单，因为云平台出口地址可能变化，代理链也可能被错误配置。
修复与上线步骤
- 先与上游确认签名原文、字符编码、摘要算法和多签名格式；
- 在测试环境保存不含敏感信息的固定向量，验证双方实现完全一致；
- 接入时间窗口和事件 ID 去重，初始只记录指标，不立即拦截；
- 观察服务器时钟偏差和上游投递延迟，再确定窗口，通常可从 5 分钟开始；
- 为关键业务表补充唯一约束或合法状态转换，验证最终幂等；
- 开启强制拦截，并对拒绝率、重复率和处理失败率告警；
- 演练密钥轮换：先部署新旧双验，再切换发送方，最后撤销旧密钥。
如果服务需要先返回 2xx 再异步处理，应在响应前把事件可靠写入数据库或持久化消息队列。仅把任务放进进程内存就返回成功，会在进程崩溃时永久丢失事件。
注意事项
- 生产环境必须使用 HTTPS，并校验证书；HMAC 不提供传输机密性；
- 明确请求体上限、请求超时和并发上限，避免验签接口成为拒绝服务入口；
- 服务端时钟应使用可靠 NTP 同步，但不要通过无限放大时间窗口掩盖时钟问题；
- 去重 TTL 至少覆盖上游最大重试周期，并结合数据保留与 Redis 容量评估；
- 多租户场景的去重键必须包含租户或发送方，避免不同来源的事件 ID 冲突；
- 不要对未验签请求返回“事件已存在”等内部状态，避免泄露可枚举信息；
- 第三方若支持非对称签名，优先按其官方协议验证公钥签名，避免自行发明协议。
总结
Webhook 安全不是“算一次 HMAC”就结束。可靠的接收端需要同时满足内容完整性、时间新鲜度、并发去重和业务幂等：原始字节参与签名，时间戳与事件 ID 也必须被签名；Redis 使用原子 SET NX 抵挡并发重复；数据库唯一约束守住最终一致性；日志和指标则帮助区分正常重试、实现错误与恶意重放。
把这些边界一次设计清楚，才能避免出现“每次验签都成功，业务却重复执行”的隐蔽事故。
Discussion
评论