Segurança de Webhooks com HMAC: Pare de Aceitar Qualquer POST Vestido de Integração

Todo webhook começa como uma promessa elegante: um serviço avisa outro serviço, tudo acontece em tempo real, ninguém precisa ficar fazendo polling como se estivesse batendo na porta de cinco em cinco segundos. Bonito. Até o dia em que você percebe que seu endpoint aceita qualquer POST que saiba o caminho certo. Aí a integração moderna vira uma caixa de correio sem fechadura.

segurança de webhooks com HMAC é o tipo de assunto que parece exagero até você receber um payload duplicado, fora de ordem ou fabricado por alguém que descobriu a URL no histórico de logs. Não precisa ser um ataque cinematográfico. Às vezes é só um replay acidental, um retry agressivo do fornecedor, um proxy mal configurado ou aquele script de teste que alguém deixou rodando na sexta-feira porque a humanidade insiste em testar limites.

Neste artigo, vamos montar uma defesa prática para webhooks: assinatura HMAC, janela de tempo contra replay, idempotência para eventos repetidos, logs úteis e uma camada simples de rate limit. O objetivo não é transformar seu endpoint em um bunker paranoico. É fazer o mínimo decente para que uma automação externa não consiga bagunçar seu banco de dados só porque conhece uma URL.

1. O Problema Real: Webhook Não É Endpoint Público Normal

Um endpoint público comum costuma ter login, sessão, token OAuth, cookie, permissão por usuário ou algum contrato explícito de autenticação. Webhook geralmente não tem nada disso. Ele é chamado por sistemas externos, às vezes de fornecedores diferentes, com payloads que chegam quando querem e com retry automático quando alguma coisa dá errado.

Isso cria uma superfície estranha: você precisa aceitar chamadas externas, mas não pode confiar apenas no fato de que a chamada veio pela internet e parece ter o formato correto. Formato correto não prova origem. JSON válido não prova intenção. Cabeçalho bonito não prova honestidade. O servidor não tem sexto sentido; ele só tem bytes.

O erro clássico é validar só o schema: se tem event_id, type, created_at e data, então processa. Isso ajuda contra payload quebrado, mas não contra payload falso. É como conferir se uma nota de resgate está bem diagramada. Tecnicamente interessante, operacionalmente inútil.

2. A Defesa Base: Assinatura HMAC

HMAC é uma assinatura calculada com uma chave secreta compartilhada entre quem envia e quem recebe. O fornecedor pega o corpo bruto da requisição, combina com um segredo e gera um hash. Seu servidor faz a mesma conta. Se o resultado bater, você sabe que aquele payload foi assinado por alguém que conhece o segredo.

A regra de ouro: assine o corpo bruto, não o JSON reformatado. Se você parsear o JSON e depois serializar de novo, pode mudar espaços, ordem de campos ou escapes. A assinatura precisa nascer dos bytes originais recebidos na requisição. É chato, mas segurança adora detalhes chatos. Eles pagam o aluguel.

import hmac
import hashlib

SECRET = b"troque-por-um-segredo-longo-e-aleatorio"

def build_signature(raw_body: bytes, timestamp: str) -> str:
    signed_payload = timestamp.encode("utf-8") + b"." + raw_body
    digest = hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()
    return "sha256=" + digest

def signature_is_valid(raw_body: bytes, timestamp: str, received: str) -> bool:
    expected = build_signature(raw_body, timestamp)
    return hmac.compare_digest(expected, received)

O uso de hmac.compare_digest não é enfeite. Comparação comum de strings pode vazar pequenas diferenças de tempo. Em um endpoint pequeno isso talvez nunca seja explorado, mas usar a função correta custa nada. Quando o custo da disciplina é baixo, eu prefiro não bancar o artista conceitual da negligência.

3. Inclua Timestamp na Assinatura

Assinar apenas o corpo impede falsificação simples, mas não impede replay. Se alguém capturar uma requisição legítima, pode reenviar exatamente o mesmo corpo com exatamente a mesma assinatura. Para reduzir esse risco, inclua um timestamp no payload assinado e rejeite chamadas antigas.

Um cabeçalho simples resolve: X-Webhook-Timestamp. O remetente assina timestamp.body. O receptor confere se o timestamp está dentro de uma janela aceitável, por exemplo cinco minutos. Se chegou com meia hora de atraso, azar. Integração que depende de aceitar evento antigo sem rastreabilidade não é robusta; é uma aposta com logs.

from datetime import datetime, timezone

MAX_SKEW_SECONDS = 300

