Verify Webhooks

import crypto from "node:crypto";
import express from "express";

const app = express();

// IMPORTANT: capture the raw body. JSON.parse() must happen AFTER verification.
app.post(
  "/webhooks/amboss",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signature = req.header("x-webhook-signature");
    const timestamp = req.header("x-webhook-timestamp");
    if (!signature || !timestamp) return res.status(400).send("missing headers");

    const body = req.body.toString("utf8"); // raw bytes as string
    const expected = crypto
      .createHmac("sha256", process.env.AMBOSS_WEBHOOK_SECRET)
      .update(`${timestamp}.${body}`)
      .digest("hex");

    const a = Buffer.from(signature, "hex");
    const b = Buffer.from(expected, "hex");
    if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
      return res.status(401).send("invalid signature");
    }

    // Optional: reject deliveries older than 5 minutes to limit replay window.
    const skewSeconds = Math.abs(Date.now() / 1000 - Number(timestamp));
    if (skewSeconds > 300) return res.status(400).send("timestamp out of range");

    const event = JSON.parse(body);
    // ...dispatch on event.event_type, dedupe on event.id...
    res.sendStatus(200);
  },
);

import hashlib
import hmac
import os
import time

from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["AMBOSS_WEBHOOK_SECRET"].encode("utf-8")

@app.post("/webhooks/amboss")
def amboss_webhook():
    signature = request.headers.get("x-webhook-signature")
    timestamp = request.headers.get("x-webhook-timestamp")
    if not signature or not timestamp:
        abort(400, "missing headers")

    body = request.get_data()  # raw bytes, before any JSON parsing
    expected = hmac.new(
        SECRET,
        f"{timestamp}.".encode("utf-8") + body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        abort(401, "invalid signature")

    if abs(time.time() - int(timestamp)) > 300:
        abort(400, "timestamp out of range")

    event = request.get_json()
    # ...dispatch on event["event_type"], dedupe on event["id"]...
    return ("", 200)

package main

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

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

func amboss(w http.ResponseWriter, r *http.Request) {
    sig := r.Header.Get("x-webhook-signature")
    ts := r.Header.Get("x-webhook-timestamp")
    if sig == "" || ts == "" {
        http.Error(w, "missing headers", http.StatusBadRequest)
        return
    }

    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "read error", http.StatusBadRequest)
        return
    }

    mac := hmac.New(sha256.New, secret)
    mac.Write([]byte(ts + "."))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))

    given, err := hex.DecodeString(sig)
    if err != nil || !hmac.Equal(given, mac.Sum(nil)) {
        http.Error(w, "invalid signature", http.StatusUnauthorized)
        return
    }

    n, _ := strconv.ParseInt(ts, 10, 64)
    if abs(time.Now().Unix()-n) > 300 {
        http.Error(w, "timestamp out of range", http.StatusBadRequest)
        return
    }

    // ...parse body, dispatch on event_type, dedupe on id...
    w.WriteHeader(http.StatusOK)
}

func abs(x int64) int64 { if x < 0 { return -x }; return x }