Receptor de webhook

Um receptor correto faz quatro coisas, nesta ordem: lê o corpo como bytes, confere a assinatura antes de interpretar o JSON, deduplica pelo ID do evento e responde 2xx rápido. O resto do trabalho vai para a sua fila.

1. A conferência da assinatura

A função não depende de framework. Ela recebe o corpo bruto e os três cabeçalhos da entrega e devolve verdadeiro ou falso.

Node.js

assinatura.mjs

import { createHmac, timingSafeEqual } from 'node:crypto'

const TOLERANCIA_SEGUNDOS = 300

/** Confere a assinatura com os bytes exatos do corpo, antes de ler o JSON. */
export function assinaturaValida({ segredo, eventoId, instante, assinatura, corpo, agora = Date.now() }) {
  if (!/^\d{1,12}$/.test(instante ?? '')) return false
  if (Math.abs(Math.floor(agora / 1000) - Number(instante)) > TOLERANCIA_SEGUNDOS) return false
  if (!/^v1=[0-9a-f]{64}$/.test(assinatura ?? '')) return false
  const esperada = createHmac('sha256', segredo).update(`${instante}.${eventoId}.`).update(corpo).digest()
  return timingSafeEqual(esperada, Buffer.from(assinatura.slice(3), 'hex'))
}

Python

assinatura.py

import hashlib
import hmac
import re
import time

TOLERANCIA_SEGUNDOS = 300
FORMATO = re.compile(r"v1=[0-9a-f]{64}")


def assinatura_valida(segredo: str, evento_id: str, instante: str, assinatura: str, corpo: bytes,
                      agora: float | None = None) -> bool:
    """Confere a assinatura com os bytes exatos do corpo, antes de ler o JSON."""
    if not instante or not instante.isdigit():
        return False
    if abs(int(agora if agora is not None else time.time()) - int(instante)) > TOLERANCIA_SEGUNDOS:
        return False
    if not assinatura or not FORMATO.fullmatch(assinatura):
        return False
    esperada = hmac.new(segredo.encode(), f"{instante}.{evento_id}.".encode() + corpo, hashlib.sha256).hexdigest()
    return hmac.compare_digest(esperada, assinatura[3:])

Regra	Por quê
HMAC-SHA256 sobre instante + "." + event_id + "." + corpo	O ID e o instante entram na assinatura: não dá para reaproveitar uma assinatura em outro evento.
Segredo completo, com o prefixo ctj_whsec_	O prefixo faz parte da chave do HMAC.
Corpo em bytes, antes de qualquer parse	Reserializar o JSON muda espaços e ordem, e a assinatura deixa de bater.
Janela de 300 segundos	Uma entrega capturada não pode ser reenviada mais tarde.
Comparação em tempo constante	Evita que o tempo de resposta revele a assinatura certa.

2. O endpoint

Node.js

receptor.mjs

// Receptor de webhooks do Cotejar com node:http (Node.js 20+), sem dependências.
import { createServer } from 'node:http'
import { assinaturaValida } from './assinatura.mjs'

const SEGREDO = process.env.COTEJAR_WEBHOOK_SECRET // ctj_whsec_..., exibido ao configurar o destino
const MAXIMO_CORPO = 16 * 1024

// Troque por uma tabela com chave única no ID do evento, gravada na mesma
// transação que aceita o processamento.
const processados = new Set()

async function lerCorpo(req) {
  const partes = []
  let tamanho = 0
  for await (const parte of req) {
    tamanho += parte.length
    if (tamanho > MAXIMO_CORPO) throw new Error('corpo excedido')
    partes.push(parte)
  }
  return Buffer.concat(partes)
}

const servidor = createServer(async (req, res) => {
  if (req.method !== 'POST' || req.url !== '/cotejar/eventos') return res.writeHead(404).end()
  let corpo
  try { corpo = await lerCorpo(req) } catch { return res.writeHead(413).end() }

  const eventoId = req.headers['cotejar-event-id']
  const valida = typeof eventoId === 'string' && assinaturaValida({
    segredo: SEGREDO, eventoId, corpo,
    instante: req.headers['cotejar-timestamp'], assinatura: req.headers['cotejar-signature'],
  })
  if (!valida) return res.writeHead(401).end()

  const evento = JSON.parse(corpo.toString('utf8'))
  if (evento.id !== eventoId || processados.has(eventoId)) return res.writeHead(204).end()
  processados.add(eventoId)
  // Responda rápido; trabalho pesado vai para a sua fila.
  console.log(`evento ${evento.dados.tipo} da empresa ${evento.empresa}`)
  res.writeHead(204).end()
})

servidor.listen(Number(process.env.PORT ?? 8080))

COTEJAR_WEBHOOK_SECRET=ctj_whsec_... PORT=8080 node receptor.mjs

Python

receptor.py

"""Receptor de webhooks do Cotejar com Flask."""
import json
import os

from flask import Flask, request

from assinatura import assinatura_valida

SEGREDO = os.environ["COTEJAR_WEBHOOK_SECRET"]  # ctj_whsec_..., exibido ao configurar o destino

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024
processados: set[str] = set()  # troque por uma tabela com chave única no ID do evento


@app.post("/cotejar/eventos")
def receber():
    corpo = request.get_data(cache=False)  # bytes exatos, antes de qualquer parse
    evento_id = request.headers.get("Cotejar-Event-Id", "")
    if not assinatura_valida(SEGREDO, evento_id, request.headers.get("Cotejar-Timestamp", ""),
                             request.headers.get("Cotejar-Signature", ""), corpo):
        return "", 401
    evento = json.loads(corpo)
    if evento["id"] != evento_id or evento_id in processados:
        return "", 204
    processados.add(evento_id)
    # Responda rápido; trabalho pesado vai para a sua fila.
    print(f"evento {evento['dados']['tipo']} da empresa {evento['empresa']}", flush=True)
    return "", 204