def timestamp_is_fresh(timestamp: str) -> bool:
    try:
        sent_at = datetime.fromtimestamp(int(timestamp), tz=timezone.utc)
    except ValueError:
        return False

    now = datetime.now(timezone.utc)
    age = abs((now - sent_at).total_seconds())
    return age <= MAX_SKEW_SECONDS

Essa janela exige relógios razoavelmente sincronizados. Em servidor decente, NTP ativo deveria ser tão normal quanto café ruim em reunião longa. Se o relógio do fornecedor vive atrasado, registre a ocorrência e pressione por correção. A solução não é deixar a porta aberta porque o relógio alheio resolveu estudar relatividade.

4. Proteção Contra Replay com Event ID

Timestamp reduz replay antigo, mas não elimina replay dentro da janela. Cinco minutos ainda bastam para duplicar cobrança, reprocessar tarefa ou disparar automação duas vezes. Por isso o webhook precisa ter um identificador único de evento: event_id, delivery_id ou algo equivalente.

Ao receber um evento, grave o ID antes ou durante o processamento em uma tabela de entregas. Se o mesmo ID aparecer de novo, responda 200 OK e não reprocesse. Sim, responda sucesso. Muitos fornecedores fazem retry quando recebem erro. Se você disser erro para duplicata, talvez ganhe mais duplicatas. Computadores são muito obedientes quando estão fazendo besteira.

CREATE TABLE webhook_deliveries (
    event_id TEXT PRIMARY KEY,
    event_type TEXT NOT NULL,
    received_at TEXT NOT NULL,
    status TEXT NOT NULL,
    payload_sha256 TEXT NOT NULL
);

import sqlite3
from datetime import datetime, timezone

def remember_delivery(conn: sqlite3.Connection, event_id: str, event_type: str, payload_hash: str) -> bool:
    try:
        conn.execute(
            """
            INSERT INTO webhook_deliveries(event_id, event_type, received_at, status, payload_sha256)
            VALUES (?, ?, ?, ?, ?)
            """,
            (event_id, event_type, datetime.now(timezone.utc).isoformat(), "received", payload_hash),
        )
        conn.commit()
        return True
    except sqlite3.IntegrityError:
        return False

Esse padrão conversa muito bem com o que já tratamos em idempotência em automações. A ideia é simples: retry não pode virar multiplicador de prejuízo. Um sistema maduro precisa conseguir ouvir a mesma notícia duas vezes sem sair gritando pela casa.

5. Pipeline de Validação: Ordem Importa

A ordem das validações evita trabalho inútil e reduz risco. Primeiro leia o corpo bruto. Depois confira presença dos cabeçalhos. Em seguida valide timestamp. Depois valide assinatura. Só então faça parse do JSON. Por fim, confira schema, tipo de evento e idempotência.

• Corpo vazio? Rejeite com 400.
• Sem timestamp ou assinatura? Rejeite com 401.
• Timestamp velho demais? Rejeite com 401 ou 400, mas registre.
• Assinatura inválida? Rejeite com 401.
• JSON inválido? Rejeite com 400.
• Evento duplicado? Responda 200 e ignore o processamento.
• Evento novo? Grave, processe e atualize status.

O detalhe importante: não parseie antes de validar assinatura. Parser é superfície de ataque também. Na prática, a maioria dos parsers modernos é segura, mas ainda é trabalho desnecessário para um payload que talvez nem tenha sido enviado por quem deveria. Primeiro identidade, depois conteúdo.

6. Implementação Flask Completa e Sem Misticismo

Abaixo vai um exemplo enxuto em Flask. Não é um framework mágico nem um produto final. É um esqueleto claro para você adaptar. Em produção, extraia segredo de variável de ambiente, use banco adequado ao volume, monitore erros e escreva testes. A diferença entre demo e produção é exatamente o número de lugares onde você admite que algo pode quebrar.

import hashlib
import hmac
import json
import os
import sqlite3
from datetime import datetime, timezone
from flask import Flask, request, jsonify

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"].encode("utf-8")
MAX_SKEW_SECONDS = 300


