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  # cabeçalho ausente ou malformado — rejeite em vez de travar
    if abs(time.time() - timestamp_seconds) > tolerance_seconds:
        return False  # desatualizado — rejeite para bloquear 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)

# Só depois que verify() retornar True é que `izap-webhook-id` é seguro como
# chave de idempotência — deduplique aqui, nunca antes.
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("assinatura inválida")
    if already_processed(delivery_id):  # seu próprio armazenamento
        return
    # ... processe o evento e então marque delivery_id como processado

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; // malformado ou desatualizado — rejeite para bloquear 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);
}

// Só depois que verify() retornar true é que `izap-webhook-id` é seguro como
// chave de idempotência — deduplique aqui, nunca antes.
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("assinatura inválida");
  }
  if (alreadyProcessed(deliveryId)) return; // seu próprio armazenamento
  // ... processe o evento e então marque deliveryId como processado
}

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  # cabeçalho ausente ou malformado — rejeite em vez de travar
    if abs(time.time() - timestamp_seconds) > tolerance_seconds:
        return False  # desatualizado — rejeite para bloquear 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; // malformado ou desatualizado — rejeite para bloquear 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);
}