fix(gateway): bypass SSRF check for Discord SDK-supplied attachment URLs

gateway/platforms/base.py

@@ -110,7 +110,12 @@ def cache_image_from_bytes(data: bytes, ext: str = ".jpg") -> str:
     return str(filepath)
 
 
-async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) -> str:
+async def cache_image_from_url(
+    url: str,
+    ext: str = ".jpg",
+    retries: int = 2,
+    trusted_source: bool = False,
+) -> str:
     """
     Download an image from a URL and save it to the local cache.
 
@@ -121,16 +126,25 @@ async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) ->
         url: The HTTP/HTTPS URL to download from.
         ext: File extension including the dot (e.g. ".jpg", ".png").
         retries: Number of retry attempts on transient failures.
+        trusted_source: If True, skip SSRF safety checks. Use ONLY for URLs
+            obtained directly from a messaging platform's SDK (e.g.
+            ``discord.Attachment.url``, ``telegram.PhotoSize.file_path``)
+            where the URL has already been authenticated by the platform.
+            DNS-rewriting proxies (Clash/Mihomo fake-ip mode) make IP-based
+            SSRF checks unreliable for legitimate platform CDNs, so trusted
+            callers MUST opt out explicitly.
 
     Returns:
         Absolute path to the cached image file as a string.
 
     Raises:
-        ValueError: If the URL targets a private/internal network (SSRF protection).
+        ValueError: If the URL targets a private/internal network (SSRF
+            protection) and ``trusted_source`` is False.
     """
-    from tools.url_safety import is_safe_url
-    if not is_safe_url(url):
-        raise ValueError(f"Blocked unsafe URL (SSRF protection): {_safe_url_for_log(url)}")
+    if not trusted_source:
+        from tools.url_safety import is_safe_url
+        if not is_safe_url(url):
+            raise ValueError(f"Blocked unsafe URL (SSRF protection): {_safe_url_for_log(url)}")
 
     import asyncio
     import httpx
@@ -225,7 +239,12 @@ def cache_audio_from_bytes(data: bytes, ext: str = ".ogg") -> str:
     return str(filepath)
 
 
-async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) -> str:
+async def cache_audio_from_url(
+    url: str,
+    ext: str = ".ogg",
+    retries: int = 2,
+    trusted_source: bool = False,
+) -> str:
     """
     Download an audio file from a URL and save it to the local cache.
 
@@ -236,16 +255,22 @@ async def cache_audio_from_url(url: str, ext: str = ".ogg", retries: int = 2) ->
         url: The HTTP/HTTPS URL to download from.
         ext: File extension including the dot (e.g. ".ogg", ".mp3").
         retries: Number of retry attempts on transient failures.
+        trusted_source: If True, skip SSRF safety checks. Use ONLY for URLs
+            obtained directly from a messaging platform's SDK where the URL
+            has already been authenticated by the platform. See
+            ``cache_image_from_url`` for the rationale.
 
     Returns:
         Absolute path to the cached audio file as a string.
 
     Raises:
-        ValueError: If the URL targets a private/internal network (SSRF protection).
+        ValueError: If the URL targets a private/internal network (SSRF
+            protection) and ``trusted_source`` is False.
     """
-    from tools.url_safety import is_safe_url
-    if not is_safe_url(url):
-        raise ValueError(f"Blocked unsafe URL (SSRF protection): {_safe_url_for_log(url)}")
+    if not trusted_source:
+        from tools.url_safety import is_safe_url
+        if not is_safe_url(url):
+            raise ValueError(f"Blocked unsafe URL (SSRF protection): {_safe_url_for_log(url)}")
 
     import asyncio
     import httpx

gateway/platforms/discord.py

@@ -2322,7 +2322,15 @@ async def _handle_message(self, message: DiscordMessage) -> None:
                     ext = "." + content_type.split("/")[-1].split(";")[0]
                     if ext not in (".jpg", ".jpeg", ".png", ".gif", ".webp"):
                         ext = ".jpg"
-                    cached_path = await cache_image_from_url(att.url, ext=ext)
+                    # trusted_source=True: att.url comes directly from the
+                    # discord.py SDK and is authenticated by Discord. Skipping
+                    # the SSRF check is necessary for users behind proxies that
+                    # use DNS rewriting (e.g. Clash/Mihomo fake-ip mode), which
+                    # would otherwise resolve cdn.discordapp.com to a private
+                    # 198.18.x.x address and trigger a false positive.
+                    cached_path = await cache_image_from_url(
+                        att.url, ext=ext, trusted_source=True,
+                    )
                     media_urls.append(cached_path)
                     media_types.append(content_type)
                     print(f"[Discord] Cached user image: {cached_path}", flush=True)
@@ -2336,7 +2344,9 @@ async def _handle_message(self, message: DiscordMessage) -> None:
                     ext = "." + content_type.split("/")[-1].split(";")[0]
                     if ext not in (".ogg", ".mp3", ".wav", ".webm", ".m4a"):
                         ext = ".ogg"
-                    cached_path = await cache_audio_from_url(att.url, ext=ext)
+                    cached_path = await cache_audio_from_url(
+                        att.url, ext=ext, trusted_source=True,
+                    )
                     media_urls.append(cached_path)
                     media_types.append(content_type)
                     print(f"[Discord] Cached user audio: {cached_path}", flush=True)

tests/gateway/test_media_download_retry.py

@@ -41,6 +41,15 @@ def _make_timeout_error() -> httpx.TimeoutException:
 class TestCacheImageFromUrl:
     """Tests for gateway.platforms.base.cache_image_from_url"""
 
