WebSocket reconnect handling and message drop fix

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #11567 fix(qqbot): Batch-2 salvage — SSRF allowlist, voice preflight, group auth, WS reconnect wait

  merged to main

  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

  Show the diff

  • gateway/platforms/qqbot.py

    @@ -64,6 +64,7 @@
         MessageEvent,
         MessageType,
         SendResult,
    +    _ssrf_redirect_guard,
         cache_document_from_bytes,
         cache_image_from_bytes,
     )
    @@ -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()
    @@ -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:
    @@ -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,
    @@ -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)
    @@ -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

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

    @@ -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",
    @@ -235,6 +236,7 @@
         "[redacted:email]": "malaiwah",
         "[redacted:email]": "v1k22",
         "[redacted:email]": "RucchiZ",
    +    "[redacted:email]": "LehaoLin",
     }
     
     

  • tests/gateway/test_qqbot.py

    @@ -1,5 +1,6 @@
     """Tests for the QQ Bot platform adapter."""
     
    +import asyncio
     import json
     import os
     import sys
    @@ -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
     # ---------------------------------------------------------------------------
    @@ -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

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

    @@ -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."""
    @@ -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

    @@ -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
    @@ -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.
     
    @@ -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
     
    @@ -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)
    @@ -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: