Event

const crypto = require('crypto');

function verifyWebhookSignature(
  webhookId,
  payload,
  signature,
  timestamp,
  secret
) {
  const signedPayload = `${webhookId}.${timestamp}.${payload}`;
  const keyBytes = Buffer.from(secret, 'base64url');
  const expectedSignature = crypto
    .createHmac('sha256', keyBytes)
    .update(signedPayload, 'utf8')
    .digest('hex');

  if (signature.length !== expectedSignature.length) {
    return false;
  }

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

app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const webhookId = req.headers['x-webhooks-id'];
  const signatureHeader = req.headers['x-webhooks-signature'];
  const timestamp = req.headers['x-webhooks-timestamp'];

  if (!webhookId || !signatureHeader || !timestamp) {
    return res.status(401).json({ error: 'Missing required headers' });
  }

  const signature = signatureHeader.replace(/^v\d+=/, '');
  const payload = req.body.toString();
  const webhookSecret = process.env.WEBHOOK_SECRET;

  if (
    !webhookSecret ||
    !verifyWebhookSignature(
      webhookId,
      payload,
      signature,
      timestamp,
      webhookSecret
    )
  ) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const event = JSON.parse(payload);
  // Process the webhook event
  res.status(200).json({ received: true });
});

import base64
import hmac
import hashlib
import os
from flask import Flask, request

def verify_webhook_signature(webhook_id: str, payload: str, signature: str, timestamp: str, secret: str) -> bool:
    signed_payload = f"{webhook_id}.{timestamp}.{payload}"
    key_bytes = base64.urlsafe_b64decode(secret + '==')
    expected_signature = hmac.new(
        key_bytes,
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected_signature)

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    webhook_id = request.headers.get('X-Webhooks-Id', '')
    signature_header = request.headers.get('X-Webhooks-Signature', '')
    timestamp = request.headers.get('X-Webhooks-Timestamp', '')

    if not webhook_id or not signature_header or not timestamp:
        return {'error': 'Missing required headers'}, 401

    signature = signature_header.split('=', 1)[1] if '=' in signature_header else signature_header
    payload = request.get_data(as_text=True)
    webhook_secret = os.environ.get('WEBHOOK_SECRET')

    if not webhook_secret or not verify_webhook_signature(webhook_id, payload, signature, timestamp, webhook_secret):
        return {'error': 'Invalid signature'}, 401

    # Process the webhook event
    return {'received': True}, 200