Webhook Security

{X-Webhook-Timestamp}.{raw_request_body}

X-Webhook-Signature: sha256=a1b2c3d4...

import hashlib
import hmac
import time

from flask import Flask, abort, request

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"


def verify_signature(payload, signature, timestamp, secret, tolerance=300):
    if not signature or not timestamp:
        return False

    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False

    if abs(int(time.time()) - timestamp_value) > tolerance:
        return False

    signed_payload = timestamp.encode("ascii") + b"." + payload
    expected = "sha256=" + hmac.new(
        secret.encode("utf-8"),
        signed_payload,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(signature, expected)


@app.route("/webhook", methods=["POST"])
def handle_webhook():
    payload = request.get_data()
    signature = request.headers.get("X-Webhook-Signature")
    timestamp = request.headers.get("X-Webhook-Timestamp")

    if not verify_signature(payload, signature, timestamp, WEBHOOK_SECRET):
        abort(401)

    event = request.get_json()
    process_event(event)
    return "", 200

const crypto = require('crypto');
const express = require('express');

const app = express();
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

function verifySignature(payload, signature, timestamp, secret, tolerance = 300) {
  if (!signature || !timestamp) return false;

  const timestampValue = Number(timestamp);
  if (!Number.isInteger(timestampValue)) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - timestampValue) > tolerance) return false;

  const signedPayload = Buffer.concat([
    Buffer.from(`${timestamp}.`, 'utf8'),
    payload,
  ]);
  const expected = `sha256=${crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex')}`;

  const providedBuffer = Buffer.from(signature, 'utf8');
  const expectedBuffer = Buffer.from(expected, 'utf8');
  return providedBuffer.length === expectedBuffer.length
    && crypto.timingSafeEqual(providedBuffer, expectedBuffer);
}

app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.header('X-Webhook-Signature');
  const timestamp = req.header('X-Webhook-Timestamp');

  if (!verifySignature(req.body, signature, timestamp, WEBHOOK_SECRET)) {
    return res.status(401).send('Unauthorized');
  }

  const event = JSON.parse(req.body.toString('utf8'));
  processEvent(event);
  return res.status(200).send('OK');
});

def test_signature():
    secret = "test_secret_123"
    timestamp = str(int(time.time()))
    payload = b'{"id":"whn_test","type":"account.created"}'
    signed_payload = timestamp.encode("ascii") + b"." + payload
    digest = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    signature = f"sha256={digest}"

    assert verify_signature(payload, signature, timestamp, secret)