Verify webhook signatures

import crypto from "node:crypto";

// Header: Fajita-Signature: t=<timestamp>,kid=<key_id>,v1=<hex>
function parseSignature(header) {
  return Object.fromEntries(
    header.split(",").map((part) => {
      const [k, v] = part.split("=");
      return [k, v];
    }),
  );
}

// Call with the RAW request body string, not a re-serialized object.
export function verify(rawBody, headers, secret, { toleranceSeconds = 300 } = {}) {
  const sig = parseSignature(headers["fajita-signature"]);
  const eventId = headers["fajita-event-id"];
  const timestamp = Number(sig.t);

  // 1. Reject requests outside the allowed time window.
  const now = Math.floor(Date.now() / 1000);
  if (!Number.isFinite(timestamp) || Math.abs(now - timestamp) > toleranceSeconds) {
    return false;
  }

  // 2. Recompute over the exact signed input.
  const signedInput = `${sig.kid}.${timestamp}.${eventId}.${rawBody}`;
  const expected = crypto
    .createHmac("sha256", secret)
    .update(signedInput)
    .digest("hex");

  // 3. Constant-time comparison.
  const a = Buffer.from(expected);
  const b = Buffer.from(sig.v1 ?? "");
  if (a.length !== b.length) return false;
  return crypto.timingSafeEqual(a, b);
}

import hashlib
import hmac
import time


def parse_signature(header: str) -> dict[str, str]:
    return dict(part.split("=", 1) for part in header.split(","))


# Pass the RAW request body bytes, not a re-serialized dict.
def verify(raw_body: bytes, headers: dict, secret: str, tolerance_seconds: int = 300) -> bool:
    sig = parse_signature(headers["fajita-signature"])
    event_id = headers["fajita-event-id"]
    timestamp = int(sig["t"])

    # 1. Reject requests outside the allowed time window.
    if abs(int(time.time()) - timestamp) > tolerance_seconds:
        return False

    # 2. Recompute over the exact signed input.
    signed_input = f"{sig['kid']}.{timestamp}.{event_id}.{raw_body.decode()}".encode()
    expected = hmac.new(secret.encode(), signed_input, hashlib.sha256).hexdigest()

    # 3. Constant-time comparison.
    return hmac.compare_digest(expected, sig.get("v1", ""))

package fajita

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"strconv"
	"strings"
	"time"
)

func parseSignature(header string) map[string]string {
	out := map[string]string{}
	for _, part := range strings.Split(header, ",") {
		kv := strings.SplitN(part, "=", 2)
		if len(kv) == 2 {
			out[kv[0]] = kv[1]
		}
	}
	return out
}

// rawBody must be the exact bytes received, not re-marshaled JSON.
func Verify(rawBody []byte, headers map[string]string, secret string, tolerance time.Duration) bool {
	sig := parseSignature(headers["fajita-signature"])
	eventID := headers["fajita-event-id"]
	ts, err := strconv.ParseInt(sig["t"], 10, 64)
	if err != nil {
		return false
	}

	// 1. Reject requests outside the allowed time window.
	if d := time.Since(time.Unix(ts, 0)); d > tolerance || d < -tolerance {
		return false
	}

	// 2. Recompute over the exact signed input.
	signedInput := fmt.Sprintf("%s.%d.%s.%s", sig["kid"], ts, eventID, string(rawBody))
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(signedInput))
	expected := hex.EncodeToString(mac.Sum(nil))

	// 3. Constant-time comparison.
	return hmac.Equal([]byte(expected), []byte(sig["v1"]))
}