Проверка доставки
`verify_webhook_signature`: сравнение за постоянное время, окно защиты от повторов и разобранное событие на выходе.
В обработчике запроса
URL вебхука публичен. Что угодно в интернете может отправить на него 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)Передавайте СЫРОЕ тело. Разбор и повторная сериализация меняют порядок ключей и пробелы, и подпись не сойдётся. Оно может быть 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. Функция обычная, а не корутина, поэтому вызывайте её без await как из синхронного, так и из асинхронного кода. Ей не нужно ничего, кроме hmac и hashlib из стандартной библиотеки.