Ideogram 异步任务 Webhook：Ed25519 验签与去重

Ideogram 的异步任务完成后，会把结果 POST 到你提供的 webhook_url，并用 Ed25519 签名。这篇按官方规范实现接收端，主要做这几件事：

• 验签：拿 Generation-Id、User-Id、Timestamp 和 body 的 sha256 拼成待验消息，逐个尝试公钥。
• 公钥缓存：JWKS 缓存 1 小时，验签失败时刷新一次，应对密钥轮换。
• 幂等：按 generation_id 去重。
• 防重放：拒绝超出时间窗口的请求，这一条是本文自加的。

先看清成品（box 上真实运行）

• 抓取了真实的 Ideogram JWKS，得到 kid v1，成功解析成 Ed25519 公钥。
• 用本地生成的密钥对签名，测了 4 种情况：正常投递 200，重复投递 200 并标记 duplicate，body 被篡改 400，时间戳过旧 400。
• pytest 4 项通过。

合规提醒：IDEOGRAM_API_KEY 只放在服务器端。海报里的文字、品牌和人物要有使用权：不要仿冒他人商标或冒充真人；需要时可以开启 enable_copyright_detection，被标记的图会返回 is_image_safe: false。在中国境内公开发布生成内容，要遵守《生成式人工智能服务管理暂行办法》，并按《人工智能生成合成内容标识办法》加上“AI 生成”标识。Ideogram API 按次计费，本文不列价格；需要价格时用 dry_run=true 实时询价，示例中的金额一律写作 x.xx。

运行说明：真实的签名私钥在 Ideogram 手里，box 上也没有 Key，所以没有收到过真实的投递。测试用的是 sign_demo.py 本地生成的密钥对，按官方给出的同一套签名格式模拟。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.9 以上，pip install flask requests cryptography pytest。真实使用时，webhook_url 必须是公网可访问的 HTTPS 地址。

签名规范（来自官方文档）

• 请求头 X-Ideogram-Webhook- 开头的一组字段：Generation-Id、User-Id、Timestamp（Unix 秒）、Key-Id、Signature（小写 hex）。
• 待验消息：Generation-Id \n User-Id \n Timestamp \n sha256_hex(原始 body)，按 UTF-8 编码。
• 公钥：https://api.ideogram.ai/v1/.well-known/jwks.json。Key-Id 只是提示，先用它对应的公钥试，失败了还要试其他公钥。
• 投递失败会重试，但次数有限，不保证一定送达，所以还要用 GET /v2/generations/{generation_id} 轮询兜底。

文件结构

50-ideogram-C-src/
├── webhook_app.py
├── sign_demo.py
└── test_webhook.py

第 1 步：接收端

"""Ideogram webhook 接收端：按官方规范做 Ed25519 验签，
公钥集（JWKS）缓存并在验签失败时刷新一次，按 generation_id 幂等，立刻下载临时图片链接。"""
import base64, hashlib, json, os, sqlite3, time, threading
import requests
from flask import Flask, request, jsonify
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

JWKS_URL = os.environ.get("IDEOGRAM_JWKS_URL") or "https://api.ideogram.ai/v1/.well-known/jwks.json"
JWKS_TTL = 3600
MAX_AGE = int(os.environ.get("WEBHOOK_MAX_AGE", "600"))  # 官方没有规定时间窗口，这是本文自加的防重放
DB = os.environ.get("WEBHOOK_DB", "webhooks.db")
H = "X-Ideogram-Webhook-"
_cache = {"keys": {}, "at": 0}
_lock = threading.Lock()


def _b64url(v):
    return base64.urlsafe_b64decode(v + "=" * (-len(v) % 4))


def load_keys(force=False):
    with _lock:
        if force or time.time() - _cache["at"] > JWKS_TTL or not _cache["keys"]:
            jwks = requests.get(JWKS_URL, timeout=5, headers={"User-Agent": "binbho-tutorial-ideogram-webhook/1.0"}).json()
            _cache["keys"] = {k["kid"]: Ed25519PublicKey.from_public_bytes(_b64url(k["x"]))
                              for k in jwks["keys"] if k.get("crv") == "Ed25519"}
            _cache["at"] = time.time()
        return _cache["keys"]


def _try(keys, kid, sig, msg):
    order = sorted(keys, key=lambda k: k != kid)        # 先试 Key-Id 指向的那把，再试其余
    for k in order:
        try:
            keys[k].verify(sig, msg); return True
        except InvalidSignature:
            continue
    return False


