Verifying webhook signatures

using System.Security.Cryptography;
using System.Text;

public static class WebhookSignature
{
    public static bool Verify(byte[] rawBody, string signatureHeader, string secret,
        TimeSpan? tolerance = null, DateTimeOffset? now = null)
    {
        string? timestamp = null;
        var signatures = new List<string>();
        foreach (var part in signatureHeader.Split(','))
        {
            var kv = part.Trim().Split('=', 2);
            if (kv.Length != 2) continue;
            if (kv[0] == "t") timestamp = kv[1];
            else if (kv[0] == "v0") signatures.Add(kv[1]);
        }
        if (timestamp is null || signatures.Count == 0 || !long.TryParse(timestamp, out var t)) return false;

        var age = (now ?? DateTimeOffset.UtcNow).ToUnixTimeSeconds() - t;
        if (Math.Abs(age) > (tolerance ?? TimeSpan.FromMinutes(5)).TotalSeconds) return false; // possible replay

        var signedPayload = Encoding.UTF8.GetBytes(timestamp + ".").Concat(rawBody).ToArray();
        var expected = HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), signedPayload);

        foreach (var s in signatures)
        {
            byte[] candidate;
            try { candidate = Convert.FromHexString(s); } catch (FormatException) { continue; }
            if (CryptographicOperations.FixedTimeEquals(expected, candidate)) return true;
        }
        return false;
    }
}

app.MapPost("/hooks/unimaze", async (HttpRequest request) =>
{
    using var ms = new MemoryStream();
    await request.Body.CopyToAsync(ms);
    var body = ms.ToArray();
    if (!WebhookSignature.Verify(body, request.Headers["UMZ-Webhook-Signature"].ToString(), secret))
        return Results.Unauthorized();
    await queue.EnqueueAsync(body);   // process in the background
    return Results.NoContent();
});

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.HexFormat;
import java.util.List;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

public final class WebhookSignature {

    public static boolean verify(byte[] rawBody, String signatureHeader, String secret,
                                 long toleranceSeconds, long nowEpochSeconds) throws Exception {
        String timestamp = null;
        List<String> signatures = new ArrayList<>();
        for (String part : signatureHeader.split(",")) {
            String[] kv = part.trim().split("=", 2);
            if (kv.length != 2) continue;
            if (kv[0].equals("t")) timestamp = kv[1];
            else if (kv[0].equals("v0")) signatures.add(kv[1]);
        }
        if (timestamp == null || signatures.isEmpty()) return false;
        if (Math.abs(nowEpochSeconds - Long.parseLong(timestamp)) > toleranceSeconds) return false; // possible replay

        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
        mac.update((timestamp + ".").getBytes(StandardCharsets.UTF_8));
        byte[] expected = mac.doFinal(rawBody);

        for (String s : signatures) {
            byte[] candidate;
            try { candidate = HexFormat.of().parseHex(s); } catch (IllegalArgumentException e) { continue; }
            if (MessageDigest.isEqual(expected, candidate)) return true;
        }
        return false;
    }
}

import hashlib
import hmac
import time


def verify_signature(raw_body: bytes, signature_header: str, secret: str,
                     tolerance_seconds: int = 300, now: int | None = None) -> bool:
    timestamp, signatures = None, []
    for part in signature_header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t":
            timestamp = value
        elif key == "v0":
            signatures.append(value)
    if timestamp is None or not signatures:
        return False

    now = int(time.time()) if now is None else now
    if abs(now - int(timestamp)) > tolerance_seconds:
        return False  # possible replay

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

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

export function verifySignature(
  rawBody: Buffer,
  signatureHeader: string,
  secret: string,
  toleranceSeconds = 300,
  now = Math.floor(Date.now() / 1000),
): boolean {
  let timestamp: string | undefined;
  const signatures: string[] = [];
  for (const part of signatureHeader.split(",")) {
    const [key, value] = part.trim().split("=", 2);
    if (key === "t") timestamp = value;
    else if (key === "v0") signatures.push(value);
  }
  if (!timestamp || signatures.length === 0) return false;
  if (Math.abs(now - Number(timestamp)) > toleranceSeconds) return false; // possible replay

  const expected = createHmac("sha256", secret)
    .update(Buffer.concat([Buffer.from(`${timestamp}.`), rawBody]))
    .digest();
  return signatures.some((s) => {
    const candidate = Buffer.from(s, "hex");
    return candidate.length === expected.length && timingSafeEqual(candidate, expected);
  });
}