Webhooks

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

export type VerifyError = "malformed" | "no_signature" | "timestamp" | "mismatch";

/**
 * Verifies a Nowgate-Signature header ("t=<unix seconds>,v1=<hex HMAC-SHA256>")
 * against the raw request body. Pass the exact bytes you received, before any JSON
 * parsing, and every endpoint secret you accept (two while rotating).
 * Returns null when the event is genuine, or the reason it is not.
 */
export function verifyNowgateSignature(
  header: string,
  rawBody: Buffer | string,
  secrets: string[],
  { toleranceSeconds = 300, now = Math.floor(Date.now() / 1000) } = {},
): VerifyError | null {
  let timestamp = 0;
  const signatures: Buffer[] = [];
  for (const part of header.split(",")) {
    const item = part.trim();
    const eq = item.indexOf("=");
    if (eq < 0) return "malformed";
    const key = item.slice(0, eq);
    const value = item.slice(eq + 1);
    if (key === "t") {
      if (timestamp !== 0 || !/^[0-9]+$/.test(value) || Number(value) <= 0) return "malformed";
      timestamp = Number(value);
    } else if (key === "v1") {
      if (!/^[0-9a-fA-F]{64}$/.test(value)) return "malformed";
      signatures.push(Buffer.from(value, "hex"));
    } // other schemes are ignored
  }
  if (timestamp === 0) return "malformed";
  if (signatures.length === 0) return "no_signature";
  if (Math.abs(now - timestamp) > toleranceSeconds) return "timestamp";

  const body = typeof rawBody === "string" ? Buffer.from(rawBody, "utf8") : rawBody;
  let match = false;
  for (const secret of secrets) {
    const expected = createHmac("sha256", secret).update(`${timestamp}.`).update(body).digest();
    for (const signature of signatures) {
      match = timingSafeEqual(expected, signature) || match; // no early exit
    }
  }
  return match ? null : "mismatch";
}

import hashlib
import hmac
import re
import time


class SignatureError(Exception):
    """Raised with the reason: malformed, no_signature, timestamp or mismatch."""


def verify_nowgate_signature(header, raw_body, secrets, tolerance=300, now=None):
    """Verifies a Nowgate-Signature header against the raw request body (bytes).

    Pass the exact bytes you received, before any JSON parsing, and every endpoint
    secret you accept (two while rotating). Raises SignatureError if the event is
    not genuine.
    """
    timestamp = None
    signatures = []
    for part in header.split(","):
        key, sep, value = part.strip().partition("=")
        if not sep:
            raise SignatureError("malformed")
        if key == "t":
            if timestamp is not None or not re.fullmatch(r"[0-9]+", value) or int(value) <= 0:
                raise SignatureError("malformed")
            timestamp = int(value)
        elif key == "v1":
            if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
                raise SignatureError("malformed")
            signatures.append(bytes.fromhex(value))
        # other schemes are ignored
    if timestamp is None:
        raise SignatureError("malformed")
    if not signatures:
        raise SignatureError("no_signature")
    now = int(time.time()) if now is None else now
    if abs(now - timestamp) > tolerance:
        raise SignatureError("timestamp")

    match = False
    for secret in secrets:
        message = str(timestamp).encode() + b"." + raw_body
        expected = hmac.new(secret.encode(), message, hashlib.sha256).digest()
        for signature in signatures:
            match = hmac.compare_digest(expected, signature) or match  # no early exit
    if not match:
        raise SignatureError("mismatch")

<?php

declare(strict_types=1);

/** Thrown with the reason: malformed, no_signature, timestamp or mismatch. */
final class SignatureError extends Exception
{
}

/**
 * Verifies a Nowgate-Signature header against the raw request body.
 *
 * Pass the exact bytes you received, before any JSON parsing (file_get_contents('php://input')),
 * and every endpoint secret you accept (two while rotating). Throws SignatureError if the event
 * is not genuine.
 *
 * @param list<string> $secrets
 */
function verify_nowgate_signature(string $header, string $rawBody, array $secrets, int $tolerance = 300, ?int $now = null): void
{
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', trim($part), 2);
        if (count($pair) !== 2) {
            throw new SignatureError('malformed');
        }
        [$key, $value] = $pair;
        if ($key === 't') {
            if ($timestamp !== null || preg_match('/\A[0-9]+\z/', $value) !== 1 || (int) $value <= 0) {
                throw new SignatureError('malformed');
            }
            $timestamp = (int) $value;
        } elseif ($key === 'v1') {
            if (preg_match('/\A[0-9a-fA-F]{64}\z/', $value) !== 1) {
                throw new SignatureError('malformed');
            }
            $signatures[] = (string) hex2bin($value);
        } // other schemes are ignored
    }
    if ($timestamp === null) {
        throw new SignatureError('malformed');
    }
    if ($signatures === []) {
        throw new SignatureError('no_signature');
    }
    $now ??= time();
    if (abs($now - $timestamp) > $tolerance) {
        throw new SignatureError('timestamp');
    }

    $match = false;
    foreach ($secrets as $secret) {
        $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret, true);
        foreach ($signatures as $signature) {
            $match = hash_equals($expected, $signature) || $match; // no early exit
        }
    }
    if (!$match) {
        throw new SignatureError('mismatch');
    }
}