Webhook Signature Verification

const crypto = require('crypto');

function verifyWebhookSignature(req, secret) {
  const sigHeader = req.headers['x-hopae-signature'];
  if (!sigHeader) {
    throw new Error('Missing X-Hopae-Signature header');
  }

  // Parse t= and v1= fields
  const parts = Object.fromEntries(
    sigHeader.split(',').map((p) => p.split('='))
  );
  const timestamp = parseInt(parts['t'], 10);
  const v1 = parts['v1'];

  if (!timestamp || !v1) {
    throw new Error('Malformed X-Hopae-Signature header');
  }

  // Replay protection: reject if timestamp is older than 5 minutes
  const nowSeconds = Math.floor(Date.now() / 1000);
  if (Math.abs(nowSeconds - timestamp) > 300) {
    throw new Error('Webhook timestamp too old — possible replay attack');
  }

  // Recompute signature using raw body bytes
  // IMPORTANT: use the raw body string, not re-serialised JSON
  const rawBody = typeof req.body === 'string' ? req.body : JSON.stringify(req.body);
  const toSign = `${timestamp}.${rawBody}`;
  const expected = crypto.createHmac('sha256', secret).update(toSign).digest('hex');

  // Constant-time comparison to prevent timing attacks
  const expectedBuf = Buffer.from(expected, 'hex');
  const receivedBuf = Buffer.from(v1, 'hex');
  if (expectedBuf.length !== receivedBuf.length || !crypto.timingSafeEqual(expectedBuf, receivedBuf)) {
    throw new Error('Signature mismatch — webhook not authenticated');
  }

  return true;
}

import hashlib
import hmac
import time

def verify_webhook_signature(headers: dict, raw_body: bytes, secret: str) -> bool:
    sig_header = headers.get('x-hopae-signature') or headers.get('X-Hopae-Signature')
    if not sig_header:
        raise ValueError('Missing X-Hopae-Signature header')

    parts = dict(p.split('=', 1) for p in sig_header.split(','))
    timestamp = int(parts.get('t', 0))
    v1 = parts.get('v1', '')

    # Replay protection: reject if older than 5 minutes
    if abs(time.time() - timestamp) > 300:
        raise ValueError('Webhook timestamp too old — possible replay attack')

    to_sign = f"{timestamp}.{raw_body.decode('utf-8')}"
    expected = hmac.new(secret.encode('utf-8'), to_sign.encode('utf-8'), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        raise ValueError('Signature mismatch — webhook not authenticated')

    return True