التحقق من طلبات webhooks

npm install svix

import { Webhook } from 'svix';

const secret = process.env.WEBHOOK_SECRET; // whsec_...

app.post('/webhooks/rntor', (req, res) => {
  const payload = req.body;
  const headers = req.headers;

  const wh = new Webhook(secret);
  
  try {
    const verified = wh.verify(JSON.stringify(payload), {
      'svix-id': headers['svix-id'],
      'svix-timestamp': headers['svix-timestamp'],
      'svix-signature': headers['svix-signature'],
    });
    
    // Process the verified payload
    handleWebhook(verified);
    res.status(200).send('OK');
    
  } catch (err) {
    console.error('Webhook verification failed:', err.message);
    res.status(401).send('Invalid signature');
  }
});

import crypto from 'crypto';

function verifyWebhookSignature(payload, headers, secret) {
  const svixId = headers['svix-id'];
  const svixTimestamp = headers['svix-timestamp'];
  const svixSignature = headers['svix-signature'];

  // Check timestamp is within 5 minutes
  const timestamp = parseInt(svixTimestamp, 10);
  const currentTime = Math.floor(Date.now() / 1000);
  if (Math.abs(currentTime - timestamp) > 300) {
    throw new Error('Timestamp too old');
  }

  // Construct signed payload
  const signedPayload = `${svixId}.${svixTimestamp}.${payload}`;

  // Decode secret (remove whsec_ prefix and base64 decode)
  const secretBytes = Buffer.from(secret.replace('whsec_', ''), 'base64');

  // Calculate expected signature
  const expectedSignature = crypto
    .createHmac('sha256', secretBytes)
    .update(signedPayload)
    .digest('base64');

  // Compare signatures
  const signatures = svixSignature.split(',');
  for (const sig of signatures) {
    const [version, signature] = sig.split(',');
    if (version === 'v1' && signature === expectedSignature) {
      return true;
    }
  }

  throw new Error('Invalid signature');
}

pip install svix

from svix.webhooks import Webhook
import os

webhook_secret = os.environ.get('WEBHOOK_SECRET')  # whsec_...

@app.route('/webhooks/rntor', methods=['POST'])
def handle_webhook():
    payload = request.get_data(as_text=True)
    headers = {
        'svix-id': request.headers.get('svix-id'),
        'svix-timestamp': request.headers.get('svix-timestamp'),
        'svix-signature': request.headers.get('svix-signature'),
    }

    wh = Webhook(webhook_secret)
    
    try:
        verified = wh.verify(payload, headers)
        # Process the verified payload
        handle_event(verified)
        return 'OK', 200
        
    except Exception as e:
        print(f'Webhook verification failed: {e}')
        return 'Invalid signature', 401

import hmac
import hashlib
import base64
import time

def verify_webhook_signature(payload, headers, secret):
    svix_id = headers.get('svix-id')
    svix_timestamp = headers.get('svix-timestamp')
    svix_signature = headers.get('svix-signature')

    # Check timestamp is within 5 minutes
    timestamp = int(svix_timestamp)
    current_time = int(time.time())
    if abs(current_time - timestamp) > 300:
        raise ValueError('Timestamp too old')

    # Construct signed payload
    signed_payload = f'{svix_id}.{svix_timestamp}.{payload}'

    # Decode secret
    secret_bytes = base64.b64decode(secret.replace('whsec_', ''))

    # Calculate expected signature
    expected_signature = base64.b64encode(
        hmac.new(secret_bytes, signed_payload.encode(), hashlib.sha256).digest()
    ).decode()

    # Compare signatures
    for sig in svix_signature.split(','):
        parts = sig.split(',')
        if len(parts) == 2 and parts[0] == 'v1' and parts[1] == expected_signature:
            return True

    raise ValueError('Invalid signature')