Webhooks

import crypto from 'crypto';
import express from 'express';

const TOLERANCE_SECONDS = 300; // reject events older than 5 minutes

function verifyQuidkeySignature(rawBody, signatureHeader, secret) {
  const match = signatureHeader?.match(/^t=(\d+),v1=([a-f0-9]+)$/);
  if (!match) throw new Error('Malformed signature header');
  const [, timestamp, providedV1] = match;

  // Reject stale events (replay protection)
  const age = Math.floor(Date.now() / 1000) - Number(timestamp);
  if (Math.abs(age) > TOLERANCE_SECONDS) throw new Error('Timestamp outside tolerance');

  // HMAC over `${timestamp}.${rawBody}`, keyed by the FULL secret (incl. whsec_)
  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${rawBody}`)
    .digest('hex');

  const a = Buffer.from(expected);
  const b = Buffer.from(providedV1);
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    throw new Error('Signature mismatch');
  }
}

const app = express();
app.post(
  '/webhooks/quidkey',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    try {
      verifyQuidkeySignature(
        req.body,                       // raw Buffer
        req.get('x-signature'),         // or req.get('stripe-signature')
        process.env.QUIDKEY_WEBHOOK_SECRET, // full "whsec_..." secret
      );
    } catch (err) {
      return res.status(400).send(`Invalid signature: ${err.message}`);
    }

    const event = JSON.parse(req.body.toString()); // parse only after verifying
    handleEvent(event);
    res.status(200).send('OK');
  },
);

import hashlib
import hmac
import time

TOLERANCE_SECONDS = 300  # reject events older than 5 minutes

def verify_quidkey_signature(raw_body: bytes, signature_header: str, secret: str) -> None:
    # Header format: t=<unix>,v1=<hex-hmac>
    parts = dict(p.split('=', 1) for p in signature_header.split(','))
    timestamp, provided_v1 = parts['t'], parts['v1']

    # Reject stale events (replay protection)
    if abs(int(time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError('Timestamp outside tolerance')

    # HMAC over f"{timestamp}.{raw_body}", keyed by the FULL secret (incl. whsec_)
    signed_payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, provided_v1):
        raise ValueError('Signature mismatch')