Verify a signature

These verifiers match the signature algorithm exactly. Pick the runtime you're on, paste, and adapt the route framework.

Capture the raw body before parsing. Express's express.json(), FastAPI's automatic JSON binding, and Next.js's req.json() all silently lose whitespace. The signature is computed over the bytes we sent — if your verifier sees a re-serialised JSON object, it will fail.

Node — no framework

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

const TOLERANCE_SEC = 300;

export interface VerifyResult {
  ok: boolean;
  reason?:
    | "missing_header"
    | "malformed_header"
    | "missing_timestamp"
    | "expired"
    | "no_v1"
    | "mismatch";
}

export function verifyTouricalSignature(args: {
  header: string | null | undefined;
  rawBody: string;
  signingKey: string;
  nowSec?: number;
  toleranceSec?: number;
}): VerifyResult {
  const tolerance = args.toleranceSec ?? TOLERANCE_SEC;
  const now = args.nowSec ?? Math.floor(Date.now() / 1000);
  if (!args.header) return { ok: false, reason: "missing_header" };

  const parts = args.header.split(",").map((p) => p.trim()).filter(Boolean);
  if (parts.length === 0) return { ok: false, reason: "malformed_header" };

  let ts: number | null = null;
  const v1Values: string[] = [];
  for (const part of parts) {
    const eq = part.indexOf("=");
    if (eq <= 0) return { ok: false, reason: "malformed_header" };
    const name = part.slice(0, eq).trim();
    const value = part.slice(eq + 1).trim();
    if (name === "t") {
      const parsed = Number.parseInt(value, 10);
      if (!Number.isFinite(parsed) || parsed <= 0) {
        return { ok: false, reason: "malformed_header" };
      }
      ts = parsed;
    } else if (name === "v1") {
      v1Values.push(value);
    }
    // Unknown names are silently ignored (forward-compat for v2+).
  }
  if (ts === null) return { ok: false, reason: "missing_timestamp" };
  if (Math.abs(now - ts) > tolerance) return { ok: false, reason: "expired" };
  if (v1Values.length === 0) return { ok: false, reason: "no_v1" };

  const expected = createHmac("sha256", args.signingKey)
    .update(`${ts}.${args.rawBody}`, "utf8")
    .digest("hex");
  for (const candidate of v1Values) {
    if (
      candidate.length === expected.length &&
      timingSafeEqual(
        Buffer.from(candidate, "utf8"),
        Buffer.from(expected, "utf8"),
      )
    ) {
      return { ok: true };
    }
  }
  return { ok: false, reason: "mismatch" };
}

Node — Express

import express from "express";
import { verifyTouricalSignature } from "./verify";

const app = express();

// Capture the raw body. DO NOT use express.json() before this point.
app.post(
  "/tourical/webhooks",
  express.raw({ type: "application/json", limit: "1mb" }),
  (req, res) => {
    const rawBody = req.body.toString("utf8");
    const result = verifyTouricalSignature({
      header: req.header("tourical-signature"),
      rawBody,
      signingKey: process.env.TOURICAL_SIGNING_KEY!,
    });
    if (!result.ok) {
      console.warn("webhook rejected:", result.reason);
      return res.status(400).end();
    }

    const event = JSON.parse(rawBody);
    // Dedupe on event.id before doing work.
    handleEvent(event);
    res.status(200).end();
  },
);

Next.js App Router

app/api/tourical/webhooks/route.ts

import { verifyTouricalSignature } from "@/lib/tourical-verify";

export async function POST(req: Request) {
  const rawBody = await req.text();   // Read raw — DO NOT call req.json() first.
  const result = verifyTouricalSignature({
    header: req.headers.get("tourical-signature"),
    rawBody,
    signingKey: process.env.TOURICAL_SIGNING_KEY!,
  });
  if (!result.ok) {
    return new Response(null, { status: 400 });
  }

  const event = JSON.parse(rawBody);
  await handleEvent(event);
  return new Response(null, { status: 200 });
}

Python

import hashlib
import hmac
import os
import time
from typing import Optional

TOLERANCE_SEC = 300

def verify_tourical_signature(
    header: Optional[str],
    raw_body: bytes,
    signing_key: str,
    now_sec: Optional[int] = None,
    tolerance_sec: int = TOLERANCE_SEC,
) -> tuple[bool, str]:
    if not header:
        return False, "missing_header"
    if now_sec is None:
        now_sec = int(time.time())

    parts = [p.strip() for p in header.split(",") if p.strip()]
    if not parts:
        return False, "malformed_header"

    ts = None
    v1_values: list[str] = []
    for part in parts:
        if "=" not in part:
            return False, "malformed_header"
        name, _, value = part.partition("=")
        name = name.strip()
        value = value.strip()
        if name == "t":
            try:
                ts = int(value)
            except ValueError:
                return False, "malformed_header"
            if ts <= 0:
                return False, "malformed_header"
        elif name == "v1":
            v1_values.append(value)
        # Unknown names ignored (forward-compat).

    if ts is None:
        return False, "missing_timestamp"
    if abs(now_sec - ts) > tolerance_sec:
        return False, "expired"
    if not v1_values:
        return False, "no_v1"

    expected = hmac.new(
        signing_key.encode("utf-8"),
        f"{ts}.{raw_body.decode('utf-8')}".encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()

    for candidate in v1_values:
        if hmac.compare_digest(candidate, expected):
            return True, "ok"
    return False, "mismatch"

FastAPI mounting:

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/tourical/webhooks")
async def tourical_webhook(request: Request):
    raw_body = await request.body()                 # Read raw bytes first.
    header = request.headers.get("tourical-signature")
    ok, reason = verify_tourical_signature(
        header, raw_body, os.environ["TOURICAL_SIGNING_KEY"]
    )
    if not ok:
        raise HTTPException(status_code=400, detail=reason)

    import json
    event = json.loads(raw_body)
    handle_event(event)
    return {"received": True}

Ruby

require "openssl"

TOLERANCE_SEC = 300

def verify_tourical_signature(header:, raw_body:, signing_key:, now_sec: nil, tolerance_sec: TOLERANCE_SEC)
  return [false, :missing_header] if header.nil? || header.empty?
  now_sec ||= Time.now.to_i

  ts = nil
  v1_values = []
  header.split(",").each do |part|
    part = part.strip
    next if part.empty?
    name, _, value = part.partition("=")
    return [false, :malformed_header] if name.nil? || value.nil?
    case name.strip
    when "t"
      ts = Integer(value.strip) rescue nil
      return [false, :malformed_header] if ts.nil? || ts <= 0
    when "v1"
      v1_values << value.strip
    end
  end

  return [false, :missing_timestamp] if ts.nil?
  return [false, :expired] if (now_sec - ts).abs > tolerance_sec
  return [false, :no_v1] if v1_values.empty?

  expected = OpenSSL::HMAC.hexdigest("sha256", signing_key, "#{ts}.#{raw_body}")
  v1_values.each do |candidate|
    return [true, :ok] if Rack::Utils.secure_compare(candidate, expected) rescue false
  end
  [false, :mismatch]
end

Testing your verifier

The fastest way to test end-to-end: click Test next to the subscription in your operator dashboard. We send a webhook.test event with a stub payload; your verifier should accept it.

You can also trigger one programmatically:

curl https://app.tourical.com/api/v1/webhooks/test \
  -H "Authorization: Bearer tk_live_..." \
  -H "Content-Type: application/json" \
  -d '{ "subscriptionId": "whsub_..." }'