Verify webhook signatures - Payments API

#!/usr/bin/env node
// Verify a RadiumOne webhook signature. Node 18+, no dependencies.
//
// Key: hex-decode the part AFTER the `whsec_` prefix -> 32 raw bytes (this
// differs from the HPP redirect signature, which uses the full secret
// STRING as the HMAC key -- see verify-redirect-signature).
// Signing string: `${v}.${t}.` + the raw request body bytes (exactly as
// received, before any JSON re-serialization).
// Header shape: X-RadiumOne-Signature: v=1,t=<unix>,k=<8hex>,s=<64hex>.
// `k` is advisory only -- never fetch/derive a key from it.
//
// Fails closed: a malformed secret, a malformed/missing header, a
// non-integer timestamp, or any unrecognized `v` all return false rather
// than throwing or silently signing with an empty/wrong-shaped key.
import { createHmac, timingSafeEqual } from "node:crypto";

const SECRET_PATTERN = /^whsec_[0-9a-fA-F]{64}$/;
const SIGNATURE_HEX_PATTERN = /^[0-9a-fA-F]{64}$/;
const TIMESTAMP_PATTERN = /^\d+$/;

/**
 * @param {{secret: string, header: string, rawBody: string|Buffer, toleranceSeconds?: number, now?: number}} args
 * @returns {boolean}
 */
export function verifyWebhookSignature({ secret, header, rawBody, toleranceSeconds = 300, now = Math.floor(Date.now() / 1000) }) {
  // Fail closed: the secret must be the whsec_<64 hex> shape (32 raw bytes)
  // -- never proceed with an empty, truncated, or wrong-type secret.
  if (typeof secret !== "string" || !SECRET_PATTERN.test(secret)) return false;
  if (typeof header !== "string" || header.length === 0) return false;

  const fields = {};
  for (const part of header.split(",")) {
    const idx = part.indexOf("=");
    if (idx < 0) return false; // malformed field, no `=`
    fields[part.slice(0, idx)] = part.slice(idx + 1);
  }

  const { v, t, s } = fields;
  if (!v || !t || !s) return false;
  if (v !== "1") return false; // reject any scheme version we don't implement
  if (!TIMESTAMP_PATTERN.test(t)) return false; // non-integer timestamp, no exception
  if (!SIGNATURE_HEX_PATTERN.test(s)) return false;

  const timestamp = Number(t);
  if (!Number.isFinite(timestamp) || Math.abs(now - timestamp) > toleranceSeconds) return false;

  const key = Buffer.from(secret.slice("whsec_".length), "hex");

  const bodyBuf = Buffer.isBuffer(rawBody) ? rawBody : Buffer.from(rawBody ?? "", "utf8");
  const signingInput = Buffer.concat([Buffer.from(`${v}.${t}.`, "utf8"), bodyBuf]);
  const expected = Buffer.from(createHmac("sha256", key).update(signingInput).digest("hex"), "hex");
  const actual = Buffer.from(s, "hex");

  if (expected.length !== actual.length) return false;
  return timingSafeEqual(expected, actual);
}

// Example: an Express-style raw-body handler.
// app.post("/webhooks/radiumone", express.raw({ type: "application/json" }), (req, res) => {
//   const ok = verifyWebhookSignature({
//     secret: process.env.RADIUMONE_WEBHOOK_SECRET,
//     header: req.header("X-RadiumOne-Signature"),
//     rawBody: req.body, // Buffer, NOT req.body after JSON parsing
//   });
//   if (!ok) return res.status(401).end();
//   res.status(200).end();
// });

#!/usr/bin/env python3
"""Verify a RadiumOne webhook signature. Python 3.10+, standard library only.

Key: hex-decode the part AFTER the ``whsec_`` prefix -> 32 raw bytes (this
differs from the HPP redirect signature, which uses the full secret STRING
as the HMAC key -- see verify-redirect-signature).
Signing string: ``{v}.{t}.`` + the raw request body bytes (exactly as
received, before any JSON re-serialization).
Header shape: ``X-RadiumOne-Signature: v=1,t=<unix>,k=<8hex>,s=<64hex>``.
``k`` is advisory only -- never fetch/derive a key from it.

Fails closed: a malformed secret, a malformed/missing header, a non-integer
timestamp, or any unrecognized ``v`` all return False rather than raising or
silently signing with an empty/wrong-shaped key.
"""
import hmac
import hashlib
import re
import time

# fullmatch (not match) + re.ASCII: plain `match()` with a trailing `$` lets
# Python accept a string with one trailing "\n" (a `$` special case, same
# class of bug as PCRE without the `D` modifier); `\d` without re.ASCII also
# matches non-ASCII Unicode digits, which `int()` happily parses too.
_SECRET_PATTERN = re.compile(r"^whsec_[0-9a-fA-F]{64}$")
_SIGNATURE_HEX_PATTERN = re.compile(r"^[0-9a-fA-F]{64}$")
_TIMESTAMP_PATTERN = re.compile(r"^\d+$", re.ASCII)


