Webhooks and signature verification

{
  "id": "evt_…",
  "object": "event",
  "type": "invoice.paid",
  "livemode": false,
  "created_at": "2026-09-27T20:15:04Z",
  "invoice_id": "inv_…",
  "data": { "object": { "id": "inv_…", "object": "invoice", "status": "paid", "…": "…" } }
}

const crypto = require("node:crypto");

/** Returns true if `header` is a valid signature of `rawBody` (a Buffer) under `secret`. */
function verifyWebhook(rawBody, header, secret, toleranceSecs = 300, nowSecs = Math.floor(Date.now() / 1000)) {
  let t = null;
  const sigs = [];
  for (const part of String(header || "").split(",")) {
    const i = part.indexOf("=");
    if (i < 0) return false;
    const k = part.slice(0, i).trim(), v = part.slice(i + 1).trim();
    if (k === "t") t = Number(v);
    else if (k === "v1") sigs.push(v);
  }
  if (!Number.isInteger(t) || sigs.length === 0) return false;
  if (Math.abs(nowSecs - t) > toleranceSecs) return false;
  const expected = crypto.createHmac("sha256", secret).update(`${t}.`).update(rawBody).digest();
  return sigs.some((s) => {
    const got = Buffer.from(s, "hex");
    return got.length === expected.length && crypto.timingSafeEqual(got, expected);
  });
}

// With node:http (the raw body is what you read off the socket):
require("node:http").createServer((req, res) => {
  const chunks = [];
  req.on("data", (c) => chunks.push(c)).on("end", () => {
    const raw = Buffer.concat(chunks);
    if (!verifyWebhook(raw, req.headers["x-cryptopay-signature"], process.env.WEBHOOK_SECRET)) {
      res.writeHead(400).end("bad signature");
      return;
    }
    const event = JSON.parse(raw.toString("utf8"));
    // Deduplicate on event.id (== the x-cryptopay-event-id header), then handle event.type.
    res.writeHead(200).end("ok");
  });
}).listen(Number(process.env.PORT) || 3000);

<?php
/** True if $header is a valid signature of the raw body $payload under $secret. */
function verify_webhook(string $payload, ?string $header, string $secret, int $tolerance = 300): bool {
    if ($header === null) return false;
    $t = null;
    $sigs = [];
    foreach (explode(',', $header) as $part) {
        $kv = explode('=', trim($part), 2);
        if (count($kv) !== 2) return false;
        if ($kv[0] === 't' && ctype_digit($kv[1])) $t = (int) $kv[1];
        elseif ($kv[0] === 'v1') $sigs[] = $kv[1];
    }
    if ($t === null || !$sigs) return false;
    if (abs(time() - $t) > $tolerance) return false;
    $expected = hash_hmac('sha256', $t . '.' . $payload, $secret);
    foreach ($sigs as $sig) {
        if (hash_equals($expected, $sig)) return true;   // constant-time
    }
    return false;
}

$payload = file_get_contents('php://input');                 // the raw body
$header  = $_SERVER['HTTP_X_CRYPTOPAY_SIGNATURE'] ?? null;
if (!verify_webhook($payload, $header, getenv('WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit('bad signature');
}
$event = json_decode($payload, true);
// Deduplicate on $event['id'], then handle $event['type'].
http_response_code(200);
echo 'ok';

import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer


def verify_webhook(raw_body: bytes, header: str | None, secret: str, tolerance: int = 300) -> bool:
    """True if `header` is a valid signature of the raw body bytes under `secret`."""
    if not header:
        return False
    t, sigs = None, []
    for part in header.split(","):
        key, sep, value = part.strip().partition("=")
        if not sep:
            return False
        if key == "t" and value.isdigit():
            t = int(value)
        elif key == "v1":
            sigs.append(value)
    if t is None or not sigs:
        return False
    if abs(int(time.time()) - t) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in sigs)  # constant-time


class Webhook(BaseHTTPRequestHandler):
    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))  # the raw body
        if not verify_webhook(raw, self.headers.get("x-cryptopay-signature"), os.environ["WEBHOOK_SECRET"]):
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"bad signature")
            return
        event = json.loads(raw)
        # Deduplicate on event["id"], then handle event["type"].
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"ok")


HTTPServer(("", int(os.environ.get("PORT", "3000"))), Webhook).serve_forever()

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

// verifyWebhook reports whether header is a valid signature of rawBody under secret.
func verifyWebhook(rawBody []byte, header, secret string, tolerance time.Duration) bool {
	t := int64(-1)
	var sigs []string
	for _, part := range strings.Split(header, ",") {
		k, v, ok := strings.Cut(strings.TrimSpace(part), "=")
		if !ok {
			return false
		}
		switch k {
		case "t":
			n, err := strconv.ParseInt(v, 10, 64)
			if err != nil || n < 0 {
				return false
			}
			t = n
		case "v1":
			sigs = append(sigs, v)
		}
	}
	if t < 0 || len(sigs) == 0 {
		return false
	}
	if age := time.Since(time.Unix(t, 0)); age > tolerance || age < -tolerance {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(strconv.FormatInt(t, 10) + "."))
	mac.Write(rawBody)
	expected := mac.Sum(nil)
	for _, s := range sigs {
		if got, err := hex.DecodeString(s); err == nil && hmac.Equal(got, expected) { // constant-time
			return true
		}
	}
	return false
}

func main() {
	secret := os.Getenv("WEBHOOK_SECRET")
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		raw, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // the raw body
		if err != nil || !verifyWebhook(raw, r.Header.Get("x-cryptopay-signature"), secret, 5*time.Minute) {
			http.Error(w, "bad signature", http.StatusBadRequest)
			return
		}
		var event struct {
			ID   string `json:"id"`
			Type string `json:"type"`
		}
		if err := json.Unmarshal(raw, &event); err != nil {
			http.Error(w, "bad json", http.StatusBadRequest)
			return
		}
		// Deduplicate on event.ID, then handle event.Type.
		w.Write([]byte("ok"))
	})
	port := os.Getenv("PORT")
	if port == "" {
		port = "3000"
	}
	log.Fatal(http.ListenAndServe(":"+port, nil))
}