Webhook

由于回调来自外部，投递本质是至少一次，因此接收端既要容忍重试，也要验证来源，而不是默认它就安全。 事件一到，回调即发 Webhook 把请求的方向整个倒了过来:不再是客户端四处轮询新数据,而是当事件发生时,由提供方往你预先注册的那个地址发一条 HTTP 请求。请求体是一小段自描述的变更说明,直接送去一个经常完全不同的服务器上的接口。因为呼叫由发布方主动发起,你作为接收方就不必一遍遍地重新连接,只需安静等新东西主动送上门,递送因此几乎瞬间可达。 重试与幂等是接收方的事 「重试与幂等」是接收方自己的本分。提供方在没收到你的及时回执时会再发一次,而一旦递达,同一条事件就可能重复出现,于是端点必须具备幂等性。接收方要把已经处理过的事件 id 记下来、把重复当空操作处理,而较重的动作最好丢进队列或后台去做,别活活堵在这条回执里同步傻等。把确认做得快、把去重与清理做得明确,再吵的网络也能变得无害,而不是一转成故障。 验证调用方而不只验 TLS 验证真伪不能只靠 TLS 这一层。由于请求来自公网回调,你必须在收到时取出签名、用共享密钥自己重新算一遍,并借助时间戳把「多旧的请求还接受」收进一个窗口。同时把回调固定在一个明确已知的地址、密钥按计划轮换,因为一个伪造的 webhook 可以做出与真回调完全一样的动作。于是最稳妥的形状是:快速回执、幂等处理、每次都验签名——默认不信任任何请求。 坏掉的样子 投递失败往往是安静发生的。如果发布方只重试两次且没有退避等待,一次短暂发布或冷启动就可能让一条订单支付通知永远丢失,而接收方根本不知道自己错过了什么。更糟的是接收端太慢:当验签、解析与写库全部堵在请求处理函数里同步执行,每次投递都会超时,发送方把每次尝试都记为失败,重试风暴又把同一批积压反复重放。正确做法是先快速回执、再把载荷推进队列、由后台工作线程慢慢消化,这样流量高峰只会表现为队列变长,而不会变成悄无声息的丢失。重试没有指数退避、也没有死信队列时,偶发抖动会被放大成系统性漏单,事后只能靠对账才能发现。 信任断裂的方式更加隐蔽。密钥轮转时如果只更新了发送方、却漏掉某个验签节点,那么所有真回调都会被拒收,在监控上看起来和伪造洪水一模一样。如果没有时间戳窗口和已处理事件编号记录,攻击者可以把截获的支付回调在几天后重放一次,让发货流程重复执行。调试接口若记录完整请求体,还可能泄露个人信息;而先返回成功、后做校验的处理器,则会让投毒事件看起来一切正常。应当把重试不足、时钟偏差、处理太慢和重复到达都当成常态,而不是边角情况。固定回调地址、按计划轮换密钥、拒绝过期与重放请求,都是同一道防线。 快速验证 把下面程序保存到临时目录的 webhook_verify_demo.py,运行 python3 webhook_verify_demo.py。它只用标准库,用 HMAC-SHA256 对带时间戳的消息做签名,并用断言检查五种行为。不需要网络,也不依赖第三方包,不会读写真实文件,适合在本地快速理解验签、容差与去重的配合。 import hashlib import hmac import time SECRET = b"whsec_test_secret_1" OLD_SECRET = b"whsec_old_secret_0" TOLERANCE_SECONDS = 300 seen_ids: set[str] = set() def sign(event_id: str, timestamp: int, payload: bytes, key: bytes = SECRET) -> str: msg = f"{event_id}.{timestamp}.".encode() + payload return hmac.new(key, msg, hashlib.sha256).hexdigest() def verify(event_id: str, timestamp: int, payload: bytes, signature: str, now: int | None = None) -> bool: if now is None: now = int(time.time()) if abs(now - timestamp) > TOLERANCE_SECONDS: return False if event_id in seen_ids: return False expected = sign(event_id, timestamp, payload) if not hmac.compare_digest(expected, signature): return False seen_ids.add(event_id) return True def main() -> None: now = int(time.time()) payload = b'{"type":"order.paid","order":"42"}' fresh = sign("evt_1", now, payload) assert verify("evt_1", now, payload, fresh, now) is True tampered = b'{"type":"order.paid","order":"43"}' assert verify("evt_2", now, tampered, fresh, now) is False stale_sig = sign("evt_3", now - 3600, payload) assert verify("evt_3", now - 3600, payload, stale_sig, now) is False assert verify("evt_1", now, payload, fresh, now) is False rotated_sig = sign("evt_5", now, payload, key=OLD_SECRET) assert verify("evt_5", now, payload, rotated_sig, now) is False print("PASS: fresh accepted; tampered, stale, replay, rotated key rejected") if __name__ == "__main__": main() 程序只接受一次新鲜投递,其余四种坏情况一律拒绝:正文被篡改导致签名对不上、时间戳早了一小时而超出五分钟容差、事件编号重复使用之前已接受的签名、用已轮转掉的老密钥签发。每种拒绝都返回否且不记入已见集合,因此只有第一次投递会进入记录。同样,超出容差的未来时间戳也会被拒绝,这样截获的回调就不会长期有效,重放攻击也就失去立足之地。 ...