def verify_webhook_signature(secret: str, header: str, raw_body: bytes, tolerance_seconds: int = 300, now: int | None = None) -> bool:
    now = int(time.time()) if now is None else now

    # Fail closed: the secret must be the whsec_<64 hex> shape (32 raw bytes)
    # -- never proceed with an empty, truncated, or wrong-type secret.
    if not isinstance(secret, str) or not _SECRET_PATTERN.fullmatch(secret):
        return False
    if not isinstance(header, str) or not header:
        return False

    fields = {}
    for part in header.split(","):
        if "=" not in part:
            return False
        k, _, v = part.partition("=")
        fields[k] = v

    v, t, s = fields.get("v"), fields.get("t"), fields.get("s")
    if not v or not t or not s:
        return False
    if v != "1":
        return False
    if not _TIMESTAMP_PATTERN.fullmatch(t):
        return False
    if not _SIGNATURE_HEX_PATTERN.fullmatch(s):
        return False

    timestamp = int(t)
    if abs(now - timestamp) > tolerance_seconds:
        return False

    key = bytes.fromhex(secret[len("whsec_"):])

    if isinstance(raw_body, str):
        raw_body = raw_body.encode("utf-8")
    signing_input = f"{v}.{t}.".encode("utf-8") + raw_body
    expected_hex = hmac.new(key, signing_input, hashlib.sha256).hexdigest()

    return hmac.compare_digest(bytes.fromhex(expected_hex), bytes.fromhex(s))


# Example: a Flask-style raw-body handler.
# @app.post("/webhooks/radiumone")
# def webhook():
#     ok = verify_webhook_signature(
#         secret=os.environ["RADIUMONE_WEBHOOK_SECRET"],
#         header=request.headers.get("X-RadiumOne-Signature", ""),
#         raw_body=request.get_data(),  # raw bytes, NOT request.json
#     )
#     if not ok:
#         return "", 401
#     return "", 200

<?php
declare(strict_types=1);

/**
 * Verify a RadiumOne webhook signature. PHP 8.1+, no framework.
 *
 * Key: hex-decode the part AFTER the `whsec_` prefix -> 32 raw bytes (this
 * differs from the HPP redirect signature, which uses the full secret
 * STRING as the HMAC key -- see verify-redirect-signature).
 * Signing string: "{v}.{t}." + the raw request body bytes (exactly as
 * received, before any JSON re-decoding).
 * Header shape: X-RadiumOne-Signature: v=1,t=<unix>,k=<8hex>,s=<64hex>.
 * `k` is advisory only -- never fetch/derive a key from it.
 *
 * Fails closed: a malformed secret, a malformed/missing header, a
 * non-integer timestamp, or any unrecognized `v` all return false rather
 * than throwing or silently signing with an empty/wrong-shaped key.
 */
function verifyWebhookSignature(string $secret, string $header, string $rawBody, int $toleranceSeconds = 300, ?int $now = null): bool
{
    $now ??= time();

    // Fail closed: the secret must be the whsec_<64 hex> shape (32 raw bytes)
    // -- never proceed with an empty, truncated, or wrong-type secret.
    // `D` anchors `$` to the absolute end of the subject -- without it PCRE
    // lets `$` match just before a single trailing "\n", so a secret/signature
    // sourced from a file with a trailing newline would wrongly pass shape
    // validation here and only fail later (or not at all).
    if (!preg_match('/^whsec_[0-9a-fA-F]{64}$/D', $secret)) {
        return false;
    }
    if ($header === '') {
        return false;
    }

    $fields = [];
    foreach (explode(',', $header) as $part) {
        $pos = strpos($part, '=');
        if ($pos === false) {
            return false;
        }
        $fields[substr($part, 0, $pos)] = substr($part, $pos + 1);
    }

    $v = $fields['v'] ?? null;
    $t = $fields['t'] ?? null;
    $s = $fields['s'] ?? null;
    if ($v === null || $t === null || $s === null || $v === '' || $t === '' || $s === '') {
        return false;
    }
    if ($v !== '1') {
        return false;
    }
    if (!ctype_digit($t)) {
        return false;
    }
    if (!preg_match('/^[0-9a-fA-F]{64}$/D', $s)) {
        return false;
    }

    $timestamp = (int) $t;
    if (abs($now - $timestamp) > $toleranceSeconds) {
        return false;
    }

    $key = hex2bin(substr($secret, strlen('whsec_')));

    $signingInput = "{$v}.{$t}." . $rawBody;
    $expectedHex = hash_hmac('sha256', $signingInput, $key);

    return hash_equals($expectedHex, strtolower($s));
}

