SmengoAPI

Проверка подписи

Каждый запрос подписан HMAC-SHA256 на секрете вашего эндпоинта. Проверка подписи — обязательный шаг: без неё любой, кто узнал адрес приёмника, сможет прислать вам выдуманное событие.

Секрет

  • Формат — whsec_ + 43 символа base64url. Генерируется на сервере Smengo; задать свой нельзя.
  • Показывается ровно один раз — при создании эндпоинта и после ротации. Повторно посмотреть его нельзя ни в интерфейсе, ни через API: хранится он под ограниченным доступом и никогда не попадает в логи.
  • Секрет потерян — нажмите «Сменить секрет» и сохраните новый.

Формат заголовка

Smengo-Signature: t=1785921323,v1=5a0f3c9e…c81b
  • t — unix-секунды момента подписи (то же значение, что в Smengo-Timestamp).
  • v1 — HMAC-SHA256 в нижнем регистре hex от строки `${t}.${rawBody}`, где rawBody — сырое тело запроса до парсинга.
  • В окне ротации подписей две, они разделены пробелом:
Smengo-Signature: t=1785921323,v1=5a0f3c9e…c81b v1=9d2b71af…40ea

Неизвестные ключи (например, v2= из будущей версии) игнорируйте — они появятся раньше, чем перестанет работать v1.

Алгоритм проверки

  1. Возьмите сырые байты тела — до JSON.parse и до любых преобразований. Пересобранный JSON даст другую строку и другую подпись.
  2. Разберите заголовок: одно значение t и все значения v1.
  3. Отвергните запрос, если |now − t| > 300 секунд (окно 5 минут, защита от повторного проигрывания старого запроса). Разбирайте t строго как ASCII-цифры: в Python isdigit() пропускает юникодные цифры вроде ١٢٣, а int() на них падает.
  4. Посчитайте HMAC-SHA256(secret, "${t}.${rawBody}") в hex.
  5. Сравните с каждым v1 в constant-time и по байтам, а не по строкам: timingSafeEqual бросает исключение на буферах разной длины (сверяйте длину заранее), а compare_digest — TypeError на строке с не-ASCII-символами (кодируйте кандидата в байты).
  6. Совпало хотя бы одно — событие подлинное. Не совпало ни одно — ответьте 401/403 и ничего не обрабатывайте.

Итерируйте по всем v1. Клиент, который берёт только первую подпись, перестанет принимать события на сутки при первой же ротации секрета.

Node.js

const crypto = require('node:crypto')

const TOLERANCE_SEC = 300

function verifySmengoSignature(rawBody, header, secret) {
  if (typeof header !== 'string' || header.length === 0) return false

  // 1. Разбор заголовка: одно t, ВСЕ v1 (при ротации их два).
  let timestamp = null
  const signatures = []
  for (const token of header.split(/[\s,]+/)) {
    const eq = token.indexOf('=')
    if (eq <= 0) continue
    const key = token.slice(0, eq)
    const value = token.slice(eq + 1)
    if (key === 't') {
      if (timestamp === null && /^\d+$/.test(value)) timestamp = Number(value)
    } else if (key === 'v1' && value.length > 0) {
      signatures.push(value)
    }
  }
  if (timestamp === null || signatures.length === 0) return false

  // 2. Окно 5 минут — ДО сравнения хешей.
  const nowSec = Math.floor(Date.now() / 1000)
  if (Math.abs(nowSec - timestamp) > TOLERANCE_SEC) return false

  // 3. Подпись считается по СЫРОМУ телу.
  const expected = Buffer.from(
    crypto.createHmac('sha256', secret).update(`${timestamp}.${rawBody}`, 'utf8').digest('hex'),
    'utf8',
  )

  // 4. Constant-time сравнение с предварительным чеком длины.
  return signatures.some((candidate) => {
    const actual = Buffer.from(candidate, 'utf8')
    return actual.length === expected.length && crypto.timingSafeEqual(actual, expected)
  })
}

Приём в Express — тело обязательно сырое:

const express = require('express')
const app = express()

