Webhooks Laboratory & Delivery Engine

import crypto from 'crypto';

export function verifyKontyraWebhook(rawBody, signatureHeader, secret) {
  const parts = signatureHeader.split(',');
  const timestamp = parts.find(p => p.startsWith('t='))?.replace('t=', '');
  const signature = parts.find(p => p.startsWith('v1='))?.replace('v1=', '');

  if (!timestamp || !signature) return false;

  // Protect against replay attacks (5 minute threshold)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp, 10)) > 300) return false;

  const payload = `${timestamp}.${rawBody}`;
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

import hmac
import hashlib
import time

def verify_kontyra_webhook(raw_body: bytes, signature_header: str, secret: str) -> bool:
    try:
        parts = dict(item.split('=') for item in signature_header.split(','))
        timestamp = int(parts['t'])
        signature = parts['v1']

        # 5-minute replay tolerance
        if abs(time.time() - timestamp) > 300:
            return False

        payload = f"{timestamp}.".encode('utf-8') + raw_body
        expected = hmac.new(secret.encode('utf-8'), payload, hashlib.sha256).hexdigest()
        return hmac.compare_digest(signature, expected)
    except Exception:
        return False