Verifying signatures

import {Custral} from "@custral/sdk";

const custral = new Custral({webhookSecret: process.env.CUSTRAL_WEBHOOK_SECRET!});

// Throws CustralSignatureVerificationError on a bad signature, a stale
// timestamp, a missing header, or a body that is not valid JSON.
const event = custral.webhooks.constructEvent(rawBody, signatureHeader);

import crypto from "node:crypto";

export function verifyCustralWebhook(rawBody: string, header: string, secret: string): boolean {
  let timestamp: number | null = null;
  let signature: string | null = null;
  for (const part of header.split(",")) {
    const eq = part.indexOf("=");
    if (eq === -1) continue;
    const key = part.slice(0, eq).trim();
    const value = part.slice(eq + 1).trim();
    if (key === "t") timestamp = Number(value);
    if (key === "v1") signature = value;
  }
  if (timestamp === null || Number.isNaN(timestamp) || !signature) return false;

  if (Math.abs(Math.floor(Date.now() / 1000) - timestamp) > 300) return false;

  const expected = crypto.createHmac("sha256", secret).update(`${timestamp}.${rawBody}`).digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(signature);
  if (a.length !== b.length) return false;
  return crypto.timingSafeEqual(a, b);
}

import hashlib, hmac, time

def verify_custral_webhook(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    try:
        timestamp = int(parts["t"])
        signature = parts["v1"]
    except (KeyError, ValueError):
        return False

    if abs(int(time.time()) - timestamp) > tolerance:
        return False

    signed = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

require "openssl"

def verify_custral_webhook(raw_body, header, secret, tolerance = 300)
  parts = header.split(",").map { |p| p.split("=", 2) }.to_h
  timestamp = Integer(parts["t"], exception: false)
  signature = parts["v1"]
  return false if timestamp.nil? || signature.nil?
  return false if (Time.now.to_i - timestamp).abs > tolerance

  expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{timestamp}.#{raw_body}")
  OpenSSL.secure_compare(expected, signature)
end