用 HMAC-SHA256 校验 Webhook 签名并防重放

问题场景

接收第三方 Webhook 时，接口通常暴露在公网。若只根据 JSON 字段判断来源，攻击者可以伪造请求；若先解析 JSON 再拼接签名，又可能因为空格、键顺序、数字格式变化导致验签失败。常见需求是：读取原始请求体，校验 X-Signature 与 X-Timestamp，拒绝时间戳偏差超过 5 分钟的请求，并尽量阻止同一签名在窗口内重复使用。

本文约定：

• 签名算法：HMAC-SHA256，输出小写十六进制；
• 待签名串：timestamp + '.' + raw_body；
• 请求头：X-Timestamp 为 Unix 秒，X-Signature 为十六进制签名；
• 密钥只放在服务端环境变量中，示例使用占位值。

最小可运行示例

安装依赖并保存为 app.py：

pip install flask requests

import hashlib
import hmac
import json
import os
import time
from collections import OrderedDict
from threading import Lock

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ.get('WEBHOOK_SECRET', 'placeholder-secret').encode('utf-8')
MAX_SKEW = 300
_seen = OrderedDict()
_lock = Lock()

def purge(now):
    while _seen:
        sig, ts = next(iter(_seen.items()))
        if ts >= now - MAX_SKEW:
            break
        _seen.popitem(last=False)

def verify(raw, ts_header, sig_header):
    if not ts_header or not sig_header:
        return False
    try:
        ts = int(ts_header)
    except ValueError:
        return False
    now = int(time.time())
    if abs(now - ts) > MAX_SKEW:
        return False

    msg = ts_header.encode('utf-8') + b'.' + raw
    expected = hmac.new(SECRET, msg, hashlib.sha256).hexdigest()
    provided = sig_header.strip().lower()
    if not hmac.compare_digest(expected, provided):
        return False

    with _lock:
        purge(now)
        if provided in _seen:
            return False
        _seen[provided] = ts
    return True

@app.post('/webhook')
def webhook():
    raw = request.get_data(cache=True)
    ts = request.headers.get('X-Timestamp')
    sig = request.headers.get('X-Signature')
    if not verify(raw, ts, sig):
        abort(401)
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        abort(400)
    # 这里按业务事件 ID 做幂等处理更稳妥
    return {'ok': True, 'event': payload.get('event')}

启动服务：

export WEBHOOK_SECRET=placeholder-secret
flask --app app run --host 127.0.0.1 --port 5000

另开终端发送带签名的请求：

import hashlib
import hmac
import json
import time

import requests

secret = b'placeholder-secret'
body = json.dumps({'event': 'ping'}, separators=(',', ':')).encode('utf-8')
ts = str(int(time.time()))
sig = hmac.new(secret, ts.encode('utf-8') + b'.' + body, hashlib.sha256).hexdigest()
resp = requests.post(
    'http://127.0.0.1:5000/webhook',
    data=body,
    headers={
        'Content-Type': 'application/json',
        'X-Timestamp': ts,
        'X-Signature': sig,
    },
    timeout=5,
)
print(resp.status_code, resp.text)

关键实现说明

1. 必须使用原始请求体。框架解析 JSON 后再序列化，可能改变空白和键顺序，导致签名不一致。示例中 request.get_data(cache=True) 在验签前取得原始字节，再用 json.loads 解析。
2. 时间戳参与签名。只校验请求头时间戳而不签名，攻击者可篡改时间戳延长有效期。将 ts_header + '.' + raw 一起计算 HMAC，才能同时约束内容和时间。
3. 比较使用 hmac.compare_digest。普通 == 可能在首个不同字符处提前返回，不利于侧信道防护。
4. 5 分钟时间窗只限制旧请求。攻击者若在窗口内原样重放，仍可能重复触发业务。示例用签名作为短期去重键，拦截完全相同的重放；生产环境建议结合事件 ID 做幂等表或缓存，并把 TTL 设为时间窗加业务处理余量。
5. 缓存清理。示例按时间顺序清理超过窗口的签名；单进程内存缓存不适合多实例部署，多实例应使用 Redis 等共享存储并设置原子写入。

边界与限制

• 服务器与发送方时钟需同步，否则合法请求可能被 5 分钟窗口拒绝；可监控偏差并使用 NTP。
• 示例从环境变量读取密钥，未包含密钥轮换。生产环境应支持新旧密钥并行验证，轮换后再停用旧密钥。
• 请求体应限制大小，并在反向代理层保留原始 body；若代理改写 body，验签会失败。
• HTTPS 用于传输加密，不能替代签名校验；签名密钥不能出现在前端或日志中。
• 内存去重只适合演示。多实例、重启和并发场景需要共享缓存或数据库唯一约束，并处理缓存不可用时的降级策略。
• 若业务允许同一事件重复投递，应使用事件 ID 做幂等，而不是仅依赖签名去重。