Webhooks

TypeScript

import { createServer } from "node:http";
import { verifyWebhook, WebhookSignatureError } from "@ravelo/sdk";

const secret = process.env.RAVELO_WEBHOOK_SECRET!;
const handled = new Set<string>(); // keep this in your database; deliveries can repeat

createServer(async (req, res) => {
  if (req.method !== "POST" || req.url !== "/webhooks/ravelo") {
    res.writeHead(404).end();
    return;
  }
  // Verify the bytes exactly as they arrived; parsing and re-serialising the JSON would break the signature.
  const chunks: Buffer[] = [];
  for await (const chunk of req) chunks.push(chunk as Buffer);
  try {
    const event = await verifyWebhook(Buffer.concat(chunks), req.headers["ravelo-signature"] as string | undefined, secret);
    if (!handled.has(event.id)) {
      handled.add(event.id);
      console.log("verified", event.type, event.id);
      if (event.type === "payment_intent.succeeded") console.log("paid:", event.data.id);
    }
    res.writeHead(200).end();
  } catch (err) {
    console.error("rejected:", err instanceof Error ? err.message : err);
    res.writeHead(err instanceof WebhookSignatureError ? 400 : 500).end();
  }
}).listen(3000, () => console.log("listening on http://localhost:3000/webhooks/ravelo"));

Python

import os
from http.server import BaseHTTPRequestHandler, HTTPServer

from ravelo import SIGNATURE_HEADER, WebhookSignatureError, verify_webhook

SECRET = os.environ["RAVELO_WEBHOOK_SECRET"]
handled = set()  # keep this in your database; deliveries can repeat


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != "/webhooks/ravelo":
            self.send_response(404)
            self.end_headers()
            return
        # Verify the bytes exactly as they arrived; parsing and re-serialising the JSON would break the signature.
        raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        try:
            event = verify_webhook(raw, self.headers.get(SIGNATURE_HEADER), SECRET)
        except WebhookSignatureError as err:
            print("rejected:", err)
            self.send_response(400)
            self.end_headers()
            return
        if event["id"] not in handled:
            handled.add(event["id"])
            print("verified", event["type"], event["id"])
            if event["type"] == "payment_intent.succeeded":
                print("paid:", event["data"]["id"])
        self.send_response(200)
        self.end_headers()


print("listening on http://localhost:3000/webhooks/ravelo")
HTTPServer(("", 3000), Handler).serve_forever()

TypeScript

import { createHmac, timingSafeEqual } from "node:crypto";

/** Verifies a Ravelo-Signature header without the SDK. `rawBody` is the request body exactly as received. */
export function verifyManually(rawBody: string, header: string, secret: string, nowMs = Date.now(), toleranceSeconds = 300): boolean {
  // kid and any name you do not know are ignored; any v1 may match, which is how a rotation overlap verifies.
  const parts = header.split(",").map((item) => item.trim().split("=", 2) as [string, string]);
  const t = parts.find(([k]) => k === "t")?.[1];
  const signatures = parts.filter(([k]) => k === "v1").map(([, v]) => v);
  if (!t || !/^\d+$/.test(t) || signatures.length === 0) return false;
  if (Math.abs(Math.floor(nowMs / 1000) - Number(t)) > toleranceSeconds) return false;

  const expected = createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");
  return signatures.some((s) => s.length === expected.length && timingSafeEqual(Buffer.from(s), Buffer.from(expected)));
}

Python

import hashlib
import hmac
import time


def verify_manually(raw_body: bytes, header: str, secret: str, now: float = None, tolerance_seconds: int = 300) -> bool:
    """Verifies a Ravelo-Signature header without the SDK. raw_body is the request body exactly as received."""
    # kid and any name you do not know are ignored; any v1 may match, which is how a rotation overlap verifies.
    pairs = [item.strip().partition("=") for item in header.split(",")]
    t = next((v for k, _, v in pairs if k == "t"), None)
    signatures = [v for k, _, v in pairs if k == "v1"]
    if not t or not t.isdigit() or not signatures:
        return False
    if abs(int(now if now is not None else time.time()) - int(t)) > tolerance_seconds:
        return False

    expected = hmac.new(secret.encode(), t.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(s, expected) for s in signatures)

curl

# Issue a new secret and keep the old one signing for one hour (the default is 24 hours).
# The response is the endpoint with its new "secret", shown once.
curl -sS "$RAVELO_API/v1/webhook-endpoints/$ENDPOINT_ID/rotate-secret" \
  -H "Authorization: Bearer $RAVELO_TEST_KEY" \
  -H "Idempotency-Key: $(uuidgen)" \
  -H "Content-Type: application/json" \
  -d '{ "overlap_seconds": 3600 }'

# Once every receiver uses the new secret, stop the old one signing:
curl -sS "$RAVELO_API/v1/webhook-endpoints/$ENDPOINT_ID/end-rotation" \
  -H "Authorization: Bearer $RAVELO_TEST_KEY" \
  -H "Idempotency-Key: $(uuidgen)" \
  -X POST