Telegram-бот на своём компьютере: настройка webhook

Вы пишете Telegram-бота и хотите проверить его на своём компьютере. Для этого Telegram нужен адрес, куда можно прислать новое сообщение. Такой способ доставки называется webhook. DevProxy даст вашей программе внешний адрес, поэтому настраивать роутер не придётся.

В примерах встретятся localhost и 127.0.0.1: оба обозначают ваш компьютер. Сообщения будет принимать небольшая программа по адресу http://127.0.0.1:8099/telegram.

Для проверки создайте отдельного бота, чтобы не мешать тому, которым уже пользуетесь. BotFather выдаст его токен: это ключ для управления ботом. Храните его отдельно от кода и никому не пересылайте.

Когда сообщение придёт, программа покажет его номер, update_id, в окне запуска. Ответа в чате не будет: в этом примере мы только проверяем доставку, а отправку ответа не добавляем.

Что понадобится

• Отдельный тестовый бот Telegram и его токен. При другом способе, long polling, бот сам спрашивает Telegram о новых сообщениях через getUpdates. Пока у бота работает webhook, получать те же события так нельзя. Поэтому не переключайте бота, если его работа не должна прерываться.

• Python 3 со стандартной библиотекой; pip и веб-фреймворки не нужны.

• DevProxy 1.0.8 или новее и ключ с вариантом «Сайт или API». Если ещё не установили программу, начните с инструкции «Как начать пользоваться DevProxy».

• Отдельный секрет для проверки входящих сообщений, secret_token. Его вы придумаете сами. Токен от BotFather для этой цели не подходит: это два разных значения.

1. Запустите приёмник на своём компьютере

Сохраните код ниже в файл receiver.py. Это программа, которая будет принимать сообщения. Она работает по адресу 127.0.0.1:8099, доступному только на вашем компьютере. Позже DevProxy свяжет его с внешним адресом для Telegram.

Программа принимает сообщения только по пути /telegram и сверяет секрет в поле X-Telegram-Bot-Api-Secret-Token. Так она отличает нужный запрос от постороннего. Для одного запроса установлен предел 256 КиБ и ожидание не больше пяти секунд.

Перед запуском задайте секрет в том же окне PowerShell, где будете запускать программу. Подойдут от 1 до 256 латинских букв, цифр, _ и -. Замените TEST_SECRET_REPLACE_ME своим случайным значением. Позже вы укажете этот же секрет при подключении Telegram. Не используйте вместо него токен бота или пароль от своего кабинета.

$env:TELEGRAM_WEBHOOK_SECRET="TEST_SECRET_REPLACE_ME"
python receiver.py

При старте в консоли появится адрес http://127.0.0.1:8099/telegram. Неизвестный путь получит ответ 404, неверный или отсутствующий секрет получит 403, а данные без целого update_id будут отклонены с ответом 400. После принятого Update (уведомления Telegram о событии) приёмник ответит 200 и напечатает только update_id. Текст сообщения, сведения о пользователе и секретный заголовок в консоль не попадают.

#!/usr/bin/env python3
"""Локальный тестовый приёмник webhook Telegram Bot API."""

from __future__ import annotations

import hmac
import json
import os
import sys
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


HOST = "127.0.0.1"
PORT = 8099
WEBHOOK_PATH = "/telegram"
MAX_BODY_BYTES = 256 * 1024
READ_TIMEOUT_SECONDS = 5


def read_webhook_secret() -> str:
    secret = os.environ.get("TELEGRAM_WEBHOOK_SECRET", "")
    if not 1 <= len(secret) <= 256:
        raise ValueError(
            "Задайте TELEGRAM_WEBHOOK_SECRET длиной от 1 до 256 символов."
        )
    if any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-" for ch in secret):
        raise ValueError(
            "TELEGRAM_WEBHOOK_SECRET допускает только A-Z, a-z, 0-9, _ и -."
        )
    return secret