def verify(headers, body: bytes):
    try:
        gid, uid, ts = headers[H + "Generation-Id"], headers[H + "User-Id"], headers[H + "Timestamp"]
        sig = bytes.fromhex(headers[H + "Signature"])
    except (KeyError, ValueError):
        return False, "缺少签名头或签名不是 hex"
    if abs(time.time() - int(ts)) > MAX_AGE:
        return False, "时间戳超出窗口"
    msg = f"{gid}\n{uid}\n{ts}\n{hashlib.sha256(body).hexdigest()}".encode()
    kid = headers.get(H + "Key-Id")
    if _try(load_keys(), kid, sig, msg) or _try(load_keys(force=True), kid, sig, msg):   # 失败时刷新一次，应对密钥轮换
        return True, "ok"
    return False, "签名不匹配"


def db():
    c = sqlite3.connect(DB)
    c.execute("create table if not exists done(gid text primary key, at real, n int)")
    return c


app = Flask(__name__)
app.json.ensure_ascii = False


@app.post("/ideogram-webhook")
def hook():
    body = request.get_data()                       # 原始字节，不能先 get_json 再序列化
    ok, why = verify(request.headers, body)
    if not ok:
        print("✗ 拒绝：", why, flush=True)
        return jsonify(error="invalid signature", reason=why), 400
    p = json.loads(body)
    gid = p["generation_id"]
    with db() as c:
        if c.execute("select 1 from done where gid=?", (gid,)).fetchone():
            print("↷ 重复投递，忽略", gid, flush=True)
            return jsonify(ok=True, duplicate=True)
        saved = 0
        for i, im in enumerate(p.get("data", [])):
            if im.get("is_image_safe") and im.get("url") and os.environ.get("WEBHOOK_DOWNLOAD", "1") == "1":
                os.makedirs("outputs", exist_ok=True)
                with open(f"outputs/{gid}_{i}.png", "wb") as f:
                    f.write(requests.get(im["url"], timeout=60).content)
                saved += 1
        c.execute("insert into done values(?,?,?)", (gid, time.time(), saved))
    print(f"✓ {gid} 已处理，{len(p.get('data', []))} 张，下载 {saved} 张", flush=True)
    return jsonify(ok=True)


if __name__ == "__main__":
    app.run("127.0.0.1", int(os.environ.get("WEBHOOK_PORT", "18502")))   # 别用通用的 PORT，很多环境里已被占用

第 2 步：签名器

"""测试用签名器：生成本地 Ed25519 密钥对，模拟 Ideogram 签名并发送 webhook（只发往本机）。
真实的私钥在 Ideogram 手里，这里的密钥对只用于本地测试。"""
import base64, hashlib, json, sys, time
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization


def new_key(kid):
    sk = Ed25519PrivateKey.generate()
    raw = sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    jwk = {"kty": "OKP", "crv": "Ed25519", "alg": "EdDSA", "use": "sig", "kid": kid,
           "x": base64.urlsafe_b64encode(raw).rstrip(b"=").decode()}
    return sk, jwk


def sign(sk, kid, payload: bytes, gid, uid="dXNlcg", ts=None):
    ts = str(int(time.time()) if ts is None else ts)
    msg = f"{gid}\n{uid}\n{ts}\n{hashlib.sha256(payload).hexdigest()}".encode()
    return {"X-Ideogram-Webhook-Generation-Id": gid, "X-Ideogram-Webhook-User-Id": uid,
            "X-Ideogram-Webhook-Timestamp": ts, "X-Ideogram-Webhook-Key-Id": kid,
            "X-Ideogram-Webhook-Signature": sk.sign(msg).hex(), "Content-Type": "application/json"}


if __name__ == "__main__":
    # 用法：python sign_demo.py jwks_dir http://127.0.0.1:18502/ideogram-webhook
    import pathlib, requests
    d, url = pathlib.Path(sys.argv[1]), sys.argv[2]
    sk, jwk = new_key("local-1")
    d.mkdir(exist_ok=True); (d / "jwks.json").write_text(json.dumps({"keys": [jwk]}))
    gid = "demo" + str(int(time.time()))
    body = json.dumps({"generation_id": gid, "created": "2026-10-21T00:00:00Z", "data": [
        {"url": "", "prompt": "p", "resolution": "1024x1024", "seed": 7, "is_image_safe": False}]}).encode()
    h = sign(sk, "local-1", body, gid)
    cases = [("正常投递", body, h), ("重复投递", body, h), ("body 被改", body.replace(b"7", b"8"), h),
             ("1 小时前的时间戳", body, sign(sk, "local-1", body, gid, ts=int(time.time()) - 3600))]
    for i, (name, b, hh) in enumerate(cases, 1):
        r = requests.post(url, data=b, headers=hh, timeout=10)
        print(f"{i} {name:<12} → HTTP {r.status_code} {r.text.strip()}")

第 3 步：测试

import json, os, time, importlib, pytest
from sign_demo import new_key, sign


