Webhook Receiver em Python Puro: O Receptor Que Valida Assinaturas HMAC e Reprocessa Falhas Automaticamente (Sem Depender de ngrok ou Serviços Pagos)
Se você integra com Stripe, GitHub, ou qualquer serviço que dispara eventos, você já precisou receber webhooks. E provavelmente fez uma das duas coisas: usou um serviço pago como ngrok/Zapier, ou criou um endpoint sem verificar a assinatura e rezou para ninguém falsificar requisições.
Hoje vamos construir um webhook receiver completo em Python puro que:
O Problema Que Webhooks Resolvem (E O Que Eles Escondem)
Webhooks são notificações push. Ao invés de sua aplicação ficar perguntando “tem atualização? tem atualização?” (polling), o serviço externo avisa quando algo acontece.
Mas isso cria três problemas críticos:
1. **Autenticidade**: Como saber se a requisição veio realmente do Stripe/GitHub e não de um atacante?
2. **Idempotência**: O que fazer se o mesmo evento chegar duas vezes (retry do serviço ou falha de rede)?
3. **Resiliência**: Se seu processamento falhar, como garantir que não perdemos o evento?
A resposta para todos eles está em três padrões: verificação HMAC, deduplicação por ID, e retry com persistência.
Box Perrengue: A Vez Que Perdi R$ 2.400 em Pagamentos Duplicados
Em 2024, eu integrava um e-commerce com Stripe. O webhook de `payment_intent.succeeded` disparava e meu código criava uma entrada no banco de dados. Funcionou por meses.
Até que um dia o Stripe reenviou o mesmo evento três vezes (falha de rede do lado deles). Meu código processou os três. Resultado: três lançamentos de R$ 800 para o mesmo pedido. O cliente recebeu o produto uma vez, mas foi cobrado três vezes.
O suporte do Stripe me devolveu o dinheiro, mas não antes de eu perder 6 horas reconciliando transações e implementar deduplicação às pressas.
Desde então, todo webhook que eu escrevo tem:
Vamos implementar isso agora.
O Código: Webhook Receiver Completo
#!/usr/bin/env python3
"""
Webhook Receiver com HMAC Validation e Retry Automático
Python 3.8+ | Zero dependências externas
"""
import hashlib
import hmac
import json
import sqlite3
import time
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import datetime, timezone
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass, asdict
from pathlib import Path
@dataclass
class WebhookEvent:
"""Representa um evento de webhook recebido"""
event_id: str
source: str # 'stripe', 'github', etc.
event_type: str
payload: Dict[str, Any]
received_at: str
signature: Optional[str] = None
status: str = 'pending' # pending, processing, completed, failed
attempts: int = 0
last_attempt_at: Optional[str] = None
error_message: Optional[str] = None
class WebhookStorage:
"""Persistência SQLite para auditoria e retry"""
def __init__(self, db_path: str = 'webhooks.db'):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Cria tabelas se não existirem"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS events (
event_id TEXT PRIMARY KEY,
source TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
received_at TEXT NOT NULL,
signature TEXT,
status TEXT NOT NULL,
attempts INTEGER DEFAULT 0,
last_attempt_at TEXT,
error_message TEXT
)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_status
ON events(status)
''')
conn.execute('''
CREATE INDEX IF NOT EXISTS idx_received_at
ON events(received_at)
''')
def save(self, event: WebhookEvent) -> bool:
"""Salva evento. Retorna False se já existir (deduplicação)"""
try:
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
INSERT INTO events (
event_id, source, event_type, payload,
received_at, signature, status, attempts,
last_attempt_at, error_message
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
event.event_id,
event.source,
event.event_type,
json.dumps(event.payload),
event.received_at,
event.signature,
event.status,
event.attempts,
event.last_attempt_at,
event.error_message
))
return True
except sqlite3.IntegrityError:
# event_id duplicado
return False
def update_status(self, event_id: str, status: str,
error_message: Optional[str] = None):
"""Atualiza status e contador de tentativas"""
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
UPDATE events
SET status = ?,
attempts = attempts + 1,
last_attempt_at = ?,
error_message = ?
WHERE event_id = ?
''', (
status,
datetime.now(timezone.utc).isoformat(),
error_message,
event_id
))
def get_pending(self, limit: int = 10) -> list[WebhookEvent]:
"""Retorna eventos pendentes para reprocessamento"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute('''
SELECT * FROM events
WHERE status IN ('pending', 'failed')
ORDER BY received_at ASC
LIMIT ?
''', (limit,))
return [self._row_to_event(row) for row in cursor.fetchall()]
def _row_to_event(self, row: tuple) -> WebhookEvent:
"""Converte linha do banco para WebhookEvent"""
return WebhookEvent(
event_id=row[0],
source=row[1],
event_type=row[2],
payload=json.loads(row[3]),
received_at=row[4],
signature=row[5],
status=row[6],
attempts=row[7],
last_attempt_at=row[8],
error_message=row[9]
)
class HMACValidator:
"""Valida assinaturas HMAC-SHA256"""
@staticmethod
def verify_stripe(payload: bytes, signature: str, secret: str) -> bool:
"""Valida webhook do Stripe"""
# Stripe usa formato: t=<timestamp>,v1=<signature>
try:
parts = signature.split(',')
timestamp = parts[0].split('=')[1]
expected_sig = parts[1].split('=')[1]
# Stripe assina: timestamp + "." + payload
signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
computed = hmac.new(
secret.encode('utf-8'),
signed_payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
except (IndexError, ValueError):
return False
@staticmethod
def verify_github(payload: bytes, signature: str, secret: str) -> bool:
"""Valida webhook do GitHub"""
# GitHub usa formato: sha256=<signature>
try:
if not signature.startswith('sha256='):
return False
expected_sig = signature[7:] # Remove "sha256="
computed = hmac.new(
secret.encode('utf-8'),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
except Exception:
return False
@staticmethod
def verify_generic(payload: bytes, signature: str, secret: str) -> bool:
"""Validação genérica HMAC-SHA256"""
computed = hmac.new(
secret.encode('utf-8'),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, signature)
class WebhookProcessor:
"""Processa eventos com retry automático"""
def __init__(self, storage: WebhookStorage, max_retries: int = 5):
self.storage = storage
self.max_retries = max_retries
self.handlers: Dict[str, Callable] = {}
self._stop_event = threading.Event()
self._retry_thread = None
def register_handler(self, event_type: str, handler: Callable):
"""Registra handler para tipo específico de evento"""
self.handlers[event_type] = handler
def process(self, event: WebhookEvent) -> bool:
"""Processa evento único. Retorna True se sucesso"""
handler = self.handlers.get(event.event_type)
if not handler:
# Handler não encontrado - marca como completado (não vai resolver)
self.storage.update_status(
event.event_id,
'completed',
'No handler registered'
)
return True
try:
handler(event.payload)
self.storage.update_status(event.event_id, 'completed')
return True
except Exception as e:
error_msg = f"{type(e).__name__}: {str(e)}"
if event.attempts >= self.max_retries:
self.storage.update_status(
event.event_id,
'failed',
f"Max retries exceeded. Last error: {error_msg}"
)
return False
self.storage.update_status(event.event_id, 'pending', error_msg)
return False
def start_retry_loop(self, interval: int = 60):
"""Inicia thread de reprocessamento em background"""
def retry_loop():
while not self._stop_event.is_set():
pending = self.storage.get_pending(limit=10)
for event in pending:
# Backoff exponencial: 1min, 2min, 4min, 8min, 16min
delay = (2 ** event.attempts) * 60
if event.last_attempt_at:
last = datetime.fromisoformat(event.last_attempt_at)
elapsed = (datetime.now(timezone.utc) - last).total_seconds()
if elapsed < delay:
continue # Ainda não é hora
self.process(event)
self._stop_event.wait(interval)
self._retry_thread = threading.Thread(target=retry_loop, daemon=True)
self._retry_thread.start()
def stop(self):
"""Para thread de retry"""
self._stop_event.set()
if self._retry_thread:
self._retry_thread.join(timeout=5)
class WebhookHandler(BaseHTTPRequestHandler):
"""Handler HTTP para receber webhooks"""
processor: WebhookProcessor
storage: WebhookStorage
secrets: Dict[str, str] # {'stripe': 'whsec_...', 'github': '...'}
def do_POST(self):
"""Recebe webhook via POST"""
try:
# Lê payload
content_length = int(self.headers.get('Content-Length', 0))
payload = self.rfile.read(content_length)
# Identifica fonte pelo path
source = self.path.strip('/').lower()
# Valida assinatura
signature = self.headers.get('X-Hub-Signature-256') or \
self.headers.get('Stripe-Signature') or \
self.headers.get('X-Webhook-Signature')
if source in self.secrets and signature:
validator = HMACValidator()
if source == 'stripe':
valid = validator.verify_stripe(
payload, signature, self.secrets[source]
)
elif source == 'github':
valid = validator.verify_github(
payload, signature, self.secrets[source]
)
else:
valid = validator.verify_generic(
payload, signature, self.secrets[source]
)
if not valid:
self.send_response(401)
self.end_headers()
self.wfile.write(b'Invalid signature')
return
# Parse payload
try:
data = json.loads(payload.decode('utf-8'))
except json.JSONDecodeError:
self.send_response(400)
self.end_headers()
self.wfile.write(b'Invalid JSON')
return
# Extrai event_id e event_type
event_id = data.get('id') or data.get('delivery') or \
hashlib.sha256(payload).hexdigest()[:16]
event_type = data.get('type') or data.get('event') or 'unknown'
# Cria evento
event = WebhookEvent(
event_id=event_id,
source=source,
event_type=event_type,
payload=data,
received_at=datetime.now(timezone.utc).isoformat(),
signature=signature
)
# Salva (deduplicação automática)
if not self.storage.save(event):
# Já existe - retorna 200 para não reenviar
self.send_response(200)
self.end_headers()
self.wfile.write(b'Already received')
return
# Processa
success = self.processor.process(event)
# Retorna 200 mesmo se falhar (para não reenviar)
# Retry acontece internamente
self.send_response(200)
self.end_headers()
self.wfile.write(b'OK')
except Exception as e:
self.send_response(500)
self.end_headers()
self.wfile.write(f'Error: {e}'.encode())
def log_message(self, format, *args):
"""Silencia logs padrão do HTTPServer"""
pass
def run_server(port: int = 8000, secrets: Dict[str, str] = None):
"""Inicia servidor de webhooks"""
storage = WebhookStorage()
processor = WebhookProcessor(storage)
# Registra handlers de exemplo
def handle_stripe_payment(payload):
print(f"✓ Pagamento processado: {payload['data']['object']['id']}")
# Sua lógica aqui
def handle_github_push(payload):
print(f"✓ Push recebido: {payload['repository']['full_name']}")
# Sua lógica aqui
processor.register_handler('payment_intent.succeeded', handle_stripe_payment)
processor.register_handler('push', handle_github_push)
# Inicia retry loop
processor.start_retry_loop(interval=60)
# Configura handler HTTP
WebhookHandler.processor = processor
WebhookHandler.storage = storage
WebhookHandler.secrets = secrets or {}
server = HTTPServer(('0.0.0.0', port), WebhookHandler)
print(f"Webhook receiver rodando em http://localhost:{port}")
print(f"Endpoints: /stripe, /github, /generic")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nDesligando...")
processor.stop()
server.shutdown()
if __name__ == '__main__':
secrets = {
'stripe': 'whsec_test123', # Substitua pelo seu secret real
'github': 'gh_webhook_secret123'
}
run_server(port=8000, secrets=secrets)Como Funciona: Os 3 Pilares da Resiliência
1. Validação HMAC (Autenticidade)
HMAC (Hash-based Message Authentication Code) é uma assinatura criptográfica que prova que:
O código suporta três formatos:
A validação usa `hmac.compare_digest()` ao invés de `==` para evitar timing attacks.
2. Deduplicação por event_id (Idempotência)
Todo webhook tem um identificador único:
O banco SQLite tem `event_id` como PRIMARY KEY. Se tentar inserir duas vezes, falha silenciosamente e retorna 200 (para não reenviar).
3. Retry com Backoff Exponencial (Resiliência)
Se o handler levantar exceção:
1. Status muda para `pending`
2. Thread de retry verifica a cada 60 segundos
3. Backoff exponencial: 1min, 2min, 4min, 8min, 16min
4. Após 5 tentativas, marca como `failed`
Isso lida com:
Testando Localmente Sem ngrok
Para testar sem expor sua máquina, use o `stripe-cli`:
# Instala Stripe CLI
curl -s https://packages.stripe.dev/api/security/keypair/stripe-cli-gpg/public | gpg --dearmor | sudo tee /usr/share/keyrings/stripe.gpg
echo "deb [signed-by=/usr/share/keyrings/stripe.gpg] https://packages.stripe.dev/stripe-cli-debian-local stable main" | sudo tee /etc/apt/sources.list.d/stripe.list
sudo apt update && sudo apt install stripe
# Forward webhooks para localhost:8000
stripe listen --forward-to localhost:8000/stripeOu teste manualmente com curl:
# Teste sem assinatura (aceito se não houver secret configurado)
curl -X POST http://localhost:8000/stripe \
-H "Content-Type: application/json" \
-d '{
"id": "evt_test123",
"type": "payment_intent.succeeded",
"data": {"object": {"id": "pi_123"}}
}'
# Teste com assinatura GitHub
SECRET="test_secret"
PAYLOAD='{"delivery":"abc123","event":"push","repository":{"full_name":"user/repo"}}'
SIGNATURE="sha256=$(echo -n "$PAYLOAD" | openssl dgst -sha256 -hmac "$SECRET" | cut -d' ' -f2)"
curl -X POST http://localhost:8000/github \
-H "Content-Type: application/json" \
-H "X-Hub-Signature-256: $SIGNATURE" \
-d "$PAYLOAD"Próximos Passos: O Que Fazer Quando Escalar
Este código funciona para milhares de eventos por hora. Quando passar disso:
1. **Mova para fila**: Substitua SQLite por Redis/RabbitMQ
2. **Separe ingestão e processamento**: Endpoint só salva, workers processam
3. **Adicione métricas**: Prometheus + Grafana para monitorar latência e falhas
4. **Use WAF**: Cloudflare ou similar para bloquear IPs maliciosos antes
Mas para 99% dos casos, esse código é mais que suficiente.
CTA: Qual Foi o Maior Perrengue Que Você Já Teve com Webhooks?
Webhooks parecem simples até falharem de formas criativas: eventos duplicados, assinaturas inválidas, timeouts misteriosos. Qual foi o pior bug que você já enfrentou?
Deixa nos comentários — sua história pode salvar alguém de perder R$ 2.400 como eu perdi.
**Código completo**: https://gist.github.com/… (link será adicionado após publicação)
**Próximo post**: Vou implementar um webhook sender com retry e assinatura HMAC do lado do emissor. Se você constrói APIs que notificam outros serviços, fica ligado.
