How to verify webhook signatures

svix-id: msg_2a3b4c5d6e7f8g9h
svix-timestamp: 1640995200
svix-signature: v1,g0hQad8xK3qE1l2m...

import { Webhook } from 'svix';

const webhookSecret = process.env.WEBHOOK_SECRET; // From your dashboard
const svix = new Webhook(webhookSecret);

app.post('/webhooks', (req, res) => {
  const signature = req.headers['svix-signature'];
  const payload = JSON.stringify(req.body);
  
  try {
    const verified = svix.verify(payload, signature);
    // Process verified webhook
    handleWebhook(verified);
    res.status(200).send('OK');
  } catch (err) {
    res.status(400).send('Invalid signature');
  }
});

const crypto = require('crypto');

function verifyWebhook(rawBody, headers, secret) {
  const id = headers['svix-id'];
  const timestamp = headers['svix-timestamp'];
  const signatureHeader = headers['svix-signature']; // space-separated "v1,<base64>" list

  if (!id || !timestamp || !signatureHeader) {
    return false;
  }

  // Replay protection: reject timestamps more than 5 minutes from now.
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp, 10)) > 300) {
    return false;
  }

  // The secret is "whsec_<base64>"; the HMAC key is the decoded base64 body.
  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
  const signedContent = `${id}.${timestamp}.${rawBody}`;
  const expected = crypto
    .createHmac('sha256', key)
    .update(signedContent)
    .digest('base64');
  const expectedBuf = Buffer.from(expected);

  // Several signatures can appear after a secret rotation; match any one.
  return signatureHeader.split(' ').some((entry) => {
    const sig = entry.split(',')[1]; // strip the "v1," version prefix
    if (!sig) {
      return false;
    }
    const sigBuf = Buffer.from(sig);
    // Length-guard first: timingSafeEqual throws on a length mismatch.
    return (
      sigBuf.length === expectedBuf.length &&
      crypto.timingSafeEqual(sigBuf, expectedBuf)
    );
  });
}

const express = require('express');
const { Webhook } = require('svix');

const app = express();
const webhookSecret = process.env.WEBHOOK_SECRET;
const svix = new Webhook(webhookSecret);

// Store processed event IDs (use Redis in production)
const processedEvents = new Set();

app.post('/webhooks', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['svix-signature'];
  
  if (!signature) {
    return res.status(400).send('Missing signature');
  }
  
  try {
    const payload = req.body.toString();
    const verified = svix.verify(payload, req.headers);

    // Check idempotency. The per-delivery ID is the svix-id header.
    const deliveryId = req.headers['svix-id'];
    if (processedEvents.has(deliveryId)) {
      return res.status(200).send('Already processed');
    }

    // Process the flat body
    handleWebhook(verified);

    // Mark as processed
    processedEvents.add(deliveryId);

    res.status(200).send('OK');
  } catch (err) {
    console.error('Webhook verification failed:', err);
    res.status(400).send('Invalid signature');
  }
});

function handleWebhook(webhook) {
  // The body is flat: read fields like webhook.command or webhook.deviceId.
  console.log('Received webhook for device:', webhook.deviceId);
  // Your webhook processing logic here
}

from flask import Flask, request, jsonify
from svix import Webhook
import os

app = Flask(__name__)
webhook_secret = os.environ.get('WEBHOOK_SECRET')
svix = Webhook(webhook_secret)

processed_events = set()

@app.route('/webhooks', methods=['POST'])
def webhook():
    signature = request.headers.get('svix-signature')
    
    if not signature:
        return jsonify({'error': 'Missing signature'}), 400
    
    try:
        payload = request.get_data(as_text=True)
        verified = svix.verify(payload, request.headers)

        # Check idempotency. The per-delivery ID is the svix-id header.
        delivery_id = request.headers.get('svix-id')
        if delivery_id in processed_events:
            return jsonify({'status': 'Already processed'}), 200

        # Process the flat body
        handle_webhook(verified)

        # Mark as processed
        processed_events.add(delivery_id)

        return jsonify({'status': 'OK'}), 200
    except Exception as e:
        print(f'Webhook verification failed: {e}')
        return jsonify({'error': 'Invalid signature'}), 400

def handle_webhook(webhook):
    # The body is flat: read fields like webhook["command"] or webhook["deviceId"].
    print(f'Received webhook for device: {webhook["deviceId"]}')
    # Your webhook processing logic here

ngrok http 3000