Monitoring Webhooks

import hashlib
import hmac
import time

def verify_webhook(secret: str, signature_header: str, raw_body: bytes,
                   tolerance_seconds: int = 300) -> bool:
    parts = dict(p.split("=", 1) for p in signature_header.split(","))
    timestamp, signature = int(parts["t"]), parts["v1"]

    if abs(time.time() - timestamp) > tolerance_seconds:
        return False  # stale — possible replay

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

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

function verifyWebhook(
  secret: string,
  signatureHeader: string,
  rawBody: string,
  toleranceSeconds = 300,
): boolean {
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((p) => p.split("=", 2) as [string, string]),
  );
  const timestamp = Number(parts.t);
  if (Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;

  const expected = createHmac("sha256", secret)
    .update(`${timestamp}.${rawBody}`)
    .digest("hex");
  return timingSafeEqual(Buffer.from(expected), Buffer.from(parts.v1));
}