# NousResearch/hermes-agent · SSRF URL safety check causing false-positive blocks

## Open now

- 3 issues, 4 PRs open on GitHub.
- #21379: test: Discord document handling tests fail due to SSRF protection blocking fake CDN URLs
- #38048: url_safety false-positively blocks DNS64/NAT64 addresses in 64:ff9b::/96
- #47123: \[qqbot\] url_safety whitelist blocks C2C file attachment downloads from grouptalk.c2c.qq.com
- PR #21386: test(gateway): mock SSRF URL guard in Discord document handling tests
- PR #38058: fix(url_safety): evaluate embedded IPv4 for NAT64 64:ff9b::/96 addresses
- PR #107724: fix(qqbot): allow grouptalk.c2c.qq.com in the url_safety private-IP bypass (#47123)
- PR #131466: fix(url_safety): classify the NAT64 well-known prefix by its embedded IPv4

Multiple issues and fixes address the SSRF protection logic incorrectly blocking legitimate traffic, including RFC 2544 and NAT64 address ranges, Discord CDN URLs, and QQ/Tencent CDN domains under TUN/DNS-rewriting proxy setups. Work spans allowlisting specific ranges, adding configurable trusted hosts, and adjusting how synthesized IPv6 addresses are evaluated.

## TL;DR

- 22 member(s): 3 open issue(s), 5 closed issue(s), 4 open PR(s), 3 merged PR(s), 7 PR(s) closed without merging.
- A fix was merged to main in PR #11567. Verify it addresses your case.
- **Recommended:** read PR #11567 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-11567,pr-11568`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
- Members outside that structural core may belong to a different underlying problem.
- Pairwise link density `0.09`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Open related issues

- #21379: test: Discord document handling tests fail due to SSRF protection blocking fake CDN URLs
- #38048: url_safety false-positively blocks DNS64/NAT64 addresses in 64:ff9b::/96
- #47123: \[qqbot\] url_safety whitelist blocks C2C file attachment downloads from grouptalk.c2c.qq.com

## Closed / duplicate issues

- #3777: \[Bug\]: RFC 2544 benchmark range (198.18.0.0/15) misclassified as private, breaking TUN proxy users
- #6511: Discord image/audio attachments fail under DNS-rewriting proxies (false-positive SSRF block from #5944)
- #6587: Discord media auto-processing fails because SSRF check blocks cdn.discordapp.com
- #11284: QQbot cannot process images: url_safety blocks Tencent CDN domain due to local proxy IP flagged as private
- #19992: url_safety.py incorrectly blocks RFC 2544 (198.18.0.0/15) used by Clash TUN Fake IP

## Prior & competing PRs

- **#11388: fix(url-safety): allow trusted QQ CDN benchmark resolution**. closed without merge · fix tier `suggested`
  - maintainer direction: category `merged_via`; implication `superseded`
  - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._
