Webhooks

import hashlib
import hmac
import time

def verify(
    signing_secret: str,
    delivery_id: str,
    timestamp: str,
    raw_body: bytes,
    header_signature_v2: str,
    tolerance_seconds: int = 300,
) -> bool:
    try:
        timestamp_seconds = int(timestamp)
    except (TypeError, ValueError):
        return False  # missing or malformed header — reject rather than crash
    if abs(time.time() - timestamp_seconds) > tolerance_seconds:
        return False  # stale — reject to block replay
    signed_material = f"{delivery_id}.{timestamp}.".encode() + raw_body
    expected = "v2=" + hmac.new(signing_secret.encode(), signed_material, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header_signature_v2)

# Only once verify() returns True is `izap-webhook-id` safe to use as your
# idempotency key — dedupe here, not before.
def handle(signing_secret: str, headers: dict, raw_body: bytes) -> None:
    delivery_id = headers["izap-webhook-id"]
    if not verify(
        signing_secret,
        delivery_id,
        headers["izap-webhook-timestamp"],
        raw_body,
        headers["izap-webhook-signature-v2"],
    ):
        raise ValueError("invalid signature")
    if already_processed(delivery_id):  # your own store
        return
    # ... handle the event, then mark delivery_id as processed

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

function verify(
  signingSecret: string,
  deliveryId: string,
  timestamp: string,
  rawBody: Buffer,
  headerSignatureV2: string,
  toleranceSeconds = 300,
): boolean {
  const timestampSeconds = Number(timestamp);
  if (!Number.isFinite(timestampSeconds) || Math.abs(Date.now() / 1000 - timestampSeconds) > toleranceSeconds) {
    return false; // malformed or stale — reject to block replay
  }
  const signedMaterial = Buffer.concat([Buffer.from(`${deliveryId}.${timestamp}.`), rawBody]);
  const expected = "v2=" + createHmac("sha256", signingSecret).update(signedMaterial).digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(headerSignatureV2);
  return a.length === b.length && timingSafeEqual(a, b);
}

// Only once verify() returns true is `izap-webhook-id` safe to use as your
// idempotency key — dedupe here, not before.
function handle(signingSecret: string, headers: Record<string, string>, rawBody: Buffer): void {
  const deliveryId = headers["izap-webhook-id"];
  if (
    !verify(
      signingSecret,
      deliveryId,
      headers["izap-webhook-timestamp"],
      rawBody,
      headers["izap-webhook-signature-v2"],
    )
  ) {
    throw new Error("invalid signature");
  }
  if (alreadyProcessed(deliveryId)) return; // your own store
  // ... handle the event, then mark deliveryId as processed
}

import hashlib
import hmac
import time

def verify_v1(signing_secret: str, timestamp: str, raw_body: bytes, header_signature: str, tolerance_seconds: int = 300) -> bool:
    try:
        timestamp_seconds = int(timestamp)
    except (TypeError, ValueError):
        return False  # missing or malformed header — reject rather than crash
    if abs(time.time() - timestamp_seconds) > tolerance_seconds:
        return False  # stale — reject to block replay
    signed_material = f"{timestamp}.".encode() + raw_body
    expected = "v1=" + hmac.new(signing_secret.encode(), signed_material, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header_signature)

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

function verifyV1(
  signingSecret: string,
  timestamp: string,
  rawBody: Buffer,
  headerSignature: string,
  toleranceSeconds = 300,
): boolean {
  const timestampSeconds = Number(timestamp);
  if (!Number.isFinite(timestampSeconds) || Math.abs(Date.now() / 1000 - timestampSeconds) > toleranceSeconds) {
    return false; // malformed or stale — reject to block replay
  }
  const signedMaterial = Buffer.concat([Buffer.from(`${timestamp}.`), rawBody]);
  const expected = "v1=" + createHmac("sha256", signingSecret).update(signedMaterial).digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(headerSignature);
  return a.length === b.length && timingSafeEqual(a, b);
}