From 9df021eb8f5c7bacfc58429169804d044210c52f Mon Sep 17 00:00:00 2001 From: Constantine <52125907+Constantine1916@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:25:19 +0800 Subject: [PATCH] fix(telegram): stop leaking bot token via Image.url (#2366) Telegram file.file_path is a full URL of the form https://api.telegram.org/file/bot/ that embeds the bot token. Since #2362 this URL was copied into Image.url, so the token was serialized into the message chain and thereby persisted to the monitoring database, shown in the dashboard, and forwarded to every installed plugin via event dispatch. Anyone with dashboard or plugin access could recover the token and take full control of the bot. Unlike the public CDN URLs used by the other adapters changed in #2362, Telegram file URLs are only usable with the embedded token, so there is no safe URL to expose. Store base64 only (as before #2362); the vision path already relies solely on base64, so nothing downstream changes. Add a regression test asserting the token never appears in the converted Image or the serialized message chain. Co-authored-by: Constantine1916 --- src/langbot/pkg/platform/sources/telegram.py | 5 +- .../platform/test_telegram_adapter.py | 69 ++++++++++++++++++- 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/langbot/pkg/platform/sources/telegram.py b/src/langbot/pkg/platform/sources/telegram.py index 7ff7109ca..f78b56893 100644 --- a/src/langbot/pkg/platform/sources/telegram.py +++ b/src/langbot/pkg/platform/sources/telegram.py @@ -179,10 +179,13 @@ class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverte ) file_format = 'image/jpeg' + # NOTE: Telegram's file.file_path is a full URL of the form + # https://api.telegram.org/file/bot/ which embeds the + # bot token. Unlike the public CDN URLs used by other adapters, it + # cannot be exposed safely, so only base64 is stored here. encoded = await asyncio.to_thread(base64.b64encode, file_bytes) message_components.append( platform_message.Image( - url=file.file_path, base64=f'data:{file_format};base64,{encoded.decode("utf-8")}', ) ) diff --git a/tests/unit_tests/platform/test_telegram_adapter.py b/tests/unit_tests/platform/test_telegram_adapter.py index 885310914..bc7af1bf5 100644 --- a/tests/unit_tests/platform/test_telegram_adapter.py +++ b/tests/unit_tests/platform/test_telegram_adapter.py @@ -1,7 +1,8 @@ """Tests for Telegram Dify form callback helpers.""" import json -from unittest.mock import AsyncMock, MagicMock +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock, MagicMock, patch import pytest from telegram import ForceReply @@ -9,8 +10,10 @@ from telegram import ForceReply import langbot_plugin.api.entities.builtin.platform.entities as platform_entities import langbot_plugin.api.entities.builtin.platform.events as platform_events import langbot_plugin.api.entities.builtin.platform.message as platform_message + from langbot.pkg.platform.sources.telegram import ( TelegramAdapter, + TelegramMessageConverter, _decode_telegram_base64_limited, _telegram_form_action_from_callback, _telegram_select_field_options, @@ -26,6 +29,70 @@ def test_telegram_base64_decode_is_bounded(monkeypatch): _decode_telegram_base64_limited('A' * 12) +TELEGRAM_BOT_TOKEN = '123456789:AAExampleBotTokenThatMustNotLeak' +TELEGRAM_FILE_URL = f'https://api.telegram.org/file/bot{TELEGRAM_BOT_TOKEN}/photos/file_0.jpg' + + +@pytest.mark.asyncio +async def test_telegram_photo_does_not_expose_bot_token_in_image_url(): + """Regression test for the Telegram bot-token leak. + + telegram.Bot builds file.file_path as + https://api.telegram.org/file/bot/, embedding the bot token. + The converter must not copy that URL into Image.url, or the token leaks to + the monitoring DB, dashboard and every installed plugin via the message + chain. Only base64 (which carries no token) may be stored. + """ + tg_file = MagicMock() + tg_file.file_path = TELEGRAM_FILE_URL + + photo_size = MagicMock() + photo_size.get_file = AsyncMock(return_value=tg_file) + + message = MagicMock() + message.text = None + message.caption = None + message.photo = [photo_size] + message.voice = None + message.document = None + + response = MagicMock() + response.headers = {} + + async def iter_chunked(_chunk_size): + yield b'\xff\xd8\xff\xe0jpeg-bytes' + + response.content.iter_chunked = iter_chunked + + @asynccontextmanager + async def fake_get(url): + yield response + + fake_session = MagicMock() + fake_session.get = fake_get + + with patch( + 'langbot.pkg.platform.sources.telegram.httpclient.get_session', + return_value=fake_session, + ): + chain = await TelegramMessageConverter.target2yiri(message, MagicMock(), 'bot-account') + + images = [c for c in chain if isinstance(c, platform_message.Image)] + assert len(images) == 1 + image = images[0] + + # The token-bearing URL must not be retained anywhere on the component. + assert not image.url + assert image.base64 is not None + assert image.base64.startswith('data:image/jpeg;base64,') + + # Belt-and-suspenders: the token must not appear in the serialized chain + # (this is what gets persisted to the monitoring DB and sent to plugins). + serialized = json.dumps(chain.model_dump(), ensure_ascii=False) + assert TELEGRAM_BOT_TOKEN not in serialized + assert 'api.telegram.org/file/bot' not in serialized + + def _select_form_data() -> dict: return { '_current_input_field': 'choice',