Verificar a assinatura

Qualquer pessoa que descobrir a URL do seu webhook pode mandar um POST para ela. A assinatura prova que o evento veio da TopTalking e não foi alterado no caminho. Sempre valide antes de processar.

Como a assinatura é feita

Cada entrega traz o cabeçalho:

X-TopTalking-Signature: t=1794409800,v1=946dcacfdaee2f0eca7a2194d9919092cac50f554189d6651b1ce5acca155cdc

• t — momento do envio, em Unix timestamp (segundos).
• v1 — assinatura em hexadecimal:

v1 = hex( HMAC_SHA256( chave = secret, mensagem = t + "." + corpo_cru ) )

O secret é o segredo de assinatura do webhook (whsec_...), usado inteiro (com o prefixo whsec_) como chave. Ele aparece quando o webhook é criado; no painel dá para Revelar segredo de novo ou Gerar novo segredo.

Passo a passo

1. Leia o corpo cru da requisição, exatamente como chegou (bytes ou texto), antes de qualquer JSON.parse. Reformatar o JSON muda os bytes e quebra a assinatura.
2. Separe t e v1 do cabeçalho X-TopTalking-Signature.
3. Rejeite se t estiver a mais de 300 segundos do horário atual (protege contra reenvio de eventos antigos).
4. Calcule HMAC-SHA256(secret, "<t>.<corpo cru>") em hexadecimal.
5. Compare com v1 usando comparação em tempo constante.
6. Se não bater, responda 401 e não processe nada.

Código pronto

Node.js (Express)

toptalking-assinatura.js

import crypto from 'node:crypto';

const TOLERANCIA_SEGUNDOS = 300; // 5 minutos

/**
 * Confere o cabeçalho X-TopTalking-Signature.
 * @param {Buffer|string} corpoCru  corpo EXATO recebido (antes de qualquer JSON.parse)
 * @param {string|undefined} cabecalho  valor de X-TopTalking-Signature
 * @param {string} secret  segredo do webhook (whsec_...)
 */
export function verificarAssinatura(corpoCru, cabecalho, secret) {
  if (!cabecalho || !secret) return false;

  // Formato: t=<unix em segundos>,v1=<hex>
  let t = null;
  const assinaturas = [];
  for (const parte of cabecalho.split(',')) {
    const i = parte.indexOf('=');
    if (i === -1) continue;
    const chave = parte.slice(0, i).trim();
    const valor = parte.slice(i + 1).trim();
    if (chave === 't') t = valor;
    else if (chave === 'v1' && valor) assinaturas.push(valor);
  }
  if (!t || !/^\d+$/.test(t) || assinaturas.length === 0) return false;

  // Rejeita eventos antigos (proteção contra replay).
  const agora = Math.floor(Date.now() / 1000);
  if (Math.abs(agora - Number(t)) > TOLERANCIA_SEGUNDOS) return false;

  // v1 = HMAC-SHA256(chave = secret, mensagem = "<t>.<corpo cru>")
  const esperado = crypto.createHmac('sha256', secret).update(`${t}.`).update(corpoCru).digest();

  // Comparação em tempo constante.
  return assinaturas.some((hex) => {
    const recebido = Buffer.from(hex, 'hex');
    return recebido.length === esperado.length && crypto.timingSafeEqual(recebido, esperado);
  });
}

server.js

import express from 'express';
import { verificarAssinatura } from './toptalking-assinatura.js';

const app = express();

// express.raw entrega o corpo como Buffer, sem parse. Não use express.json() nesta rota.
app.post('/webhooks/toptalking', express.raw({ type: 'application/json' }), (req, res) => {
  const ok = verificarAssinatura(
    req.body,
    req.get('X-TopTalking-Signature'),
    process.env.TOPTALKING_WEBHOOK_SECRET,
  );
  if (!ok) return res.status(401).send('assinatura inválida');

  const evento = JSON.parse(req.body.toString('utf8'));
  res.sendStatus(200); // responda rápido…

  setImmediate(() => processarEvento(evento)); // …e processe depois (ideal: uma fila)
});