pip install flask
COTEJAR_WEBHOOK_SECRET=ctj_whsec_... flask --app receptor run --port 8080

Em produção, sirva com um servidor WSGI (gunicorn, uWSGI) atrás de HTTPS.

Resposta do receptor	Efeito no Cotejar
2xx concluído	Entrega confirmada.
401 (assinatura inválida)	Tentativa falha e entra no ciclo de novas tentativas.
Redirecionamento	Não é seguido; conta como falha.
Resposta acima de 16 KiB ou demora	Falha da tentativa. Responda vazio e cedo.

204 para um evento repetido é intencional: outra tentativa pode chegar com novo instante e nova assinatura para o mesmo ID depois de uma resposta perdida. Confirme a entrega sem aplicar o efeito de novo.

Deduplicação de verdade

O Set em memória serve para o exemplo. Em produção, grave o Cotejar-Event-Id numa tabela com chave única, na mesma transação que registra o trabalho. Assim um evento nunca é processado duas vezes, mesmo com dois processos recebendo em paralelo.

3. O teste com o vetor publicado

Rode este teste antes de expor o endpoint. Ele usa o vetor fixo da página de testes e injeta o relógio, então não depende de rede nem de horário.

Node.js

testar-assinatura.mjs

import assert from 'node:assert/strict'
import { test } from 'node:test'
import { assinaturaValida } from './assinatura.mjs'

// Vetor publicado em docs.cotejar.com.br/inicio/testes; o segredo é fictício.
const vetor = {
  segredo: 'ctj_whsec_exemplo_de_teste_nao_use_em_producao_000000',
  eventoId: '3f2b8c1e-5d4a-4b7e-9c2f-1a6d8e0b4c71',
  instante: '1792000000',
  assinatura: 'v1=536f51b3d450c13da9a5c021f67bf7599fd4923d5fb32c910e01b728d7e5340a',
  corpo: Buffer.from('{"id":"3f2b8c1e-5d4a-4b7e-9c2f-1a6d8e0b4c71","versao":1,"empresa":"12345678","ocorridoEm":"2026-10-14T12:26:40.000Z","dados":{"tipo":"webhook.teste","destinoId":"9b1c2d3e-4f50-4a61-8b72-93a4b5c6d7e8"}}'),
  agora: 1792000030 * 1000,
}

test('aceita o vetor publicado', () => assert.equal(assinaturaValida(vetor), true))
test('recusa um byte diferente no corpo', () => {
  const corpo = Buffer.from(vetor.corpo); corpo[corpo.length - 2] ^= 1
  assert.equal(assinaturaValida({ ...vetor, corpo }), false)
})
test('recusa fora da janela de cinco minutos', () => assert.equal(assinaturaValida({ ...vetor, agora: (1792000000 + 301) * 1000 }), false))
test('recusa outro evento com a mesma assinatura', () => assert.equal(assinaturaValida({ ...vetor, eventoId: '00000000-0000-4000-8000-000000000000' }), false))

node --test testar-assinatura.mjs

Python

test_assinatura.py

import unittest

from assinatura import assinatura_valida

# Vetor publicado em docs.cotejar.com.br/inicio/testes; o segredo é fictício.
VETOR = dict(
    segredo="ctj_whsec_exemplo_de_teste_nao_use_em_producao_000000",
    evento_id="3f2b8c1e-5d4a-4b7e-9c2f-1a6d8e0b4c71",
    instante="1792000000",
    assinatura="v1=536f51b3d450c13da9a5c021f67bf7599fd4923d5fb32c910e01b728d7e5340a",
    corpo=b'{"id":"3f2b8c1e-5d4a-4b7e-9c2f-1a6d8e0b4c71","versao":1,"empresa":"12345678","ocorridoEm":"2026-10-14T12:26:40.000Z","dados":{"tipo":"webhook.teste","destinoId":"9b1c2d3e-4f50-4a61-8b72-93a4b5c6d7e8"}}',
    agora=1792000030,
)


class TesteAssinatura(unittest.TestCase):
    def test_aceita_o_vetor_publicado(self):
        self.assertTrue(assinatura_valida(**VETOR))

    def test_recusa_corpo_alterado(self):
        self.assertFalse(assinatura_valida(**{**VETOR, "corpo": VETOR["corpo"].replace(b"12345678", b"12345679")}))

    def test_recusa_fora_da_janela(self):
        self.assertFalse(assinatura_valida(**{**VETOR, "agora": 1792000000 + 301}))


if __name__ == "__main__":
    unittest.main()

python -m unittest test_assinatura

4. A prova de ponta a ponta

1. Publique o receptor num endereço HTTPS público, na porta 443.
2. Configure o destino em Integrações › Webhooks e guarde o segredo ctj_whsec_... exibido.
3. Use Enviar evento de teste e confira o histórico: o webhook.teste precisa chegar a Entregue.
4. Se a tentativa falhar, abra Ver tentativas: o status HTTP e o código de falha mostram se o problema foi rede, TLS, redirecionamento ou a resposta do seu receptor.

Os dois receptores foram exercitados com entregas assinadas pelo mesmo código que assina as entregas do Cotejar: evento válido (204), repetido (204), corpo alterado (401), instante de dez minutos atrás (401) e sem assinatura (401).