Verifying signatures

webhook-signature: v1,fqM1zvDrPfiCYH4CqlC1k+5CYE9qx8TG3bFYaIop8Bc= v1,oFR+dH53Jk3A2ucgB2cyDkc5btd5S1G1uMQ9k0xjtAU=

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

const TOLERANCE_SECONDS = 5 * 60;

/**
 * Verifies a Tanitar webhook delivery.
 *
 * @param secret   the endpoint's signing secret, as shown once when you created or rotated it (`whsec_…`)
 * @param headers  the request headers (lower-case names)
 * @param rawBody  the request body exactly as received, before any JSON parsing
 * @returns true when one of the signatures matches and the timestamp is fresh
 */
export function verifyWebhook(
  secret: string,
  headers: Record<string, string | undefined>,
  rawBody: string,
  now: number = Math.floor(Date.now() / 1000),
): boolean {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];
  if (!id || !timestamp || !signatures) return false;

  // Reject old or future-dated deliveries so a captured request cannot be replayed later.
  const sentAt = Number(timestamp);
  if (!Number.isInteger(sentAt) || Math.abs(now - sentAt) > TOLERANCE_SECONDS) return false;

  // The key is the secret without its `whsec_` prefix, base64url-decoded.
  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64url');
  const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${rawBody}`).digest();

  // During a secret rotation the header carries one `v1,<signature>` entry per active secret.
  return signatures.split(' ').some((entry) => {
    const [version, signature] = entry.split(',', 2);
    if (version !== 'v1' || !signature) return false;
    const received = Buffer.from(signature, 'base64');
    return received.length === expected.length && timingSafeEqual(received, expected);
  });
}

import base64
import hashlib
import hmac
import time

TOLERANCE_SECONDS = 5 * 60


def verify_webhook(secret: str, headers: dict, raw_body: bytes, now: int | None = None) -> bool:
    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:
        return False
    try:
        sent_at = int(timestamp)
    except ValueError:
        return False
    now = int(time.time()) if now is None else now
    if abs(now - sent_at) > TOLERANCE_SECONDS:
        return False

    # The key is the secret without its `whsec_` prefix, base64url-decoded.
    encoded = secret.removeprefix("whsec_")
    key = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
    expected = hmac.new(key, f"{msg_id}.{timestamp}.".encode() + raw_body, hashlib.sha256).digest()

    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version != "v1" or not signature:
            continue
        try:
            received = base64.b64decode(signature)
        except ValueError:
            continue
        if hmac.compare_digest(received, expected):
            return True
    return False

package webhooks

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

const toleranceSeconds = 5 * 60

// VerifyWebhook reports whether rawBody was signed by Tanitar with secret (whsec_...).
func VerifyWebhook(secret string, header http.Header, rawBody []byte, now time.Time) bool {
	id := header.Get("webhook-id")
	timestamp := header.Get("webhook-timestamp")
	signatures := header.Get("webhook-signature")
	if id == "" || timestamp == "" || signatures == "" {
		return false
	}
	sentAt, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false
	}
	if diff := now.Unix() - sentAt; diff > toleranceSeconds || diff < -toleranceSeconds {
		return false
	}

	// The key is the secret without its whsec_ prefix, base64url-decoded (no padding).
	key, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, key)
	mac.Write([]byte(id + "." + timestamp + "."))
	mac.Write(rawBody)
	expected := mac.Sum(nil)

	for _, entry := range strings.Split(signatures, " ") {
		version, signature, found := strings.Cut(entry, ",")
		if !found || version != "v1" {
			continue
		}
		received, err := base64.StdEncoding.DecodeString(signature)
		if err != nil {
			continue
		}
		if hmac.Equal(received, expected) {
			return true
		}
	}
	return false
}