// express.raw, а не express.json: нужен исходный Buffer.
app.post('/hooks/smengo', express.raw({ type: 'application/json' }), (req, res) => {
  const rawBody = req.body.toString('utf8')
  if (!verifySmengoSignature(rawBody, req.get('Smengo-Signature'), process.env.SMENGO_WEBHOOK_SECRET)) {
    return res.status(401).end()
  }

  const event = JSON.parse(rawBody)
  // Дедупликация по event.id (= заголовок Smengo-Event-Id) и АСИНХРОННАЯ обработка:
  // ответить нужно в пределах 15 секунд.
  enqueue(event)
  res.status(200).end()
})

Python

import hashlib
import hmac
import time

TOLERANCE_SEC = 300


def verify_smengo_signature(raw_body: bytes, header: str, secret: str) -> bool:
    if not header:
        return False

    # 1. Разбор заголовка: одно t, ВСЕ v1 (при ротации их два).
    timestamp = None
    signatures = []
    for token in header.replace(",", " ").split():
        key, _, value = token.partition("=")
        if key == "t":
            # НЕ isdigit(): он пропускает юникодные цифры («١٢٣», «²³»), на
            # которых int() бросает ValueError. Ограничение длины — от
            # заголовка с тысячами цифр (int() на нём тоже падает).
            if timestamp is None and value.isascii() and value.isdecimal() and len(value) <= 20:
                timestamp = int(value)
        elif key == "v1" and value:
            signatures.append(value)
    if timestamp is None or not signatures:
        return False

    # 2. Окно 5 минут — ДО сравнения хешей.
    if abs(int(time.time()) - timestamp) > TOLERANCE_SEC:
        return False

    # 3. Подпись считается по СЫРЫМ байтам тела.
    expected = hmac.new(
        secret.encode("utf-8"),
        f"{timestamp}.".encode("utf-8") + raw_body,
        hashlib.sha256,
    ).hexdigest().encode("ascii")

    # 4. Constant-time сравнение по БАЙТАМ. compare_digest на СТРОКЕ с не-ASCII
    #    бросает TypeError (чек длины не спасает: len("é" * 64) == 64), а
    #    заголовок присылает кто угодно. На байтах разной длины он просто
    #    возвращает False — отдельный чек длины не нужен.
    return any(
        hmac.compare_digest(candidate.encode("utf-8", "ignore"), expected)
        for candidate in signatures
    )

Приём во Flask — request.get_data() отдаёт сырые байты:

from flask import Flask, request

app = Flask(__name__)


@app.post("/hooks/smengo")
def smengo_webhook():
    raw_body = request.get_data()
    header = request.headers.get("Smengo-Signature", "")
    if not verify_smengo_signature(raw_body, header, SMENGO_WEBHOOK_SECRET):
        return "", 401

    event = request.get_json()
    enqueue(event)  # дедупликация по event["id"], обработка асинхронная
    return "", 200

Ротация секрета

Кнопка «Сменить секрет» в карточке вебхука выдаёт новый секрет и запускает окно на 24 часа:

  1. Сразу после ротации каждая доставка подписывается обоими секретами — старым и новым (два v1 через пробел).
  2. В течение суток обновите секрет на своей стороне. Приём не прерывается: код, итерирующий по всем v1, принимает события и до, и после переключения.
  3. По истечении окна старый секрет перестаёт использоваться, в заголовке остаётся одна подпись.

Пока окно не закрылось, повторная ротация недоступна. Если секрет скомпрометирован и старый нужно отключить немедленно — удалите эндпоинт и создайте новый.

Частые ошибки

Симптом Причина
Подпись не сходится, тело выглядит правильным тело пересобрано после парсинга (JSON.stringify(JSON.parse(body))) — считайте HMAC по сырым байтам
Приём сломался ровно после смены секрета код берёт первое v1 вместо перебора всех
Исключение внутри сравнения timingSafeEqual вызван на буферах разной длины (сверяйте длину заранее) или compare_digest — на строке с не-ASCII (сравнивайте байты)
На мусорный заголовок приёмник отвечает 500 вместо 401 t разобран через isdigit() либо подпись сравнивается строками — и то и другое падает на заголовке, который может прислать кто угодно, даже не зная секрета
Все запросы отвергаются как «старые» часы приёмника разъехались; окно допуска — 5 минут, синхронизируйте время по NTP
Подпись сошлась, но события приходят дважды это штатный режим at-least-once — дедуплицируйте по Smengo-Event-Id

Что дальше

Была ли статья полезна?