class TelegramWebhookHandler(BaseHTTPRequestHandler):
    webhook_secret = b""

    def setup(self) -> None:
        super().setup()
        self.connection.settimeout(READ_TIMEOUT_SECONDS)

    def send_result(self, status: HTTPStatus, ok: bool) -> None:
        body = json.dumps({"ok": ok}, separators=(",", ":")).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def secret_is_valid(self) -> bool:
        received = self.headers.get("X-Telegram-Bot-Api-Secret-Token")
        if received is None:
            return False
        try:
            received_bytes = received.encode("utf-8")
        except UnicodeEncodeError:
            return False
        return hmac.compare_digest(received_bytes, self.webhook_secret)

    def do_POST(self) -> None:
        if self.path != WEBHOOK_PATH:
            self.send_result(HTTPStatus.NOT_FOUND, False)
            return
        if not self.secret_is_valid():
            self.send_result(HTTPStatus.FORBIDDEN, False)
            return

        content_length = self.headers.get("Content-Length")
        try:
            body_length = int(content_length) if content_length is not None else -1
        except ValueError:
            self.send_result(HTTPStatus.BAD_REQUEST, False)
            return
        if body_length <= 0:
            self.send_result(HTTPStatus.BAD_REQUEST, False)
            return
        if body_length > MAX_BODY_BYTES:
            self.send_result(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, False)
            return

        try:
            body = self.rfile.read(body_length)
        except (TimeoutError, OSError):
            self.send_result(HTTPStatus.REQUEST_TIMEOUT, False)
            return
        if len(body) != body_length:
            self.send_result(HTTPStatus.BAD_REQUEST, False)
            return

        try:
            update = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            self.send_result(HTTPStatus.BAD_REQUEST, False)
            return
        if not isinstance(update, dict) or type(update.get("update_id")) is not int:
            self.send_result(HTTPStatus.BAD_REQUEST, False)
            return

        print(f"Учебный Update принят: update_id={update['update_id']}", flush=True)
        self.send_result(HTTPStatus.OK, True)

    def log_message(self, format: str, *args: object) -> None:
        return


class TelegramReceiverServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True


def main() -> int:
    try:
        secret = read_webhook_secret()
    except ValueError as error:
        print(error, file=sys.stderr)
        return 2

    TelegramWebhookHandler.webhook_secret = secret.encode("ascii")
    server = TelegramReceiverServer((HOST, PORT), TelegramWebhookHandler)
    print(f"Локальный приёмник слушает http://{HOST}:{PORT}{WEBHOOK_PATH}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nПриёмник остановлен.")
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Сначала отправьте программе два пробных сообщения со своего компьютера: одно с неверным секретом, другое с правильным. Для этого откройте ещё одно окно PowerShell. В следующем примере замените TEST_SECRET_REPLACE_ME тем значением, с которым запустили приёмник:

try {
  Invoke-WebRequest -UseBasicParsing -Uri "http://127.0.0.1:8099/telegram" -Method Post -Headers @{ "X-Telegram-Bot-Api-Secret-Token" = "WRONG_SECRET" } -ContentType "application/json" -Body '{"update_id":900001}' | Out-Null
} catch {
  "wrong=$([int]$_.Exception.Response.StatusCode)"
}
$response = Invoke-WebRequest -UseBasicParsing -Uri "http://127.0.0.1:8099/telegram" -Method Post -Headers @{ "X-Telegram-Bot-Api-Secret-Token" = "TEST_SECRET_REPLACE_ME" } -ContentType "application/json" -Body '{"update_id":900001}'
"right=$($response.StatusCode)"

В ответе должны появиться wrong=403 и right=200: чужой секрет отклонён, правильный принят. В окне, где запущен receiver.py, появится строка Учебный Update принят: update_id=900001. Пока сообщение отправили вы сами. Telegram подключим на следующем шаге.

2. Дайте программе внешний адрес через DevProxy

В клиенте 1.0.8 или новее откройте локальную панель по адресу и одноразовому коду из его консоли. Обычно адрес такой: http://127.0.0.1:4040, но занятый порт может отличаться. Сверяйтесь со строкой в консоли. Если одноразовый код истёк, нажмите R, чтобы получить новый.

Telegram не может пройти браузерную форму пароля. На отдельном канале доступ проверяет приёмник с помощью secret_token; форма входа перед ним помешает доставке. Защиту существующих каналов оставьте как есть.

В панели выберите Каналы → Создать канал → HTTP. Укажите имя tg-hook и локальный адрес http://127.0.0.1:8099. Сохранение в исходный YAML оставьте включённым. Нажмите Сохранить, а затем отдельно Запустить: одно сохранение канал не запускает. Новый канал ведёт только к приёмнику. Защиту других каналов и остальной конфиг сохраните.

Если запускаете DevProxy впервые и файла настроек ещё нет, сохраните такой файл в формате YAML:

server: https://control.devtunnels.ru
token: ВАШ_HTTP_КЛЮЧ
reconnect: true
tunnels:
  - name: tg-hook
    kind: http
    local: http://127.0.0.1:8099

