Webhook Signature Verification

const crypto = require("crypto");

function verifyWebhookSignature(req, secretKey) {
  // Extract headers
  const signature = req.headers["x-signature"];
  const publicKey = req.headers["x-public-key"];

  if (!signature || !publicKey) {
    throw new Error("Missing required signature headers");
  }

  // Get raw body (must be the exact string, not parsed JSON)
  const rawBody = typeof req.body === "string" ? req.body : JSON.stringify(req.body);

  // Compute signature from payload
  const computedSignature = crypto.createHmac("sha256", secretKey).update(rawBody).digest("hex");

  // Compare signatures (use constant-time comparison to prevent timing attacks)
  if (signature.length !== computedSignature.length) {
    throw new Error("Invalid signature");
  }

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

// Express.js middleware example
app.post("/webhook", express.raw({ type: "application/json" }), (req, res) => {
  try {
    const publicKey = req.headers["x-public-key"];
    const secretKey = getSecretKeyByPublicKey(publicKey); // Your function to retrieve secret key

    if (verifyWebhookSignature(req, secretKey)) {
      // Signature is valid, parse and use the payload data
      const payload = JSON.parse(req.body.toString());
      console.log("Webhook verified:", payload);

      // Use the payload data - it's verified and trusted
      // Process the event: payload.event.type, payload.event.data, etc.

      res.status(200).json({ received: true });
    } else {
      res.status(401).json({ error: "Invalid signature" });
    }
  } catch (error) {
    console.error("Webhook verification error:", error);
    res.status(401).json({ error: error.message });
  }
});

import hmac
import hashlib
import time
import json
from flask import Flask, request, jsonify

app = Flask(__name__)

def verify_webhook_signature(request, secret_key):
    """
    Verify webhook signature using HMAC-SHA256

    Args:
        request: Flask request object
        secret_key: Organization's secret key (str)

    Returns:
        bool: True if signature is valid, False otherwise
    """
    # Extract headers
    signature = request.headers.get('x-signature')
    public_key = request.headers.get('x-public-key')

    if not all([signature, public_key]):
        raise ValueError('Missing required signature headers')

    # Get raw body (must be bytes, not parsed JSON)
    raw_body = request.get_data()

    # Compute signature from payload
    computed_signature = hmac.new(
        secret_key.encode('utf-8'),
        raw_body,
        hashlib.sha256
    ).hexdigest()

    # Compare signatures (use constant-time comparison)
    return hmac.compare_digest(signature, computed_signature)

@app.route('/webhook', methods=['POST'])
def webhook_handler():
    try:
        public_key = request.headers.get('x-public-key')
        secret_key = get_secret_key_by_public_key(public_key)  # Your function to retrieve secret key

        if verify_webhook_signature(request, secret_key):
            # Signature is valid, parse and use the payload data
            payload = request.get_json()
            print(f'Webhook verified: {payload}')

            # Use the payload data - it's verified and trusted
            # Process the event: payload['event']['type'], payload['event']['data'], etc.

            return jsonify({'received': True}), 200
        else:
            return jsonify({'error': 'Invalid signature'}), 401
    except Exception as e:
        print(f'Webhook verification error: {e}')
        return jsonify({'error': str(e)}), 401

def get_secret_key_by_public_key(public_key):
    """
    Retrieve secret key by public key.
    Replace this with your actual implementation.
    """
    # TODO: Implement your logic to retrieve secret key from database/storage
    pass