Webhook signature examples

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;
}

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 False