Vérifier une livraison
`verify_webhook_signature` : à temps constant, avec une fenêtre de rejeu, et l'événement analysé en retour.
Dans un gestionnaire de requêtes
Une URL de webhook est publique. N'importe quoi sur Internet peut y envoyer en POST du JSON de la bonne forme : un gestionnaire qui lit payload['type'] sans vérifier la signature est donc une API d'écriture ouverte.
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)Transmettez le corps BRUT. L'analyser puis le re-sérialiser modifie l'ordre des clés et les espaces, et la signature ne correspondra plus. Il peut être bytes ou str : await request.body() dans FastAPI, request.body dans Django, request.get_data() dans Flask. headers accepte n'importe quel mapping, comme le request.headers de chacun de ces frameworks ou un simple dict, et X-OpenEmail-Signature est trouvé quelle que soit sa casse.
Deux points dont cette vérification s'occupe et qu'un contrôle écrit à la main néglige d'ordinaire : elle compare le MAC en temps constant, si bien qu'on ne peut pas en retrouver le préfixe correct en mesurant le temps de réponse, et elle rejette une livraison décalée de plus de tolerance_seconds dans un sens comme dans l'autre (cinq minutes sauf indication contraire), de sorte qu'une requête capturée ne reste pas rejouable indéfiniment. tolerance_seconds=0 désactive le contrôle anti-rejeu. Ces deux bogues sont silencieux. Un gestionnaire qui présente l'un ou l'autre passe tous les tests que vous auriez l'idée d'écrire.
Elle lève WebhookVerificationError à chaque échec : secret vide, en-tête X-OpenEmail-Signature absent, en-tête qui ne respecte pas la forme t=<seconds>,v1=<hex>, horodatage hors de la fenêtre, signature qui ne correspond pas, ou corps qui n'est pas du JSON. En cas de succès, elle renvoie le corps analysé sous forme de WebhookPayload : il n'y a donc pas de second json.loads à rater. Annotez le résultat, par exemple WebhookPayload[EmailOpenedData], pour typer 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 hérite à la fois d'OpenEmailError et de ValueError. La fonction est une fonction ordinaire et non une coroutine : appelez-la donc sans await, depuis du code synchrone comme asynchrone. Elle n'a besoin de rien d'autre que hmac et hashlib de la bibliothèque standard.