التحقق من تسليم
`verify_webhook_signature`: مقارنة بزمن ثابت، مع نافذة إعادة تشغيل، وإعادة الحدث محلَّلًا.
داخل معالج طلب
عنوان URL الخاص بـwebhook عام. فأي شيء على الإنترنت يستطيع أن يرسل إليه بـPOST بنية JSON بالشكل الصحيح، ولذلك فإن معالجًا يقرأ payload['type'] دون التحقق من التوقيع هو 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)مرّر المتن الخام (RAW). فالتحليل وإعادة التسلسل يغيّران ترتيب المفاتيح والمسافات، ولن يتطابق التوقيع. ويجوز أن يكون bytes أو str: await request.body() في FastAPI، وrequest.body في Django، وrequest.get_data() في Flask. ويقبل headers أي كائن من نوع mapping، مثل request.headers في كل إطار من أطر العمل تلك أو dict عاديًا، ويُعثر على X-OpenEmail-Signature مهما كانت حالة أحرفه.
أمران يعالجهما هذا ولا يعالجهما عادةً فحص مكتوب يدويًا: فهو يقارن الـMAC بزمن ثابت، فلا يمكن استخراج البادئة الصحيحة بقياس الزمن، وهو يرفض تسليمًا أقدم من tolerance_seconds في أي من الاتجاهين، أي خمس دقائق ما لم تقل غير ذلك، فلا يبقى طلب ملتقَط قابلًا لإعادة التشغيل إلى الأبد. والقيمة tolerance_seconds=0 تعطّل فحص إعادة التشغيل. وكلا الخللين صامت. والمعالج المصاب بأي منهما يجتاز كل اختبار قد يخطر لك أن تكتبه.
يرفع WebhookVerificationError عند كل إخفاق: secret فارغ، أو ترويسة X-OpenEmail-Signature مفقودة، أو ترويسة ليست على الصيغة t=<seconds>,v1=<hex>، أو طابع وقت خارج النافذة، أو توقيع لا يتطابق، أو متن ليس JSON. وعند النجاح يعيد المتن محلَّلًا كـ WebhookPayload، فلا يوجد json.loads ثانٍ يمكن أن تخطئ فيه. وأضِف تلميح نوع إلى النتيجة، مثل WebhookPayload[EmailOpenedData]، لتحديد نوع 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 من OpenEmailError وValueError كليهما. والدالة عادية لا coroutine، فاستدعِها دون await من الشيفرة المتزامنة وغير المتزامنة على حدّ سواء. ولا تحتاج إلى شيء سوى hmac وhashlib من المكتبة القياسية.