用 Replicate Webhook 接收预测事件：HMAC 验签、重放防护与去重

预测时间长时，与其一直轮询，不如在创建时传 webhook，让 Replicate 在状态变化时 POST 给你。 这篇做 replicate_webhook.py：按官方方法验证 HMAC-SHA256 签名，拒绝过期请求，按 webhook-id 去重，忽略终态之后迟到的事件，并把结果写进 predictions.jsonl。

先看清成品

• 用 Standard Webhooks 的官方测试数据验证了验签实现（同一个签名也出现在 Replicate 文档的示例里）。
• 在 box 上真实运行：5 次投递（处理中、成功、重复、迟到、错误密钥）全部按预期处理。
• 真实问题：直接用测试数据验签返回“时间戳超出容差”，因为那个时间戳是 2021 年的；验签函数因此加了可注入的 now 参数。5 项 pytest 通过。

合规提醒：只用 Replicate 官方 HTTP API 和你自己的 API Token（replicate.com/account/api-tokens），Token 只放服务端，不共享账号。按运行时间计费，价格以 replicate.com/pricing 为准，本文不写价格。每个模型有自己的许可证和使用限制（在模型页面查看），生成内容要遵守 Replicate 服务条款，不生成侵权或违法内容，不冒用真人肖像。

运行说明：我没有 Replicate Token，没有跑过真实预测（也就没有产生费用）。真实跑过的是：用无效 Token 调用官方 API（2026-10-21 21:09–21:40，UTC+8，返回原样贴在文中），以及全部本地代码。webhook 内容和签名是我用自己生成的测试密钥做的，字段照官方 HTTP 参考里的 prediction 对象。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

85-replicate-B-src/
├── replicate_webhook.py
├── send_test_webhook.py
├── samples/
│   ├── processing.json
│   └── succeeded.json
└── test_replicate_webhook.py

第 1 步：接收服务

"""接收 Replicate 预测的 webhook：验签（HMAC-SHA256）、防重放、按 webhook-id 去重、
忽略终态之后迟到的事件，并在成功时记录输出文件（API 生成的文件 1 小时后删除）。

签名内容：{webhook-id}.{webhook-timestamp}.{原始 body}
密钥：GET /v1/webhooks/default/secret 返回的 key，去掉 whsec_ 前缀后 Base64 解码。"""
import base64, hashlib, hmac, json, os, time
from http.server import BaseHTTPRequestHandler, HTTPServer

TOLERANCE = 300      # 时间容差由你决定，官方只要求“在你的容差范围内”
TERMINAL = {"succeeded", "failed", "canceled", "aborted"}   # aborted 见“预测生命周期”页


def verify(secret, headers, body, now=None):
    wid, ts, sigs = headers.get("webhook-id"), headers.get("webhook-timestamp"), headers.get("webhook-signature")
    if not (wid and ts and sigs):
        return False, "缺少 webhook-* 请求头"
    now = int(time.time()) if now is None else now
    try:
        if abs(now - int(ts)) > TOLERANCE:
            return False, "时间戳超出容差"
    except ValueError:
        return False, "时间戳格式错误"
    key = base64.b64decode(secret.split("_", 1)[1] if secret.startswith("whsec_") else secret)
    expected = base64.b64encode(hmac.new(key, f"{wid}.{ts}.".encode() + body, hashlib.sha256).digest()).decode()
    for item in sigs.split():                        # 可能有多个，空格分隔，形如 v1,xxxx
        _, _, sig = item.partition(",")
        if hmac.compare_digest(sig, expected):      # 常量时间比较
            return True, "ok"
    return False, "签名不匹配"


class State:
    def __init__(self, path):
        self.path, self.seen, self.final = path, set(), {}
        if os.path.exists(path):
            for line in open(path, encoding="utf-8"):
                e = json.loads(line)
                self.seen.add(e["webhook_id"])
                if e["status"] in TERMINAL:
                    self.final[e["prediction_id"]] = e["status"]

    def handle(self, wid, pred):
        pid, st = pred["id"], pred["status"]
        if wid in self.seen:
            return f"重复投递 {wid}，跳过"
        self.seen.add(wid)
        if pid in self.final:
            return f"{pid} 已是终态 {self.final[pid]}，忽略迟到的 {st}"
        rec = {"webhook_id": wid, "prediction_id": pid, "status": st, "output": pred.get("output"), "error": pred.get("error")}
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        if st in TERMINAL:
            self.final[pid] = st
        if st == "succeeded":
            return f"{pid} 成功，输出 {pred.get('output')}（1 小时内下载保存）"
        if st == "failed":
            return f"{pid} 失败：{pred.get('error')}"
        return f"{pid} 状态 {st}"


