Webhook
Terima event yang sama dengan SSE lewat HTTP POST ke server Anda, ditandatangani HMAC-SHA256.
Mendaftarkan endpoint
- Developer Portal → Webhook → Tambah subscription: URL HTTPS publik (port 443) dan event yang dilanggan.
- Secret ditampilkan sekali; simpan di environment server Anda.
- Respons 2xx = diterima. Gagal → dicoba ulang maksimal 8 kali dengan backoff; setelahnya status
dead. Redirect tidak diikuti.
Verifikasi tanda tangan
Header: x-webhook-id (ID event, untuk deduplikasi), x-webhook-timestamp (detik Unix), x-webhook-signature (v1=<hex>). Tanda tangan = HMAC-SHA256 dengan secret atas timestamp + "." + raw body. Bandingkan secara constant-time dan tolak timestamp di luar ±5 menit.
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