Headers y firma

import crypto from 'node:crypto';
import express from 'express';

const SECRET = process.env.PRUFF_WEBHOOK_SECRET;
const MAX_AGE_SECONDS = 30 * 60;

function isValidSignature(rawBody, signatureHeader) {
  const parts = {};
  for (const part of (signatureHeader ?? '').split(',')) {
    const separator = part.indexOf('=');
    if (separator > 0) {
      parts[part.slice(0, separator)] = part.slice(separator + 1);
    }
  }
  const timestamp = Number(parts.t);
  if (!Number.isInteger(timestamp) || !parts.v1) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > MAX_AGE_SECONDS) return false;

  const expected = crypto
    .createHmac('sha256', SECRET)
    .update(`${parts.t}.`)
    .update(rawBody)
    .digest('hex');
  const received = parts.v1;
  return (
    expected.length === received.length &&
    crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(received))
  );
}

const app = express();

app.post(
  '/pruff/invoices/new',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    if (!isValidSignature(req.body, req.get('X-Webhook-Signature'))) {
      return res.sendStatus(401);
    }
    const event = JSON.parse(req.body.toString('utf8'));
    // Encola el evento para procesarlo fuera del handler.
    res.sendStatus(204);
  },
);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["PRUFF_WEBHOOK_SECRET"].encode()
MAX_AGE_SECONDS = 30 * 60


def is_valid_signature(raw_body: bytes, signature_header: str) -> bool:
    parts = {}
    for part in signature_header.split(","):
        key, separator, value = part.partition("=")
        if separator:
            parts[key] = value
    timestamp = parts.get("t", "")
    received = parts.get("v1", "")
    if not timestamp.isdigit() or not received:
        return False
    if abs(time.time() - int(timestamp)) > MAX_AGE_SECONDS:
        return False

    expected = hmac.new(
        SECRET, timestamp.encode() + b"." + raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, received)


app = Flask(__name__)


@app.post("/pruff/invoices/new")
def new_invoices():
    if not is_valid_signature(
        request.get_data(), request.headers.get("X-Webhook-Signature", "")
    ):
        abort(401)
    event = request.get_json()
    # Encola el evento para procesarlo fuera del handler.
    return "", 204