- **#11567: fix(qqbot): Batch-2 salvage — SSRF allowlist, voice preflight, group auth, WS reconnect wait**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `gateway/platforms/qqbot.py`, `gateway/run.py`, `scripts/release.py`, `tests/gateway/test_qqbot.py`, `tests/gateway/test_unauthorized_dm_behavior.py`, `tests/tools/test_url_safety.py`, `tools/url_safety.py`
  - fix hunks:
  - `gateway/platforms/qqbot.py`

    ```diff
    @@ -64,6 +64,7 @@
         MessageEvent,
         MessageType,
         SendResult,
    +    _ssrf_redirect_guard,
         cache_document_from_bytes,
         cache_image_from_bytes,
     )
    ```

    ```diff
    @@ -226,7 +227,11 @@ async def connect(self) -> bool:
                 return False
     
             try:
    -            self._http_client = httpx.AsyncClient(timeout=30.0, follow_redirects=True)
    +            self._http_client = httpx.AsyncClient(
    +                timeout=30.0,
    +                follow_redirects=True,
    +                event_hooks={"response": [_ssrf_redirect_guard]},
    +            )
     
                 # 1. Get access token
                 await self._ensure_token()
    ```

    ```diff
    @@ -1101,6 +1106,11 @@ async def _stt_voice_attachment(
                 is_pre_wav = True
                 logger.info("[QQ] STT: using voice_wav_url (pre-converted WAV)")
     
    +        from tools.url_safety import is_safe_url
    +        if not is_safe_url(download_url):
    +            logger.warning("[QQ] STT blocked unsafe URL: %s", download_url[:80])
    +            return None
    +
             try:
                 # 2. Download audio (QQ CDN requires Authorization header)
                 if not self._http_client:
    ```

    ```diff
    @@ -1525,6 +1535,33 @@ async def _upload_media(
     
             raise last_exc  # type: ignore[misc]
     
    +    # Maximum time (seconds) to wait for reconnection before giving up on send.
    +    _RECONNECT_WAIT_SECONDS = 15.0
    +    # How often (seconds) to poll is_connected while waiting.
    +    _RECONNECT_POLL_INTERVAL = 0.5
    +
    +    async def _wait_for_reconnection(self) -> bool:
    +        """Wait for the WebSocket listener to reconnect.
    +
    +        The listener loop (_listen_loop) auto-reconnects on disconnect, but
    +        there is a race window where send() is called right after a disconnect
    +        and before the reconnect completes.  This method polls is_connected
    +        for up to _RECONNECT_WAIT_SECONDS.
    +
    +        Returns True if reconnected, False if still disconnected.
    +        """
    +        logger.info("[%s] Not connected — waiting for reconnection (up to %.0fs)",
    +                    self.name, self._RECONNECT_WAIT_SECONDS)
    +        waited = 0.0
    +        while waited < self._RECONNECT_WAIT_SECONDS:
    +            await asyncio.sleep(self._RECONNECT_POLL_INTERVAL)
    +            waited += self._RECONNECT_POLL_INTERVAL
    +            if self.is_connected:
    +                logger.info("[%s] Reconnected after %.1fs", self.name, waited)
    +                return True
    +        logger.warning("[%s] Still not connected after %.0fs", self.name, self._RECONNECT_WAIT_SECONDS)
    +        return False
    +
         async def send(
             self,
             chat_id: str,
    ```

    ```diff
    @@ -1540,7 +1577,8 @@ async def send(
             del metadata
     
             if not self.is_connected:
    -            return SendResult(success=False, error="Not connected")
    +            if not await self._wait_for_reconnection():
    +                return SendResult(success=False, error="Not connected", retryable=True)
     
             if not content or not content.strip():
                 return SendResult(success=True)
    ```

    ```diff
    @@ -1741,7 +1779,8 @@ async def _send_media(
         ) -> SendResult:
             """Upload media and send as a native message."""
             if not self.is_connected:
    -            return SendResult(success=False, error="Not connected")
    +            if not await self._wait_for_reconnection():
    +                return SendResult(success=False, error="Not connected", retryable=True)
     
             try:
                 # Resolve media source
    ```
  - `gateway/run.py`

    ```diff
    @@ -2618,6 +2618,9 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
                 Platform.BLUEBUBBLES: "BLUEBUBBLES_ALLOWED_USERS",
                 Platform.QQBOT: "QQ_ALLOWED_USERS",
             }
    +        platform_group_env_map = {
    +            Platform.QQBOT: "QQ_GROUP_ALLOWED_USERS",
    +        }
             platform_allow_all_map = {
                 Platform.TELEGRAM: "TELEGRAM_ALLOW_ALL_USERS",
                 Platform.DISCORD: "DISCORD_ALLOW_ALL_USERS",
    ```

    ```diff
    @@ -2649,12 +2652,23 @@ def _is_user_authorized(self, source: SessionSource) -> bool:
     
             # Check platform-specific and global allowlists
             platform_allowlist = os.getenv(platform_env_map.get(source.platform, ""), "").strip()
    +        group_allowlist = ""
    +        if source.chat_type == "group":
    +            group_allowlist = os.getenv(platform_group_env_map.get(source.platform, ""), "").strip()
             global_allowlist = os.getenv("GATEWAY_ALLOWED_USERS", "").strip()
     
    -        if not platform_allowlist and not global_allowlist:
    +        if not platform_allowlist and not group_allowlist and not global_allowlist:
                 # No allowlists configured -- check global allow-all flag
                 return os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in ("true", "1", "yes")
     
    +        # Some platforms authorize group traffic by chat ID rather than sender ID.
    +        if group_allowlist and source.chat_type == "group" and source.chat_id:
    +            allowed_group_ids = {
    +                chat_id.strip() for chat_id in group_allowlist.split(",") if chat_id.strip()
    +            }
    +            if "*" in allowed_group_ids or source.chat_id in allowed_group_ids:
    +                return True
    +
             # Check if user is in any allowlist
             allowed_ids = set()
             if platform_allowlist:
    ```
  - `scripts/release.py`

    ```diff
    @@ -96,6 +96,7 @@
         "[redacted:email]": "wakamex",
         "[redacted:email]": "clawdia-nash",
         "[redacted:email]": "austinpickett",
    +    "[redacted:email]": "dieutx",
         "[redacted:email]": "jaisup",
         "[redacted:email]": "PercyDikec",
         "[redacted:email]": "deankerr",
    ```

    ```diff
    @@ -235,6 +236,7 @@
         "[redacted:email]": "malaiwah",
         "[redacted:email]": "v1k22",
         "[redacted:email]": "RucchiZ",
    +    "[redacted:email]": "LehaoLin",
     }
     
     
    ```
  - `tests/gateway/test_qqbot.py`

    ```diff
    @@ -1,5 +1,6 @@
     """Tests for the QQ Bot platform adapter."""
     
    +import asyncio
     import json
     import os
     import sys
    ```

    ```diff
    @@ -149,6 +150,47 @@ def test_audio_extension_amr(self):
             assert self._fn("", "recording.amr") is True
     
     
    +# ---------------------------------------------------------------------------
    +# Voice attachment SSRF protection
    +# ---------------------------------------------------------------------------
    +
    +class TestVoiceAttachmentSSRFProtection:
    +    def _make_adapter(self, **extra):
    +        from gateway.platforms.qqbot import QQAdapter
    +        return QQAdapter(_make_config(**extra))
    +
    +    def test_stt_blocks_unsafe_download_url(self):
    +        adapter = self._make_adapter(app_id="a", client_secret="b")
    +        adapter._http_client = mock.AsyncMock()
    +
    +        with mock.patch("tools.url_safety.is_safe_url", return_value=False):
    +            transcript = asyncio.run(
    +                adapter._stt_voice_attachment(
    +                    "http://127.0.0.1/voice.silk",
    +                    "audio/silk",
    +                    "voice.silk",
    +                )
    +            )
    +
    +        assert transcript is None
    +        adapter._http_client.get.assert_not_called()
    +
    +    def test_connect_uses_redirect_guard_hook(self):
    +        from gateway.platforms.qqbot import QQAdapter, _ssrf_redirect_guard
    +
    +        client = mock.AsyncMock()
    +        with mock.patch("gateway.platforms.qqbot.httpx.AsyncClient", return_value=client) as async_client_cls:
    +            adapter = QQAdapter(_make_config(app_id="a", client_secret="b"))
    +            adapter._ensure_token = [redacted:secret-assignment]"stop after client creation"))
    +
    +            connected = asyncio.run(adapter.connect())
    +
    +        assert connected is False
    +        assert async_client_cls.call_count == 1
    +        kwargs = async_client_cls.call_args.kwargs
    +        assert kwargs.get("follow_redirects") is True
    +        assert kwargs.get("event_hooks", {}).get("response") == [_ssrf_redirect_guard]
    +
     # ---------------------------------------------------------------------------
     # _strip_at_mention
     # ---------------------------------------------------------------------------
    ```

    ```diff
    @@ -458,3 +500,85 @@ def test_reply_to(self):
             adapter = self._make_adapter(app_id="a", client_secret="b", markdown_support=False)
             body = adapter._build_text_body("reply text", reply_to="msg_123")
             assert body.get("message_reference", {}).get("message_id") == "msg_123"
    +
    +
    +# ---------------------------------------------------------------------------
    +# _wait_for_reconnection / send reconnection wait
    +# ---------------------------------------------------------------------------
    +
    +class TestWaitForReconnection:
    +    """Test that send() waits for reconnection instead of silently dropping."""
    +
    +    def _make_adapter(self, **extra):
    +        from gateway.platforms.qqbot import QQAdapter
    +        return QQAdapter(_make_config(**extra))
    +
    +    @pytest.mark.asyncio
    +    async def test_send_waits_and_succeeds_on_reconnect(self):
    +        """send() should wait for reconnection and then deliver the message."""
    +        adapter = self._make_adapter(app_id="a", client_secret="b")
    +        # Initially disconnected
    +        adapter._running = False
    +        adapter._http_client = mock.MagicMock()
    +
    +        # Simulate reconnection after 0.3s (faster than real interval)
    +        async def fake_api_request(*args, **kwargs):
    +            return {"id": "msg_123"}
    +
    +        adapter._api_request = fake_api_request
    +        adapter._ensure_token = [redacted:secret-assignment]
    +        adapter._RECONNECT_POLL_INTERVAL = 0.1
    +        adapter._RECONNECT_WAIT_SECONDS = 5.0
    +
    +        # Schedule reconnection after a short delay
    +        async def reconnect_after_delay():
    +            await asyncio.sleep(0.3)
    +            adapter._running = True
    +
    +        asyncio.get_event_loop().create_task(reconnect_after_delay())
    +
    +        result = await adapter.send("test_openid", "Hello, world!")
    +        assert result.success
    +        assert result.message_id == "msg_123"
    +
    +    @pytest.mark.asyncio
    +    async def test_send_returns_retryable_after_timeout(self):
    +        """send() should return retryable=True if reconnection takes too long."""
    +        adapter = self._make_adapter(app_id="a", client_secret="b")
    +        adapter._running = False
    +        adapter._RECONNECT_POLL_INTERVAL = 0.05
    +        adapter._RECONNECT_WAIT_SECONDS = 0.2
    +
    +        result = await adapter.send("test_openid", "Hello, world!")
    +        assert not result.success
    +        assert result.retryable is True
    +        assert "Not connected" in result.error
    +
    +    @pytest.mark.asyncio
    +    async def test_send_succeeds_immediately_when_connected(self):
    +        """send() should not wait when already connected."""
    +        adapter = self._make_adapter(app_id="a", client_secret="b")
    +        adapter._running = True
    +        adapter._http_client = mock.MagicMock()
    +
    +        async def fake_api_request(*args, **kwargs):
    +            return {"id": "msg_immediate"}
    +
    +        adapter._api_request = fake_api_request
    +
    +        result = await adapter.send("test_openid", "Hello!")
    +        assert result.success
    +        assert result.message_id == "msg_immediate"
    +
    +    @pytest.mark.asyncio
    +    async def test_send_media_waits_for_reconnect(self):
    +        """_send_media should also wait for reconnection."""
    +        adapter = self._make_adapter(app_id="a", client_secret="b")
    +        adapter._running = False
    +        adapter._RECONNECT_POLL_INTERVAL = 0.05
    +        adapter._RECONNECT_WAIT_SECONDS = 0.2
    +
    +        result = await adapter._send_media("test_openid", "http://example.com/img.jpg", 1, "image")
    +        assert not result.success
    +        assert result.retryable is True
    +        assert "Not connected" in result.error
    ```
  - `tests/gateway/test_unauthorized_dm_behavior.py`

    ```diff
    @@ -21,6 +21,7 @@ def _clear_auth_env(monkeypatch) -> None:
             "MATTERMOST_ALLOWED_USERS",
             "MATRIX_ALLOWED_USERS",
             "DINGTALK_ALLOWED_USERS", "FEISHU_ALLOWED_USERS", "WECOM_ALLOWED_USERS",
    +        "QQ_ALLOWED_USERS", "QQ_GROUP_ALLOWED_USERS",
             "GATEWAY_ALLOWED_USERS",
             "TELEGRAM_ALLOW_ALL_USERS",
             "DISCORD_ALLOW_ALL_USERS",
    ```

    ```diff
    @@ -32,6 +33,7 @@ def _clear_auth_env(monkeypatch) -> None:
             "MATTERMOST_ALLOW_ALL_USERS",
             "MATRIX_ALLOW_ALL_USERS",
             "DINGTALK_ALLOW_ALL_USERS", "FEISHU_ALLOW_ALL_USERS", "WECOM_ALLOW_ALL_USERS",
    +        "QQ_ALLOW_ALL_USERS",
             "GATEWAY_ALLOW_ALL_USERS",
         ):
             monkeypatch.delenv(key, raising=False)
    ```

    ```diff
    @@ -130,6 +132,46 @@ def test_star_wildcard_works_for_any_platform(monkeypatch):
         assert runner._is_user_authorized(source) is True
     
     
    +def test_qq_group_allowlist_authorizes_group_chat_without_user_allowlist(monkeypatch):
    +    _clear_auth_env(monkeypatch)
    +    monkeypatch.setenv("QQ_GROUP_ALLOWED_USERS", "group-openid-1")
    +
    +    runner, _adapter = _make_runner(
    +        Platform.QQBOT,
    +        GatewayConfig(platforms={Platform.QQBOT: PlatformConfig(enabled=True)}),
    +    )
    +
    +    source = SessionSource(
    +        platform=Platform.QQBOT,
    +        user_id="member-openid-999",
    +        chat_id="group-openid-1",
    +        user_name="tester",
    +        chat_type="group",
    +    )
    +
    +    assert runner._is_user_authorized(source) is True
    +
    +
    +def test_qq_group_allowlist_does_not_authorize_other_groups(monkeypatch):
    +    _clear_auth_env(monkeypatch)
    +    monkeypatch.setenv("QQ_GROUP_ALLOWED_USERS", "group-openid-1")
    +
    +    runner, _adapter = _make_runner(
    +        Platform.QQBOT,
    +        GatewayConfig(platforms={Platform.QQBOT: PlatformConfig(enabled=True)}),
    +    )
    +
    +    source = SessionSource(
    +        platform=Platform.QQBOT,
    +        user_id="member-openid-999",
    +        chat_id="group-openid-2",
    +        user_name="tester",
    +        chat_type="group",
    +    )
    +
    +    assert runner._is_user_authorized(source) is False
    +
    +
     @pytest.mark.asyncio
     async def test_unauthorized_dm_pairs_by_default(monkeypatch):
         _clear_auth_env(monkeypatch)
    ```
  - `tests/tools/test_url_safety.py`

    ```diff
    @@ -152,6 +152,34 @@ def test_non_cgnat_100_allowed(self):
                 # 100.0.0.1 is a global IP, not in CGNAT range
                 assert is_safe_url("http://legit-host.example/") is True
     
    +    def test_benchmark_ip_blocked_for_non_allowlisted_host(self):
    +        with patch("socket.getaddrinfo", return_value=[
    +            (2, 1, 6, "", ("198.18.0.23", 0)),
    +        ]):
    +            assert is_safe_url("https://example.com/file.jpg") is False
    +
    +    def test_qq_multimedia_hostname_allowed_with_benchmark_ip(self):
    +        with patch("socket.getaddrinfo", return_value=[
    +            (2, 1, 6, "", ("198.18.0.23", 0)),
    +        ]):
    +            assert is_safe_url("https://multimedia.nt.qq.com.cn/download?id=123") is True
    +
    +    def test_qq_multimedia_hostname_exception_is_exact_match(self):
    +        with patch("socket.getaddrinfo", return_value=[
    +            (2, 1, 6, "", ("198.18.0.23", 0)),
    +        ]):
    +            assert is_safe_url("https://sub.multimedia.nt.qq.com.cn/download?id=123") is False
    +
    +    def test_qq_multimedia_hostname_exception_requires_https(self):
    +        with patch("socket.getaddrinfo", return_value=[
    +            (2, 1, 6, "", ("198.18.0.23", 0)),
    +        ]):
    +            assert is_safe_url("http://multimedia.nt.qq.com.cn/download?id=123") is False
    +
    +    def test_qq_multimedia_hostname_dns_failure_still_blocked(self):
    +        with patch("socket.getaddrinfo", side_effect=socket.gaierror("Name resolution failed")):
    +            assert is_safe_url("https://multimedia.nt.qq.com.cn/download?id=123") is False
    +
     
     class TestIsBlockedIp:
         """Direct tests for the _is_blocked_ip helper."""
    ```

    ```diff
    @@ -159,7 +187,7 @@ class TestIsBlockedIp:
         @pytest.mark.parametrize("ip_str", [
             "127.0.0.1", "10.0.0.1", "172.16.0.1", "192.168.1.1",
             "169.254.169.254", "0.0.0.0", "224.0.0.1", "255.255.255.255",
    -        "100.64.0.1", "100.100.100.100", "100.127.255.254",
    +        "100.64.0.1", "100.100.100.100", "100.127.255.254", "198.18.0.23",
             "::1", "fe80::1", "fc00::1", "fd12::1", "ff02::1",
             "::ffff:127.0.0.1", "::ffff:169.254.169.254",
         ])
    ```
  - `tools/url_safety.py`

    ```diff
    @@ -29,6 +29,13 @@
         "metadata.goog",
     })
     
    +# Exact HTTPS hostnames allowed to resolve to private/benchmark-space IPs.
    +# This is intentionally narrow: QQ media downloads can legitimately resolve
    +# to 198.18.0.0/15 behind local proxy/benchmark infrastructure.
    +_TRUSTED_PRIVATE_IP_HOSTS = frozenset({
    +    "multimedia.nt.qq.com.cn",
    +})
    +
     # 100.64.0.0/10 (CGNAT / Shared Address Space, RFC 6598) is NOT covered by
     # ipaddress.is_private — it returns False for both is_private and is_global.
     # Must be blocked explicitly. Used by carrier-grade NAT, Tailscale/WireGuard
    ```

    ```diff
    @@ -48,6 +55,11 @@ def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
         return False
     
     
    +def _allows_private_ip_resolution(hostname: str, scheme: str) -> bool:
    +    """Return True when a trusted HTTPS hostname may bypass IP-class blocking."""
    +    return scheme == "https" and hostname in _TRUSTED_PRIVATE_IP_HOSTS
    +
    +
     def is_safe_url(url: str) -> bool:
         """Return True if the URL target is not a private/internal address.
     
    ```

    ```diff
    @@ -56,7 +68,8 @@ def is_safe_url(url: str) -> bool:
         """
         try:
             parsed = urlparse(url)
    -        hostname = (parsed.hostname or "").strip().lower()
    +        hostname = (parsed.hostname or "").strip().lower().rstrip(".")
    +        scheme = (parsed.scheme or "").strip().lower()
             if not hostname:
                 return False
     
    ```

    ```diff
    @@ -65,6 +78,8 @@ def is_safe_url(url: str) -> bool:
                 logger.warning("Blocked request to internal hostname: %s", hostname)
                 return False
     
    +        allow_private_ip = _allows_private_ip_resolution(hostname, scheme)
    +
             # Try to resolve and check IP
             try:
                 addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
    ```

    ```diff
    @@ -81,13 +96,19 @@ def is_safe_url(url: str) -> bool:
                 except ValueError:
                     continue
     
    -            if _is_blocked_ip(ip):
    +            if not allow_private_ip and _is_blocked_ip(ip):
                     logger.warning(
                         "Blocked request to private/internal address: %s -> %s",
                         hostname, ip_str,
                     )
                     return False
     
    +        if allow_private_ip:
    +            logger.debug(
    +                "Allowing trusted hostname despite private/internal resolution: %s",
    +                hostname,
    +            )
    +
             return True
     
         except Exception as exc:
    ```