В token вставьте свой ключ «Сайт или API». Сохраните отступы, как в примере: по ним программа понимает, какие настройки относятся к одному каналу. Если DevProxy уже работает с вашим файлом, не заменяйте его целиком.

В существующем файле найдите список tunnels и добавьте в него только этот канал:

  - name: tg-hook
    kind: http
    local: http://127.0.0.1:8099

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

Перед регистрацией webhook запустите приёмник на 127.0.0.1:8099, канал tg-hook и клиент DevProxy. В разделе Туннели дождитесь состояния «в сети». Скопируйте выданный для tg-hook адрес https:// и добавьте в конце /telegram. Из одного имени tg-hook адрес не складывается: домен, префикс и порт выдаёт сервис.

На иллюстрации HTTP-канал называется site. Для своего теста скопируйте HTTPS-адрес строки tg-hook.

Итоговый адрес должен начинаться с https:// и заканчиваться на /telegram, например https://your-address.example/telegram. Если после перезапуска адрес канала изменился, передайте новый адрес в Bot API. Старый адрес уже не ведёт к текущему каналу.

3. Подключите webhook для отдельного бота

Сохраните следующий код во второй файл, bot_api.py, рядом с первым. Он сообщит Telegram, куда присылать сообщения. У него три команды: set подключает адрес, info показывает состояние, delete отключает доставку на этот адрес.

Токен бота и секрет программа попросит ввести отдельно. Во время ввода символы не видны, и в самой команде указывать их не нужно. Программа также проверит адрес и допустимые символы секрета. Накопившиеся сообщения автоматически не удаляются.

#!/usr/bin/env python3
"""Управление webhook отдельного учебного бота через Telegram Bot API."""

from __future__ import annotations

import argparse
import getpass
import json
import re
import sys
import urllib.error
import urllib.parse
import urllib.request


API_BASE = "https://api.telegram.org"
REQUEST_TIMEOUT_SECONDS = 15
MAX_RESPONSE_BYTES = 1024 * 1024
SECRET_PATTERN = re.compile(r"[A-Za-z0-9_-]{1,256}")
BOT_TOKEN_PATTERN = re.compile(r"[0-9]+:[A-Za-z0-9_-]+")


class SafeOperationError(Exception):
    """Ошибка с безопасным для вывода текстом."""


def ask_bot_token() -> str:
    token = getpass.getpass("Bot token (ввод не отображается): ").strip()
    if BOT_TOKEN_PATTERN.fullmatch(token) is None:
        raise SafeOperationError("Bot token пустой или содержит недопустимые символы.")
    return token


def ask_webhook_secret() -> str:
    secret = getpass.getpass("Secret token webhook (ввод не отображается): ").strip()
    if SECRET_PATTERN.fullmatch(secret) is None:
        raise SafeOperationError(
            "Secret token должен содержать 1–256 символов A-Z, a-z, 0-9, _ или -."
        )
    return secret


def validate_webhook_url(url: str) -> str:
    try:
        parts = urllib.parse.urlsplit(url)
        hostname = parts.hostname
        port = parts.port
    except ValueError:
        raise SafeOperationError("URL webhook имеет неверную структуру.") from None
    if (
        parts.scheme != "https"
        or not hostname
        or parts.username is not None
        or parts.password is not None
        or parts.fragment
    ):
        raise SafeOperationError("Укажите внешний HTTPS URL без логина, пароля и fragment.")
    if port not in (None, 80, 88, 443, 8443):
        raise SafeOperationError(
            "Telegram документирует webhook на портах 443, 80, 88 и 8443."
        )
    return url


