Verifikimi i një dërgese
`verify_webhook_signature`: në kohë konstante, me një dritare replay, dhe me ngjarjen e parsuar si kthim.
Në një handler kërkese
Një URL webhook-u është publike. Çdo gjë në internet mund të bëjë POST me JSON të formës së duhur drejt saj, ndaj një handler që lexon payload['type'] pa kontrolluar nënshkrimin është një API shkrimi e hapur.
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)Jepni trupin E PAPËRPUNUAR. Parsimi dhe riserializimi ndryshojnë renditjen e çelësave dhe hapësirat, dhe nënshkrimi nuk do të përputhet. Mund të jetë bytes ose str: await request.body() në FastAPI, request.body në Django, request.get_data() në Flask. headers merr çdo mapping, si request.headers i secilit prej këtyre framework-eve ose një dict i thjeshtë, dhe X-OpenEmail-Signature gjendet pavarësisht shkronjave të mëdha a të vogla.
Dy gjëra që kjo i trajton e që një kontroll i bërë me dorë zakonisht nuk i trajton: e krahason MAC-un në kohë konstante, që parashtesa e saktë të mos nxirret duke i matur kohën, dhe e refuzon një dërgesë më të vjetër se tolerance_seconds në të dy drejtimet (pesë minuta, veç nëse thoni ndryshe), që një kërkesë e kapur të mos jetë e riluajtshme përgjithmonë. tolerance_seconds=0 e çaktivizon kontrollin e replay-t. Të dy këta defekte janë të heshtur. Një handler me cilindo prej tyre i kalon të gjitha testet që do t'ju shkonte mendja të shkruanit.
Funksioni ngre WebhookVerificationError në çdo dështim: një secret bosh, një header X-OpenEmail-Signature që mungon, një që nuk është në formën t=<seconds>,v1=<hex>, një vulë kohore jashtë dritares, një nënshkrim që nuk përputhet, ose një trup që nuk është JSON. Në sukses kthen trupin të analizuar si një WebhookPayload, kështu që nuk ka një json.loads të dytë për ta gabuar. Anotojeni rezultatin, si WebhookPayload[EmailOpenedData], për të tipizuar 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 është nënklasë si e OpenEmailError, ashtu edhe e ValueError. Funksioni është i zakonshëm dhe jo korutinë, ndaj thirreni pa await si nga kodi sinkron, ashtu edhe nga ai asinkron. Nuk ka nevojë për asgjë përtej hmac dhe hashlib nga biblioteka standarde.