+    @pytest.fixture(autouse=True)
+    def _mock_safe_url(self):
+        # Force is_safe_url to True so these tests don't depend on the
+        # local DNS resolver. Some proxy configurations (e.g. Clash/Mihomo
+        # fake-ip mode) rewrite example.com to a private 198.18.x.x IP,
+        # which would otherwise trip the SSRF guard and fail every test.
+        with patch("tools.url_safety.is_safe_url", return_value=True):
+            yield
+
     def test_success_on_first_attempt(self, tmp_path, monkeypatch):
         """A clean 200 response caches the image and returns a path."""
         monkeypatch.setattr("gateway.platforms.base.IMAGE_CACHE_DIR", tmp_path / "img")
@@ -178,6 +187,12 @@ async def run():
 class TestCacheAudioFromUrl:
     """Tests for gateway.platforms.base.cache_audio_from_url"""
 
+    @pytest.fixture(autouse=True)
+    def _mock_safe_url(self):
+        # See TestCacheImageFromUrl._mock_safe_url for rationale.
+        with patch("tools.url_safety.is_safe_url", return_value=True):
+            yield
+
     def test_success_on_first_attempt(self, tmp_path, monkeypatch):
         """A clean 200 response caches the audio and returns a path."""
         monkeypatch.setattr("gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path / "audio")
@@ -721,3 +736,98 @@ async def run():
         # No sleep — fell back on first attempt
         mock_sleep.assert_not_called()
         assert adapter._session.get.call_count == 1
+
+
+# ---------------------------------------------------------------------------
+# trusted_source SSRF bypass — for platform-SDK-supplied URLs
+# (See: fix for Discord attachments under DNS-rewriting proxies / fake-ip)
+# ---------------------------------------------------------------------------
+
+class TestTrustedSourceBypass:
+    """Tests for trusted_source=True bypassing SSRF checks.
+
+    Platform SDKs (discord.py, telegram, etc.) hand us pre-authenticated
+    attachment URLs. Users behind DNS-rewriting proxies (Clash/Mihomo
+    fake-ip mode) see these URLs resolve to private 198.18.x.x addresses,
+    which trips the SSRF guard with a false positive. Trusted callers can
+    opt out via ``trusted_source=True``.
+    """
+
+    def _build_mock_client(self, content: bytes = b"\xff\xd8\xff data"):
+        fake_response = MagicMock()
+        fake_response.content = content
+        fake_response.raise_for_status = MagicMock()
+        mock_client = AsyncMock()
+        mock_client.get = AsyncMock(return_value=fake_response)
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+        return mock_client
+
+    def test_image_blocked_by_default_when_unsafe(self, tmp_path, monkeypatch):
+        """Without trusted_source, an unsafe URL raises ValueError."""
+        monkeypatch.setattr("gateway.platforms.base.IMAGE_CACHE_DIR", tmp_path / "img")
+
+        async def run():
+            with patch("tools.url_safety.is_safe_url", return_value=False):
+                from gateway.platforms.base import cache_image_from_url
+                return await cache_image_from_url(
+                    "https://cdn.discordapp.com/attachments/x/y.png", ext=".png"
+                )
+
+        with pytest.raises(ValueError, match="SSRF"):
+            asyncio.run(run())
+
+    def test_image_trusted_source_bypasses_ssrf(self, tmp_path, monkeypatch):
+        """trusted_source=True skips the SSRF check entirely."""
+        monkeypatch.setattr("gateway.platforms.base.IMAGE_CACHE_DIR", tmp_path / "img")
+        mock_client = self._build_mock_client()
+
+        async def run():
+            # is_safe_url would return False, but it must NOT be called.
+            with patch("httpx.AsyncClient", return_value=mock_client), \
+                 patch("tools.url_safety.is_safe_url",
+                       side_effect=AssertionError("must not be called")):
+                from gateway.platforms.base import cache_image_from_url
+                return await cache_image_from_url(
+                    "https://cdn.discordapp.com/attachments/x/y.png",
+                    ext=".png",
+                    trusted_source=True,
+                )
+
+        path = asyncio.run(run())
+        assert path.endswith(".png")
+        mock_client.get.assert_called_once()
+
+    def test_audio_blocked_by_default_when_unsafe(self, tmp_path, monkeypatch):
+        """Audio variant: SSRF guard fires when trusted_source is omitted."""
+        monkeypatch.setattr("gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path / "aud")
+
+        async def run():
+            with patch("tools.url_safety.is_safe_url", return_value=False):
+                from gateway.platforms.base import cache_audio_from_url
+                return await cache_audio_from_url(
+                    "https://cdn.discordapp.com/attachments/x/y.ogg", ext=".ogg"
+                )
+
+        with pytest.raises(ValueError, match="SSRF"):
+            asyncio.run(run())
+
+    def test_audio_trusted_source_bypasses_ssrf(self, tmp_path, monkeypatch):
+        """Audio variant: trusted_source=True allows the download."""
+        monkeypatch.setattr("gateway.platforms.base.AUDIO_CACHE_DIR", tmp_path / "aud")
+        mock_client = self._build_mock_client(content=b"OggS audio data")
+
+        async def run():
+            with patch("httpx.AsyncClient", return_value=mock_client), \
+                 patch("tools.url_safety.is_safe_url",
+                       side_effect=AssertionError("must not be called")):
+                from gateway.platforms.base import cache_audio_from_url
+                return await cache_audio_from_url(
+                    "https://cdn.discordapp.com/attachments/x/y.ogg",
+                    ext=".ogg",
+                    trusted_source=True,
+                )
+
+        path = asyncio.run(run())
+        assert path.endswith(".ogg")
+        mock_client.get.assert_called_once()