Swytcho Webhooks: Receive Real-Time Event Notifications

import hashlib
import hmac
import time

def verify_webhook(payload_body: bytes, signature_header: str, secret: str) -> bool:
    """
    Verify a Swytcho webhook signature.

    :param payload_body: The raw request body as bytes.
    :param signature_header: The value of the Swytcho-Signature header.
    :param secret: Your endpoint's signing secret.
    :raises ValueError: If the signature is invalid or the timestamp is stale.
    """
    parts = dict(part.split("=", 1) for part in signature_header.split(","))
    timestamp = parts.get("t")
    received_sig = parts.get("v1")

    if not timestamp or not received_sig:
        raise ValueError("Malformed Swytcho-Signature header.")

    # Reject payloads older than 5 minutes to prevent replay attacks
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("Webhook timestamp is too old.")

    signed_payload = f"{timestamp}.".encode() + payload_body
    expected_sig = hmac.new(
        secret.encode(), signed_payload, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected_sig, received_sig):
        raise ValueError("Webhook signature verification failed.")

    return True

import crypto from "crypto";

/**
 * Verify a Swytcho webhook signature.
 * @param {Buffer} payloadBody - The raw request body.
 * @param {string} signatureHeader - The value of the Swytcho-Signature header.
 * @param {string} secret - Your endpoint's signing secret.
 */
function verifyWebhook(payloadBody, signatureHeader, secret) {
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((p) => p.split("="))
  );

  const timestamp = parts["t"];
  const receivedSig = parts["v1"];

  if (!timestamp || !receivedSig) {
    throw new Error("Malformed Swytcho-Signature header.");
  }

  // Reject payloads older than 5 minutes to prevent replay attacks
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp)) > 300) {
    throw new Error("Webhook timestamp is too old.");
  }

  const signedPayload = Buffer.concat([
    Buffer.from(`${timestamp}.`),
    payloadBody,
  ]);

  const expectedSig = crypto
    .createHmac("sha256", secret)
    .update(signedPayload)
    .digest("hex");

  if (!crypto.timingSafeEqual(Buffer.from(expectedSig), Buffer.from(receivedSig))) {
    throw new Error("Webhook signature verification failed.");
  }

  return true;
}