- **#11568: fix(discord): route attachment downloads through authenticated bot session**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `gateway/platforms/discord.py`, `tests/gateway/test_discord_attachment_download.py`
  - fix hunks:
  - `gateway/platforms/discord.py`

    ```diff
    @@ -51,7 +51,9 @@
         ProcessingOutcome,
         SendResult,
         cache_image_from_url,
    +    cache_image_from_bytes,
         cache_audio_from_url,
    +    cache_audio_from_bytes,
         cache_document_from_bytes,
         SUPPORTED_DOCUMENT_TYPES,
     )
    ```

    ```diff
    @@ -2489,6 +2491,124 @@ def _format_thread_chat_name(self, thread: Any) -> str:
                 return f"{parent_name} / {thread_name}"
             return thread_name
     
    +    # ------------------------------------------------------------------
    +    # Attachment download helpers
    +    #
    +    # Discord attachments (images / audio / documents) are fetched via the
    +    # authenticated bot session whenever the Attachment object exposes
    +    # ``read()``. That sidesteps two classes of bug that hit the older
    +    # plain-HTTP path:
    +    #
    +    #   1. ``cdn.discordapp.com`` URLs increasingly require bot auth on
    +    #      download — unauthenticated httpx sees 403 Forbidden.
    +    #      (issue #8242)
    +    #   2. Some user environments (VPNs, corporate DNS, tunnels) resolve
    +    #      ``cdn.discordapp.com`` to private-looking IPs that our
    +    #      ``is_safe_url`` guard classifies as SSRF risks. Routing the
    +    #      fetch through discord.py's own HTTP client handles DNS
    +    #      internally so our guard isn't consulted for the attachment
    +    #      path. (issue #6587)
    +    #
    +    # If ``att.read()`` is unavailable (unexpected object shape / test
    +    # stub) or the bot session fetch fails, we fall back to the existing
    +    # SSRF-gated URL downloaders. The fallback keeps defense-in-depth
    +    # against any future Discord payload-schema drift that could slip a
    +    # non-CDN URL into the ``att.url`` field. (issue #11345)
    +    # ------------------------------------------------------------------
    +
    +    async def _read_attachment_bytes(self, att) -> Optional[bytes]:
    +        """Read an attachment via discord.py's authenticated bot session.
    +
    +        Returns the raw bytes on success, or ``None`` if ``att`` doesn't
    +        expose a callable ``read()`` or the read itself fails. Callers
    +        should treat ``None`` as a signal to fall back to the URL-based
    +        downloaders.
    +        """
    +        reader = getattr(att, "read", None)
    +        if reader is None or not callable(reader):
    +            return None
    +        try:
    +            return await reader()
    +        except Exception as e:
    +            logger.warning(
    +                "[Discord] Authenticated attachment read failed for %s: %s",
    +                getattr(att, "filename", None) or getattr(att, "url", "<unknown>"),
    +                e,
    +            )
    +            return None
    +
    +    async def _cache_discord_image(self, att, ext: str) -> str:
    +        """Cache a Discord image attachment to local disk.
    +
    +        Primary path: ``att.read()`` + ``cache_image_from_bytes``
    +        (authenticated, no SSRF gate).
    +
    +        Fallback: ``cache_image_from_url`` (plain httpx, SSRF-gated).
    +        """
    +        raw_bytes = await self._read_attachment_bytes(att)
    +        if raw_bytes is not None:
    +            try:
    +                return cache_image_from_bytes(raw_bytes, ext=ext)
    +            except Exception as e:
    +                logger.debug(
    +                    "[Discord] cache_image_from_bytes rejected att.read() data; falling back to URL: %s",
    +                    e,
    +                )
    +        return await cache_image_from_url(att.url, ext=ext)
    +
    +    async def _cache_discord_audio(self, att, ext: str) -> str:
    +        """Cache a Discord audio attachment to local disk.
    +
    +        Primary path: ``att.read()`` + ``cache_audio_from_bytes``
    +        (authenticated, no SSRF gate).
    +
    +        Fallback: ``cache_audio_from_url`` (plain httpx, SSRF-gated).
    +        """
    +        raw_bytes = await self._read_attachment_bytes(att)
    +        if raw_bytes is not None:
    +            try:
    +                return cache_audio_from_bytes(raw_bytes, ext=ext)
    +            except Exception as e:
    +                logger.debug(
    +                    "[Discord] cache_audio_from_bytes failed; falling back to URL: %s",
    +                    e,
    +                )
    +        return await cache_audio_from_url(att.url, ext=ext)
    +
    +    async def _cache_discord_document(self, att, ext: str) -> bytes:
    +        """Download a Discord document attachment and return the raw bytes.
    +
    +        Primary path: ``att.read()`` (authenticated, no SSRF gate).
    +
    +        Fallback: SSRF-gated ``aiohttp`` download. This closes the gap
    +        where the old document path made raw ``aiohttp.ClientSession``
    +        requests with no safety check (#11345). The caller is responsible
    +        for passing the returned bytes to ``cache_document_from_bytes``
    +        (and, where applicable, for injecting text content).
    +        """
    +        raw_bytes = await self._read_attachment_bytes(att)
    +        if raw_bytes is not None:
    +            return raw_bytes
    +
    +        # Fallback: SSRF-gated URL download.
    +        if not is_safe_url(att.url):
    +            raise ValueError(
    +                f"Blocked unsafe attachment URL (SSRF protection): {att.url}"
    +            )
    +        import aiohttp
    +        from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp
    +        _proxy = resolve_proxy_url(platform_env_var="DISCORD_PROXY")
    +        _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(_proxy)
    +        async with aiohttp.ClientSession(**_sess_kw) as session:
    +            async with session.get(
    +                att.url,
    +                timeout=aiohttp.ClientTimeout(total=30),
    +                **_req_kw,
    +            ) as resp:
    +                if resp.status != 200:
    +                    raise Exception(f"HTTP {resp.status}")
    +                return await resp.read()
    +
         async def _handle_message(self, message: DiscordMessage) -> None:
             """Handle incoming Discord messages."""
             # In server channels (not DMs), require the bot to be @mentioned
    ```

    ```diff
    @@ -2642,7 +2762,7 @@ 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)
    +                    cached_path = await self._cache_discord_image(att, ext)
                         media_urls.append(cached_path)
                         media_types.append(content_type)
                         print(f"[Discord] Cached user image: {cached_path}", flush=True)
    ```

    ```diff
    @@ -2656,7 +2776,7 @@ 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 self._cache_discord_audio(att, ext)
                         media_urls.append(cached_path)
                         media_types.append(content_type)
                         print(f"[Discord] Cached user audio: {cached_path}", flush=True)
    ```

    ```diff
    @@ -2687,19 +2807,7 @@ async def _handle_message(self, message: DiscordMessage) -> None:
                             )
                         else:
                             try:
    -                            import aiohttp
    -                            from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp
    -                            _proxy = resolve_proxy_url(platform_env_var="DISCORD_PROXY")
    -                            _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(_proxy)
    -                            async with aiohttp.ClientSession(**_sess_kw) as session:
    -                                async with session.get(
    -                                    att.url,
    -                                    timeout=aiohttp.ClientTimeout(total=30),
    -                                    **_req_kw,
    -                                ) as resp:
    -                                    if resp.status != 200:
    -                                        raise Exception(f"HTTP {resp.status}")
    -                                    raw_bytes = await resp.read()
    +                            raw_bytes = await self._cache_discord_document(att, ext)
                                 cached_path = cache_document_from_bytes(
                                     raw_bytes, att.filename or f"document{ext}"
                                 )
    ```
  - `tests/gateway/test_discord_attachment_download.py`

    ```diff
    @@ -0,0 +1,360 @@
    +"""Tests for Discord attachment downloads via the authenticated bot session.
    +
    +Covers the three download paths (image / audio / document) in
    +``DiscordAdapter._handle_message()`` and the shared ``_cache_discord_*``
    +helpers. Verifies that:
    +
    +- ``att.read()`` is preferred over the legacy URL-based downloaders so
    +  that Discord's CDN auth (and user-environment DNS quirks) can't block
    +  media caching. (issues #8242 image 403s, #6587 CDN SSRF false-positives)
    +- Falls back cleanly to the SSRF-gated ``cache_*_from_url`` helpers
    +  (image/audio) or SSRF-gated aiohttp (documents) when ``att.read()``
    +  isn't available or fails.
    +- The document fallback path now runs through the SSRF gate for
    +  defense-in-depth. (issue #11345)
    +"""
    +
    +import sys
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +from gateway.config import PlatformConfig
    +
    +
    +def _ensure_discord_mock():
    +    """Install a mock discord module when discord.py isn't available."""
    +    if "discord" in sys.modules and hasattr(sys.modules["discord"], "__file__"):
    +        return
    +
    +    discord_mod = MagicMock()
    +    discord_mod.Intents.default.return_value = MagicMock()
    +    discord_mod.Client = MagicMock
    +    discord_mod.File = MagicMock
    +    discord_mod.DMChannel = type("DMChannel", (), {})
    +    discord_mod.Thread = type("Thread", (), {})
    +    discord_mod.ForumChannel = type("ForumChannel", (), {})
    +    discord_mod.ui = SimpleNamespace(View=object, button=lambda *a, **k: (lambda fn: fn), Button=object)
    +    discord_mod.ButtonStyle = SimpleNamespace(success=1, primary=2, secondary=2, danger=3, green=1, grey=2, blurple=2, red=3)
    +    discord_mod.Color = SimpleNamespace(orange=lambda: 1, green=lambda: 2, blue=lambda: 3, red=lambda: 4, purple=lambda: 5)
    +    discord_mod.Interaction = object
    +    discord_mod.Embed = MagicMock
    +    discord_mod.app_commands = SimpleNamespace(
    +        describe=lambda **kwargs: (lambda fn: fn),
    +        choices=lambda **kwargs: (lambda fn: fn),
    +        Choice=lambda **kwargs: SimpleNamespace(**kwargs),
    +    )
    +
    +    ext_mod = MagicMock()
    +    commands_mod = MagicMock()
    +    commands_mod.Bot = MagicMock
    +    ext_mod.commands = commands_mod
    +
    +    sys.modules.setdefault("discord", discord_mod)
    +    sys.modules.setdefault("discord.ext", ext_mod)
    +    sys.modules.setdefault("discord.ext.commands", commands_mod)
    +
    +
    +_ensure_discord_mock()
    +
    +from gateway.platforms.discord import DiscordAdapter  # noqa: E402
    +
    +
    +# Minimal valid image / audio / PDF bytes so the cache_*_from_bytes
    +# validators accept them. cache_image_from_bytes runs _looks_like_image()
    +# which checks for magic bytes; PNG's magic is sufficient.
    +_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
    +_OGG_BYTES = b"OggS" + b"\x00" * 60
    +_PDF_BYTES = b"%PDF-1.4\n" + b"fake pdf body" + b"\n%%EOF"
    +
    +
    +def _make_adapter() -> DiscordAdapter:
    +    return DiscordAdapter(PlatformConfig(enabled=True, token="***"))
    +
    +
    +def _make_attachment_with_read(payload: bytes) -> SimpleNamespace:
    +    """Attachment stub that exposes .read() — the happy-path primary."""
    +    return SimpleNamespace(
    +        url="https://cdn.discordapp.com/attachments/fake/file.png",
    +        filename="file.png",
    +        size=len(payload),
    +        read=AsyncMock(return_value=payload),
    +    )
    +
    +
    +def _make_attachment_without_read() -> SimpleNamespace:
    +    """Attachment stub that has no .read() — exercises the URL fallback."""
    +    return SimpleNamespace(
    +        url="https://cdn.discordapp.com/attachments/fake/file.png",
    +        filename="file.png",
    +        size=1024,
    +    )
    +
    +
    +# ---------------------------------------------------------------------------
    +# _read_attachment_bytes
    +# ---------------------------------------------------------------------------
    +
    +class TestReadAttachmentBytes:
    +    """Unit tests for the low-level att.read() wrapper."""
    +
    +    @pytest.mark.asyncio
    +    async def test_returns_bytes_on_successful_read(self):
    +        adapter = _make_adapter()
    +        att = _make_attachment_with_read(b"hello world")
    +
    +        result = await adapter._read_attachment_bytes(att)
    +
    +        assert result == b"hello world"
    +        att.read.assert_awaited_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_returns_none_when_read_missing(self):
    +        adapter = _make_adapter()
    +        att = _make_attachment_without_read()
    +
    +        result = await adapter._read_attachment_bytes(att)
    +
    +        assert result is None
    +
    +    @pytest.mark.asyncio
    +    async def test_returns_none_when_read_raises(self):
    +        """Bot-session fetch failures are swallowed so callers fall back."""
    +        adapter = _make_adapter()
    +        att = SimpleNamespace(
    +            url="https://cdn.discordapp.com/attachments/fake/file.png",
    +            filename="file.png",
    +            read=AsyncMock(side_effect=RuntimeError("403 Forbidden")),
    +        )
    +
    +        result = await adapter._read_attachment_bytes(att)
    +
    +        assert result is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# _cache_discord_image
    +# ---------------------------------------------------------------------------
    +
    +class TestCacheDiscordImage:
    +    @pytest.mark.asyncio
    +    async def test_prefers_att_read_over_url(self):
    +        """Primary path: att.read() bytes → cache_image_from_bytes, no URL fetch."""
    +        adapter = _make_adapter()
    +        att = _make_attachment_with_read(_PNG_BYTES)
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_image_from_bytes",
    +            return_value="/tmp/cached.png",
    +        ) as mock_bytes, patch(
    +            "gateway.platforms.discord.cache_image_from_url",
    +            new_callable=AsyncMock,
    +        ) as mock_url:
    +            result = await adapter._cache_discord_image(att, ".png")
    +
    +        assert result == "/tmp/cached.png"
    +        mock_bytes.assert_called_once_with(_PNG_BYTES, ext=".png")
    +        mock_url.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_falls_back_to_url_when_no_read(self):
    +        """No .read() → URL path is used (existing SSRF-gated behavior)."""
    +        adapter = _make_adapter()
    +        att = _make_attachment_without_read()
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_image_from_bytes",
    +        ) as mock_bytes, patch(
    +            "gateway.platforms.discord.cache_image_from_url",
    +            new_callable=AsyncMock,
    +            return_value="/tmp/from_url.png",
    +        ) as mock_url:
    +            result = await adapter._cache_discord_image(att, ".png")
    +
    +        assert result == "/tmp/from_url.png"
    +        mock_bytes.assert_not_called()
    +        mock_url.assert_awaited_once_with(att.url, ext=".png")
    +
    +    @pytest.mark.asyncio
    +    async def test_falls_back_to_url_when_bytes_validator_rejects(self):
    +        """If att.read() returns garbage that cache_image_from_bytes rejects
    +        (e.g. an HTML error page), fall back to the URL downloader instead
    +        of surfacing the validation error to the caller."""
    +        adapter = _make_adapter()
    +        att = _make_attachment_with_read(b"<html>forbidden</html>")
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_image_from_bytes",
    +            side_effect=ValueError("not a valid image"),
    +        ), patch(
    +            "gateway.platforms.discord.cache_image_from_url",
    +            new_callable=AsyncMock,
    +            return_value="/tmp/fallback.png",
    +        ) as mock_url:
    +            result = await adapter._cache_discord_image(att, ".png")
    +
    +        assert result == "/tmp/fallback.png"
    +        mock_url.assert_awaited_once()
    +
    +
    +# ---------------------------------------------------------------------------
    +# _cache_discord_audio
    +# ---------------------------------------------------------------------------
    +
    +class TestCacheDiscordAudio:
    +    @pytest.mark.asyncio
    +    async def test_prefers_att_read_over_url(self):
    +        adapter = _make_adapter()
    +        att = _make_attachment_with_read(_OGG_BYTES)
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_audio_from_bytes",
    +            return_value="/tmp/voice.ogg",
    +        ) as mock_bytes, patch(
    +            "gateway.platforms.discord.cache_audio_from_url",
    +            new_callable=AsyncMock,
    +        ) as mock_url:
    +            result = await adapter._cache_discord_audio(att, ".ogg")
    +
    +        assert result == "/tmp/voice.ogg"
    +        mock_bytes.assert_called_once_with(_OGG_BYTES, ext=".ogg")
    +        mock_url.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_falls_back_to_url_when_no_read(self):
    +        adapter = _make_adapter()
    +        att = _make_attachment_without_read()
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_audio_from_url",
    +            new_callable=AsyncMock,
    +            return_value="/tmp/from_url.ogg",
    +        ) as mock_url:
    +            result = await adapter._cache_discord_audio(att, ".ogg")
    +
    +        assert result == "/tmp/from_url.ogg"
    +        mock_url.assert_awaited_once_with(att.url, ext=".ogg")
    +
    +
    +# ---------------------------------------------------------------------------
    +# _cache_discord_document
    +# ---------------------------------------------------------------------------
    +
    +class TestCacheDiscordDocument:
    +    @pytest.mark.asyncio
    +    async def test_prefers_att_read_returns_bytes_directly(self):
    +        """Primary path: att.read() → raw bytes, no aiohttp involvement."""
    +        adapter = _make_adapter()
    +        att = _make_attachment_with_read(_PDF_BYTES)
    +
    +        with patch("aiohttp.ClientSession") as mock_session:
    +            result = await adapter._cache_discord_document(att, ".pdf")
    +
    +        assert result == _PDF_BYTES
    +        mock_session.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_fallback_blocked_by_ssrf_guard(self):
    +        """Document fallback path now honors is_safe_url — was missing before.
    +
    +        Regression guard for #11345: the old aiohttp block skipped the
    +        SSRF check entirely; a non-CDN ``att.url`` could have reached
    +        internal-looking hosts. The fallback must now refuse unsafe URLs.
    +        """
    +        adapter = _make_adapter()
    +        att = _make_attachment_without_read()  # no .read → forces fallback
    +
    +        with patch(
    +            "gateway.platforms.discord.is_safe_url", return_value=False
    +        ) as mock_safe, patch("aiohttp.ClientSession") as mock_session:
    +            with pytest.raises(ValueError, match="SSRF"):
    +                await adapter._cache_discord_document(att, ".pdf")
    +
    +        mock_safe.assert_called_once_with(att.url)
    +        # aiohttp must NOT be contacted when the URL is blocked.
    +        mock_session.assert_not_called()
    +
    +    @pytest.mark.asyncio
    +    async def test_fallback_aiohttp_when_safe_url(self):
    +        """Safe URL + no att.read() → aiohttp fallback executes."""
    +        adapter = _make_adapter()
    +        att = _make_attachment_without_read()
    +
    +        # Build an aiohttp session mock that returns 200 + payload.
    +        resp = AsyncMock()
    +        resp.status = 200
    +        resp.read = AsyncMock(return_value=_PDF_BYTES)
    +        resp.__aenter__ = AsyncMock(return_value=resp)
    +        resp.__aexit__ = AsyncMock(return_value=False)
    +
    +        session = AsyncMock()
    +        session.get = MagicMock(return_value=resp)
    +        session.__aenter__ = AsyncMock(return_value=session)
    +        session.__aexit__ = AsyncMock(return_value=False)
    +
    +        with patch(
    +            "gateway.platforms.discord.is_safe_url", return_value=True
    +        ), patch("aiohttp.ClientSession", return_value=session):
    +            result = await adapter._cache_discord_document(att, ".pdf")
    +
    +        assert result == _PDF_BYTES
    +
    +
    +# ---------------------------------------------------------------------------
    +# Integration: end-to-end via _handle_message
    +# ---------------------------------------------------------------------------
    +
    +class TestHandleMessageUsesAuthenticatedRead:
    +    """E2E: verify _handle_message routes image/audio downloads through
    +    att.read() so cdn.discordapp.com 403s (#8242) and SSRF false-positives
    +    on mangled DNS (#6587) no longer block media caching.
    +    """
    +
    +    @pytest.mark.asyncio
    +    async def test_image_downloads_via_att_read_not_url(self, monkeypatch):
    +        """Image attachments with .read() never call cache_image_from_url."""
    +        adapter = _make_adapter()
    +        adapter._client = SimpleNamespace(user=SimpleNamespace(id=999))
    +        adapter.handle_message = AsyncMock()
    +
    +        with patch(
    +            "gateway.platforms.discord.cache_image_from_bytes",
    +            return_value="/tmp/img_from_read.png",
    +        ), patch(
    +            "gateway.platforms.discord.cache_image_from_url",
    +            new_callable=AsyncMock,
    +        ) as mock_url_download:
    +            att = SimpleNamespace(
    +                url="https://cdn.discordapp.com/attachments/fake/file.png",
    +                filename="file.png",
    +                content_type="image/png",
    +                size=len(_PNG_BYTES),
    +                read=AsyncMock(return_value=_PNG_BYTES),
    +            )
    +            # Minimal Discord message stub for _handle_message.
    +            from datetime import datetime, timezone
    +
    +            class _FakeDMChannel:
    +                id = 100
    +                name = "dm"
    +
    +            # Patch the DMChannel isinstance check so our fake counts as DM.
    +            monkeypatch.setattr(
    +                "gateway.platforms.discord.discord.DMChannel",
    +                _FakeDMChannel,
    +            )
    +            chan = _FakeDMChannel()
    +            msg = SimpleNamespace(
    +                id=1, content="", attachments=[att], mentions=[],
    +                reference=None,
    +                created_at=datetime.now(timezone.utc),
    +                channel=chan,
    +                author=SimpleNamespace(id=42, display_name="U", name="U"),
    +            )
    +            await adapter._handle_message(msg)
    +
    +        mock_url_download.assert_not_called()
    +        event = adapter.handle_message.call_args[0][0]
    +        assert event.media_urls == ["/tmp/img_from_read.png"]
    +        assert event.media_types == ["image/png"]
    ```
