fix(weixin): clamp server-suggested long-poll timeout

gateway/platforms/weixin.py

@@ -85,6 +85,10 @@
 EP_GET_QR_STATUS = "ilink/bot/get_qrcode_status"
 
 LONG_POLL_TIMEOUT_MS = 35_000
+# Hard ceiling for server-suggested long-poll budgets (see
+# ``_apply_long_poll_timeout_hint``). Default boot value stays 35s; iLink may
+# ask for longer, but never for an unbounded hang.
+MAX_LONG_POLL_TIMEOUT_MS = 120_000
 API_TIMEOUT_MS = 15_000
 CONFIG_TIMEOUT_MS = 10_000
 QR_TIMEOUT_MS = 35_000
@@ -108,6 +112,20 @@ def _is_stale_session_ret(
     return (errmsg or "").lower() == "unknown error"
 
 
+def _apply_long_poll_timeout_hint(current_ms: int, suggested) -> int:
+    """Return the next long-poll wait budget for getUpdates.
+
+    iLink may return ``longpolling_timeout_ms`` after each poll. Trust
+    positive ints so the server can lengthen the wait, but clamp to
+    ``MAX_LONG_POLL_TIMEOUT_MS`` so a bad or malicious suggestion cannot
+    pin ``_poll_loop`` (and delay disconnect cancellation) for an
+    unbounded wall-clock hang via ``asyncio.wait_for``.
+    """
+    if isinstance(suggested, int) and suggested > 0:
+        return min(suggested, MAX_LONG_POLL_TIMEOUT_MS)
+    return current_ms
+
+
 MEDIA_IMAGE = 1
 MEDIA_VIDEO = 2
 MEDIA_FILE = 3
@@ -1365,9 +1383,9 @@ async def _poll_loop(self) -> None:
                     sync_buf=sync_buf,
                     timeout_ms=timeout_ms,
                 )
-                suggested_timeout = response.get("longpolling_timeout_ms")
-                if isinstance(suggested_timeout, int) and suggested_timeout > 0:
-                    timeout_ms = suggested_timeout
+                timeout_ms = _apply_long_poll_timeout_hint(
+                    timeout_ms, response.get("longpolling_timeout_ms")
+                )
 
                 ret = response.get("ret", 0)
                 errcode = response.get("errcode", 0)

tests/gateway/test_weixin.py

@@ -646,6 +646,42 @@ def test_get_updates_returns_empty_sentinel_on_timeout(self):
         assert result == {"ret": 0, "msgs": [], "get_updates_buf": "buf-123"}
 
 
+class TestWeixinLongPollTimeoutClamp:
+    """Regression: iLink may return ``longpolling_timeout_ms`` after getUpdates.
+
+    The value feeds ``asyncio.wait_for`` for the next poll. Without a ceiling,
+    a huge positive suggestion pins the poll task (and delays disconnect)
+    for an unbounded wall-clock hang.
+    """
+
+    def test_huge_server_suggestion_is_clamped(self):
+        clamped = weixin._apply_long_poll_timeout_hint(
+            weixin.LONG_POLL_TIMEOUT_MS,
+            10**12,
+        )
+        assert clamped == weixin.MAX_LONG_POLL_TIMEOUT_MS
+        assert clamped < 10**12
+
+    def test_normal_server_suggestion_is_kept(self):
+        assert (
+            weixin._apply_long_poll_timeout_hint(weixin.LONG_POLL_TIMEOUT_MS, 35_000)
+            == 35_000
+        )
+
+    def test_non_positive_or_non_int_keeps_current(self):
+        current = weixin.LONG_POLL_TIMEOUT_MS
+        assert weixin._apply_long_poll_timeout_hint(current, 0) == current
+        assert weixin._apply_long_poll_timeout_hint(current, -5) == current
+        assert weixin._apply_long_poll_timeout_hint(current, None) == current
+        assert weixin._apply_long_poll_timeout_hint(current, "35000") == current
+        assert weixin._apply_long_poll_timeout_hint(current, 35_000.5) == current
+
+    def test_ceiling_constant_is_above_default(self):
+        # Server may legitimately ask for longer than the default boot value,
+        # but still within a sane upper bound.
+        assert weixin.MAX_LONG_POLL_TIMEOUT_MS >= weixin.LONG_POLL_TIMEOUT_MS
+
+
 class TestWeixinVoiceAlwaysDownloaded:
     """Regression tests for #27300: when WeChat (Weixin) returns a
     ``voice_item.text`` (Tencent Cloud's STT) we must still download