Verifying signatures

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

const TOLERANCE_SECONDS = 300;

/**
 * @param {Buffer|string} rawBody  The exact request body bytes.
 * @param {string|undefined} header  The X-ChargeExpress-Signature header.
 * @param {string|string[]} secrets  Your whsec_… secret, or several during a rotation.
 * @param {number} [now]  Current Unix time in seconds (for tests).
 */
function verifyChargeExpressSignature(rawBody, header, secrets, now = Math.floor(Date.now() / 1000)) {
  if (typeof header !== 'string' || header.length === 0) return false;

  let timestamp = null;
  const signatures = [];
  for (const part of header.split(',')) {
    const index = part.indexOf('=');
    if (index === -1) continue;
    const key = part.slice(0, index).trim();
    const value = part.slice(index + 1).trim();
    if (key === 't') timestamp = /^\d+$/.test(value) ? Number(value) : null;
    else if (key === 'v1' && /^[0-9a-f]{64}$/i.test(value)) signatures.push(Buffer.from(value, 'hex'));
  }
  if (timestamp === null || signatures.length === 0) return false;
  if (Math.abs(now - timestamp) > TOLERANCE_SECONDS) return false;

  const body = Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(rawBody, 'utf8');
  for (const secret of Array.isArray(secrets) ? secrets : [secrets]) {
    const expected = crypto
      .createHmac('sha256', secret)
      .update(`${timestamp}.`)
      .update(body)
      .digest();
    if (signatures.some((given) => crypto.timingSafeEqual(given, expected))) return true;
  }
  return false;
}

module.exports = { verifyChargeExpressSignature };

import hashlib
import hmac
import time

TOLERANCE_SECONDS = 300


def verify_chargeexpress_signature(raw_body, header, secrets, now=None):
    """raw_body: the exact request body (bytes or str).
    header: the X-ChargeExpress-Signature header value.
    secrets: your whsec_... secret, or a list of secrets during a rotation."""
    if not header:
        return False
    if isinstance(raw_body, str):
        raw_body = raw_body.encode("utf-8")
    if isinstance(secrets, str):
        secrets = [secrets]

    timestamp = None
    signatures = []
    for part in header.split(","):
        key, sep, value = part.partition("=")
        if not sep:
            continue
        key, value = key.strip(), value.strip()
        if key == "t":
            timestamp = int(value) if value.isdigit() else None
        elif key == "v1" and len(value) == 64:
            signatures.append(value.lower())
    if timestamp is None or not signatures:
        return False

    now = int(time.time()) if now is None else now
    if abs(now - timestamp) > TOLERANCE_SECONDS:
        return False

    signed = str(timestamp).encode("ascii") + b"." + raw_body
    for secret in secrets:
        expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, given) for given in signatures):
            return True
    return False

<?php

const CHARGEEXPRESS_TOLERANCE_SECONDS = 300;

/**
 * @param string          $rawBody The exact request body: file_get_contents('php://input')
 * @param string|null     $header  $_SERVER['HTTP_X_CHARGEEXPRESS_SIGNATURE'] ?? null
 * @param string|string[] $secrets Your whsec_... secret, or several during a rotation
 */
function verify_chargeexpress_signature(string $rawBody, ?string $header, $secrets, ?int $now = null): bool
{
    if ($header === null || $header === '') {
        return false;
    }
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', $part, 2);
        if (count($pair) !== 2) {
            continue;
        }
        [$key, $value] = array_map('trim', $pair);
        if ($key === 't') {
            $timestamp = ctype_digit($value) ? (int) $value : null;
        } elseif ($key === 'v1' && strlen($value) === 64) {
            $signatures[] = strtolower($value);
        }
    }
    if ($timestamp === null || count($signatures) === 0) {
        return false;
    }
    $now = $now ?? time();
    if (abs($now - $timestamp) > CHARGEEXPRESS_TOLERANCE_SECONDS) {
        return false;
    }
    foreach ((array) $secrets as $secret) {
        $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
        foreach ($signatures as $given) {
            if (hash_equals($expected, $given)) {
                return true;
            }
        }
    }
    return false;
}

