Do not log the Telegram bot token (#182452)

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()

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)

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"
     },

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"),
     [