Webhook seguro: validação de assinatura, retries e idempotência

1. Fundamentos de Webhooks e Riscos de Segurança

Webhooks são callbacks HTTP automatizados que notificam sistemas sobre eventos em tempo real, diferentemente do polling tradicional onde o cliente consulta periodicamente o servidor. Essa abordagem reduz latência e tráfego, mas introduz riscos específicos de segurança.

Os principais vetores de ataque incluem:

  • Interceptação: invasores capturam o payload durante a transmissão
  • Replay attack: um evento legítimo é reenviado múltiplas vezes para causar duplicação
  • Falsificação de origem: atacantes enviam requisições fingindo ser o emissor legítimo

Um cenário típico de ataque envolve um atacante que intercepta um webhook de pagamento aprovado e o reenvia dezenas de vezes, creditando valores múltiplos na conta de destino. Sem mecanismos de segurança, o sistema receptor processa cada requisição como um evento válido.

2. Validação de Assinatura com HMAC

A validação de assinatura usa uma chave secreta compartilhada entre emissor e receptor, aplicando HMAC (Hash-based Message Authentication Code) com SHA-256.

Geração da assinatura no emissor

O emissor calcula o HMAC do payload JSON e inclui no cabeçalho:

POST /webhook HTTP/1.1
Host: receptor.exemplo.com
Content-Type: application/json
X-Signature-256: sha256=abc123def456...
X-Timestamp: 1700000000
X-Nonce: 7a8b9c0d1e2f

{"evento": "pagamento.criado", "valor": 100}

Implementação da verificação no receptor

import hmac
import hashlib
import time

CHAVE_SECRETA = "sua_chave_secreta_aqui"
JANELA_TEMPO = 300  # 5 minutos

def verificar_assinatura(payload, assinatura_recebida, timestamp, nonce):
    # 1. Verificar timestamp contra replay
    if abs(time.time() - timestamp) > JANELA_TEMPO:
        return False

    # 2. Construir mensagem com timestamp e nonce
    mensagem = f"{timestamp}.{nonce}.{payload}"

    # 3. Calcular HMAC esperado
    assinatura_esperada = hmac.new(
        CHAVE_SECRETA.encode(),
        mensagem.encode(),
        hashlib.sha256
    ).hexdigest()

    # 4. Comparação segura contra timing attacks
    return hmac.compare_digest(
        f"sha256={assinatura_esperada}",
        assinatura_recebida
    )

A inclusão de timestamp e nonce no cálculo da assinatura impede ataques de replay: mesmo que o payload seja reenviado, o timestamp expirado ou o nonce repetido invalidam a requisição.

3. Implementação de Retries com Backoff Inteligente

Emissores confiáveis implementam retry automático quando o receptor retorna erro. A estratégia de backoff exponencial com jitter evita sobrecarga:

import random
import time

BASE_DELAY = 2  # segundos
MAX_RETRIES = 5
MAX_TOTAL_TIME = 300  # 5 minutos

def calcular_delay(tentativa):
    delay = BASE_DELAY * (2 ** tentativa)
    jitter = random.uniform(0, delay * 0.1)
    return min(delay + jitter, MAX_TOTAL_TIME)

for tentativa in range(MAX_RETRIES):
    resposta = enviar_webhook(payload)

    if resposta.status_code == 200:
        break  # Sucesso

    if resposta.status_code >= 400 and resposta.status_code < 500:
        break  # Erro do cliente, não retentar

    delay = calcular_delay(tentativa)
    time.sleep(delay)

Regras importantes:
- Códigos 5xx (erro do servidor) disparam retry
- Códigos 4xx (erro do cliente) não disparam retry
- Limite máximo de 5 tentativas em até 5 minutos
- Logging de todas as falhas para auditoria

Após exceder o limite, o evento deve ser enviado para uma dead letter queue para análise manual.

4. Idempotência na Entrega de Webhooks

