Verify webhook signatures

import express from 'express';
import crypto  from 'node:crypto';

const app    = express();
const SECRET = process.env.KEY_WEBHOOK_SECRET;

app.post(
  '/webhooks/key',
  express.raw({ type: 'application/json' }), // raw Buffer, not parsed
  (req, res) => {
    const sig = req.headers['x-webhook-signature'] || '';
    const expected =
      'sha256=' +
      crypto.createHmac('sha256', SECRET).update(req.body).digest('hex');

    if (!crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected))) {
      return res.status(401).send('invalid signature');
    }

    const event = JSON.parse(req.body);
    // route by event.eventType…
    res.status(200).end();
  }
);

app.listen(3000);

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

app    = Flask(__name__)
SECRET = os.environ["KEY_WEBHOOK_SECRET"].encode()

@app.post("/webhooks/key")
def handle():
    body = request.get_data()                       # raw bytes
    sig  = request.headers.get("X-Webhook-Signature", "")
    expected = "sha256=" + hmac.new(SECRET, body, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(sig, expected):
        abort(401)

    event = request.get_json(force=True)
    # route by event["eventType"]…
    return "", 200

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "io"
    "net/http"
    "os"
)

var secret = []byte(os.Getenv("KEY_WEBHOOK_SECRET"))

func handle(w http.ResponseWriter, r *http.Request) {
    body, _ := io.ReadAll(r.Body)
    mac     := hmac.New(sha256.New, secret)
    mac.Write(body)
    expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))

    if !hmac.Equal([]byte(r.Header.Get("X-Webhook-Signature")), []byte(expected)) {
        http.Error(w, "invalid signature", 401)
        return
    }
    // JSON-unmarshal and route…
    w.WriteHeader(200)
}