Verificar uma entrega
`verify_webhook_signature`: tempo constante, com uma janela de replay, e o evento já analisado de volta.
Num handler de pedidos
Um URL de webhook é público. Qualquer coisa na internet lhe pode fazer POST de JSON com a forma certa, por isso um handler que lê payload['type'] sem verificar a assinatura é uma API de escrita aberta.
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)Passe o corpo EM BRUTO. Analisá-lo e voltar a serializá-lo altera a ordem das chaves e os espaços, e a assinatura não vai corresponder. Pode ser bytes ou str: await request.body() no FastAPI, request.body no Django, request.get_data() no Flask. headers aceita qualquer mapeamento, como o request.headers de cada um desses frameworks ou um dict simples, e X-OpenEmail-Signature é encontrado seja qual for a caixa das letras.
Duas coisas que isto trata e que uma verificação feita à mão costuma não tratar: compara o MAC em tempo constante, para que o prefixo correto não possa ser recuperado cronometrando-o, e rejeita uma entrega com mais de tolerance_seconds em qualquer das direções, cinco minutos salvo indicação em contrário, para que um pedido capturado não seja reproduzível para sempre. tolerance_seconds=0 desliga a verificação de replay. Ambos os bugs são silenciosos. Um handler com qualquer um deles passa todos os testes que lhe ocorreria escrever.
Lança WebhookVerificationError em todas as falhas: um secret vazio, um cabeçalho X-OpenEmail-Signature em falta, um que não esteja na forma t=<seconds>,v1=<hex>, um timestamp fora da janela, uma assinatura que não corresponde ou um corpo que não é JSON. Em caso de sucesso devolve o corpo analisado como um WebhookPayload, por isso não há um segundo json.loads para errar. Anote o resultado, por exemplo 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 é subclasse tanto de OpenEmailError como de ValueError. A função é uma função normal e não uma corrotina, por isso chame-a sem await, tanto a partir de código síncrono como assíncrono. Não precisa de nada além de hmac e hashlib da biblioteca padrão.