Безопасный Telegram-бот для бизнес-агента

Почему одной проверки команды недостаточно

Команда /report может выглядеть безобидно, но её смысл зависит от отправителя, чата и подключённых инструментов. Если бот доверяет любому входящему сообщению, посторонний пользователь способен запросить данные или запустить действие от имени компании.

Первый рубеж — allowlist: явный список идентификаторов, которым разрешён доступ. Проверять нужно числовой user_id, а не имя пользователя. Telegram-имя можно изменить, оно может отсутствовать и не является надёжным идентификатором.

Модель доступа

Для компактного примера разделим команды на два класса:

• Безопасные для чтения: /help, /status. Они не меняют внешнее состояние и возвращают ограниченный набор данных.
• Опасные: /send_report, /cancel_order. Они отправляют сведения, меняют записи или запускают необратимый процесс, поэтому требуют подтверждения.

Реальная классификация зависит от возможностей вашего агента. Команда чтения тоже становится чувствительной, если раскрывает персональные данные, финансовые показатели или внутренние инструкции.

Шаг 1. Храните настройки вне кода

Ниже — пример конфигурации окружения. Значения условны: замените их собственными идентификаторами. Не публикуйте токен бота и не добавляйте файл .env в репозиторий.

TELEGRAM_BOT_TOKEN=значение_из_защищённого_хранилища
TELEGRAM_ALLOWED_USER_IDS=100001,100002
TELEGRAM_ALLOWED_CHAT_IDS=-100200001
CONFIRMATION_TTL_SECONDS=120

Отдельный список чатов не заменяет список пользователей. В разрешённую группу можно добавить нового участника, поэтому для бизнес-команд должны одновременно пройти обе проверки.

Шаг 2. Проверяйте отправителя и чат до обработки текста

Пример ниже использует обычный Python и не привязан к конкретной библиотеке Telegram. Объект update здесь обозначает нормализованное входящее обновление.

import os

def parse_id_set(variable_name: str) -> set[int]:
    raw = os.environ.get(variable_name, "")
    result = set()

    for item in raw.split(","):
        item = item.strip()
        if not item:
            continue
        try:
            result.add(int(item))
        except ValueError as exc:
            raise RuntimeError(
                f"Некорректное значение в {variable_name}"
            ) from exc

    return result


ALLOWED_USERS = parse_id_set("TELEGRAM_ALLOWED_USER_IDS")
ALLOWED_CHATS = parse_id_set("TELEGRAM_ALLOWED_CHAT_IDS")


def is_authorized(update) -> bool:
    user = getattr(update, "effective_user", None)
    chat = getattr(update, "effective_chat", None)

    if user is None or chat is None:
        return False

    return (
        user.id in ALLOWED_USERS
        and chat.id in ALLOWED_CHATS
    )


def handle_update(update):
    if not is_authorized(update):
        audit_denial(update)
        return safe_reply(
            update,
            "Команда недоступна."
        )

    text = (update.effective_message.text or "").strip()
    return route_command(update, text)

Отказ должен происходить по умолчанию. Пустой allowlist означает отсутствие доступа, а не разрешение для всех. Также не сообщайте неизвестному пользователю, какой именно идентификатор или список не прошёл проверку.

Шаг 3. Добавьте одноразовое подтверждение

Опасную команду нельзя выполнять сразу после получения текста. Сначала создайте короткоживущий запрос, покажите пользователю точное последствие и потребуйте подтверждение через кнопку.

from dataclasses import dataclass
from secrets import token_urlsafe
from time import time

CONFIRMATION_TTL = int(
    os.environ.get("CONFIRMATION_TTL_SECONDS", "120")
)

@dataclass
class PendingAction:
    user_id: int
    chat_id: int
    action: str
    resource_id: str
    expires_at: float


pending_actions: dict[str, PendingAction] = {}


def request_confirmation(
    user_id: int,
    chat_id: int,
    action: str,
    resource_id: str,
) -> str:
    nonce = token_urlsafe(24)
    pending_actions[nonce] = PendingAction(
        user_id=user_id,
        chat_id=chat_id,
        action=action,
        resource_id=resource_id,
        expires_at=time() + CONFIRMATION_TTL,
    )
    return nonce


def consume_confirmation(
    nonce: str,
    user_id: int,
    chat_id: int,
) -> PendingAction | None:
    item = pending_actions.pop(nonce, None)

    if item is None or item.expires_at < time():
        return None

    if item.user_id != user_id or item.chat_id != chat_id:
        return None

    return item

При нажатии кнопки обработчик повторно проверяет allowlist, извлекает одноразовую запись и выполняет только сохранённое действие. Нельзя брать название операции или идентификатор ресурса прямо из текста кнопки без серверной проверки.

def confirm_callback(update, nonce: str):
    if not is_authorized(update):
        return safe_reply(update, "Действие недоступно.")

    item = consume_confirmation(
        nonce=nonce,
        user_id=update.effective_user.id,
        chat_id=update.effective_chat.id,
    )

    if item is None:
        return safe_reply(
            update,
            "Подтверждение недействительно или устарело."
        )

    if item.action == "send_report":
        send_report(report_id=item.resource_id)
        return safe_reply(update, "Отчёт отправлен.")

    return safe_reply(update, "Неизвестное действие.")

Словарь в памяти подходит только для демонстрации механики. В нескольких процессах или при перезапусках храните подтверждения в общем хранилище с TTL и атомарным удалением при чтении.

Шаг 4. Ограничьте команды явным маршрутизатором

