Webhook Signature Verification

import { Webhook } from "standardwebhooks";

const wh = new Webhook(process.env.MACROPAY_WEBHOOK_SECRET!);

export async function handleWebhook(request: Request): Promise<Response> {
  const body = await request.text(); // raw body — do not pre-parse

  try {
    const event = wh.verify(body, {
      "webhook-id": request.headers.get("webhook-id")!,
      "webhook-timestamp": request.headers.get("webhook-timestamp")!,
      "webhook-signature": request.headers.get("webhook-signature")!,
    });

    // Verified + parsed. Safe to act on.
    if (event.type === "order.paid") {
      await grantLicense(event.data.customer_id);
    }

    return new Response("OK", { status: 200 });
  } catch (err) {
    console.error("Rejected unverified webhook:", err);
    return new Response("Invalid signature", { status: 401 });
  }
}

import os
from standardwebhooks.webhooks import Webhook

wh = Webhook(os.environ["MACROPAY_WEBHOOK_SECRET"])

def handle_webhook(request):
    body = request.body.decode("utf-8")  # raw body — do not pre-parse

    try:
        event = wh.verify(
            body,
            {
                "webhook-id": request.headers["webhook-id"],
                "webhook-timestamp": request.headers["webhook-timestamp"],
                "webhook-signature": request.headers["webhook-signature"],
            },
        )

        # Verified + parsed. Safe to act on.
        if event["type"] == "order.paid":
            grant_license(event["data"]["customer_id"])

        return 200, "OK"
    except Exception as e:
        print(f"Rejected unverified webhook: {e}")
        return 401, "Invalid signature"

npm install standardwebhooks

pip install standardwebhooks

import { createHmac, timingSafeEqual } from "crypto";

function verifyWebhook(
  body: string,
  headers: Record<string, string>,
  secret: string
): boolean {
  const msgId = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];

  // 1. Reject anything outside a 5-minute window (replay protection).
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp, 10)) > 300) {
    throw new Error("Webhook timestamp too old or too new");
  }

  // 2. Rebuild the exact string Macropay signed.
  const signedContent = `${msgId}.${timestamp}.${body}`;

  // 3. Recompute the HMAC with the Base64-decoded secret (drop "whsec_").
  const secretBytes = Buffer.from(secret.replace("whsec_", ""), "base64");
  const expected = createHmac("sha256", secretBytes)
    .update(signedContent)
    .digest("base64");

  // 4. Match against each provided signature; values look like "v1,<base64>".
  for (const sig of signatures.split(" ")) {
    const [, sigValue] = sig.split(",");
    if (
      sigValue &&
      timingSafeEqual(Buffer.from(expected), Buffer.from(sigValue))
    ) {
      return true;
    }
  }

  return false;
}

import base64
import hashlib
import hmac
import time

def verify_webhook(
    body: str,
    headers: dict[str, str],
    secret: str,
) -> bool:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]

    # 1. Reject anything outside a 5-minute window (replay protection).
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("Webhook timestamp too old or too new")

    # 2. Rebuild the exact string Macropay signed.
    signed_content = f"{msg_id}.{timestamp}.{body}"

    # 3. Recompute the HMAC with the Base64-decoded secret (drop "whsec_").
    secret_bytes = base64.b64decode(secret.removeprefix("whsec_"))
    expected = base64.b64encode(
        hmac.new(secret_bytes, signed_content.encode("utf-8"), hashlib.sha256).digest()
    ).decode("utf-8")

    # 4. Match against each provided signature; values look like "v1,<base64>".
    for sig in signatures.split(" "):
        _, _, sig_value = sig.partition(",")
        if sig_value and hmac.compare_digest(expected, sig_value):
            return True

    return False