Webhooks

import crypto from "node:crypto";

function verifyVestrapaySignature(rawBody, signatureHeader, secret) {
  const expected =
    "sha256=" + crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  const a = Buffer.from(signatureHeader ?? "", "utf8");
  const b = Buffer.from(expected, "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// Express: app.use("/webhooks/vestrapay", express.raw({ type: "application/json" }))
app.post("/webhooks/vestrapay", (req, res) => {
  const rawBody = req.body; // Buffer
  if (!verifyVestrapaySignature(rawBody, req.get("X-VestraPay-Signature"), process.env.VESTRAPAY_WEBHOOK_SECRET)) {
    return res.status(401).end();
  }
  const payload = JSON.parse(rawBody.toString("utf8"));
  res.status(200).end();
  // enqueue: verify payload.data.reference, then fulfill once
});

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

app = Flask(__name__)

def verify_signature(raw: bytes, header: str | None, secret: str) -> bool:
    expected = "sha256=" + hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(header or "", expected)

@app.post("/webhooks/vestrapay")
def webhook():
    raw = request.get_data()
    if not verify_signature(raw, request.headers.get("X-VestraPay-Signature"), os.environ["VESTRAPAY_WEBHOOK_SECRET"]):
        return ("", 401)
    payload = request.get_json(force=True)
    return ("", 200)

$raw = file_get_contents("php://input");
$header = $_SERVER["HTTP_X_VESTRAPAY_SIGNATURE"] ?? "";
$expected = "sha256=" . hash_hmac("sha256", $raw, getenv("VESTRAPAY_WEBHOOK_SECRET"));
if (!hash_equals($expected, $header)) {
  http_response_code(401);
  exit;
}
$payload = json_decode($raw, true);
http_response_code(200);