Webhooks

import crypto from "node:crypto";

const TOLERANCE_SECONDS = 5 * 60;

// headers: the request's headers, with lowercase names.
// rawBody: a Buffer holding the body exactly as it arrived.
export function verifyWebhook(secret, headers, rawBody) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures || !secret.startsWith("whsec_")) {
    return false;
  }

  const sent = Number(timestamp);
  const now = Math.floor(Date.now() / 1000);
  if (!Number.isInteger(sent) || Math.abs(now - sent) > TOLERANCE_SECONDS) {
    return false;
  }

  const key = Buffer.from(secret.slice("whsec_".length), "base64");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest();

  return signatures.split(" ").some((entry) => {
    const [version, value] = entry.split(",");
    if (version !== "v1" || !value) {
      return false;
    }
    const given = Buffer.from(value, "base64");
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
}

import base64
import hashlib
import hmac
import time

TOLERANCE_SECONDS = 5 * 60


def verify_webhook(secret: str, headers, raw_body: bytes) -> bool:
    """headers: the request's headers. raw_body: the body exactly as it arrived."""
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures or not secret.startswith("whsec_"):
        return False

    try:
        sent = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - sent) > TOLERANCE_SECONDS:
        return False

    key = base64.b64decode(secret[len("whsec_"):])
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()

    for entry in signatures.split(" "):
        version, _, value = entry.partition(",")
        if version == "v1" and hmac.compare_digest(value, expected):
            return True
    return False

package webhooks

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

const tolerance = 5 * time.Minute

// Verify reports whether an event came from us. body is the request body
// exactly as it arrived.
func Verify(secret string, h http.Header, body []byte) bool {
	id, ts, signatures := h.Get("webhook-id"), h.Get("webhook-timestamp"), h.Get("webhook-signature")
	encoded, isSecret := strings.CutPrefix(secret, "whsec_")
	if id == "" || signatures == "" || !isSecret {
		return false
	}

	sent, err := strconv.ParseInt(ts, 10, 64)
	if err != nil {
		return false
	}
	if age := time.Since(time.Unix(sent, 0)); age > tolerance || age < -tolerance {
		return false
	}

	key, err := base64.StdEncoding.DecodeString(encoded)
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, key)
	mac.Write([]byte(id + "." + ts + "."))
	mac.Write(body)
	expected := mac.Sum(nil)

	for _, entry := range strings.Split(signatures, " ") {
		version, value, _ := strings.Cut(entry, ",")
		given, err := base64.StdEncoding.DecodeString(value)
		if version == "v1" && err == nil && hmac.Equal(given, expected) {
			return true
		}
	}
	return false
}