def make_handler(secret, state, log=print):
    class H(BaseHTTPRequestHandler):
        def do_POST(self):
            body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
            ok, why = verify(secret, {k.lower(): v for k, v in self.headers.items()}, body)
            if not ok:
                log(f"拒绝：{why}"); return self._send(401)
            log(state.handle(self.headers["webhook-id"], json.loads(body)))
            self._send(200)      # 尽快回 2xx，否则终态 webhook 会被重试

        def _send(self, code):
            self.send_response(code); self.end_headers()

        def log_message(self, *a):
            pass
    return H


if __name__ == "__main__":
    secret = os.environ["REPLICATE_WEBHOOK_SECRET"]
    port = int(os.environ.get("PORT", "8090"))
    print(f"监听 127.0.0.1:{port}", flush=True)
    HTTPServer(("127.0.0.1", port), make_handler(secret, State("predictions.jsonl"), log=lambda s: print(s, flush=True))).serve_forever()

第 2 步：模拟投递

"""本地模拟 Replicate 投递（用我自己生成的测试密钥签名，不是 Replicate 的）。
用法：python send_test_webhook.py <URL> <payload.json> [webhook-id]"""
import base64, hashlib, hmac, os, sys, time, urllib.request, urllib.error

url, payload = sys.argv[1], sys.argv[2]
wid = sys.argv[3] if len(sys.argv) > 3 else f"msg_{int(time.time() * 1000)}"
secret = os.environ["REPLICATE_WEBHOOK_SECRET"]
key = base64.b64decode(secret.removeprefix("whsec_"))
body, ts = open(payload, "rb").read(), str(int(time.time()))
sig = base64.b64encode(hmac.new(key, f"{wid}.{ts}.".encode() + body, hashlib.sha256).digest()).decode()
req = urllib.request.Request(url, data=body, method="POST", headers={"Content-Type": "application/json",
      "webhook-id": wid, "webhook-timestamp": ts, "webhook-signature": f"v1,{sig}"})
try:
    print(urllib.request.urlopen(req).status)
except urllib.error.HTTPError as e:
    print(e.code)

第 3 步：示例事件

{"id": "ufawqhfynnddngldkgtslldrkq", "model": "black-forest-labs/flux-schnell", "version": "v", "status": "processing", "input": {"prompt": "水彩橘猫"}, "output": null, "error": null, "logs": "", "metrics": {}}
{"id": "ufawqhfynnddngldkgtslldrkq", "model": "black-forest-labs/flux-schnell", "version": "v", "status": "succeeded", "input": {"prompt": "水彩橘猫"}, "output": ["https://replicate.delivery/example/out-0.webp"], "error": null, "logs": "", "metrics": {"predict_time": 1.1}}

第 4 步：测试

import base64, hashlib, hmac, json
import replicate_webhook as w

# Standard Webhooks 官方测试数据（msg id / payload / secret 来自其 Python 库测试），
# 签名 v1,g0hM9SsE… 也出现在 Replicate 文档的 webhook-signature 示例里
VEC = {"webhook-id": "msg_p5jXN8AQM9LWM0D4loKWxJek", "webhook-timestamp": "1614265330",
       "webhook-signature": "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE="}
BODY = b'{"test": 2432232314}'
SECRET = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"


def test_reference_vector():
    assert w.verify(SECRET, VEC, BODY, now=1614265330) == (True, "ok")


def test_old_timestamp_rejected_without_now():
    assert w.verify(SECRET, VEC, BODY) == (False, "时间戳超出容差")


def test_multiple_signatures_one_valid():
    h = dict(VEC, **{"webhook-signature": "v1,bm9ldHUjKzFob2VudXRob2VodWUzMjRvdWVvdW9ldQo= " + VEC["webhook-signature"]})
    assert w.verify(SECRET, h, BODY, now=1614265330)[0]


def test_tampered_body_and_missing_headers():
    assert w.verify(SECRET, VEC, BODY + b" ", now=1614265330) == (False, "签名不匹配")
    assert w.verify(SECRET, {}, BODY)[1] == "缺少 webhook-* 请求头"