def db():
    conn = sqlite3.connect("webhooks.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS webhook_deliveries (
            event_id TEXT PRIMARY KEY,
            event_type TEXT NOT NULL,
            received_at TEXT NOT NULL,
            status TEXT NOT NULL,
            payload_sha256 TEXT NOT NULL
        )
    """)
    return conn


def fresh_timestamp(value: str) -> bool:
    try:
        sent_at = datetime.fromtimestamp(int(value), tz=timezone.utc)
    except ValueError:
        return False
    age = abs((datetime.now(timezone.utc) - sent_at).total_seconds())
    return age <= MAX_SKEW_SECONDS


def valid_signature(raw_body: bytes, timestamp: str, received: str) -> bool:
    signed = timestamp.encode("utf-8") + b"." + raw_body
    digest = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    expected = "sha256=" + digest
    return hmac.compare_digest(expected, received or "")


def insert_once(conn, event_id, event_type, payload_hash):
    try:
        conn.execute(
            "INSERT INTO webhook_deliveries VALUES (?, ?, ?, ?, ?)",
            (event_id, event_type, datetime.now(timezone.utc).isoformat(), "received", payload_hash),
        )
        conn.commit()
        return True
    except sqlite3.IntegrityError:
        return False


@app.post("/webhooks/vendor")
def receive_vendor_webhook():
    raw = request.get_data(cache=False)
    timestamp = request.headers.get("X-Webhook-Timestamp")
    signature = request.headers.get("X-Webhook-Signature")

    if not raw or not timestamp or not signature:
        return jsonify({"error": "missing webhook authentication"}), 401

    if not fresh_timestamp(timestamp):
        return jsonify({"error": "stale webhook"}), 401

    if not valid_signature(raw, timestamp, signature):
        return jsonify({"error": "invalid signature"}), 401

    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        return jsonify({"error": "invalid json"}), 400

    event_id = payload.get("event_id")
    event_type = payload.get("type")
    if not event_id or not event_type:
        return jsonify({"error": "missing event identity"}), 400

    payload_hash = hashlib.sha256(raw).hexdigest()
    conn = db()
    if not insert_once(conn, event_id, event_type, payload_hash):
        return jsonify({"status": "duplicate_ignored"}), 200

    # Aqui entra sua regra de negócio real.
    # Faça pouco dentro da request; fila é melhor para trabalho pesado.
    conn.execute("UPDATE webhook_deliveries SET status = ? WHERE event_id = ?", ("processed", event_id))
    conn.commit()
    return jsonify({"status": "processed"}), 200

Esse código é propositalmente pequeno, mas já evita a tragédia comum: endpoint público sem autenticação forte, sem janela temporal e sem memória de eventos. É impressionante o quanto de caos operacional nasce de um @app.post inocente demais.

7. Teste Local com curl: Porque Fé Não É Estratégia

Você precisa testar assinatura válida, assinatura inválida, timestamp velho, JSON quebrado e evento duplicado. Testar só o caminho feliz é basicamente perguntar para o ladrão se ele pretende tocar a campainha. Abaixo vai um gerador simples de assinatura para usar junto com curl.

BODY='{"event_id":"evt_123","type":"invoice.paid","data":{"amount":9900}}'
TS=$(date +%s)
SIG=$(python3 - <<'PY_SIG'
import hmac, hashlib, os
secret = os.environ["WEBHOOK_SECRET"].encode()
body = os.environ["BODY"].encode()
ts = os.environ["TS"]
print("sha256=" + hmac.new(secret, ts.encode() + b"." + body, hashlib.sha256).hexdigest())
PY_SIG
)

curl -i http://localhost:5000/webhooks/vendor   -H "Content-Type: application/json"   -H "X-Webhook-Timestamp: $TS"   -H "X-Webhook-Signature: $SIG"   --data "$BODY"

Depois rode a mesma requisição duas vezes. A primeira deve processar. A segunda deve retornar duplicata ignorada. Esse é o momento em que você descobre se a sua idempotência está funcionando ou se ela era só uma palavra elegante no quadro da sprint.

8. Box Perrengue: O Replay Que Virou Cobrança Duplicada

Esse tipo de falha dói porque parece injusta. O webhook era legítimo. A assinatura, se existisse, passaria. O problema era semântico: a entrega era repetida e o sistema tratava repetição como novidade. Por isso assinatura não basta. Ela responde “quem enviou?”. Idempotência responde “eu já lidei com isso?”. Você precisa das duas perguntas.

9. Logs Que Ajudam Sem Vazar Segredo

Log de webhook precisa ser útil, mas não fofoqueiro. Nunca registre o segredo. Evite gravar payload completo quando ele contém dados pessoais, tokens ou informações financeiras. Em vez disso, registre ID do evento, tipo, hash do payload, status, latência, IP de origem e motivo de rejeição. O objetivo é depurar sem criar um vazamento em câmera lenta.

• event_id: identifica a entrega.
• event_type: ajuda a filtrar comportamento.
• payload_sha256: comprova se dois corpos eram iguais sem expor o corpo.
• decision: accepted, duplicate, invalid_signature, stale_timestamp.
• duration_ms: mostra gargalo antes que ele vire timeout.

Se você gostou dessa linha, vale ler também triagem de logs de erro em automações Python e observabilidade de logs para debugar produção. Webhook sem log decente vira arqueologia: você fica escovando vestígios tentando descobrir quem derrubou o templo.

10. Rate Limit e Firewall: Camada Extra, Não Muleta

Rate limit não substitui HMAC. IP allowlist não substitui HMAC. Firewall não substitui HMAC. Mas todos ajudam. Defesa boa é redundante sem virar carnaval. Se o fornecedor publica faixas de IP confiáveis, você pode limitar tráfego. Se usa Nginx, dá para reduzir abuso básico com limit_req.

limit_req_zone $binary_remote_addr zone=webhook_limit:10m rate=10r/s;

server {
    location /webhooks/vendor {
        limit_req zone=webhook_limit burst=20 nodelay;
        proxy_pass http://app:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

O cuidado aqui é não bloquear retries legítimos. Webhooks costumam chegar em rajadas quando um lote de eventos é liberado. Comece com limites generosos, monitore e ajuste. Segurança que derruba o fluxo de negócio todo dia vira inimiga da operação, e aí alguém desliga “temporariamente”. Temporariamente, como sabemos, é uma unidade de tempo que pode durar até a aposentadoria.

11. Rotação de Segredo Sem Derrubar Integração

Segredo eterno é senha escrita em pedra. Em algum momento você vai precisar trocar. O jeito adulto de fazer isso é aceitar dois segredos por um período: o atual e o anterior. O remetente passa a assinar com o novo; seu receptor aceita ambos durante a janela de migração; depois você remove o antigo.

SECRETS = [
    os.environ["WEBHOOK_SECRET_CURRENT"].encode(),
    os.environ.get("WEBHOOK_SECRET_PREVIOUS", "").encode(),
]

def valid_with_any_secret(raw_body, timestamp, received):
    signed = timestamp.encode() + b"." + raw_body
    for secret in filter(None, SECRETS):
        digest = hmac.new(secret, signed, hashlib.sha256).hexdigest()
        if hmac.compare_digest("sha256=" + digest, received or ""):
            return True
    return False

Documente a data de remoção do segredo antigo. Sem data, você não criou rotação; criou uma coleção. E coleção de segredos antigos é basicamente um museu de incidentes esperando curador.

12. Checklist de Produção

Antes de colocar no ar, eu usaria esta lista. Ela é curta de propósito. Checklist enorme vira decoração corporativa; checklist bom machuca um pouco porque obriga decisão.

• O corpo bruto é usado na assinatura, sem reserializar JSON.
• A assinatura usa HMAC-SHA256 com segredo longo e aleatório.
• A comparação usa função constante, como hmac.compare_digest.
• O timestamp faz parte do conteúdo assinado.
• Eventos fora da janela temporal são rejeitados.
• O event_id é único e armazenado antes do processamento perigoso.
• Duplicatas retornam sucesso sem repetir efeito colateral.
• Logs registram decisão e hash, não segredo.
• Há teste automatizado para assinatura inválida, replay e duplicata.
• Existe plano de rotação de segredo.

Conclusão: Webhook Seguro É Webhook Que Desconfia com Educação

segurança de webhooks com HMAC não é uma frescura de arquitetura. É a diferença entre uma integração confiável e um endpoint público aceitando ordens de qualquer pessoa com conexão à internet. A parte boa é que a implementação não precisa ser enorme: HMAC no corpo bruto, timestamp assinado, janela anti-replay, event ID idempotente, logs úteis e limites razoáveis já resolvem uma quantidade absurda de problemas.

O ponto central é parar de tratar webhook como “só mais uma rota”. Ele é uma fronteira entre sistemas. Fronteira sem controle vira ponto turístico para bug. E bug, quando encontra automação, ganha perna, agenda e senso de oportunidade.

Se você quer aprofundar a parte operacional, veja a categoria Fortaleza Digital e compare com os padrões de resiliência em automação resiliente com Python. Segurança e confiabilidade não são departamentos separados; são o mesmo adulto responsável usando crachás diferentes.

CTA: qual automação você quer ver blindada no próximo artigo? Webhook de pagamento, bot de Telegram, integração com planilha, pipeline de deploy ou rotina de backup? Manda o cenário real. Quanto mais concreto o perrengue, melhor o tutorial.