- **#12256: fix(url_safety): allow NAT64 well-known prefix (RFC 6052)**. closed without merge · fix tier `suggested`
  - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#51758`, `#51758`
  - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._
- **#14166: feat(security): global toggle to allow private/internal URL resolution**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `hermes_cli/config.py`, `tests/tools/test_url_safety.py`, `tools/url_safety.py`
- **#38071: fix(url_safety): handle DNS64/NAT64 synthesized addresses in SSRF check**. closed without merge · fix tier `suggested`
  - maintainer direction: category `uncategorized`; implication `needs_llm`; refs `#38058`, `#12256`
  - _suggested-tier fix detail withheld; re-request with `?tier=all` to include it._
- **#3779: fix(security): allow RFC 2544 range bypass for TUN-mode proxy / Fake-IP users          **. closed without merge
  - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#14166`
- **#3833: fix(security): allow RFC 2544 benchmark range for TUN/Fake-IP proxy users**. closed without merge
  - maintainer direction: category `stale_abandoned`; implication `active`
- **#6512: fix(gateway): bypass SSRF check for Discord SDK-supplied attachment URLs**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `53da34a4f`, `53da34a4fc2e408d3545b607b3c0b298a6d21e84`, `#11568`, `#6587`, `#6511`, `#11568`
- **#21386: test(gateway): mock SSRF URL guard in Discord document handling tests**. open
  - maintainer direction: category `uncategorized`; implication `needs_llm`; refs `cc8e5ec2a`
  - files: `cli.py`, `tests/gateway/test_discord_document_handling.py`
- **#36700: feat(security): make trusted private-IP hosts configurable via security.trusted_private_hosts**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`
- **#38058: fix(url_safety): evaluate embedded IPv4 for NAT64 64:ff9b::/96 addresses**. open
  - maintainer direction: category `uncategorized`; implication `needs_llm`
  - files: `tests/tools/test_url_safety.py`, `tools/url_safety.py`
- **#107724: fix(qqbot): allow grouptalk.c2c.qq.com in the url_safety private-IP bypass (#47123)**. open
  - maintainer direction: category `uncategorized`; implication `needs_llm`; refs `#47177`, `#47123`, `#85256`, `#47177`
  - files: `tests/tools/test_url_safety.py`, `tools/url_safety.py`
- **#131466: fix(url_safety): classify the NAT64 well-known prefix by its embedded IPv4**. open
  - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#12256`, `#51758`, `#51758`
  - files: `tests/tools/test_url_safety.py`, `tools/url_safety.py`

## Fix status

- Merged to main in PR #11567 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-04-22`. This confirmation is old and may be out of date (a revert/reopen would not have invalidated it here); re-verify.
- Subsystem: `agent`
