Webhooks na prática

POST https://api.suaempresa.com/WovePay/webhook

curl -X POST https://api.wovepay.com/v1/webhooks/create \
  -H "X-API-Key: wp_live_SUA_CHAVE" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://api.suaempresa.com/WovePay/webhook",
    "events": [
      "payment.paid",
      "payment.created",
      "payment.pix.expired",
      "transfer.completed",
      "payment_link.paid"
    ]
  }'

import crypto from "node:crypto";
import express from "express";

const app = express();

// Raw body obrigatório para HMAC
app.post(
  "/WovePay/webhook",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["x-WovePay-signature"] as string | undefined;
    const secret = process.env.WovePay_WEBHOOK_SECRET!;

    if (!verifySignature(req.body, signature, secret)) {
      return res.status(401).send("Invalid signature");
    }

    const event = JSON.parse(req.body.toString());
    const deliveryId = event.id;

    if (await alreadyProcessed(deliveryId)) {
      return res.status(200).json({ ok: true });
    }

    switch (event.event) {
      case "payment.paid":
        await handlePaymentPaid(event.data);
        break;
      case "payment.pix.expired":
        await handlePixExpired(event.data);
        break;
      // ...
    }

    await markProcessed(deliveryId);
    return res.status(200).json({ ok: true });
  },
);

function verifySignature(
  rawBody: Buffer,
  signatureHeader: string | undefined,
  webhookSecret: string,
): boolean {
  if (!signatureHeader?.startsWith("sha256=")) return false;
  const received = signatureHeader.slice("sha256=".length);
  const expected = crypto
    .createHmac("sha256", webhookSecret)
    .update(rawBody)
    .digest("hex");
  const a = Buffer.from(expected, "hex");
  const b = Buffer.from(received, "hex");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/WovePay/webhook")
async def WovePay_webhook(request: Request):
    raw = await request.body()
    sig = request.headers.get("x-WovePay-signature", "")
    secret = os.environ["WovePay_WEBHOOK_SECRET"]

    if not verify(raw, sig, secret):
        raise HTTPException(401, "Invalid signature")

    event = json.loads(raw)
    # idempotência + processamento...
    return {"ok": True}

def verify(raw: bytes, header: str, secret: str) -> bool:
    if not header.startswith("sha256="):
        return False
    received = header[7:]
    expected = hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, received)

if (await db.webhookDelivery.exists(deliveryId)) {
  return res.status(200).json({ ok: true });
}