Webhooks

import crypto from "crypto"

function verifyWebhook(rawBody, signatureHeader, secret) {
  const expected = crypto
    .createHmac("sha256", secret)
    .update(rawBody)          // rawBody must be the raw Buffer, not parsed JSON
    .digest("hex")

  const trusted = Buffer.from(`sha256=${expected}`, "utf8")
  const received = Buffer.from(signatureHeader, "utf8")

  // Constant-time comparison prevents timing attacks
  if (trusted.length !== received.length) return false
  return crypto.timingSafeEqual(trusted, received)
}

// Express.js example
app.post("/webhooks/ababilpay", express.raw({ type: "application/json" }), (req, res) => {
  const sig = req.headers["ababilpay-signature"]

  if (!verifyWebhook(req.body, sig, process.env.ABABILPAY_WEBHOOK_SECRET)) {
    return res.status(401).send("Invalid signature")
  }

  const event = JSON.parse(req.body)

  switch (event.event) {
    case "payment.completed":
      // fulfil order using event.data.order_id
      break
    case "invoice.paid":
      // mark invoice settled
      break
  }

  res.status(200).send("ok")
})

import hmac
import hashlib
import os
from flask import Flask, request, abort

app = Flask(__name__)

def verify_webhook(raw_body: bytes, signature_header: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header)

@app.route("/webhooks/ababilpay", methods=["POST"])
def webhook():
    sig = request.headers.get("AbabilPay-Signature", "")
    raw = request.get_data()

    if not verify_webhook(raw, sig, os.environ["ABABILPAY_WEBHOOK_SECRET"]):
        abort(401)

    event = request.get_json(force=True)

    if event["event"] == "payment.completed":
        pass  # fulfil order using event["data"]["order_id"]
    elif event["event"] == "invoice.paid":
        pass  # mark invoice settled

    return "ok", 200