def call_bot_api(token: str, method: str, payload: dict[str, object] | None = None) -> dict[str, object]:
    url = f"{API_BASE}/bot{token}/{method}"
    body = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(
        url,
        data=body,
        method="POST",
        headers={"Content-Type": "application/json; charset=utf-8"},
    )
    try:
        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            response_body = response.read(MAX_RESPONSE_BYTES + 1)
    except urllib.error.HTTPError as error:
        raise SafeOperationError(f"Bot API вернул HTTP {error.code}.") from None
    except (urllib.error.URLError, TimeoutError, OSError):
        raise SafeOperationError("Не удалось соединиться с Bot API.") from None

    if len(response_body) > MAX_RESPONSE_BYTES:
        raise SafeOperationError("Ответ Bot API превышает допустимый размер.")
    try:
        result = json.loads(response_body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise SafeOperationError("Bot API вернул ответ не в формате JSON.") from None
    if not isinstance(result, dict) or result.get("ok") is not True:
        raise SafeOperationError("Bot API не подтвердил успешное выполнение.")
    return result


def run_set(url: str) -> None:
    webhook_url = validate_webhook_url(url)
    token = ask_bot_token()
    secret = ask_webhook_secret()
    call_bot_api(
        token,
        "setWebhook",
        {
            "url": webhook_url,
            "secret_token": secret,
            "allowed_updates": ["message"],
        },
    )
    print("setWebhook: Bot API подтвердил регистрацию подписки.")
    print("Факт доставки проверьте отдельно по событию в приёмнике.")


def run_info() -> None:
    token = ask_bot_token()
    response = call_bot_api(token, "getWebhookInfo")
    info = response.get("result")
    if not isinstance(info, dict):
        raise SafeOperationError("Bot API не вернул объект WebhookInfo.")
    print(f"url: {info.get('url', '')}")
    print(f"pending_update_count: {info.get('pending_update_count', '')}")
    print(f"last_error_message: {info.get('last_error_message', '')}")


def run_delete() -> None:
    token = ask_bot_token()
    print("Будет снята webhook-подписка этого учебного бота.")
    confirmation = input("Введите delete для продолжения: ").strip()
    if confirmation != "delete":
        print("Операция отменена.")
        return
    call_bot_api(token, "deleteWebhook")
    print("deleteWebhook: Bot API подтвердил снятие webhook.")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest="command", required=True)
    set_parser = commands.add_parser("set", help="зарегистрировать webhook учебного бота")
    set_parser.add_argument(
        "--url",
        required=True,
        help="внешний HTTPS URL вида https://…/telegram",
    )
    commands.add_parser("info", help="показать состояние webhook")
    commands.add_parser("delete", help="явно снять webhook учебного бота")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        if args.command == "set":
            run_set(args.url)
        elif args.command == "info":
            run_info()
        else:
            run_delete()
    except SafeOperationError as error:
        print(str(error), file=sys.stderr)
        return 1
    except (EOFError, KeyboardInterrupt):
        print("Операция отменена.", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Запустите регистрацию и подставьте адрес HTTPS-канала из предыдущего шага:

python bot_api.py set --url https://ВАШ_АДРЕС_ИЗ_ТУННЕЛЕЙ/telegram

Сначала введите токен от BotFather, затем секрет, с которым запущен receiver.py. Символы при вводе не отображаются. Если соединиться не удалось, программа покажет короткую ошибку без токена.

Успешный ответ setWebhook подтверждает только подписку, а не доставку события. Отправьте боту обычное сообщение со своего аккаунта. В консоли приёмника появится строка с новым update_id. Текст сообщения и данные пользователя программа не печатает. Запомните этот идентификатор: строка покажет, что конкретное событие дошло до вашей программы, а приёмник ответил 200.

Параметры регистрации и проверки состояния описаны в официальной справке Telegram Bot API.

Состояние можно посмотреть отдельной командой:

python bot_api.py info

В ответе будут адрес доставки (url), число ожидающих сообщений (pending_update_count) и последняя ошибка (last_error_message). Ноль в очереди ещё не значит, что всё работает: возможно, боту просто никто не писал. А сообщение о старой ошибке может остаться и после её исправления. Поэтому проверяйте результат по новому update_id в окне приёмника после сообщения, которое вы только что отправили.

Если событие не пришло

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

Для ответа 404 проверьте путь /telegram, состояние HTTP-канала и адрес, скопированный из Туннелей. Ответ 400 во время локального теста значит, что JSON не удалось разобрать или в нём нет целого update_id. Если после перезапуска или пересоздания канала адрес изменился, снова выполните set с новым адресом https://. Когда программа не может подключиться к 127.0.0.1:8099, проверьте, что приёмник запущен на этом компьютере, слушает 127.0.0.1 и порт не занят другим процессом.

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

Как закончить тест

Сначала снимите webhook только со своего тестового бота командой python bot_api.py delete и подтвердите действие словом delete. Затем остановите канал tg-hook и только после этого завершите приёмник по Ctrl+C. Другие каналы и действующего бота оставьте как были. Старые ожидающие события автоматически не удаляются: drop_pending_updates не применяется без отдельного решения владельца бота.

Похожим способом можно получать уведомления от GitHub: об этом есть инструкция GitHub webhook на localhost. А если нужно просто показать сайт заказчику, достаточно открыть его по ссылке через DevProxy.