using System;
using System.Collections.Generic;
using System.Security.Cryptography;
using System.Text;

public static class ChargeExpressSignature
{
    private const long ToleranceSeconds = 300;

    /// <param name="rawBody">The exact request body bytes.</param>
    /// <param name="header">The X-ChargeExpress-Signature header value.</param>
    /// <param name="secrets">Your whsec_... secret, or several during a rotation.</param>
    public static bool Verify(byte[] rawBody, string? header, IEnumerable<string> secrets, long? now = null)
    {
        if (string.IsNullOrEmpty(header)) return false;

        long? timestamp = null;
        var signatures = new List<byte[]>();
        foreach (var part in header.Split(','))
        {
            var index = part.IndexOf('=');
            if (index < 0) continue;
            var key = part[..index].Trim();
            var value = part[(index + 1)..].Trim();
            if (key == "t")
                timestamp = long.TryParse(value, out var t) && t >= 0 ? t : null;
            else if (key == "v1" && value.Length == 64)
            {
                try { signatures.Add(Convert.FromHexString(value)); }
                catch (FormatException) { }
            }
        }
        if (timestamp is null || signatures.Count == 0) return false;

        var current = now ?? DateTimeOffset.UtcNow.ToUnixTimeSeconds();
        if (Math.Abs(current - timestamp.Value) > ToleranceSeconds) return false;

        var prefix = Encoding.ASCII.GetBytes($"{timestamp.Value}.");
        var signed = new byte[prefix.Length + rawBody.Length];
        prefix.CopyTo(signed, 0);
        rawBody.CopyTo(signed, prefix.Length);

        foreach (var secret in secrets)
        {
            var expected = HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), signed);
            foreach (var given in signatures)
                if (CryptographicOperations.FixedTimeEquals(expected, given)) return true;
        }
        return false;
    }
}

const express = require('express');
const { verifyChargeExpressSignature } = require('./verify');

const app = express();

// Register before any app-wide express.json() so this route gets a Buffer.
app.post('/chargeexpress', express.raw({ type: 'application/json' }), (req, res) => {
  const ok = verifyChargeExpressSignature(
    req.body,
    req.get('X-ChargeExpress-Signature'),
    process.env.CHARGEEXPRESS_WEBHOOK_SECRET,
  );
  if (!ok) return res.status(400).send('Invalid signature');

  const event = JSON.parse(req.body.toString('utf8'));
  res.sendStatus(200);
  handleEvent(event); // dedupe on event.id, then process
});

import json
import os

from flask import Flask, abort, request

from verify import verify_chargeexpress_signature

app = Flask(__name__)


@app.post("/chargeexpress")
def chargeexpress_webhook():
    raw = request.get_data()  # bytes, before any parsing
    if not verify_chargeexpress_signature(
        raw,
        request.headers.get("X-ChargeExpress-Signature"),
        os.environ["CHARGEEXPRESS_WEBHOOK_SECRET"],
    ):
        abort(400)
    event = json.loads(raw)
    enqueue(event)  # dedupe on event["id"], then process
    return "", 200

<?php
require __DIR__ . '/verify.php';

$raw = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_CHARGEEXPRESS_SIGNATURE'] ?? null;

if (!verify_chargeexpress_signature($raw, $header, getenv('CHARGEEXPRESS_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}

$event = json_decode($raw, true);
http_response_code(200);
// dedupe on $event['id'], then process

app.MapPost("/chargeexpress", async (HttpRequest request) =>
{
    using var buffer = new MemoryStream();
    await request.Body.CopyToAsync(buffer);
    var raw = buffer.ToArray();

    var secret = Environment.GetEnvironmentVariable("CHARGEEXPRESS_WEBHOOK_SECRET")!;
    if (!ChargeExpressSignature.Verify(raw, request.Headers["X-ChargeExpress-Signature"], new[] { secret }))
        return Results.BadRequest();

    var evt = System.Text.Json.JsonDocument.Parse(raw);
    // dedupe on evt.RootElement.GetProperty("id"), then process
    return Results.Ok();
});