Verificar una entrega
`verify_webhook_signature`: en tiempo constante, con una ventana de repetición, y devuelve el evento ya analizado.
En un handler de solicitudes
La URL de un webhook es pública. Cualquier cosa en internet puede hacerle POST de un JSON con la forma correcta, así que un handler que lee payload['type'] sin comprobar la firma es una API de escritura abierta.
import os from fastapi import FastAPI, Request, Responsefrom openemail import WebhookVerificationError, verify_webhook_signature app = FastAPI() @app.post('/webhooks/openemail')async def webhook(request: Request) -> Response: try: event = verify_webhook_signature( payload=await request.body(), headers=request.headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], tolerance_seconds=300, ) except WebhookVerificationError: return Response('bad signature', status_code=400) print(event['type'], event['data']) return Response(status_code=204)Pasa el cuerpo EN CRUDO. Analizarlo y volver a serializarlo cambia el orden de las claves y los espacios, y la firma no coincidirá. Puede ser bytes o str: await request.body() en FastAPI, request.body en Django, request.get_data() en Flask. headers acepta cualquier mapeo, como el request.headers de cada uno de esos frameworks o un dict simple, y X-OpenEmail-Signature se encuentra sea cual sea su capitalización.
Dos cosas que esto resuelve y que una comprobación casera no suele resolver: compara el MAC en tiempo constante, de modo que el prefijo correcto no puede recuperarse midiendo tiempos, y rechaza una entrega con más de tolerance_seconds de antigüedad en cualquiera de las dos direcciones, cinco minutos salvo que digas otra cosa, así que una solicitud capturada no se puede repetir para siempre. tolerance_seconds=0 desactiva la comprobación de repetición. Ambos fallos son silenciosos. Un handler con cualquiera de los dos pasa todas las pruebas que se te ocurriría escribir.
Lanza WebhookVerificationError ante cualquier fallo: un secret vacío, una cabecera X-OpenEmail-Signature ausente, una que no tenga la forma t=<seconds>,v1=<hex>, una marca de tiempo fuera de la ventana, una firma que no coincide o un cuerpo que no es JSON. Si tiene éxito, devuelve el cuerpo analizado como un WebhookPayload, así que no hay un segundo json.loads en el que equivocarse. Anota el resultado, por ejemplo como WebhookPayload[EmailOpenedData], para tipar data.
import os from openemail import verify_webhook_signaturefrom openemail.types import EmailOpenedData, WebhookPayload def on_open(body: bytes, headers: dict[str, str]) -> None: event: WebhookPayload[EmailOpenedData] = verify_webhook_signature( payload=body, headers=headers, secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'], ) print(event['data']['recipient'], event['data']['first'], event['data']['country'])WebhookVerificationError es subclase tanto de OpenEmailError como de ValueError. La función es una función normal y no una corrutina, así que llámala sin await tanto desde código síncrono como asíncrono. No necesita nada más que hmac y hashlib de la biblioteca estándar.