HMAC verification

Node.js (TypeScript)

No dependencies beyond Node's built-in crypto. Works in standalone Node, Express, Fastify, and Next.js route handlers (use await req.text() in App Router).

import { createHmac, timingSafeEqual } from "crypto";

/**
 * Verify a Crispy v1 webhook signature.
 *
 * Header:        Webhook-Signature: v1,t=<unix_seconds>,s=<hex_sha256>
 * HMAC input:    "v1." + t + "." + rawBody   (HMAC-SHA256, hex)
 * Idempotency:   Webhook-Event-Id (stable across retries)
 *
 * Pass one or more secrets to support zero-downtime secret rotation
 * (try the primary, then the secondary). Crispy signs with the primary;
 * during a rotation window accept either.
 */
export function verifyCrispyV1(
  rawBody: string,
  signatureHeader: string | undefined, // Webhook-Signature
  ...secrets: string[]
): boolean {
  if (!signatureHeader) return false;

  // Parse "v1,t=<ts>,s=<hex>" -> { t, s }. Reject anything malformed.
  const parts = signatureHeader.split(",").map((p) => p.trim());
  if (parts[0] !== "v1") return false;
  let ts = "";
  let sig = "";
  for (const part of parts.slice(1)) {
    const eq = part.indexOf("=");
    if (eq === -1) continue;
    const key = part.slice(0, eq);
    const val = part.slice(eq + 1);
    if (key === "t") ts = val;
    else if (key === "s") sig = val;
  }
  if (!ts || !/^[0-9]+$/.test(ts)) return false;
  if (!sig || !/^[a-f0-9]+$/i.test(sig)) return false;

  const signedInput = "v1." + ts + "." + rawBody;
  for (const secret of secrets) {
    if (!secret) continue;
    const expected = createHmac("sha256", secret).update(signedInput).digest("hex");
    if (timingSafeHexEqual(expected, sig)) return true;
  }
  return false;
}

function timingSafeHexEqual(a: string, b: string): boolean {
  if (a.length !== b.length) return false;
  try {
    return timingSafeEqual(Buffer.from(a, "hex"), Buffer.from(b, "hex"));
  } catch {
    return false;
  }
}

// --- Idempotency ---
// Webhook-Event-Id is identical across retries of the same event. Store
// delivered IDs for 7 days and short-circuit duplicates. Redis SET NX EX
// example; swap for your store of choice.
async function alreadyProcessed(
  redis: { set: (k: string, v: string, opts: { NX: true; EX: number }) => Promise<string | null> },
  eventId: string
): Promise<boolean> {
  // Returns "OK" on first insert, null if key already exists.
  const result = await redis.set("webhook:" + eventId, "1", { NX: true, EX: 7 * 24 * 60 * 60 });
  return result === null;
}

// --- Express handler ---
import type { Request, Response } from "express";

export async function handleCrispyWebhook(req: Request, res: Response) {
  // IMPORTANT: rawBody must be the unparsed body string. In Express, register
  // express.raw({ type: "application/json" }) on this route, not express.json().
  const rawBody = (req.body as Buffer).toString("utf8");

  const ok = verifyCrispyV1(
    rawBody,
    req.header("Webhook-Signature"),
    process.env.CRISPY_WEBHOOK_SECRET!,
    process.env.CRISPY_WEBHOOK_SECRET_SECONDARY ?? "" // optional, for rotation
  );
  if (!ok) {
    res.status(401).send("invalid signature");
    return;
  }

  const eventId = req.header("Webhook-Event-Id");
  if (!eventId || (await alreadyProcessed(redisClient, eventId))) {
    res.status(200).send("ok"); // idempotent: ack but do nothing
    return;
  }

  const payload = JSON.parse(rawBody);
  // ... do the actual work
  res.status(200).send("ok");
}

Python

Standard library only (hmac + hashlib). Flask example included; the verifier itself is framework-agnostic.

import hmac
import hashlib
from typing import Optional

