Webhook «генерация готова» ещё не повод доверять JSON: HMAC на Python и ловушка повторной доставки

Асинхронная AI-задача закончилась, callback принёс статус done — и бэкенд готов продолжать workflow. Но валидный JSON не доказывает происхождение события, а верная подпись не делает его новым. Покажу проверку исходных байтов на Python и тест, в котором повторный webhook честно проходит HMAC ещё раз.

Две проверки, которые нельзя заменять друг другом

Представим учебный обработчик результатов генерации: он принимает job_id и status, сохраняет артефакт и запускает следующий шаг. У такого входа как минимум две границы доверия. Первая: сообщение действительно соответствует данным, подписанным нашим контрагентом. Вторая: это событие относится к нашей задаче и его ещё не применяли.

HMAC помогает с первой границей при условии, что секрет известен только доверенным сторонам. Он не подтверждает качество результата модели, не даёт права менять произвольный tenant и сам по себе не препятствует повтору старого подписанного сообщения.

Сначала байты, затем JSON

Подпись считается по конкретному сообщению, а не по смыслу объекта. Пробелы, порядок ключей и экранирование Unicode могут измениться при json.loads → json.dumps. Получится эквивалентный JSON, но другая последовательность байтов. Проверка такого пересобранного тела начнёт отклонять настоящие доставки.

Правильная граница: получить исходное тело запроса в bytes → проверить подпись по контракту отправителя → разобрать JSON → проверить схему и принадлежность задачи. Не записывать бизнес-изменения и не передавать содержимое агенту до завершения этих проверок.

Один конкретный адаптер, а не «универсальная HMAC-проверка»

Для воспроизводимости ниже использую реальную схему GitHub: X-Hub-Signature-256 содержит sha256= и hex HMAC-SHA256 от тела. Это не утверждение, что DataEyesAI или любой AI-провайдер присылает именно такой заголовок. Для их callback нужен отдельный адаптер по официальному контракту: состав подписанного сообщения, кодирование секрета, timestamp, версии подписи и правила ротации могут отличаться. Если есть официальный verifier, я бы начинал с него.

Функция намеренно проверяет только подпись тела. Ожидает один уже однозначно извлечённый заголовок; HTTP-слой должен отвергать дублирующиеся или неоднозначные заголовки. Пустой секрет — ошибка конфигурации, а не способ временно отключить защиту. Секрет загружаем из защищённой конфигурации, не из payload.

import hashlib import hmac import re class BadSignature(ValueError): pass def verify_github_body(raw: bytes, signature: str | None, secret: bytes) -> None: # GitHub body-only HMAC; not a universal provider adapter. if not isinstance(secret, bytes) or not secret: raise ValueError('webhook secret is not configured') if not isinstance(raw, bytes): raise TypeError('raw request body must be bytes') if not isinstance(signature, str) or not re.fullmatch( r'sha256=[0-9a-f]{64}', signature ): raise BadSignature('invalid signature header') expected = 'sha256=' + hmac.new(secret, raw, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise BadSignature('signature mismatch')

Три теста, которые показывают границу защиты

Ниже — тесты на стандартной библиотеке Python 3.10+. Их можно добавить под функцию и запустить как один файл. Первый использует опубликованный GitHub тестовый вектор, а не подпись, вычисленную той же функцией. Второй оставляет смысл JSON прежним, но меняет байты. Третий специально показывает, чего HMAC не делает: один и тот же подписанный запрос принимается дважды. Это ожидаемое поведение, не ошибка криптографии.

import json import unittest class SignatureTests(unittest.TestCase): def sign(self, body): return 'sha256=' + hmac.new( b'test-only-secret', body, hashlib.sha256 ).hexdigest() def test_official_vector(self): verify_github_body( b'Hello, World!', 'sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17', b"It's a Secret to Everybody", ) def test_same_json_different_bytes(self): raw = b'{"job_id": "j1", "status": "done"}' rewritten = json.dumps( json.loads(raw), separators=(',', ':') ).encode() self.assertEqual(json.loads(raw), json.loads(rewritten)) self.assertNotEqual(raw, rewritten) with self.assertRaises(BadSignature): verify_github_body( rewritten, self.sign(raw), b'test-only-secret' ) def test_replay_still_passes(self): raw = b'{"event_id":"e1","status":"done"}' for _ in range(2): verify_github_body(raw, self.sign(raw), b'test-only-secret') if __name__ == '__main__': unittest.main(verbosity=2)

