Webhook
通过 HTTP POST 将与 SSE 相同的事件发送到您的服务器，使用 HMAC-SHA256 签名。
注册端点
- 开发者门户 → Webhook → 添加订阅：公开的 HTTPS URL（443 端口）和订阅的事件。
- Secret 仅显示一次；请保存在服务器环境变量中。
- 返回 2xx 表示已接收。失败时最多以退避方式重试 8 次；之后状态为
dead。不会跟随重定向。
验证签名
请求头：x-webhook-id（事件 ID，用于去重）、x-webhook-timestamp（Unix 秒）、x-webhook-signature（v1=<hex>）。签名 = 使用 secret 对 timestamp + "." + 原始请求体 计算的 HMAC-SHA256。请使用常量时间比较，并拒绝超出 ±5 分钟的时间戳。
ts
import { createHmac, timingSafeEqual } from 'node:crypto';
import express from 'express';
const app = express();
app.post('/banipool-webhook', express.raw({ type: 'application/json' }), (req, res) => {
const ts = req.header('x-webhook-timestamp') ?? '';
const sig = (req.header('x-webhook-signature') ?? '').replace(/^v1=/, '');
const expected = createHmac('sha256', process.env.BANIPOOL_WEBHOOK_SECRET!)
.update(`${ts}.${req.body.toString('utf8')}`).digest('hex');
const fresh = Math.abs(Date.now() / 1000 - Number(ts)) < 300;
const ok = fresh && sig.length === expected.length && timingSafeEqual(Buffer.from(sig), Buffer.from(expected));
if (!ok) return res.status(401).end();
const event = JSON.parse(req.body.toString('utf8')); // { id, type, data, createdAt }
// Deduplikasi: simpan event.id yang sudah diproses
console.log(event.type, event.data);
res.status(204).end();
});
app.listen(3000);python
import hmac, hashlib, os, time, json
from flask import Flask, request, abort
app = Flask(__name__)
SECRET = os.environ["BANIPOOL_WEBHOOK_SECRET"].encode()
@app.post("/banipool-webhook")
def webhook():
ts = request.headers.get("x-webhook-timestamp", "")
sig = request.headers.get("x-webhook-signature", "").removeprefix("v1=")
raw = request.get_data()
expected = hmac.new(SECRET, f"{ts}.".encode() + raw, hashlib.sha256).hexdigest()
if abs(time.time() - int(ts or 0)) > 300 or not hmac.compare_digest(sig, expected):
abort(401)
event = json.loads(raw) # {id, type, data, createdAt}
print(event["type"], event["data"])
return "", 204