fix(telegram): validate remote media URLs

nanobot/channels/telegram.py

@@ -19,6 +19,7 @@ from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.paths import get_media_dir
 from nanobot.config.schema import Base
+from nanobot.security.network import validate_url_target
 from nanobot.utils.helpers import split_message
 
 TELEGRAM_MAX_MESSAGE_LEN = 4000  # Telegram message character limit
@@ -313,6 +314,10 @@ class TelegramChannel(BaseChannel):
             return "audio"
         return "document"
 
+    @staticmethod
+    def _is_remote_media_url(path: str) -> bool:
+        return path.startswith(("http://", "https://"))
+
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Telegram."""
         if not self._app:
@@ -356,7 +361,10 @@ class TelegramChannel(BaseChannel):
                 param = "photo" if media_type == "photo" else media_type if media_type in ("voice", "audio") else "document"
 
                 # Telegram Bot API accepts HTTP(S) URLs directly for media params.
-                if media_path.startswith(("http://", "https://")):
+                if self._is_remote_media_url(media_path):
+                    ok, error = validate_url_target(media_path)
+                    if not ok:
+                        raise ValueError(f"unsafe media URL: {error}")
                     await sender(
                         chat_id=chat_id,
                         **{param: media_path},

tests/test_telegram_channel.py

@@ -30,6 +30,7 @@ class _FakeUpdater:
 class _FakeBot:
     def __init__(self) -> None:
         self.sent_messages: list[dict] = []
+        self.sent_media: list[dict] = []
         self.get_me_calls = 0
 
     async def get_me(self):
@@ -42,6 +43,18 @@ class _FakeBot:
     async def send_message(self, **kwargs) -> None:
         self.sent_messages.append(kwargs)
 
+    async def send_photo(self, **kwargs) -> None:
+        self.sent_media.append({"kind": "photo", **kwargs})
+
+    async def send_voice(self, **kwargs) -> None:
+        self.sent_media.append({"kind": "voice", **kwargs})
+
+    async def send_audio(self, **kwargs) -> None:
+        self.sent_media.append({"kind": "audio", **kwargs})
+
+    async def send_document(self, **kwargs) -> None:
+        self.sent_media.append({"kind": "document", **kwargs})
+
     async def send_chat_action(self, **kwargs) -> None:
         pass
 
@@ -231,6 +244,65 @@ async def test_send_reply_infers_topic_from_message_id_cache() -> None:
     assert channel._app.bot.sent_messages[0]["reply_parameters"].message_id == 10
 
 
+@pytest.mark.asyncio
+async def test_send_remote_media_url_after_security_validation(monkeypatch) -> None:
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
+        MessageBus(),
+    )
+    channel._app = _FakeApp(lambda: None)
+    monkeypatch.setattr("nanobot.channels.telegram.validate_url_target", lambda url: (True, ""))
+
+    await channel.send(
+        OutboundMessage(
+            channel="telegram",
+            chat_id="123",
+            content="",
+            media=["https://example.com/cat.jpg"],
+        )
+    )
+
+    assert channel._app.bot.sent_media == [
+        {
+            "kind": "photo",
+            "chat_id": 123,
+            "photo": "https://example.com/cat.jpg",
+            "reply_parameters": None,
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_send_blocks_unsafe_remote_media_url(monkeypatch) -> None:
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
+        MessageBus(),
+    )
+    channel._app = _FakeApp(lambda: None)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.validate_url_target",
+        lambda url: (False, "Blocked: example.com resolves to private/internal address 127.0.0.1"),
+    )
+
+    await channel.send(
+        OutboundMessage(
+            channel="telegram",
+            chat_id="123",
+            content="",
+            media=["http://example.com/internal.jpg"],
+        )
+    )
+
+    assert channel._app.bot.sent_media == []
+    assert channel._app.bot.sent_messages == [
+        {
+            "chat_id": 123,
+            "text": "[Failed to send: internal.jpg]",
+            "reply_parameters": None,
+        }
+    ]
+
+
 @pytest.mark.asyncio
 async def test_group_policy_mention_ignores_unmentioned_group_message() -> None:
     channel = TelegramChannel(