Получайте события EVIR на свой сервер

import asyncio
import base64
import binascii
import hashlib
import hmac
import json
import os
import re
import sqlite3
import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response

# 1. Настройки
# Берём секрет из переменной сервера — не вставляйте whsec_ прямо в код.
app = FastAPI()
bootstrap = os.getenv("EVIR_WEBHOOK_BOOTSTRAP") == "true"
database_path = os.getenv("EVIR_WEBHOOK_DB", "evir-webhooks.sqlite")
secrets = [
    value.strip()
    for value in os.getenv("EVIR_WEBHOOK_SECRETS", "").split(",")
    if value.strip().startswith("whsec_")
]

if not bootstrap and not secrets:
    raise RuntimeError("EVIR webhook signing secret is missing")


# 2. Надёжное хранение
# SQLite запоминает event.id до ответа EVIR, поэтому повторная доставка не запустит действие дважды.
def initialize_database():
    with sqlite3.connect(database_path) as database:
        database.execute("PRAGMA journal_mode = WAL")
        database.execute("PRAGMA synchronous = FULL")
        database.execute("PRAGMA busy_timeout = 5000")
        database.execute("""CREATE TABLE IF NOT EXISTS evir_webhook_events (
            id TEXT PRIMARY KEY,
            project_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            body_json TEXT NOT NULL,
            received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
            status TEXT NOT NULL DEFAULT 'queued',
            processed_at TEXT
        )""")


def persist_once(event_id: str, project_id: str, event_type: str, raw_body: bytes):
    with sqlite3.connect(database_path) as database:
        database.execute(
            """INSERT OR IGNORE INTO evir_webhook_events
               (id, project_id, event_type, body_json) VALUES (?, ?, ?, ?)""",
            (event_id, project_id, event_type, raw_body.decode("utf-8")),
        )


initialize_database()

accepted_event_types = {
    "op.completed",
    "op.unsubscribed",
    "transition.opened",
    "impression.completed",
    "impression.served",
    "webhook.test",
}


def parse_object(raw_body: bytes):
    try:
        value = json.loads(raw_body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return None
    return value if isinstance(value, dict) else None


def event_project_id(value):
    project = value.get("project") if isinstance(value, dict) else None
    project_id = project.get("id") if isinstance(project, dict) else None
    if not isinstance(project_id, str) or re.fullmatch(r"[A-Za-z0-9_-]{1,96}", project_id) is None:
        return None
    return project_id


def verification_challenge(value):
    if not isinstance(value, dict) or set(value) != {
        "type", "challenge", "project", "createdAt"
    }:
        return None
    project = value.get("project")
    challenge = value.get("challenge")
    if value.get("type") != "webhook.endpoint.verification":
        return None
    if not isinstance(challenge, str) or re.fullmatch(r"[A-Za-z0-9_-]{32}", challenge) is None:
        return None
    if not isinstance(value.get("createdAt"), str) or len(value["createdAt"]) > 64:
        return None
    if not isinstance(project, dict) or set(project) != {"id"}:
        return None
    if event_project_id(value) is None:
        return None
    return challenge


def decode_base64url(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


@app.post("/webhooks/evir")
async def evir_webhook(request: Request):
    # 3. Читаем исходное тело
    # Подпись считается по этим bytes. Нельзя сначала разобрать JSON и собрать его заново.
    raw_body_buffer = bytearray()
    async for chunk in request.stream():
        if len(raw_body_buffer) + len(chunk) > 64 * 1024:
            return Response(status_code=413)
        raw_body_buffer.extend(chunk)
    raw_body = bytes(raw_body_buffer)

    # 4. Проверяем адрес
    # Временно включите bootstrap перед подключением или изменением вебхука и сразу выключите после.
    if bootstrap and len(raw_body) <= 2_048:
        challenge = verification_challenge(parse_object(raw_body))
        if challenge is not None:
            return JSONResponse({"challenge": challenge})

    if not secrets:
        return Response(status_code=503)

    event_id = request.headers.get("X-EVIR-Webhook-Id", "")
    timestamp = request.headers.get("X-EVIR-Webhook-Timestamp", "")
    signature = request.headers.get("X-EVIR-Webhook-Signature", "")
    match = re.fullmatch(r"v1=([A-Za-z0-9_-]+)", signature)
    if not event_id or not timestamp.isdigit() or match is None:
        return Response(status_code=401)

    # 5. Проверяем подпись
    # При неверной подписи или времени запрос отклоняется и событие не сохраняется.
    try:
        timestamp_seconds = int(timestamp)
        actual = decode_base64url(match.group(1))
    except (binascii.Error, ValueError, TypeError):
        return Response(status_code=401)
    if abs(int(time.time()) - timestamp_seconds) > 300:
        return Response(status_code=401)

    signed = f"{event_id}.{timestamp}.".encode() + raw_body
    valid = False
    for secret in secrets:
        expected = hmac.new(secret.encode(), signed, hashlib.sha256).digest()
        current = hmac.compare_digest(actual, expected)
        valid = current or valid
    if not valid:
        return Response(status_code=401)

    event = parse_object(raw_body)
    if event is None:
        return Response(status_code=400)
    if event.get("type") == "webhook.endpoint.verification":
        challenge = verification_challenge(event)
        return Response(status_code=400) if challenge is None else JSONResponse({"challenge": challenge})
    project_id = event_project_id(event)
    if (event.get("id") != event_id or event.get("apiVersion") != "v1"
            or project_id is None
            or event.get("type") not in accepted_event_types):
        return Response(status_code=400)

    # 6. Сохраняем один раз
    # Сначала надёжно записываем событие, только потом быстро отвечаем HTTP 204.
    try:
        await asyncio.to_thread(persist_once, event_id, project_id, event["type"], raw_body)
    except (sqlite3.Error, UnicodeDecodeError):
        return Response(status_code=503)

    # 7. Этот обработчик только принимает и сохраняет событие.
    # Фоновая задача обрабатывает строки со status = 'queued'.
    return Response(status_code=204)