Verificar a assinatura

Precificador-Assinatura: t=1759510800,v1=5f2b…

Node.js

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

const SEGREDO = process.env.PC3D_WEBHOOK_SEGREDO; // whsec_…
const JANELA_S = 300;

export function assinaturaValida(corpoCru, cabecalho, segredo, agora = Math.floor(Date.now() / 1000)) {
  if (typeof cabecalho !== 'string' || cabecalho.length > 300) return false;
  const partes = cabecalho.split(',').map((p) => p.trim());
  const t = Number(partes.find((p) => p.startsWith('t='))?.slice(2));
  const assinaturas = partes.filter((p) => p.startsWith('v1=')).map((p) => p.slice(3));
  if (!Number.isInteger(t) || Math.abs(agora - t) > JANELA_S || assinaturas.length === 0) return false;

  const esperada = crypto.createHmac('sha256', segredo).update(`${t}.`).update(corpoCru).digest();
  return assinaturas.some((hex) => {
    if (!/^[0-9a-f]{64}$/.test(hex)) return false;
    return crypto.timingSafeEqual(Buffer.from(hex, 'hex'), esperada);
  });
}

const app = express();

// express.raw: o corpo chega como Buffer, sem parse
app.post('/webhooks/precificador', express.raw({ type: 'application/json', limit: '16kb' }), (req, res) => {
  if (!assinaturaValida(req.body, req.get('Precificador-Assinatura'), SEGREDO)) {
    return res.sendStatus(401);
  }
  const evento = JSON.parse(req.body.toString('utf8'));
  // 1) deduplique por evento.id  2) coloque numa fila  3) reconsulte evento.url com a sua chave
  res.sendStatus(204);
});

app.listen(3000);

Python

import hashlib
import hmac
import os
import re
import time

from flask import Flask, abort, request

SEGREDO = os.environ["PC3D_WEBHOOK_SEGREDO"].encode()  # whsec_…
JANELA_S = 300


def assinatura_valida(corpo_cru: bytes, cabecalho: str | None, segredo: bytes, agora: int | None = None) -> bool:
    if not cabecalho or len(cabecalho) > 300:
        return False
    partes = [p.strip() for p in cabecalho.split(",")]
    t_txt = next((p[2:] for p in partes if p.startswith("t=")), "")
    assinaturas = [p[3:] for p in partes if p.startswith("v1=")]
    if not t_txt.isdigit() or not assinaturas:
        return False
    t = int(t_txt)
    if abs((agora or int(time.time())) - t) > JANELA_S:
        return False
    esperada = hmac.new(segredo, f"{t}.".encode() + corpo_cru, hashlib.sha256).hexdigest()
    return any(re.fullmatch(r"[0-9a-f]{64}", a) and hmac.compare_digest(a, esperada) for a in assinaturas)


app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024


@app.post("/webhooks/precificador")
def receber():
    corpo = request.get_data(cache=True)  # corpo cru, antes do parse
    if not assinatura_valida(corpo, request.headers.get("Precificador-Assinatura"), SEGREDO):
        abort(401)
    evento = request.get_json(force=True)
    # 1) deduplique por evento["id"]  2) coloque numa fila  3) reconsulte evento["url"] com a sua chave
    return "", 204