Verify signatures

import crypto from 'node:crypto';

const TOLERANCE_MS = 5 * 60 * 1000;

/**
 * @param {Record<string, string | string[] | undefined>} headers  lower-cased (Node does this)
 * @param {Buffer} rawBody  the exact bytes received
 * @param {string} secret   the endpoint's signing secret
 */
export function verifyGaugeSignature(headers, rawBody, secret, now = Date.now()) {
  const header = (name) => [headers[name]].flat()[0] ?? '';
  let timestamp = '';
  let signatures = [];

  if (header('gauge-signature')) {
    for (const part of header('gauge-signature').split(',')) {
      const [key, value = ''] = part.trim().split('=', 2);
      if (key === 't') timestamp = value;
      else if (key === 'v1') signatures.push(value);
    }
  } else {
    timestamp = header('x-pulse-timestamp');
    signatures = header('x-pulse-signature')
      .split(',')
      .map((s) => s.trim().replace(/^sha256=/, ''))
      .filter(Boolean);
  }
  if (!/^\d+$/.test(timestamp) || signatures.length === 0) return false;

  // `t` is epoch milliseconds.
  if (Math.abs(now - Number(timestamp)) > TOLERANCE_MS) return false;

  const expected = crypto.createHmac('sha256', secret).update(`${timestamp}.`).update(rawBody).digest();
  return signatures.some((sig) => {
    const given = Buffer.from(sig, 'hex');
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
}

app.post('/gauge/webhooks', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verifyGaugeSignature(req.headers, req.body, process.env.GAUGE_WEBHOOK_SECRET)) {
    return res.sendStatus(400);
  }
  const event = JSON.parse(req.body);
  res.sendStatus(204);
  // …handle `event` after answering
});

import hashlib
import hmac
import time

TOLERANCE_MS = 5 * 60 * 1000


def verify_gauge_signature(headers, raw_body: bytes, secret: str, now_ms=None) -> bool:
    h = {k.lower(): v for k, v in headers.items()}
    timestamp, signatures = "", []

    if h.get("gauge-signature"):
        for part in h["gauge-signature"].split(","):
            key, _, value = part.strip().partition("=")
            if key == "t":
                timestamp = value
            elif key == "v1":
                signatures.append(value)
    else:
        timestamp = h.get("x-pulse-timestamp", "")
        signatures = [
            s.strip()[len("sha256="):] if s.strip().startswith("sha256=") else s.strip()
            for s in h.get("x-pulse-signature", "").split(",")
            if s.strip()
        ]
    if not timestamp.isdigit() or not signatures:
        return False

    # `t` is epoch milliseconds.
    now_ms = int(time.time() * 1000) if now_ms is None else now_ms
    if abs(now_ms - int(timestamp)) > TOLERANCE_MS:
        return False

    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, sig) for sig in signatures)

<?php

function verify_gauge_signature(array $headers, string $rawBody, string $secret, ?int $nowMs = null): bool
{
    $h = array_change_key_case($headers, CASE_LOWER);
    $timestamp = '';
    $signatures = [];

    if (!empty($h['gauge-signature'])) {
        foreach (explode(',', $h['gauge-signature']) as $part) {
            [$key, $value] = array_pad(explode('=', trim($part), 2), 2, '');
            if ($key === 't') {
                $timestamp = $value;
            } elseif ($key === 'v1') {
                $signatures[] = $value;
            }
        }
    } else {
        $timestamp = $h['x-pulse-timestamp'] ?? '';
        foreach (explode(',', $h['x-pulse-signature'] ?? '') as $sig) {
            $sig = trim($sig);
            if ($sig !== '') {
                $signatures[] = preg_replace('/^sha256=/', '', $sig);
            }
        }
    }
    if (!ctype_digit($timestamp) || $signatures === []) {
        return false;
    }

    // `t` is epoch milliseconds.
    $nowMs = $nowMs ?? (int) floor(microtime(true) * 1000);
    if (abs($nowMs - (int) $timestamp) > 300000) {
        return false;
    }

    $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
    foreach ($signatures as $sig) {
        if (hash_equals($expected, $sig)) {
            return true;
        }
    }
    return false;
}

// Plain PHP:
// verify_gauge_signature(getallheaders(), file_get_contents('php://input'), getenv('GAUGE_WEBHOOK_SECRET'));

$headers = array_map(fn ($values) => $values[0] ?? '', $request->headers->all());
$ok = verify_gauge_signature($headers, $request->getContent(), config('services.gauge.webhook_secret'));