किसी delivery का सत्यापन
`verify_webhook_signature`: स्थिर-समय में, replay विंडो के साथ, और पार्स किया गया event वापस।
किसी request handler में
webhook URL सार्वजनिक होता है। इंटरनेट पर कोई भी चीज़ उस पर सही आकार वाला JSON POST कर सकती है, इसलिए वह handler जो signature जाँचे बिना payload['type'] पढ़ता है, एक खुला write 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 body पास करें। पार्स करके दोबारा serialise करने से keys का क्रम और whitespace बदल जाते हैं, और हस्ताक्षर मेल नहीं खाएगा। यह bytes या str हो सकती है: FastAPI में await request.body(), Django में request.body, Flask में request.get_data()। headers कोई भी mapping लेता है, जैसे इन फ़्रेमवर्क में से हर एक का request.headers या एक सादा dict, और X-OpenEmail-Signature चाहे जिस case में हो, मिल जाता है।
दो चीज़ें जो यह संभालता है और हाथ से लिखी जाँच आमतौर पर नहीं: यह MAC की तुलना स्थिर समय में करता है, इसलिए समय नापकर सही उपसर्ग निकाला नहीं जा सकता, और यह किसी भी दिशा में tolerance_seconds से पुरानी delivery अस्वीकार कर देता है (जब तक आप कुछ और न कहें, पाँच मिनट), इसलिए पकड़ा गया request हमेशा के लिए replay करने लायक नहीं रहता। tolerance_seconds=0 replay जाँच बंद कर देता है। दोनों बग चुप हैं। जिस handler में इनमें से कोई एक भी हो, वह हर उस टेस्ट में पास हो जाता है जो आपके मन में आएगा।
यह हर विफलता पर WebhookVerificationError raise करता है: ख़ाली secret, अनुपस्थित X-OpenEmail-Signature header, ऐसा header जो t=<seconds>,v1=<hex> रूप में नहीं है, विंडो से बाहर का timestamp, ऐसा हस्ताक्षर जो मेल नहीं खाता, या ऐसी body जो JSON नहीं है। सफलता पर यह body को WebhookPayload के रूप में पार्स करके लौटाता है, इसलिए ग़लती करने को कोई दूसरा json.loads बचता ही नहीं। data को type देने के लिए परिणाम को annotate करें, जैसे WebhookPayload[EmailOpenedData]।
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 दोनों की subclass है। यह फ़ंक्शन coroutine नहीं बल्कि सादा फ़ंक्शन है, इसलिए इसे सिंक और एसिंक दोनों तरह के कोड से बिना await के कॉल करें। इसे standard library के hmac और hashlib के अलावा कुछ नहीं चाहिए।