def verify_crispy_v1(
    raw_body: bytes,
    signature_header: Optional[str],  # Webhook-Signature
    *secrets: str,
) -> bool:
    """
    Verify a Crispy v1 webhook signature.

    Header:      Webhook-Signature: v1,t=<unix_seconds>,s=<hex_sha256>
    HMAC input:  b"v1." + t + b"." + raw_body   (HMAC-SHA256, hex)
    Idempotency: Webhook-Event-Id (stable across retries)

    Pass one or more secrets to support secret rotation (primary, then
    secondary). Crispy signs with the primary; accept either during a window.
    """
    if not signature_header:
        return False

    parts = [p.strip() for p in signature_header.split(",")]
    if not parts or parts[0] != "v1":
        return False
    ts = ""
    sig = ""
    for part in parts[1:]:
        if "=" not in part:
            continue
        key, _, val = part.partition("=")
        if key == "t":
            ts = val
        elif key == "s":
            sig = val
    if not ts or not ts.isdigit():
        return False
    if not sig or not all(c in "0123456789abcdefABCDEF" for c in sig):
        return False

    signed_input = b"v1." + ts.encode("utf-8") + b"." + raw_body
    for secret in secrets:
        if not secret:
            continue
        expected = hmac.new(
            secret.encode("utf-8"), signed_input, hashlib.sha256
        ).hexdigest()
        # constant-time comparison
        if hmac.compare_digest(expected, sig):
            return True
    return False


# --- Idempotency ---
# Webhook-Event-Id is identical across retries. Store delivered IDs for 7
# days and short-circuit duplicates. Redis SET NX EX example:

import redis as redis_lib  # pip install redis

def already_processed(r: redis_lib.Redis, event_id: str) -> bool:
    # SET key value NX EX 604800 -> returns True on first insert, None if exists
    result = r.set(f"webhook:{event_id}", "1", nx=True, ex=7 * 24 * 60 * 60)
    return result is None


# --- Flask handler ---
from flask import Flask, request, abort
import os
import json

app = Flask(__name__)
r = redis_lib.Redis.from_url(os.environ["REDIS_URL"])

@app.route("/webhooks/crispy", methods=["POST"])
def handle_crispy_webhook():
    # IMPORTANT: request.get_data() returns the raw bytes; do not call .json
    # before verification, otherwise a tampered body will be silently parsed.
    raw_body = request.get_data()
    signature = request.headers.get("Webhook-Signature")
    event_id = request.headers.get("Webhook-Event-Id")

    ok = verify_crispy_v1(
        raw_body,
        signature,
        os.environ["CRISPY_WEBHOOK_SECRET"],
        os.environ.get("CRISPY_WEBHOOK_SECRET_SECONDARY", ""),  # optional
    )
    if not ok:
        abort(401)

    if not event_id or already_processed(r, event_id):
        return ("", 200)  # idempotent: ack but do nothing

    payload = json.loads(raw_body)
    # ... do the actual work
    return ("", 200)

Go

Standard library only for verification. The handler example uses github.com/redis/go-redis/v9 for idempotency; swap for your store of choice.

package crispy

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"io"
	"net/http"
	"os"
	"regexp"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

var (
	hexRe = regexp.MustCompile(`^[a-fA-F0-9]+$`)
	tsRe  = regexp.MustCompile(`^[0-9]+$`)
)

// VerifyCrispyV1 verifies a Crispy v1 webhook signature.
//
// Header:      Webhook-Signature: v1,t=<unix_seconds>,s=<hex_sha256>
// HMAC input:  "v1." + t + "." + rawBody   (HMAC-SHA256, hex)
// Idempotency: Webhook-Event-Id (stable across retries)
//
// Pass one or more secrets to support secret rotation (primary, then
// secondary). Crispy signs with the primary; accept either during a window.
func VerifyCrispyV1(rawBody []byte, signatureHeader string, secrets ...string) bool {
	if signatureHeader == "" {
		return false
	}
	parts := strings.Split(signatureHeader, ",")
	if strings.TrimSpace(parts[0]) != "v1" {
		return false
	}
	var ts, sig string
	for _, part := range parts[1:] {
		kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch kv[0] {
		case "t":
			ts = kv[1]
		case "s":
			sig = kv[1]
		}
	}
	if ts == "" || !tsRe.MatchString(ts) {
		return false
	}
	if sig == "" || !hexRe.MatchString(sig) {
		return false
	}

	signedInput := append([]byte("v1."+ts+"."), rawBody...)
	for _, secret := range secrets {
		if secret == "" {
			continue
		}
		expected := computeHmac(signedInput, secret)
		if hmacEqual(expected, sig) {
			return true
		}
	}
	return false
}

