Template: Webhook handler universal en Python

Webhook Handler Universal en Python

Un handler reutilizable que soporta webhooks de cualquier servicio (Stripe, GitHub, WhatsApp, etc.) con validación de firmas HMAC. Implementaciones para Django y FastAPI.

Django: Handler con validación de firma

import hmac
import hashlib
import json
import logging
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from functools import wraps

logger = logging.getLogger(__name__)

WEBHOOK_CONFIGS = {
    'stripe': {
        'secret': 'whsec_...',
        'header': 'HTTP_STRIPE_SIGNATURE',
        'algo': 'sha256',
    },
    'github': {
        'secret': 'gh_webhook_secret',
        'header': 'HTTP_X_HUB_SIGNATURE_256',
        'algo': 'sha256',
    },
}

def verify_signature(payload: bytes, signature: str, secret: str, algo: str) -> bool:
    expected = hmac.new(
        secret.encode(), payload, getattr(hashlib, algo)
    ).hexdigest()

    # Stripe incluye prefijo "sha256=", GitHub también
    if '=' in signature:
        signature = signature.split('=', 1)[-1]

    # Stripe usa formato especial: t=...,v1=... — extraer v1
    if ',v1=' in signature:
        parts = dict(p.split('=', 1) for p in signature.split(','))
        signature = parts.get('v1', signature)

    return hmac.compare_digest(expected, signature)

def webhook_handler(service: str):
    config = WEBHOOK_CONFIGS[service]

    def decorator(func):
        @wraps(func)
        @csrf_exempt
        @require_POST
        def wrapper(request, *args, **kwargs):
            signature = request.META.get(config['header'], '')
            if not verify_signature(
                request.body, signature, config['secret'], config['algo']
            ):
                logger.warning(f"Webhook {service}: firma inválida")
                return HttpResponse(status=403)

            try:
                payload = json.loads(request.body)
            except json.JSONDecodeError:
                return HttpResponse(status=400)

            return func(request, payload, *args, **kwargs)
        return wrapper
    return decorator

Uso en Django

@webhook_handler('stripe')
def stripe_webhook(request, payload):
    event_type = payload.get('type')

    if event_type == 'checkout.session.completed':
        handle_checkout(payload['data']['object'])
    elif event_type == 'customer.subscription.deleted':
        handle_cancellation(payload['data']['object'])

    return JsonResponse({'status': 'ok'})

@webhook_handler('github')
def github_webhook(request, payload):
    action = payload.get('action')
    if action == 'push':
        trigger_deploy(payload['repository']['full_name'])
    return JsonResponse({'status': 'ok'})

FastAPI: Handler equivalente

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

async def verify_webhook(request: Request, service: str) -> dict:
    config = WEBHOOK_CONFIGS[service]
    body = await request.body()
    signature = request.headers.get(
        config['header'].replace('HTTP_', '').replace('_', '-').title()
    , '')

    if not verify_signature(body, signature, config['secret'], config['algo']):
        raise HTTPException(403, "Firma inválida")

    return json.loads(body)

@app.post("/webhooks/stripe")
async def stripe_wh(request: Request):
    payload = await verify_webhook(request, 'stripe')
    # Procesar evento...
    return {"status": "ok"}

Testing de webhooks

import hmac, hashlib, requests, json

def send_test_webhook(url: str, payload: dict, secret: str):
    body = json.dumps(payload).encode()
    sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    resp = requests.post(url, data=body, headers={
        'Content-Type': 'application/json',
        'X-Hub-Signature-256': f'sha256={sig}',
    })
    print(f"Status: {resp.status_code}, Body: {resp.text}")

send_test_webhook(
    'http://localhost:8000/webhooks/github',
    {'action': 'push', 'repository': {'full_name': 'user/repo'}},
    'gh_webhook_secret',
)