Webhooks

import crypto from "crypto";

function verifyWebhookSignature(
  payload: string,
  headers: Record<string, string>,
  secret: string
): boolean {
  const timestamp = headers["webhook-timestamp"];
  const signature = headers["webhook-signature"];

  if (!timestamp || !signature) return false;

  // Reject requests older than 5 minutes
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > 300) return false;

  const message = `${timestamp}.${payload}`;
  const expected = crypto
    .createHmac("sha256", secret)
    .update(message)
    .digest("base64");

  // During secret rotation, multiple signatures may be
  // space-delimited (one per active secret). Try each.
  const signatures = signature.split(" ");
  return signatures.some((sig) => {
    const received = sig.replace("v1,", "");
    return crypto.timingSafeEqual(
      Buffer.from(expected),
      Buffer.from(received)
    );
  });
}

// Usage in an Express handler
app.post("/webhook", (req, res) => {
  const rawBody = req.body; // use raw body string, not parsed JSON
  const isValid = verifyWebhookSignature(
    rawBody,
    req.headers,
    process.env.WEBHOOK_SECRET
  );

  if (!isValid) {
    return res.status(401).send("Invalid signature");
  }

  const event = JSON.parse(rawBody);

  switch (event.type) {
    case "user.created":
      // handle user creation
      break;
    case "wallet.created":
      // handle wallet creation
      break;
    // ...
  }

  res.status(200).send("OK");
});

import hmac
import hashlib
import base64
import time
import json

def verify_webhook_signature(payload: str, headers: dict, secret: str) -> bool:
    timestamp = headers.get("webhook-timestamp", "")
    signature = headers.get("webhook-signature", "")

    if not timestamp or not signature:
        return False

    # Reject requests older than 5 minutes
    if abs(time.time() - int(timestamp)) > 300:
        return False

    message = f"{timestamp}.{payload}"
    expected = base64.b64encode(
        hmac.new(secret.encode(), message.encode(), hashlib.sha256).digest()
    ).decode()

    # During secret rotation, multiple signatures may be
    # space-delimited (one per active secret). Try each.
    signatures = signature.split(" ")
    return any(
        hmac.compare_digest(expected, sig.replace("v1,", ""))
        for sig in signatures
    )

# Usage in a Flask handler
@app.route("/webhook", methods=["POST"])
def webhook():
    raw_body = request.get_data(as_text=True)
    is_valid = verify_webhook_signature(
        raw_body,
        request.headers,
        os.environ["WEBHOOK_SECRET"],
    )

    if not is_valid:
        return "Invalid signature", 401

    event = json.loads(raw_body)

    if event["type"] == "user.created":
        pass  # handle user creation
    elif event["type"] == "wallet.created":
        pass  # handle wallet creation

    return "OK", 200