Verify a webhook signature

{
  "id": "evt_a1b2c3d4e5...",
  "type": "order.created",
  "createdAt": "2026-05-19T11:23:45.000Z",
  "data": {
    "orderId": "ord_xyz",
    "order": {
      /* enriched resource */
    }
  }
}

import crypto from 'crypto';
import express from 'express';

const SECRET = process.env.BRAINERCE_WEBHOOK_SECRET!;
const MAX_AGE_MS = 5 * 60 * 1000; // 5 minutes

const app = express();

// IMPORTANT: capture the RAW body (not parsed JSON) for HMAC.
// `express.json()` mutates whitespace — the signature won't match.
app.post('/webhooks/brainerce', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.header('X-Brainerce-Signature');
  const timestamp = req.header('X-Brainerce-Timestamp');
  if (!signature || !timestamp) {
    return res.status(400).send('Missing signature headers');
  }

  // 1. Replay window — reject anything older than 5 min.
  const ageMs = Date.now() - parseInt(timestamp, 10);
  if (Number.isNaN(ageMs) || ageMs > MAX_AGE_MS || ageMs < -60_000) {
    return res.status(400).send('Timestamp out of window');
  }

  // 2. Recompute signature over `${timestamp}.${rawBody}`.
  const expected = crypto
    .createHmac('sha256', SECRET)
    .update(`${timestamp}.${req.body.toString('utf8')}`)
    .digest('hex');

  // 3. Constant-time compare.
  const sigBuf = Buffer.from(signature, 'hex');
  const expectedBuf = Buffer.from(expected, 'hex');
  if (sigBuf.length !== expectedBuf.length || !crypto.timingSafeEqual(sigBuf, expectedBuf)) {
    return res.status(400).send('Invalid signature');
  }

  // 4. Idempotency — dedupe on envelope id.
  const event = JSON.parse(req.body.toString('utf8'));
  if (await alreadyHandled(event.id)) {
    return res.status(200).send('OK (duplicate)');
  }
  await markHandled(event.id);

  // 5. Do your work.
  await handleEvent(event);
  res.status(200).send('OK');
});

import hmac
import hashlib
import time
from fastapi import FastAPI, Request, HTTPException

SECRET = os.environ["BRAINERCE_WEBHOOK_SECRET"].encode()
MAX_AGE_MS = 5 * 60 * 1000

app = FastAPI()

@app.post("/webhooks/brainerce")
async def webhook(request: Request):
    raw_body = await request.body()  # bytes; NOT request.json()
    signature = request.headers.get("X-Brainerce-Signature")
    timestamp = request.headers.get("X-Brainerce-Timestamp")
    if not signature or not timestamp:
        raise HTTPException(400, "Missing signature headers")

    age_ms = int(time.time() * 1000) - int(timestamp)
    if age_ms > MAX_AGE_MS or age_ms < -60_000:
        raise HTTPException(400, "Timestamp out of window")

    signed_payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()

    # Constant-time compare.
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(400, "Invalid signature")

    # ... idempotency + handler
    return {"ok": True}

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

const maxAgeMs = 5 * 60 * 1000

func verify(w http.ResponseWriter, r *http.Request) {
    sig := r.Header.Get("X-Brainerce-Signature")
    ts := r.Header.Get("X-Brainerce-Timestamp")
    body, _ := io.ReadAll(r.Body)

    tsMs, err := strconv.ParseInt(ts, 10, 64)
    if err != nil {
        http.Error(w, "bad ts", 400)
        return
    }
    if time.Now().UnixMilli()-tsMs > maxAgeMs {
        http.Error(w, "stale", 400)
        return
    }

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

    if !hmac.Equal([]byte(sig), []byte(expected)) {
        http.Error(w, "bad sig", 400)
        return
    }
    // ... process
}