Verify webhook signatures

import { createHmac, timingSafeEqual } from 'node:crypto';

// rawBody: the body exactly as received (string or Buffer), before any JSON parsing.
// secret: the endpoint's signing secret, whsec_…
export function verifyBizisyWebhook(rawBody, headers, secret) {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];
  if (!id || !timestamp || !signatures) return false;
  // Refuse old or future requests (replays).
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
  const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${rawBody}`).digest();
  // During a secret rotation the header holds two signatures: accept either.
  return signatures.split(' ').some((part) => {
    const [version, sig] = part.split(',');
    const got = Buffer.from(sig ?? '', 'base64');
    return version === 'v1' && got.length === expected.length && timingSafeEqual(got, expected);
  });
}

import base64, hashlib, hmac, time

# raw_body: the body exactly as received (bytes), before any JSON parsing.
# secret: the endpoint's signing secret, whsec_...
def verify_bizisy_webhook(raw_body: bytes, headers, secret: str) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature", "")
    if not msg_id or not timestamp:
        return False
    try:
        # Refuse old or future requests (replays).
        if abs(time.time() - int(timestamp)) > 300:
            return False
        key = base64.b64decode(secret.removeprefix("whsec_"))
        signed = f"{msg_id}.{timestamp}.".encode() + raw_body
        expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest())
        # During a secret rotation the header holds two signatures: accept either.
        return any(
            version == "v1" and hmac.compare_digest(sig.encode(), expected)
            for version, _, sig in (part.partition(",") for part in signatures.split(" "))
        )
    except (ValueError, UnicodeEncodeError):
        # A malformed timestamp or signature is a refusal, not an error.
        return False

package webhooks

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"net/http"
	"strconv"
	"strings"
	"time"
)

// VerifyBizisyWebhook checks a request's signature.
// body: the raw body exactly as received; secret: the endpoint's signing secret, whsec_…
func VerifyBizisyWebhook(body []byte, header http.Header, secret string) bool {
	id, ts, sigs := header.Get("webhook-id"), header.Get("webhook-timestamp"), header.Get("webhook-signature")
	if id == "" || ts == "" || sigs == "" {
		return false
	}
	// Refuse old or future requests (replays).
	t, err := strconv.ParseInt(ts, 10, 64)
	if err != nil || time.Since(time.Unix(t, 0)).Abs() > 5*time.Minute {
		return false
	}
	key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, key)
	mac.Write([]byte(id + "." + ts + "."))
	mac.Write(body)
	expected := mac.Sum(nil)
	// During a secret rotation the header holds two signatures: accept either.
	for _, part := range strings.Split(sigs, " ") {
		version, sig, _ := strings.Cut(part, ",")
		got, err := base64.StdEncoding.DecodeString(sig)
		if version == "v1" && err == nil && hmac.Equal(got, expected) {
			return true
		}
	}
	return false
}

app.post('/webhooks/bizisy', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verifyBizisyWebhook(req.body, req.headers, process.env.BIZISY_WEBHOOK_SECRET)) return res.sendStatus(400);
  res.sendStatus(204); // answer within 10 seconds, then do the work
  queue.add(JSON.parse(req.body));
});

@app.post("/webhooks/bizisy")
def bizisy_webhook():
    if not verify_bizisy_webhook(request.get_data(), request.headers, os.environ["BIZISY_WEBHOOK_SECRET"]):
        abort(400)
    queue.put(request.get_json())  # do the work after answering
    return "", 204

http.HandleFunc("/webhooks/bizisy", func(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
	if err != nil || !webhooks.VerifyBizisyWebhook(body, r.Header, os.Getenv("BIZISY_WEBHOOK_SECRET")) {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	w.WriteHeader(http.StatusNoContent)
	go handle(body) // answer within 10 seconds, then do the work
})