Mesmo com retries controlados, o mesmo evento pode ser entregue múltiplas vezes. A idempotência garante que cada evento seja processado exatamente uma vez.

Implementação com chave de idempotência

O emissor inclui um cabeçalho Idempotency-Key único por evento:

POST /webhook HTTP/1.1
Idempotency-Key: evt_20231115_001
X-Signature-256: sha256=...

Verificação no receptor

import redis

cache = redis.Redis(host='localhost', port=6379, db=0)
TEMPO_EXPIRACAO = 86400  # 24 horas

def processar_com_idempotencia(payload, idempotency_key):
    # 1. Verificar se chave já foi processada
    if cache.exists(f"processed:{idempotency_key}"):
        return {"status": "already_processed", "http_code": 409}

    # 2. Registrar chave com lock atômico
    if cache.setnx(f"lock:{idempotency_key}", "1"):
        try:
            # 3. Processar evento
            resultado = processar_evento(payload)

            # 4. Marcar como processado
            cache.setex(
                f"processed:{idempotency_key}",
                TEMPO_EXPIRACAO,
                "1"
            )
            return {"status": "success", "http_code": 200}
        finally:
            cache.delete(f"lock:{idempotency_key}")
    else:
        return {"status": "in_progress", "http_code": 202}

Respostas HTTP:
- 200 OK — evento processado com sucesso
- 202 Accepted — evento em processamento (outra requisição simultânea)
- 409 Conflict — evento já processado anteriormente

5. Orquestração do Fluxo Completo de Recebimento

O pipeline de validação segue a ordem:

  1. Verificação de assinatura — rejeitar requisições não autenticadas
  2. Verificação de idempotência — evitar processamento duplicado
  3. Processamento do evento — lógica de negócio
  4. Resposta HTTP — indicar resultado
def webhook_handler(request):
    # 1. Extrair cabeçalhos
    assinatura = request.headers.get('X-Signature-256')
    timestamp = int(request.headers.get('X-Timestamp', 0))
    nonce = request.headers.get('X-Nonce', '')
    idempotency_key = request.headers.get('Idempotency-Key', '')

    # 2. Validar assinatura
    if not verificar_assinatura(request.body, assinatura, timestamp, nonce):
        return {"status": "invalid_signature", "http_code": 401}

    # 3. Validar idempotência
    resultado = processar_com_idempotencia(request.body, idempotency_key)
    return resultado

Para payloads grandes, inclua um checksum SHA-256 do corpo no cabeçalho Content-MD5 ou X-Content-SHA256 para verificar integridade antes do processamento.

6. Boas Práticas de Operação e Manutenção

Rotação de chaves secretas

Implemente versionamento de chaves com sufixos:

CHAVES_SECRETAS = {
    "v1": "chave_ativa_atual",
    "v2": "chave_nova_em_transicao"
}

Mantenha chaves antigas por 30 dias para garantir que eventos em trânsito sejam validados.

Testes de integração

Simule cenários críticos:

# Teste 1: assinatura inválida
payload = {"evento": "teste"}
assinatura_invalida = "sha256=000000..."
assert webhook_handler(payload, assinatura_invalida)["http_code"] == 401

# Teste 2: evento duplicado
chave = "teste_duplicado"
primeiro = webhook_handler(payload, assinatura_valida, idempotency_key=chave)
segundo = webhook_handler(payload, assinatura_valida, idempotency_key=chave)
assert primeiro["http_code"] == 200
assert segundo["http_code"] == 409

Ferramentas de debug

  • webhook.site — testa recebimento de webhooks com interface visual
  • ngrok — expõe servidor local para receber webhooks de serviços externos
  • Logs estruturados — registre timestamp, idempotency_key, código HTTP e tempo de processamento

Documente o contrato completo do webhook: formato do payload, cabeçalhos obrigatórios, política de retry e códigos de resposta esperados.

Referências