Bir teslimatı doğrulamak
`verify_webhook_signature`: sabit zamanlı, bir yeniden oynatma penceresiyle ve ayrıştırılmış olayı geri vererek.
Bir istek işleyicisinde
Bir webhook URL'si herkese açıktır. İnternetteki herhangi bir şey ona doğru biçimde JSON POST edebilir; bu yüzden imzayı denetlemeden payload['type'] okuyan bir işleyici, açık bir yazma API'sidir.
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)HAM gövdeyi geçirin. Ayrıştırıp yeniden serileştirmek anahtar sırasını ve boşlukları değiştirir ve imza tutmaz. bytes ya da str olabilir: FastAPI'de await request.body(), Django'da request.body, Flask'ta request.get_data(). headers, bu framework'lerin her birinin request.headers nesnesi ya da düz bir dict gibi herhangi bir eşleme alır ve X-OpenEmail-Signature, harf durumu ne olursa olsun bulunur.
Elle yazılmış bir denetimin genellikle ele almadığı iki şey: MAC'i sabit zamanda karşılaştırır, böylece doğru ön ek zamanlama ölçülerek geri elde edilemez; ve her iki yönde de tolerance_seconds'tan, yani siz aksini söylemedikçe beş dakikadan, daha eski bir teslimatı reddeder, böylece yakalanmış bir istek sonsuza dek yeniden oynatılamaz. tolerance_seconds=0 yeniden oynatma denetimini kapatır. Bu hataların ikisi de sessizdir. Hangisini barındırırsa barındırsın bir işleyici, yazmayı akıl edeceğiniz her testi geçer.
Her başarısızlıkta WebhookVerificationError fırlatır: boş bir secret, eksik bir X-OpenEmail-Signature başlığı, t=<seconds>,v1=<hex> biçiminde olmayan bir başlık, pencerenin dışındaki bir zaman damgası, eşleşmeyen bir imza ya da JSON olmayan bir gövde. Başarı durumunda gövdeyi WebhookPayload olarak ayrıştırılmış hâliyle döndürür; böylece yanlış yapılacak ikinci bir json.loads kalmaz. data alanını türlemek için sonucu WebhookPayload[EmailOpenedData] gibi bir türle işaretleyin.
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, hem OpenEmailError hem de ValueError sınıfının alt sınıfıdır. Fonksiyon bir coroutine değil, düz bir fonksiyondur; bu yüzden onu senkron ve asenkron koddan aynı şekilde await olmadan çağırın. Standart kütüphanedeki hmac ve hashlib dışında hiçbir şeye ihtiyaç duymaz.