Webhooks

import crypto from "node:crypto"

// Keep the signing secret on your server (env var), never in a repo or browser.
const SECRET = process.env.ZAHREX_WEBHOOK_SECRET
const TOLERANCE_SECONDS = 300 // reject old timestamps (replay protection)

// rawBody must be the exact bytes received, before any JSON parsing.
export function verifyZahrexWebhook(rawBody, headers) {
  const header = headers["zahrex-signature"] ?? ""
  // Header looks like: t=<unix timestamp>,v1=<hex signature>
  const parts = Object.fromEntries(header.split(",").map((p) => p.split("=", 2)))
  const t = Number(parts.t)
  const v1 = parts.v1
  if (!Number.isFinite(t) || !v1) return false

  if (Math.abs(Date.now() / 1000 - t) > TOLERANCE_SECONDS) return false

  const expected = crypto
    .createHmac("sha256", SECRET)
    .update(`${t}.${rawBody}`)
    .digest("hex")

  const a = Buffer.from(expected)
  const b = Buffer.from(v1)
  // Constant-time comparison; lengths must match first or timingSafeEqual throws.
  return a.length === b.length && crypto.timingSafeEqual(a, b)
}

import hashlib
import hmac
import os
import time

# Keep the signing secret on your server (env var), never in a repo or browser.
SECRET = os.environ["ZAHREX_WEBHOOK_SECRET"].encode()
TOLERANCE_SECONDS = 300  # reject old timestamps (replay protection)


def verify_zahrex_webhook(raw_body: bytes, headers: dict) -> bool:
    # raw_body must be the exact bytes received, before any JSON parsing.
    # Header names can arrive in any case, depending on your framework.
    header = {k.lower(): v for k, v in headers.items()}.get("zahrex-signature", "")
    # Header looks like: t=<unix timestamp>,v1=<hex signature>
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts["t"])
        v1 = parts["v1"]
    except (KeyError, ValueError):
        return False

    if abs(time.time() - t) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(SECRET, f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    # Constant-time comparison.
    return hmac.compare_digest(expected, v1)