Webhooks

import { createHmac, timingSafeEqual } from 'node:crypto'

export function parseHeader(
  signatureHeader: string | undefined,
): { t: string; v1: string } | null {
  if (typeof signatureHeader !== 'string' || !signatureHeader) return null
  const parts: Record<string, string> = Object.create(null)
  for (const segment of signatureHeader.split(',')) {
    const eq = segment.indexOf('=')
    if (eq <= 0) continue
    parts[segment.slice(0, eq).trim()] = segment.slice(eq + 1).trim()
  }
  if (!parts.t || !parts.v1) return null
  return { t: parts.t, v1: parts.v1 }
}

function hexToBuffer(hex: string): Buffer | null {
  if (!hex || hex.length % 2 !== 0 || !/^[0-9a-fA-F]+$/.test(hex)) return null
  try {
    return Buffer.from(hex, 'hex')
  } catch {
    return null
  }
}

export function verifyProceedsWebhook({
  rawBody,
  signatureHeader,
  secret,
}: {
  rawBody: Buffer | string
  signatureHeader: string | undefined
  secret: string
}): boolean {
  if (!secret.startsWith('whsec_')) {
    throw new TypeError('secret must start with whsec_')
  }
  const parsed = parseHeader(signatureHeader)
  if (!parsed) return false
  const provided = hexToBuffer(parsed.v1)
  if (!provided) return false
  const bodyUtf8 = Buffer.isBuffer(rawBody) ? rawBody.toString('utf8') : String(rawBody)
  const expectedHex = createHmac('sha256', secret)
    .update(`${parsed.t}.${bodyUtf8}`, 'utf8')
    .digest('hex')
  const expected = Buffer.from(expectedHex, 'hex')
  if (provided.length !== expected.length) return false
  return timingSafeEqual(provided, expected)
}

import hashlib
import hmac


def verify_proceeds_webhook(
    *,
    raw_body: bytes | str,
    signature_header: str | None,
    secret: str,
) -> bool:
    if not secret.startswith("whsec_"):
        raise TypeError("secret must start with whsec_")
    if not isinstance(signature_header, str) or not signature_header:
        return False

    parts: dict[str, str] = {}
    for segment in signature_header.split(","):
        eq = segment.find("=")
        if eq <= 0:
            continue
        key = segment[:eq].strip()
        value = segment[eq + 1 :].strip()
        if key:
            parts[key] = value

    t = parts.get("t")
    v1 = parts.get("v1")
    if not t or not v1:
        return False

    body_utf8 = (
        raw_body.decode("utf-8")
        if isinstance(raw_body, (bytes, bytearray))
        else str(raw_body)
    )
    signed_string = f"{t}.{body_utf8}".encode("utf-8")
    expected = hmac.new(secret.encode("utf-8"), signed_string, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1.lower())