Verify signatures

import crypto from "node:crypto";

// rawBody MUST be the exact bytes Ionic sent (a string or Buffer),
// captured before any JSON parsing.
export function verifyWebhook(secret: string, headers: Record<string, string>, rawBody: string) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatureHeader = headers["webhook-signature"]; // "v1,xxx v1,yyy"

  // 1. Replay protection: reject timestamps outside a 5-minute window.
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > 300) {
    throw new Error("Webhook timestamp outside tolerance");
  }

  // 2 + 3. Recompute the signature over `{id}.{timestamp}.{body}`.
  const secretBytes = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const signedContent = `${id}.${timestamp}.${rawBody}`;
  const expected = crypto.createHmac("sha256", secretBytes).update(signedContent).digest("base64");

  // 4. Constant-time compare against each provided v1 signature.
  const provided = signatureHeader.split(" ").map((part) => part.split(",")[1]);
  const matched = provided.some(
    (sig) =>
      sig.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected)),
  );
  if (!matched) throw new Error("No matching webhook signature");
}

import base64, hashlib, hmac, time

def verify_webhook(secret: str, headers: dict, raw_body: bytes) -> None:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signature_header = headers["webhook-signature"]  # "v1,xxx v1,yyy"

    # 1. Replay protection.
    if abs(int(time.time()) - int(timestamp)) > 300:
        raise ValueError("Webhook timestamp outside tolerance")

    # 2 + 3. Recompute over `{id}.{timestamp}.{body}`.
    secret_bytes = base64.b64decode(secret.removeprefix("whsec_"))
    signed_content = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(
        hmac.new(secret_bytes, signed_content, hashlib.sha256).digest()
    ).decode()

    # 4. Constant-time compare against each provided signature.
    provided = [part.split(",", 1)[1] for part in signature_header.split()]
    if not any(hmac.compare_digest(sig, expected) for sig in provided):
        raise ValueError("No matching webhook signature")

package webhooks

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

// Verify checks a webhook delivery. body MUST be the raw request bytes.
func Verify(secret string, h http.Header, body []byte) error {
	id := h.Get("webhook-id")
	ts := h.Get("webhook-timestamp")
	sigHeader := h.Get("webhook-signature") // "v1,xxx v1,yyy"

	// 1. Replay protection.
	t, err := strconv.ParseInt(ts, 10, 64)
	if err != nil {
		return fmt.Errorf("bad timestamp: %w", err)
	}
	if math.Abs(float64(time.Now().Unix()-t)) > 300 {
		return errors.New("timestamp outside tolerance")
	}

	// 2 + 3. Recompute over `{id}.{timestamp}.{body}`.
	key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
	if err != nil {
		return fmt.Errorf("bad secret: %w", err)
	}
	mac := hmac.New(sha256.New, key)
	fmt.Fprintf(mac, "%s.%s.%s", id, ts, body)
	expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))

	// 4. Constant-time compare against each provided signature.
	for _, part := range strings.Fields(sigHeader) {
		_, sig, _ := strings.Cut(part, ",")
		if hmac.Equal([]byte(sig), []byte(expected)) {
			return nil
		}
	}
	return errors.New("no matching webhook signature")
}