app.listen(3000);

Cuidado com o express.json() global

Se o seu app usa app.use(express.json()), ele consome o corpo antes da sua rota. Registre a rota do webhook antes do express.json() ou exclua o caminho dela do parser.

Python (Flask)

app.py

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["TOPTALKING_WEBHOOK_SECRET"]
TOLERANCIA_SEGUNDOS = 300


def verificar_assinatura(corpo: bytes, cabecalho, secret: str) -> bool:
    """corpo = bytes EXATOS recebidos; cabecalho = X-TopTalking-Signature."""
    if not cabecalho or not secret:
        return False
    t, assinaturas = None, []
    for parte in cabecalho.split(","):
        chave, _, valor = parte.strip().partition("=")
        if chave == "t":
            t = valor
        elif chave == "v1" and valor:
            assinaturas.append(valor)
    if not t or not t.isdigit() or not assinaturas:
        return False
    if abs(time.time() - int(t)) > TOLERANCIA_SEGUNDOS:
        return False
    esperado = hmac.new(secret.encode(), f"{t}.".encode() + corpo, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(esperado.encode(), a.encode()) for a in assinaturas)


@app.post("/webhooks/toptalking")
def webhook_toptalking():
    corpo = request.get_data()  # bytes crus
    if not verificar_assinatura(corpo, request.headers.get("X-TopTalking-Signature"), SECRET):
        abort(401)
    evento = request.get_json()
    # Enfileire o processamento e responda rápido.
    fila.enqueue(processar_evento, evento)
    return "", 200

Python (FastAPI)

main.py

import hashlib
import hmac
import json
import os
import time

from fastapi import BackgroundTasks, FastAPI, HTTPException, Request

app = FastAPI()
SECRET = os.environ["TOPTALKING_WEBHOOK_SECRET"]
TOLERANCIA_SEGUNDOS = 300


def verificar_assinatura(corpo: bytes, cabecalho, secret: str) -> bool:
    if not cabecalho or not secret:
        return False
    t, assinaturas = None, []
    for parte in cabecalho.split(","):
        chave, _, valor = parte.strip().partition("=")
        if chave == "t":
            t = valor
        elif chave == "v1" and valor:
            assinaturas.append(valor)
    if not t or not t.isdigit() or not assinaturas:
        return False
    if abs(time.time() - int(t)) > TOLERANCIA_SEGUNDOS:
        return False
    esperado = hmac.new(secret.encode(), f"{t}.".encode() + corpo, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(esperado.encode(), a.encode()) for a in assinaturas)


@app.post("/webhooks/toptalking")
async def webhook_toptalking(request: Request, background: BackgroundTasks):
    corpo = await request.body()  # bytes crus, antes de qualquer parse
    if not verificar_assinatura(corpo, request.headers.get("x-toptalking-signature"), SECRET):
        raise HTTPException(status_code=401, detail="assinatura inválida")
    evento = json.loads(corpo)
    background.add_task(processar_evento, evento)  # roda depois da resposta
    return {"ok": True}

Deno / Supabase Edge Function

Ideal para apps do Lovable e projetos Supabase. Usa só Web Crypto (sem dependências).

supabase/functions/toptalking-webhook/index.ts

// Recebe os webhooks da TopTalking e valida a assinatura.
// Secret necessário: TOPTALKING_WEBHOOK_SECRET (o whsec_... do webhook).
const SECRET = Deno.env.get('TOPTALKING_WEBHOOK_SECRET') ?? '';
const TOLERANCIA_SEGUNDOS = 300;
const encoder = new TextEncoder();

function hexParaBytes(hex: string): Uint8Array | null {
  if (hex.length % 2 !== 0 || !/^[0-9a-f]+$/i.test(hex)) return null;
  const bytes = new Uint8Array(hex.length / 2);
  for (let i = 0; i < bytes.length; i++) bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16);
  return bytes;
}

