Webhooks Overview

import { createHmac, timingSafeEqual } from 'crypto'

function verifyWebhook(secret, signatureHeader, rawBody) {
  // Parse "t=<unix>,v1=<hex>,v1=<hex>" header
  const parts = signatureHeader.split(',')
  const timestamp = parts.find(p => p.startsWith('t=')).slice(2)
  const signatures = parts.filter(p => p.startsWith('v1=')).map(p => p.slice(3))

  // Reject stale requests (5 minute tolerance)
  const age = Math.floor(Date.now() / 1000) - parseInt(timestamp, 10)
  if (age > 300) throw new Error('Timestamp too old')

  // Compute expected signature
  const mac = createHmac('sha256', secret)
  mac.update(timestamp)
  mac.update('.')
  mac.update(rawBody)
  const expected = mac.digest('hex')

  // Compare against each v1 signature (supports secret rotation overlap)
  const match = signatures.some(sig =>
    timingSafeEqual(Buffer.from(expected, 'hex'), Buffer.from(sig, 'hex'))
  )
  if (!match) throw new Error('Invalid signature')
}

// In your webhook handler:
const rawBody = req.body // must be the raw string, not parsed JSON
const signature = req.headers['x-nexio-signature']
verifyWebhook('whsec_...', signature, rawBody)

import hmac
import hashlib
import time

def verify_webhook(secret: str, signature_header: str, raw_body: bytes) -> None:
    # Parse "t=<unix>,v1=<hex>,v1=<hex>" header
    parts = signature_header.split(",")
    timestamp = next(p[2:] for p in parts if p.startswith("t="))
    signatures = [p[3:] for p in parts if p.startswith("v1=")]

    # Reject stale requests (5 minute tolerance)
    if int(time.time()) - int(timestamp) > 300:
        raise ValueError("Timestamp too old")

    # Compute expected signature
    message = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()

    # Compare against each v1 signature (supports secret rotation overlap)
    if not any(hmac.compare_digest(expected, sig) for sig in signatures):
        raise ValueError("Invalid signature")

# In your webhook handler:
raw_body = request.get_data()  # raw bytes, not parsed JSON
signature = request.headers["X-Nexio-Signature"]
verify_webhook("whsec_...", signature, raw_body)

package main

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

func VerifyWebhook(secret, signatureHeader string, rawBody []byte) error {
	// Parse "t=<unix>,v1=<hex>,v1=<hex>" header
	parts := strings.Split(signatureHeader, ",")
	var timestamp string
	var signatures []string
	for _, p := range parts {
		if strings.HasPrefix(p, "t=") {
			timestamp = p[2:]
		} else if strings.HasPrefix(p, "v1=") {
			signatures = append(signatures, p[3:])
		}
	}

	// Reject stale requests (5 minute tolerance)
	ts, _ := strconv.ParseInt(timestamp, 10, 64)
	if time.Now().Unix()-ts > 300 {
		return errors.New("timestamp too old")
	}

	// Compute expected signature
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(timestamp))
	mac.Write([]byte("."))
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))

	// Compare against each v1 signature (supports secret rotation overlap)
	for _, sig := range signatures {
		if hmac.Equal([]byte(expected), []byte(sig)) {
			return nil
		}
	}
	return errors.New("invalid signature")
}