def test_state_dedupe_and_ordering(tmp_path):
    s = w.State(str(tmp_path / "p.jsonl"))
    p = lambda st: {"id": "p1", "status": st, "output": ["https://replicate.delivery/x.webp"] if st == "succeeded" else None}
    assert "状态 processing" in s.handle("m1", p("processing"))
    assert "成功" in s.handle("m2", p("succeeded"))
    assert "重复投递" in s.handle("m2", p("succeeded"))
    assert "忽略迟到" in s.handle("m3", p("processing"))
    s2 = w.State(str(tmp_path / "p.jsonl"))           # 重启后仍记得
    assert "重复投递" in s2.handle("m1", p("processing")) and "忽略迟到" in s2.handle("m9", p("processing"))

第 5 步：运行

$ export REPLICATE_WEBHOOK_SECRET="whsec_$(openssl rand -base64 24)" PORT=8090   # 本地测试密钥
$ python replicate_webhook.py &
21:37 UTC+8
$ python send_test_webhook.py http://127.0.0.1:8090/webhooks/replicate samples/processing.json msg_001
200
$ python send_test_webhook.py http://127.0.0.1:8090/webhooks/replicate samples/succeeded.json msg_002
200
$ python send_test_webhook.py http://127.0.0.1:8090/webhooks/replicate samples/succeeded.json msg_002
200
$ python send_test_webhook.py http://127.0.0.1:8090/webhooks/replicate samples/processing.json msg_003
200
$ REPLICATE_WEBHOOK_SECRET=whsec_<另一个密钥> python send_test_webhook.py http://127.0.0.1:8090/webhooks/replicate samples/succeeded.json msg_004
401
# 服务端输出
监听 127.0.0.1:8090
ufawqhfynnddngldkgtslldrkq 状态 processing
ufawqhfynnddngldkgtslldrkq 成功，输出 ['https://replicate.delivery/example/out-0.webp']（1 小时内下载保存）
重复投递 msg_002，跳过
ufawqhfynnddngldkgtslldrkq 已是终态 succeeded，忽略迟到的 processing
拒绝：签名不匹配
$ wc -l predictions.jsonl
2 predictions.jsonl

# 真实密钥接口（无效 Token，21:38 UTC+8）
$ curl -s https://api.replicate.com/v1/webhooks/default/secret -H 'Authorization: Bearer r8_invalid_demo'
{"title":"Unauthenticated","detail":"You did not pass a valid authentication token","status":401}   ← HTTP 401
$ python -m pytest -q
5 passed

怎么检查结果？

• 第 3 次投递（同一个 msg_002）只打印“重复投递”，jsonl 不会多一行。
• 第 4 次是迟到的 processing，被忽略，不会把已成功的预测改回处理中。
• 换一个密钥签名返回 401。
• 成功事件里的输出链接 1 小时后失效：在处理函数里立即下载（可以复用 A 篇的 save_outputs），不要只存链接。

调试示例

1. 用官方测试数据验签，返回“时间戳超出容差”。 我不想只靠自己签、自己验（两边写错同一处也能通过），于是用了 Standard Webhooks 官方 Python 库测试里的数据：msg id msg_p5jXN8AQM9LWM0D4loKWxJek、payload {"test": 2432232314}、secret MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw，签名 v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=（Replicate 文档的示例 header 里第一个签名就是它），时间戳 1614265330。第一次调用：

>>> verify('whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw', h, b'{"test": 2432232314}')
(False, '时间戳超出容差')
>>> verify(..., now=1614265330)
(True, 'ok')

时间戳是 2021 年的，防重放检查生效了。这正说明两项检查是分开的：给 verify() 加上可注入的 now 参数后，签名部分通过，证明 HMAC 的拼接方式和 Base64 处理是对的。

2. 乱序和重复要分别处理。 官方说同一 webhook 可能重发（webhook-id 相同），也可能乱序到达。重发按 webhook-id 去重；乱序按预测 id 记住终态，终态之后来的任何事件都忽略。两者都要落盘，否则服务重启后会忘掉，测试 test_state_dedupe_and_ordering 覆盖了重启的情况。

常见问题

模板规则

• 用原始 body 验签。
• 常量时间比较签名。
• 按 webhook-id 去重，按预测 id 记终态。
• 尽快返回 2xx，重活放到后台。

项目资源

验签先用官方测试数据过一遍，比自己签自己验更可靠。