From 0cb1d21d32d4281dd4834f303ec6404402f3044d Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Thu, 17 Sep 2026 18:36:11 +0200 Subject: [PATCH] Do not log the Telegram bot token (#182452) --- .../components/telegram_bot/__init__.py | 16 +++- .../components/telegram_bot/log_filter.py | 78 +++++++++++++++++++ .../components/telegram_bot/strings.json | 3 + .../telegram_bot/test_telegram_bot.py | 29 +++++++ 4 files changed, 124 insertions(+), 2 deletions(-) create mode 100644 homeassistant/components/telegram_bot/log_filter.py diff --git a/homeassistant/components/telegram_bot/__init__.py b/homeassistant/components/telegram_bot/__init__.py index 54d567aa8a01..feeb9c6c2b8d 100644 --- a/homeassistant/components/telegram_bot/__init__.py +++ b/homeassistant/components/telegram_bot/__init__.py @@ -7,7 +7,7 @@ import telegram from telegram import Bot from telegram.error import InvalidToken, TelegramError -from homeassistant.const import CONF_PLATFORM, Platform +from homeassistant.const import CONF_API_KEY, CONF_PLATFORM, Platform from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.helpers import ( @@ -41,6 +41,7 @@ from .const import ( PLATFORM_POLLING, PLATFORM_WEBHOOKS, ) +from .log_filter import async_redact_token, async_unredact_token from .services import async_setup_services _LOGGER = logging.getLogger(__name__) @@ -157,6 +158,10 @@ def bot_device_info(config_entry: TelegramBotConfigEntry, bot_id: int) -> dr.Dev async def async_setup_entry(hass: HomeAssistant, entry: TelegramBotConfigEntry) -> bool: """Create the Telegram bot from config entry.""" + # Registered before the bot is built: the library logs the token as soon as + # it constructs the API URLs. + async_redact_token(entry.data[CONF_API_KEY]) + bot: Bot = await hass.async_add_executor_job(initialize_bot, hass, entry.data) try: await bot.get_me() @@ -164,7 +169,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: TelegramBotConfigEntry) # pylint: disable-next=home-assistant-exception-not-translated raise ConfigEntryAuthFailed("Invalid API token for Telegram Bot.") from err except TelegramError as err: - raise ConfigEntryNotReady from err + # Do not let the message through: the Telegram API URL embeds the bot + # token, and library errors quote that URL. + raise ConfigEntryNotReady( + translation_domain=DOMAIN, + translation_key="cannot_connect", + ) from err p_type: str = entry.data[CONF_PLATFORM] @@ -212,6 +222,8 @@ async def async_unload_entry( hass: HomeAssistant, entry: TelegramBotConfigEntry ) -> bool: """Unload Telegram app.""" + async_unredact_token(entry.data[CONF_API_KEY]) + # broadcast platform has no app if entry.runtime_data.app: await entry.runtime_data.app.shutdown() diff --git a/homeassistant/components/telegram_bot/log_filter.py b/homeassistant/components/telegram_bot/log_filter.py new file mode 100644 index 000000000000..f834ec4a2a74 --- /dev/null +++ b/homeassistant/components/telegram_bot/log_filter.py @@ -0,0 +1,78 @@ +"""Keep Telegram bot tokens out of the logs. + +The Telegram API embeds the bot token in the URL path, so the library, the HTTP +stack underneath it and any traceback quoting that URL all carry the token. None +of that passes through this integration's own log calls, so the only way to keep +it out of `home-assistant.log` is to scrub the records on their way out. +""" + +import logging +from typing import override + +REDACTED = "**REDACTED**" + + +class TokenRedactingFilter(logging.Filter): + """Replace known bot tokens in log records with a placeholder.""" + + def __init__(self) -> None: + """Initialize the filter with no tokens to redact.""" + super().__init__() + # Replaced rather than mutated, so the logging thread always reads a + # consistent snapshot. + self._tokens: frozenset[str] = frozenset() + + def add_token(self, token: str) -> None: + """Start redacting a token.""" + self._tokens |= {token} + + def remove_token(self, token: str) -> None: + """Stop redacting a token.""" + self._tokens -= {token} + + @override + def filter(self, record: logging.LogRecord) -> bool: + """Redact any known token in the message and the traceback.""" + if not (tokens := self._tokens): + return True + + message = record.getMessage() + if any(token in message for token in tokens): + for token in tokens: + message = message.replace(token, REDACTED) + record.msg = message + record.args = None + + if record.exc_info: + traceback = logging.Formatter().formatException(record.exc_info) + if any(token in traceback for token in tokens): + for token in tokens: + traceback = traceback.replace(token, REDACTED) + # Hand the handler finished text so it cannot re-expand the + # original exception. + record.exc_text = traceback + record.exc_info = None + + return True + + +_FILTER = TokenRedactingFilter() + + +def async_redact_token(token: str) -> None: + """Redact a bot token from every log record from now on.""" + # The filter goes on the root handlers, not on a logger: a logger's filters + # only see records logged through that logger, so a filter on "telegram" + # would miss "telegram.Bot", which is where the library logs the token. + # Handler filters see every record that reaches them. + root = logging.getLogger() + for handler in root.handlers: + if _FILTER not in handler.filters: + handler.addFilter(_FILTER) + + _FILTER.add_token(token) + + +def async_unredact_token(token: str) -> None: + """Stop redacting a bot token.""" + _FILTER.remove_token(token) diff --git a/homeassistant/components/telegram_bot/strings.json b/homeassistant/components/telegram_bot/strings.json index 7a9428cabe5e..439ee965ad09 100644 --- a/homeassistant/components/telegram_bot/strings.json +++ b/homeassistant/components/telegram_bot/strings.json @@ -186,6 +186,9 @@ "allowlist_external_dirs_error": { "message": "File path has not been configured in allowlist_external_dirs." }, + "cannot_connect": { + "message": "Could not connect to Telegram." + }, "entry_not_loaded": { "message": "{telegram_bot} is not loaded" }, diff --git a/tests/components/telegram_bot/test_telegram_bot.py b/tests/components/telegram_bot/test_telegram_bot.py index e60de9e71115..0ecea74df54e 100644 --- a/tests/components/telegram_bot/test_telegram_bot.py +++ b/tests/components/telegram_bot/test_telegram_bot.py @@ -145,6 +145,35 @@ async def test_polling_platform_init_failed( assert mock_polling_config_entry.state is ConfigEntryState.SETUP_RETRY +async def test_polling_platform_init_failed_does_not_log_token( + hass: HomeAssistant, + mock_polling_config_entry: MockConfigEntry, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test that a connection failure does not put the bot token in the log.""" + api_key = mock_polling_config_entry.data[CONF_API_KEY] + # The Telegram API URL embeds the bot token, and library errors quote it. + error = NetworkError( + "httpx.HTTPStatusError: Client error '401 Unauthorized' for url " + f"'https://api.telegram.org/bot{api_key}/getMe'" + ) + + with patch( + "homeassistant.components.telegram_bot.bot.Bot.get_me", side_effect=error + ): + mock_polling_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_polling_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_polling_config_entry.state is ConfigEntryState.SETUP_RETRY + # Home Assistant strips the trailing period from translated messages. + assert mock_polling_config_entry.reason == "Could not connect to Telegram" + + # Nothing at any level may carry the token: not the info line, not the + # traceback config entry setup logs, not the library's own debug output. + assert api_key not in caplog.text + + @pytest.mark.parametrize( ("service", "input"), [