Verificar la firma de un webhook

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

const SECRETO = process.env.SAFEONUBA_WEBHOOK_SECRET; // whsec_…
const TOLERANCIA_S = 5 * 60;

export function firmaValida(cuerpo, marca, cabecera, secreto = SECRETO) {
  if (!cuerpo || !marca || !cabecera) return false;

  // 1. El timestamp, a menos de 5 minutos de nuestro reloj.
  const t = Number(marca);
  if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > TOLERANCIA_S) return false;

  // 2. La firma esperada, sobre el cuerpo en bruto (un Buffer).
  const esperada = createHmac('sha256', secreto).update(`${marca}.`).update(cuerpo).digest();

  // 3. Vale cualquiera de las firmas de la cabecera, comparadas en tiempo constante.
  return cabecera.split(',').some((parte) => {
    const [version, hex] = parte.trim().split('=');
    if (version !== 'v1' || !/^[0-9a-f]{64}$/i.test(hex ?? '')) return false;
    return timingSafeEqual(Buffer.from(hex, 'hex'), esperada);
  });
}

createServer((req, res) => {
  if (req.method !== 'POST' || req.url !== '/safeonuba') {
    res.writeHead(404).end();
    return;
  }
  const trozos = [];
  req.on('data', (trozo) => trozos.push(trozo));
  req.on('end', () => {
    const cuerpo = Buffer.concat(trozos); // en bruto: nada de JSON.parse antes de verificar
    const valida = firmaValida(
      cuerpo,
      req.headers['x-safeonuba-timestamp'],
      req.headers['x-safeonuba-signature'],
    );
    if (!valida) {
      res.writeHead(401).end();
      return;
    }
    encolar(JSON.parse(cuerpo.toString('utf8'))); // se procesa después, fuera de la petición
    res.writeHead(204).end();
  });
}).listen(8080);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

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

app = Flask(__name__)


def firma_valida(cuerpo: bytes, marca: str | None, cabecera: str | None) -> bool:
    if not marca or not cabecera:
        return False

    # 1. El timestamp, a menos de 5 minutos de nuestro reloj.
    try:
        t = int(marca)
    except ValueError:
        return False
    if abs(time.time() - t) > TOLERANCIA_S:
        return False

    # 2. La firma esperada, sobre el cuerpo en bruto.
    esperada = hmac.new(SECRETO, marca.encode() + b"." + cuerpo, hashlib.sha256).hexdigest()

    # 3. Vale cualquiera de las firmas de la cabecera, comparadas en tiempo constante.
    for parte in cabecera.split(","):
        version, _, firma = parte.strip().partition("=")
        if version == "v1" and hmac.compare_digest(firma.lower(), esperada):
            return True
    return False


@app.post("/safeonuba")
def recibir():
    cuerpo = request.get_data()  # en bruto, antes de tocar request.json
    if not firma_valida(
        cuerpo,
        request.headers.get("X-SafeOnuba-Timestamp"),
        request.headers.get("X-SafeOnuba-Signature"),
    ):
        abort(401)
    encolar(request.get_json())  # se procesa después, fuera de la petición
    return "", 204

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

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

app.MapPost("/safeonuba", async (HttpRequest req) =>
{
    // En bruto: se lee el cuerpo entero antes de deserializar nada.
    using var ms = new MemoryStream();
    await req.Body.CopyToAsync(ms);
    var cuerpo = ms.ToArray();

    if (!FirmaValida(cuerpo, req.Headers["X-SafeOnuba-Timestamp"], req.Headers["X-SafeOnuba-Signature"], secreto))
        return Results.Unauthorized();

    Encolar(JsonDocument.Parse(cuerpo)); // se procesa después, fuera de la petición
    return Results.NoContent();
});

app.Run();

static bool FirmaValida(byte[] cuerpo, string? marca, string? cabecera, byte[] secreto)
{
    if (string.IsNullOrEmpty(marca) || string.IsNullOrEmpty(cabecera)) return false;

    // 1. El timestamp, a menos de 5 minutos de nuestro reloj.
    if (!long.TryParse(marca, out var t)) return false;
    if (Math.Abs(DateTimeOffset.UtcNow.ToUnixTimeSeconds() - t) > 5 * 60) return false;

    // 2. La firma esperada, sobre el cuerpo en bruto.
    var firmado = Encoding.UTF8.GetBytes(marca + ".").Concat(cuerpo).ToArray();
    var esperada = HMACSHA256.HashData(secreto, firmado);

    // 3. Vale cualquiera de las firmas de la cabecera, comparadas en tiempo constante.
    foreach (var parte in cabecera.Split(','))
    {
        var trozos = parte.Trim().Split('=', 2);
        if (trozos.Length != 2 || trozos[0] != "v1") continue;
        byte[] recibida;
        try { recibida = Convert.FromHexString(trozos[1]); }
        catch (FormatException) { continue; }
        if (CryptographicOperations.FixedTimeEquals(recibida, esperada)) return true;
    }
    return false;
}