// Example: a plain PHP handler reading the raw POST body.
// $rawBody = file_get_contents('php://input');
// $header = $_SERVER['HTTP_X_RADIUMONE_SIGNATURE'] ?? '';
// if (!verifyWebhookSignature(getenv('RADIUMONE_WEBHOOK_SECRET'), $header, $rawBody)) {
//     http_response_code(401);
//     exit;
// }

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

/**
 * Verify a RadiumOne webhook signature. Java 17+, no framework
 * ({@code javax.crypto.Mac} only).
 *
 * Key: hex-decode the part AFTER the {@code whsec_} prefix -> 32 raw bytes
 * (this differs from the HPP redirect signature, which uses the full
 * secret STRING as the HMAC key -- see verify-redirect-signature).
 * Signing string: {@code "{v}.{t}."} + the raw request body bytes (exactly
 * as received, before any JSON re-serialization).
 * Header shape: {@code X-RadiumOne-Signature: v=1,t=<unix>,k=<8hex>,s=<64hex>}.
 * {@code k} is advisory only -- never fetch/derive a key from it.
 *
 * Fails closed: a malformed secret, a malformed/missing header, a
 * non-integer timestamp, or any unrecognized {@code v} all return false
 * rather than throwing or silently signing with an empty/wrong-shaped key.
 */
public final class VerifyWebhookSignature {

    private static final Pattern SECRET_PATTERN = Pattern.compile("^whsec_[0-9a-fA-F]{64}$");
    private static final Pattern SIGNATURE_HEX_PATTERN = Pattern.compile("^[0-9a-fA-F]{64}$");
    private static final Pattern TIMESTAMP_PATTERN = Pattern.compile("^\\d+$");

    private VerifyWebhookSignature() {
    }

    public static boolean verify(String secret, String header, byte[] rawBody, int toleranceSeconds, Long now) {
        long nowSeconds = now != null ? now : System.currentTimeMillis() / 1000L;

        // Fail closed: the secret must be the whsec_<64 hex> shape (32 raw
        // bytes) -- never proceed with an empty, truncated, or wrong-type secret.
        if (secret == null || !SECRET_PATTERN.matcher(secret).matches()) {
            return false;
        }
        if (header == null || header.isEmpty()) {
            return false;
        }

        Map<String, String> fields = new HashMap<>();
        for (String part : header.split(",")) {
            int idx = part.indexOf('=');
            if (idx < 0) {
                return false;
            }
            fields.put(part.substring(0, idx), part.substring(idx + 1));
        }

        String v = fields.get("v");
        String t = fields.get("t");
        String s = fields.get("s");
        if (v == null || v.isEmpty() || t == null || t.isEmpty() || s == null || s.isEmpty()) {
            return false;
        }
        if (!"1".equals(v)) {
            return false;
        }
        if (!TIMESTAMP_PATTERN.matcher(t).matches()) {
            return false;
        }
        if (!SIGNATURE_HEX_PATTERN.matcher(s).matches()) {
            return false;
        }

        // TIMESTAMP_PATTERN only bounds the shape (all digits), not the
        // length -- a 20+ digit value overflows a long and must be rejected,
        // not thrown, same as every other malformed-input path here.
        long timestamp;
        try {
            timestamp = Long.parseLong(t);
        } catch (NumberFormatException e) {
            return false;
        }
        if (Math.abs(nowSeconds - timestamp) > toleranceSeconds) {
            return false;
        }

        byte[] key = hexToBytes(secret.substring("whsec_".length()));

        byte[] prefix = (v + "." + t + ".").getBytes(StandardCharsets.UTF_8);
        byte[] signingInput = new byte[prefix.length + rawBody.length];
        System.arraycopy(prefix, 0, signingInput, 0, prefix.length);
        System.arraycopy(rawBody, 0, signingInput, prefix.length, rawBody.length);

        byte[] expected;
        try {
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(new SecretKeySpec(key, "HmacSHA256"));
            expected = mac.doFinal(signingInput);
        } catch (Exception e) {
            return false;
        }

        byte[] actual = hexToBytes(s);
        return java.security.MessageDigest.isEqual(expected, actual);
    }

    private static byte[] hexToBytes(String hex) {
        byte[] out = new byte[hex.length() / 2];
        for (int i = 0; i < out.length; i++) {
            out[i] = (byte) Integer.parseInt(hex.substring(i * 2, i * 2 + 2), 16);
        }
        return out;
    }

    // Example: a plain servlet reading the raw request body before any JSON parsing.
    // boolean ok = VerifyWebhookSignature.verify(
    //     System.getenv("RADIUMONE_WEBHOOK_SECRET"),
    //     request.getHeader("X-RadiumOne-Signature"),
    //     rawBodyBytes,
    //     300,
    //     null);
    // if (!ok) { response.sendError(401); return; }
}