@pytest.fixture()
def env(tmp_path, monkeypatch):
    monkeypatch.setenv("WEBHOOK_DB", str(tmp_path / "w.db"))
    monkeypatch.setenv("WEBHOOK_DOWNLOAD", "0")
    import webhook_app; importlib.reload(webhook_app)
    keys = {}
    monkeypatch.setattr(webhook_app, "load_keys", lambda force=False: {
        k: __import__("cryptography.hazmat.primitives.asymmetric.ed25519", fromlist=["x"]).Ed25519PublicKey
        .from_public_bytes(webhook_app._b64url(j["x"])) for k, j in keys.items()})
    return webhook_app, webhook_app.app.test_client(), keys


def payload(gid):
    return json.dumps({"generation_id": gid, "created": "2026-10-21T00:00:00Z",
                       "data": [{"url": "", "prompt": "p", "resolution": "1024x1024", "seed": 1, "is_image_safe": False}]},
                      ensure_ascii=False, indent=2).encode()


def test_valid_then_duplicate(env):
    app, c, keys = env
    sk, jwk = new_key("v1"); keys["v1"] = jwk
    b = payload("g1"); h = sign(sk, "v1", b, "g1")
    assert c.post("/ideogram-webhook", data=b, headers=h).status_code == 200
    r = c.post("/ideogram-webhook", data=b, headers=h)
    assert r.status_code == 200 and r.json["duplicate"]


def test_reserialized_body_fails(env):
    app, c, keys = env
    sk, jwk = new_key("v1"); keys["v1"] = jwk
    b = payload("g2"); h = sign(sk, "v1", b, "g2")
    compact = json.dumps(json.loads(b), separators=(",", ":")).encode()   # 解析后重新序列化
    assert c.post("/ideogram-webhook", data=compact, headers=h).status_code == 400


def test_rotation_kid_hint_is_only_a_hint(env):
    app, c, keys = env
    old, jo = new_key("v1"); new, jn = new_key("v2"); keys.update(v1=jo, v2=jn)
    b = payload("g3"); h = sign(new, "v1", b, "g3")     # Key-Id 写的是 v1，实际用 v2 签
    assert c.post("/ideogram-webhook", data=b, headers=h).status_code == 200


def test_old_timestamp_and_foreign_key(env):
    app, c, keys = env
    sk, jwk = new_key("v1"); keys["v1"] = jwk
    b = payload("g4")
    assert c.post("/ideogram-webhook", data=b, headers=sign(sk, "v1", b, "g4", ts=int(time.time()) - 3600)).status_code == 400
    evil, _ = new_key("v1")
    assert c.post("/ideogram-webhook", data=b, headers=sign(evil, "v1", b, "g4")).status_code == 400

第 4 步：运行

$ python -c "import webhook_app as w; k=w.load_keys(); print('真实 JWKS kid：', list(k), type(list(k.values())[0]).__name__)"
真实 JWKS kid： ['v1'] Ed25519PublicKey

# 本地演示：用本地 JWKS 代替官方的
$ mkdir jw && (cd jw && python3 -m http.server 18503 --bind 127.0.0.1 &)
$ IDEOGRAM_JWKS_URL=http://127.0.0.1:18503/jwks.json python webhook_app.py &
$ python sign_demo.py jw http://127.0.0.1:18502/ideogram-webhook
1 正常投递         → HTTP 200 {"ok":true}
2 重复投递         → HTTP 200 {"duplicate":true,"ok":true}
3 body 被改      → HTTP 400 {"error":"invalid signature","reason":"签名不匹配"}
4 1 小时前的时间戳    → HTTP 400 {"error":"invalid signature","reason":"时间戳超出窗口"}

# 接收端日志
✓ demo1792575608 已处理，1 张，下载 0 张
↷ 重复投递，忽略 demo1792575608
✗ 拒绝： 签名不匹配
✗ 拒绝： 时间戳超出窗口

$ python -m pytest -q
4 passed in 0.15s

调试示例

1. 服务器启动在了 3099 端口。 第一版用 PORT 环境变量指定端口，默认 18502，结果演示脚本一直报 Connection refused。看日志才发现服务跑在 http://127.0.0.1:3099：box 上本来就设置了 PORT=3099。PORT 这种通用变量名，在很多托管平台和容器里都会预先设置，所以改用专门的 WEBHOOK_PORT。

2. 重新序列化 JSON 后验签失败。 test_reserialized_body_fails 复现了这个问题：同样的内容，换成紧凑格式重新序列化后，sha256 变了，签名自然对不上。所以接收端只能用 request.get_data() 拿到的原始字节做验签。

常见问题

模板规则

• 验签用原始字节；Key-Id 只作提示，所有公钥都要试一遍。
• JWKS 要缓存，失败时刷新。
• 幂等处理加上轮询兜底，回调处理要快。

项目资源

接收端验签可靠，再加上轮询兜底，异步生成就可以放心接进生产流程了。