В локальном запуске прошли семь тестов: три показанных выше, а также проверки изменённого тела и неверного секрета, некорректных заголовков, Unicode в теле и ошибок конфигурации/типа. Платные API не вызывались. Эти тесты проверяют verifier, но не заменяют интеграционный тест с реальным proxy, HTTP-фреймворком и callback провайдера.

Почему нельзя спасти replay случайным заголовком

В показанной схеме подписано тело — не все заголовки HTTP. Если строить анти-replay только по неподписанному delivery_id, изменение этого заголовка у перехваченного сообщения может обойти такую дедупликацию. Нужен стабильный идентификатор из аутентифицированных данных либо другой явно обоснованный ключ и проверка перехода состояния своей задачи.

У некоторых протоколов в подпись входят timestamp и ID. Тогда verifier обязан проверить именно подписанный timestamp и допустимое окно по правилам поставщика. Просто прочитать неподписанное время или добавить собственный заголовок недостаточно. В body-only схеме GitHub такого окна в этой функции нет; нельзя приписывать ей защиту от старой доставки.

После HMAC — сверка задачи и надёжный приём

Для callback генерации я бы разделил маршрут на короткий приём и фоновую обработку. После проверки подписи разбираем JSON, валидируем типы и разрешённые события. Затем находим job по связке провайдер + аккаунт интеграции + внешний job_id, сохранённой при отправке задачи. Tenant берём из своей записи, а не переносим вслепую из входного JSON. Неизвестные job не должны создавать оплачиваемые действия.

Проверяем допустимый переход состояния и поколение попытки. Поздний ответ от предыдущей генерации не должен перезаписать результат новой только потому, что у обоих указан status=done. Один job_id также не всегда годится как единственный ключ события: у задачи могут быть разные законные обновления.

Уже проверенное событие сохраняем в durable inbox с уникальностью по доверенному scope и стабильному event_id. Если контракт не даёт подписанного ID, отдельно выбираем и документируем ключ: хеш точного тела ловит идентичные повторы, но не объединит семантически одинаковые события с разными timestamp или представлением JSON. Для них остаётся идемпотентная бизнес-обработка.

Не делать «SELECT — если нет, то INSERT»: конкурентные доставки пройдут обе проверки. Уникальность должна обеспечиваться хранилищем атомарно. Сохранение inbox и постановку работы связываем одной транзакцией — например, worker читает сам inbox или используется outbox. Подтверждать успешный приём до commit опасно: после падения событие может исчезнуть.

Повтор уже надёжно сохранённого события подтверждаем без повторного side effect. Состояние обработки хранится отдельно от факта приёма: запись «видели» не означает «успешно применили». Если worker вызывает внешнюю систему, inbox не превращает это в exactly-once: нужны её идемпотентный ключ и сверка неизвестного результата. При конфликте ID с другим содержимым нужен разбор, а не тихое признание успеха.

Что ещё ограничить на входе

Лимит размера тела нужно применять во время чтения и на ingress, а не после загрузки произвольного payload в память. Добавляем ограничения времени и нагрузки; не выводим в логи секреты, подписи и полное содержимое prompt. При ротации принимаем только заранее настроенные ключи в ограниченное окно, не выбирая произвольный secret по данным запроса.

Действительная подпись не делает безопасным result_url внутри JSON. Скачивание результата — отдельная граница с проверкой адреса, перенаправлений и лимитов ответа. А текст результата, переданный LLM-агенту, остаётся внешними данными, не новыми системными инструкциями.

Практический вывод

Перед запуском callback-маршрута я бы проверил три независимых свойства: подпись сходится на реальном исходном теле; повтор не запускает бизнес-действие второй раз; позднее событие не откатывает новую попытку задачи. Криптографический verifier решает только первую часть. Как раз смешение этих трёх свойств и создаёт опасное ощущение: «HMAC проверили — значит, webhook уже безопасен».

Первоисточник для схемы и фиксированного тестового вектора — GitHub Docs, Validating webhook deliveries: https://docs.github.com/en/webhooks/using-webhooks/validating-webhook-deliveries. Архитектура inbox выше — рекомендуемый подход для разбора, а не готовый обработчик конкретного AI-сервиса.

Какая проверка чаще теряется в ваших callback-интеграциях: исходные байты, защита от повторов или сопоставление с текущей попыткой задачи?