Eine Zustellung verifizieren
`verify_webhook_signature`: in konstanter Zeit, mit einem Replay-Fenster und dem geparsten Event zurück.
In einem Request-Handler
Eine Webhook-URL ist öffentlich. Alles im Internet kann JSON in der richtigen Form dorthin POSTen; ein Handler, der payload['type'] liest, ohne die Signatur zu prüfen, ist eine offene Schreib-API.
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)Übergeben Sie den ROHEN Body. Parsen und erneutes Serialisieren ändert Schlüsselreihenfolge und Whitespace, und die Signatur passt dann nicht mehr. Er darf bytes oder str sein: await request.body() in FastAPI, request.body in Django, request.get_data() in Flask. headers nimmt jedes Mapping entgegen, etwa die request.headers jedes dieser Frameworks oder ein einfaches dict, und X-OpenEmail-Signature wird unabhängig von der Schreibweise gefunden.
Zwei Dinge, die das hier erledigt und die eine selbstgebaute Prüfung meist nicht leistet: Es vergleicht den MAC in konstanter Zeit, sodass sich das korrekte Präfix nicht über die Laufzeit rekonstruieren lässt, und es weist eine Zustellung zurück, die in eine der beiden Richtungen mehr als tolerance_seconds alt ist – fünf Minuten, sofern nicht anders angegeben –, sodass eine mitgeschnittene Anfrage nicht für immer wiederholbar ist. tolerance_seconds=0 schaltet die Replay-Prüfung ab. Beide Fehler sind stumm. Ein Handler mit einem von beiden besteht jeden Test, auf den man zu schreiben käme.
Die Funktion löst bei jedem Fehlschlag WebhookVerificationError aus: ein leeres secret, ein fehlender Header X-OpenEmail-Signature, einer, der nicht die Form t=<seconds>,v1=<hex> hat, ein Zeitstempel außerhalb des Fensters, eine Signatur, die nicht passt, oder ein Body, der kein JSON ist. Bei Erfolg gibt sie den Body geparst als WebhookPayload zurück, es gibt also kein zweites json.loads, bei dem etwas schiefgehen könnte. Annotieren Sie das Ergebnis, etwa als WebhookPayload[EmailOpenedData], um data zu typisieren.
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 ist eine Unterklasse sowohl von OpenEmailError als auch von ValueError. Die Funktion ist eine gewöhnliche Funktion und keine Koroutine, rufen Sie sie also aus synchronem wie asynchronem Code ohne await auf. Sie braucht nichts außer hmac und hashlib aus der Standardbibliothek.