async function verificarAssinatura(corpo: string, cabecalho: string | null, secret: string) {
  if (!cabecalho || !secret) return false;

  // Formato: t=<unix em segundos>,v1=<hex>
  let t: string | null = null;
  const assinaturas: string[] = [];
  for (const parte of cabecalho.split(',')) {
    const i = parte.indexOf('=');
    if (i === -1) continue;
    const chave = parte.slice(0, i).trim();
    const valor = parte.slice(i + 1).trim();
    if (chave === 't') t = valor;
    else if (chave === 'v1' && valor) assinaturas.push(valor);
  }
  if (!t || !/^\d+$/.test(t) || assinaturas.length === 0) return false;

  // Rejeita eventos com mais de 5 minutos de diferença (replay).
  if (Math.abs(Date.now() / 1000 - Number(t)) > TOLERANCIA_SEGUNDOS) return false;

  const chave = await crypto.subtle.importKey(
    'raw',
    encoder.encode(secret),
    { name: 'HMAC', hash: 'SHA-256' },
    false,
    ['verify'],
  );
  const dados = encoder.encode(`${t}.${corpo}`);
  for (const hex of assinaturas) {
    const assinatura = hexParaBytes(hex);
    // crypto.subtle.verify já compara em tempo constante.
    if (assinatura && (await crypto.subtle.verify('HMAC', chave, assinatura, dados))) return true;
  }
  return false;
}

Deno.serve(async (req) => {
  if (req.method !== 'POST') return new Response('Método não permitido', { status: 405 });

  const corpo = await req.text(); // corpo CRU: não use req.json() antes de validar
  const valido = await verificarAssinatura(corpo, req.headers.get('X-TopTalking-Signature'), SECRET);
  if (!valido) return new Response('assinatura inválida', { status: 401 });

  const evento = JSON.parse(corpo);

  // Faça aqui o que precisar (ex.: salvar numa tabela). Mantenha rápido: < 10 s.
  console.log('Evento recebido:', evento.type, evento.id);

  return new Response(JSON.stringify({ ok: true }), {
    headers: { 'Content-Type': 'application/json' },
  });
});

A TopTalking não envia o token de login da Supabase, então desligue a verificação de JWT desta função:

supabase/config.toml

[functions.toptalking-webhook]
verify_jwt = false

A URL para cadastrar no painel fica assim: https://<seu-projeto>.supabase.co/functions/v1/toptalking-webhook. O guia completo, com tabela e prompt pronto, está em Lovable.

Teste a sua implementação

Use estes valores para conferir se o seu código calcula a assinatura certa (desligue a checagem de tempo só neste teste, pois t é fixo):

Item	Valor
secret	whsec_d8d10826ecb93b4679de7409553d10982e864a172f6df97b
t	1794409800
corpo cru	o JSON abaixo, exatamente assim (uma linha, sem espaços)
v1 esperado	946dcacfdaee2f0eca7a2194d9919092cac50f554189d6651b1ce5acca155cdc

{"id":"252a2c18-7b60-47a6-9feb-9c8322f2ba50","object":"event","type":"ping","created_at":"2026-11-11T15:10:00Z","organization_id":"e044a615-e879-44bb-8510-948b10b5e93c","data":{"webhook_id":"f03b0773-5912-4a45-9139-c29375485843","message":"Evento de teste enviado pela TopTalking."}}

Depois, clique em Enviar teste no painel e confira em Ver entregas se o seu endpoint respondeu 200.

Problemas comuns

Sintoma	Causa provável
A assinatura nunca bate	O corpo foi convertido em objeto e depois em texto de novo (JSON.stringify(req.body)). Use o corpo cru.
Bate no teste, falha em produção	Um middleware (body parser, proxy, framework) alterou o corpo antes da sua rota.
Falha só às vezes	Relógio do servidor fora de hora. Sincronize com NTP; a tolerância é de 300 s.
Parou de bater depois de um tempo	Alguém clicou em Gerar novo segredo. Atualize a variável de ambiente.
401 vindo da Supabase, não do seu código	A função está com verify_jwt ligado. Desligue para esta função.