func computeHmac(body []byte, secret string) string {
	h := hmac.New(sha256.New, []byte(secret))
	h.Write(body)
	return hex.EncodeToString(h.Sum(nil))
}

// hmacEqual decodes both as hex then uses hmac.Equal for constant-time compare.
func hmacEqual(expected string, got string) bool {
	if len(expected) != len(got) {
		return false
	}
	expectedBytes, err := hex.DecodeString(expected)
	if err != nil {
		return false
	}
	gotBytes, err := hex.DecodeString(got)
	if err != nil {
		return false
	}
	return hmac.Equal(expectedBytes, gotBytes)
}

// --- Idempotency ---
// Webhook-Event-Id is identical across retries. Store delivered IDs for 7
// days and short-circuit duplicates.

func alreadyProcessed(r *redis.Client, eventID string) (bool, error) {
	// SET key value NX EX 604800 -> true on first insert, false if exists.
	ok, err := r.SetNX(ctxBackground(), "webhook:"+eventID, "1", 7*24*time.Hour).Result()
	if err != nil {
		return false, err
	}
	return !ok, nil
}

// --- net/http handler ---

func HandleCrispyWebhook(w http.ResponseWriter, req *http.Request) {
	// IMPORTANT: read the raw body BEFORE parsing JSON. The HMAC is computed
	// over the unparsed bytes.
	rawBody, err := io.ReadAll(req.Body)
	if err != nil {
		http.Error(w, "read error", http.StatusBadRequest)
		return
	}
	defer req.Body.Close()

	ok := VerifyCrispyV1(
		rawBody,
		req.Header.Get("Webhook-Signature"),
		os.Getenv("CRISPY_WEBHOOK_SECRET"),
		os.Getenv("CRISPY_WEBHOOK_SECRET_SECONDARY"), // optional, for rotation
	)
	if !ok {
		http.Error(w, "invalid signature", http.StatusUnauthorized)
		return
	}

	eventID := req.Header.Get("Webhook-Event-Id")
	if eventID == "" {
		w.WriteHeader(http.StatusOK)
		return
	}
	dup, err := alreadyProcessed(redisClient, eventID)
	if err == nil && dup {
		w.WriteHeader(http.StatusOK) // idempotent: ack but do nothing
		return
	}

	// ... json.Unmarshal(rawBody, &payload); do the actual work
	w.WriteHeader(http.StatusOK)
}

The v1 payload (Hybrid shape)

v1 events carry a stable identifiers block plus a curated set of fields — never raw internal database rows. The top-level id equals the Webhook-Event-Id header.

The identifiers block carries multiple keys so you can match a delivery to your own records: linkedin_provider_id is the hashed ACoA… form, while member_id is the numeric LinkedIn member ID (the value behind urn:li:member:<id>) that many CRMs use as their canonical person key. member_idis null when Crispy hasn't captured it for that contact.

{
  "payload_version": "1",
  "id": "3f1c2a90-8b4e-4d21-9f0a-2c7d5e6b1a34",
  "event": "contact.connection_status_changed",
  "timestamp": "2026-01-15T10:30:00.000Z",
  "webhook_id": "8d2a7c14-3e9b-4f56-8a01-9c4d2e1f7b60",
  "account_id": "a1b2c3d4-e5f6-4789-90ab-1c2d3e4f5a6b",
  "data": {
    "identifiers": {
      "crispy_id": "contact_01J8Z9K2QY",
      "linkedin_url": "https://www.linkedin.com/in/jane-doe",
      "linkedin_provider_id": "ACoAAB1cD2eF",
      "member_id": "121354090",
      "email": "[email protected]",
      "external_ids": {}
    },
    "name": "Jane Doe",
    "first_name": "Jane",
    "last_name": "Doe",
    "headline": "VP Engineering at Acme",
    "company_name": "Acme",
    "company_domain": "acme.com",
    "location": "San Francisco, CA",
    "country": "United States",
    "phone": null,
    "connection_status": "connected",
    "tags": ["target-account"],
    "custom_attrs": {},
    "source_event": "invitation.accepted",
    "triggered_at": "2026-01-15T10:30:00.000Z"
  }
}