HMAC signing

const crypto = require('crypto');
const zlib = require('zlib');

function verifyWebhookSignature({
  secret,
  rawBody,            // Buffer holding the body exactly as received
  signatureHeader,    // x-webhook-signature
  idempotencyHeader,  // x-idempotency-key
  contentEncoding,    // content-encoding, if present
  toleranceSeconds = 300,
}) {
  // 1. If the body arrived compressed, decompress it before signing.
  const body = contentEncoding === 'gzip' ? zlib.gunzipSync(rawBody) : rawBody;

  // 2. Pull t and v1 out of the signature header.
  const parts = {};
  for (const pair of signatureHeader.split(',')) {
    const i = pair.indexOf('=');
    parts[pair.slice(0, i).trim()] = pair.slice(i + 1).trim();
  }
  const { t, v1 } = parts;

  // 3. Recover topic, partition, and offset from the idempotency header.
  const last = idempotencyHeader.lastIndexOf(':');
  const prev = idempotencyHeader.lastIndexOf(':', last - 1);
  const topic = idempotencyHeader.slice(0, prev);
  const partition = idempotencyHeader.slice(prev + 1, last);
  const offset = idempotencyHeader.slice(last + 1);

  // 4. Build the canonical string and compute the HMAC.
  const canonical = Buffer.concat([
    Buffer.from(`${t}\n${topic}\n${partition}\n${offset}\n`, 'utf8'),
    body,
  ]);
  const expected = crypto.createHmac('sha256', secret).update(canonical).digest();
  const received = Buffer.from(v1, 'hex');

  // 5. Compare in constant time and reject stale signatures.
  const valid =
    expected.length === received.length && crypto.timingSafeEqual(expected, received);
  const fresh = Math.abs(Math.floor(Date.now() / 1000) - Number(t)) <= toleranceSeconds;

  return valid && fresh;
}

import gzip
import hashlib
import hmac
import time


def verify_webhook_signature(
    secret: str,
    raw_body: bytes,            # the body exactly as received
    signature_header: str,      # x-webhook-signature
    idempotency_header: str,    # x-idempotency-key
    content_encoding: str | None = None,
    tolerance_seconds: int = 300,
) -> bool:
    # 1. If the body arrived compressed, decompress it before signing.
    body = gzip.decompress(raw_body) if content_encoding == "gzip" else raw_body

    # 2. Pull t and v1 out of the signature header.
    parts = dict(p.strip().split("=", 1) for p in signature_header.split(","))
    t, v1 = parts["t"], parts["v1"]

    # 3. Recover topic, partition, and offset from the idempotency header.
    topic, partition, offset = idempotency_header.rsplit(":", 2)

    # 4. Build the canonical string and compute the HMAC.
    canonical = b"\n".join([
        t.encode(), topic.encode(), partition.encode(), offset.encode(), body,
    ])
    expected = hmac.new(secret.encode(), canonical, hashlib.sha256).hexdigest()

    # 5. Compare in constant time and reject stale signatures.
    fresh = abs(int(time.time()) - int(t)) <= tolerance_seconds
    return hmac.compare_digest(expected, v1) and fresh