Не передавайте произвольный пользовательский текст агенту с правом вызывать инструменты. Разрешайте только известные команды и проверяйте аргументы по строгой схеме.

SAFE_COMMANDS = {"/help", "/status"}
DANGEROUS_COMMANDS = {"/send_report"}


def route_command(update, text: str):
    command, *args = text.split(maxsplit=1)

    if command == "/help":
        return safe_reply(
            update,
            "Доступно: /status, /send_report <report_id>"
        )

    if command == "/status":
        status = get_public_service_status()
        return safe_reply(update, f"Состояние сервиса: {status}")

    if command == "/send_report":
        if len(args) != 1 or not valid_report_id(args[0]):
            return safe_reply(
                update,
                "Формат: /send_report <report_id>"
            )

        report_id = args[0]
        nonce = request_confirmation(
            user_id=update.effective_user.id,
            chat_id=update.effective_chat.id,
            action="send_report",
            resource_id=report_id,
        )

        return reply_with_confirmation_button(
            update=update,
            text=f"Отправить отчёт {report_id}?",
            callback_data=f"confirm:{nonce}",
        )

    return safe_reply(update, "Неизвестная команда.")

Функция valid_report_id должна принимать только формат, используемый вашей системой. Например, регулярное выражение с ограниченной длиной безопаснее, чем попытка удалить «плохие» символы из произвольной строки.

Шаг 5. Возвращайте безопасные ответы

В ответах пользователю не должно быть токенов, трассировок, SQL-запросов, системных промптов и полных ответов внутренних API. Подробности ошибки нужны журналу, а не Telegram-чату.

import logging

logger = logging.getLogger(__name__)


def execute_safely(update, operation):
    try:
        return operation()
    except Exception:
        logger.exception(
            "Ошибка операции Telegram-бота",
            extra={
                "user_id": update.effective_user.id,
                "chat_id": update.effective_chat.id,
            },
        )
        return safe_reply(
            update,
            "Операция не выполнена. Обратитесь к администратору."
        )

Не записывайте в журнал токен бота, содержимое секретных полей и полный текст документов. Для аудита обычно достаточно времени, идентификаторов пользователя и чата, типа действия, идентификатора ресурса и результата.

Проверка результата

Проведите ручную проверку в отдельном тестовом окружении, не вызывающем реальные бизнес-операции:

1. Отправьте /status от пользователя вне allowlist. Ожидается нейтральный отказ без раскрытия правил доступа.
2. Повторите команду разрешённым пользователем в запрещённом чате. Ожидается такой же отказ.
3. В разрешённом контексте вызовите /send_report с некорректным идентификатором. Бот должен показать формат и не создавать действие.
4. Вызовите корректную опасную команду. До нажатия кнопки бизнес-операция выполняться не должна.
5. Подтвердите действие другим пользователем или из другого чата. Запрос должен быть отклонён.
6. Нажмите правильную кнопку дважды. Успешным может быть только первое нажатие.
7. Дождитесь истечения TTL. Просроченное подтверждение не должно выполнять операцию.
8. Искусственно вызовите контролируемую ошибку адаптера. В чате должен появиться общий ответ, а подробность — только в защищённом журнале.

Это чек-лист для воспроизведения, а не заявление о выполненных автоматических тестах. Для вашего проекта его стоит оформить как интеграционные тесты с подменёнными бизнес-инструментами.

Типовые ошибки

• Проверка по @username. Используйте числовой user_id; отображаемое имя годится только для интерфейса.
• Разрешение по одному chat_id. Проверяйте и чат, и конкретного отправителя.
• Подтверждение словом «да». Старое сообщение или параллельный диалог создают неоднозначность. Связывайте кнопку с одноразовым запросом.
• Данные операции внутри callback_data. Храните критичные параметры на сервере, а в кнопку помещайте непрозрачный случайный идентификатор.
• Выполнение до подтверждения. На первом шаге разрешены только валидация и создание ожидающего запроса.
• Передача исключения пользователю. Возвращайте нейтральное сообщение, а технические детали направляйте в защищённый журнал.
• Безграничные аргументы. Ограничивайте тип, длину и допустимый формат каждого параметра.
• Секреты в коде. Загружайте их из защищённого хранилища и предусмотрите ротацию.

Ограничения решения

Allowlist не защищает от компрометации разрешённого аккаунта. Для особенно чувствительных операций добавьте второй независимый фактор или подтверждение в корпоративной системе.

Пример также не заменяет защиту webhook: проверку секретного пути или заголовка, TLS, ограничение размера запросов и контроль частоты. Права самого бизнес-агента должны быть минимальными: Telegram-бот не должен получать административный доступ только потому, что интерфейс ограничен списком пользователей.

Если операция затрагивает деньги, персональные данные или юридически значимые действия, одного нажатия кнопки может быть недостаточно. Используйте разделение ролей, правило двух участников, идемпотентные операции и неизменяемый аудит согласно требованиям вашей организации.

Итоговая схема

1. Получить обновление и извлечь числовые идентификаторы.
2. Проверить пользователя и чат по allowlist.
3. Разобрать только известную команду и строго проверить аргументы.
4. Для опасного действия создать одноразовое подтверждение с TTL.
5. При подтверждении повторить авторизацию и атомарно погасить запрос.
6. Выполнить разрешённую операцию с минимальными правами.
7. Вернуть безопасный ответ, а технический результат записать в защищённый аудит.

Другие практические материалы собраны в разделе «Руководства», а определения терминов — в глоссарии Agent Lab Journal.