Verifikasi tiap POST yang masuk ke URL webhook kamu sebelum kamu proses body-nya.
Yang kamu butuhin
- Baca raw body — byte mentah dari request, jangan kamu stringify ulang JSON-nya.
- Baca header signature dari request (biasanya
webhook-id/svix-id,webhook-timestamp/svix-timestamp, danwebhook-signature/svix-signature). - Pakai endpoint signing secret dari Wazapin (formatnya
whsec_…). Simpan sebagai secret di server, jangan taruh di code front-end. - Tolak request yang timestamp-nya udah kelewat jauh — biar aman dari replay attack.
- Bandingin signature yang kamu hitung sama yang dikirim pakai constant-time comparison.
Skema signing-nya ngikutin standar Svix yang dipakai buat outbound delivery. Kamu bisa pakai library resmi Svix atau ikutin langkah HMAC yang sama di bawah.
Node.js
import { Webhook } from 'svix';
const wh = new Webhook(process.env.WAZAPIN_WEBHOOK_SECRET);
app.post('/webhooks/wazapin', express.raw({ type: 'application/json' }), (req, res) => {
try {
wh.verify(req.body, req.headers);
} catch {
return res.status(403).send('invalid signature');
}
const event = JSON.parse(req.body.toString('utf8'));
res.status(200).send('ok');
});import crypto from 'crypto';
function verifyWazapinWebhook(rawBody, headers, secret) {
const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
const msgId = headers['svix-id'] || headers['webhook-id'];
const timestamp = headers['svix-timestamp'] || headers['webhook-timestamp'];
const sigHeader = headers['svix-signature'] || headers['webhook-signature'];
if (!msgId || !timestamp || !sigHeader) return false;
const signed = `${msgId}.${timestamp}.${rawBody.toString('utf8')}`;
const expected = crypto.createHmac('sha256', key).update(signed).digest('base64');
for (const part of sigHeader.split(' ')) {
const [version, sig] = part.split(',');
if (version !== 'v1' || !sig) continue;
const a = Buffer.from(sig);
const b = Buffer.from(expected);
if (a.length === b.length && crypto.timingSafeEqual(a, b)) return true;
}
return false;
}Python
from svix.webhooks import Webhook, WebhookVerificationError
wh = Webhook(os.environ["WAZAPIN_WEBHOOK_SECRET"])
@app.post("/webhooks/wazapin")
async def wazapin_webhook(request: Request):
payload = await request.body()
try:
wh.verify(payload, dict(request.headers))
except WebhookVerificationError:
raise HTTPException(status_code=403)
return {"ok": True}import hmac
import hashlib
import base64
def verify_wazapin_webhook(raw_body: bytes, headers: dict, secret: str) -> bool:
key = base64.b64decode(secret.removeprefix("whsec_"))
msg_id = headers.get("svix-id") or headers.get("webhook-id")
timestamp = headers.get("svix-timestamp") or headers.get("webhook-timestamp")
sig_header = headers.get("svix-signature") or headers.get("webhook-signature")
if not msg_id or not timestamp or not sig_header:
return False
signed = f"{msg_id}.{timestamp}.{raw_body.decode('utf-8')}".encode("utf-8")
expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode("ascii")
for part in sig_header.split():
version, sig = part.split(",", 1)
if version == "v1" and hmac.compare_digest(sig, expected):
return True
return FalseGo
Pakai github.com/svix/svix-webhooks dengan secret whsec_ endpoint kamu, atau bikin sendiri dengan format yang sama: msgID + "." + timestamp + "." + string(body) pakai HMAC-SHA256 dan base64, terus cocokin sama entry v1 di header signature.
Kalau gagal gimana?
- Balas
403kalau signature-nya nggak valid. - Balas
400kalau JSON-nya rusak setelah verifikasi berhasil. - Balas
200kalau event-nya dobel (udah kamu proses sebelumnya) — habis cek idempotency.