Verifying a webhook signature

import { createHmac, timingSafeEqual } from 'node:crypto';
import { createServer } from 'node:http';

const SECRET = process.env.SAFEONUBA_WEBHOOK_SECRET; // whsec_…
const TOLERANCE_S = 5 * 60;

export function isValidSignature(body, timestamp, header, secret = SECRET) {
  if (!body || !timestamp || !header) return false;

  // 1. The timestamp, within 5 minutes of our clock.
  const t = Number(timestamp);
  if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > TOLERANCE_S) return false;

  // 2. The expected signature, over the raw body (a Buffer).
  const expected = createHmac('sha256', secret).update(`${timestamp}.`).update(body).digest();

  // 3. Any of the signatures in the header will do, compared in constant time.
  return header.split(',').some((part) => {
    const [version, hex] = part.trim().split('=');
    if (version !== 'v1' || !/^[0-9a-f]{64}$/i.test(hex ?? '')) return false;
    return timingSafeEqual(Buffer.from(hex, 'hex'), expected);
  });
}

createServer((req, res) => {
  if (req.method !== 'POST' || req.url !== '/safeonuba') {
    res.writeHead(404).end();
    return;
  }
  const chunks = [];
  req.on('data', (chunk) => chunks.push(chunk));
  req.on('end', () => {
    const body = Buffer.concat(chunks); // raw: no JSON.parse before verifying
    const valid = isValidSignature(
      body,
      req.headers['x-safeonuba-timestamp'],
      req.headers['x-safeonuba-signature'],
    );
    if (!valid) {
      res.writeHead(401).end();
      return;
    }
    enqueue(JSON.parse(body.toString('utf8'))); // processed later, outside the request
    res.writeHead(204).end();
  });
}).listen(8080);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["SAFEONUBA_WEBHOOK_SECRET"].encode()  # whsec_…
TOLERANCE_S = 5 * 60

app = Flask(__name__)


def is_valid_signature(body: bytes, timestamp: str | None, header: str | None) -> bool:
    if not timestamp or not header:
        return False

    # 1. The timestamp, within 5 minutes of our clock.
    try:
        t = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - t) > TOLERANCE_S:
        return False

    # 2. The expected signature, over the raw body.
    expected = hmac.new(SECRET, timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()

    # 3. Any of the signatures in the header will do, compared in constant time.
    for part in header.split(","):
        version, _, signature = part.strip().partition("=")
        if version == "v1" and hmac.compare_digest(signature.lower(), expected):
            return True
    return False


@app.post("/safeonuba")
def receive():
    body = request.get_data()  # raw, before touching request.json
    if not is_valid_signature(
        body,
        request.headers.get("X-SafeOnuba-Timestamp"),
        request.headers.get("X-SafeOnuba-Signature"),
    ):
        abort(401)
    enqueue(request.get_json())  # processed later, outside the request
    return "", 204

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

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();
var secret = Encoding.UTF8.GetBytes(builder.Configuration["SAFEONUBA_WEBHOOK_SECRET"]!); // whsec_…

app.MapPost("/safeonuba", async (HttpRequest req) =>
{
    // Raw: the whole body is read before deserialising anything.
    using var ms = new MemoryStream();
    await req.Body.CopyToAsync(ms);
    var body = ms.ToArray();

    if (!IsValidSignature(body, req.Headers["X-SafeOnuba-Timestamp"], req.Headers["X-SafeOnuba-Signature"], secret))
        return Results.Unauthorized();

    Enqueue(JsonDocument.Parse(body)); // processed later, outside the request
    return Results.NoContent();
});

app.Run();

static bool IsValidSignature(byte[] body, string? timestamp, string? header, byte[] secret)
{
    if (string.IsNullOrEmpty(timestamp) || string.IsNullOrEmpty(header)) return false;

    // 1. The timestamp, within 5 minutes of our clock.
    if (!long.TryParse(timestamp, out var t)) return false;
    if (Math.Abs(DateTimeOffset.UtcNow.ToUnixTimeSeconds() - t) > 5 * 60) return false;

    // 2. The expected signature, over the raw body.
    var signed = Encoding.UTF8.GetBytes(timestamp + ".").Concat(body).ToArray();
    var expected = HMACSHA256.HashData(secret, signed);

    // 3. Any of the signatures in the header will do, compared in constant time.
    foreach (var part in header.Split(','))
    {
        var pieces = part.Trim().Split('=', 2);
        if (pieces.Length != 2 || pieces[0] != "v1") continue;
        byte[] received;
        try { received = Convert.FromHexString(pieces[1]); }
        catch (FormatException) { continue; }
        if (CryptographicOperations.FixedTimeEquals(received, expected)) return true;
    }
    return false;
}