fix(weixin): require fresh credentials for forced login

nanobot/channels/weixin/connect.py

@@ -22,6 +22,7 @@ class WeixinConnectSession:
     channel: WeixinChannel
     current_poll_base_url: str
     refresh_count: int
+    force: bool
     created_wall: float
     deadline: float
     last_error: str | None = None
@@ -72,7 +73,7 @@ class WeixinConnectStore:
 
         channel.connect_open_client()
         try:
-            qrcode_id, qr_url = await channel.connect_fetch_qr_code()
+            qrcode_id, qr_url = await channel.connect_fetch_qr_code(force=force)
         except Exception as exc:
             await self._close_channel(channel)
             raise ChannelConnectError(
@@ -89,6 +90,7 @@ class WeixinConnectStore:
             channel=channel,
             current_poll_base_url=channel.connect_base_url,
             refresh_count=0,
+            force=force,
             created_wall=now_wall,
             deadline=time.monotonic() + 600,
         )
@@ -187,7 +189,7 @@ class WeixinConnectStore:
                 }
             try:
                 session.qrcode_id, session.qr_url = (
-                    await session.channel.connect_fetch_qr_code()
+                    await session.channel.connect_fetch_qr_code(force=session.force)
                 )
             except Exception as exc:
                 self._sessions.pop(session_id, None)
@@ -204,6 +206,17 @@ class WeixinConnectStore:
             )
 
         if status == "binded_redirect":
+            if session.force:
+                self._sessions.pop(session_id, None)
+                await self._close_channel(session.channel)
+                return {
+                    "session_id": session_id,
+                    "status": "failed",
+                    "message": (
+                        "Unable to complete a new WeChat login. "
+                        "Start again and scan with the account you want to connect."
+                    ),
+                }
             if not session.channel.connect_load_state():
                 self._sessions.pop(session_id, None)
                 await self._close_channel(session.channel)
@@ -234,7 +247,7 @@ class WeixinConnectStore:
                 }
             try:
                 session.qrcode_id, session.qr_url = (
-                    await session.channel.connect_fetch_qr_code()
+                    await session.channel.connect_fetch_qr_code(force=session.force)
                 )
             except Exception as exc:
                 self._sessions.pop(session_id, None)

nanobot/channels/weixin/runtime.py

@@ -726,9 +726,9 @@ class WeixinChannel(BaseChannel):
                 break
         return tokens
 
-    async def _fetch_qr_code(self) -> tuple[str, str]:
+    async def _fetch_qr_code(self, *, force: bool = False) -> tuple[str, str]:
-        """Fetch a fresh QR code. Returns (qrcode_id, scan_url)."""
+        """Fetch a QR code without existing credentials when forced."""
-        local_tokens = self._local_token_list()
+        local_tokens = [] if force else self._local_token_list()
         data = await self._api_post(
             "ilink/bot/get_bot_qrcode?bot_type=3",
             {"local_token_list": local_tokens},
@@ -755,11 +755,11 @@ class WeixinChannel(BaseChannel):
             raise RuntimeError(f"Failed to get QR code from WeChat API: {data}")
         return qrcode_id, (qrcode_img_content or qrcode_id)
 
-    async def _qr_login(self) -> bool:
+    async def _qr_login(self, *, force: bool = False) -> bool:
-        """Perform QR code login flow. Returns True on success."""
+        """Perform QR login; forced flows accept only newly confirmed credentials."""
         try:
             refresh_count = 0
-            qrcode_id, scan_url = await self._fetch_qr_code()
+            qrcode_id, scan_url = await self._fetch_qr_code(force=force)
             self._print_qr_code(scan_url)
             current_poll_base_url = self.config.base_url
             verify_code = ""
@@ -825,11 +825,16 @@ class WeixinChannel(BaseChannel):
                     if refresh_count > MAX_QR_REFRESH_COUNT:
                         self.logger.warning("WeChat verification failed too many times")
                         return False
-                    qrcode_id, scan_url = await self._fetch_qr_code()
+                    qrcode_id, scan_url = await self._fetch_qr_code(force=force)
                     current_poll_base_url = self.config.base_url
                     self._print_qr_code(scan_url)
                     continue
                 elif status == "binded_redirect":
+                    if force:
+                        self.logger.error(
+                            "Forced WeChat login returned an existing binding without new credentials"
+                        )
+                        return False
                     if self._token or self._load_state():
                         self.logger.info("WeChat account is already connected")
                         return True
@@ -846,7 +851,7 @@ class WeixinChannel(BaseChannel):
                             MAX_QR_REFRESH_COUNT,
                         )
                         return False
-                    qrcode_id, scan_url = await self._fetch_qr_code()
+                    qrcode_id, scan_url = await self._fetch_qr_code(force=force)
                     current_poll_base_url = self.config.base_url
                     verify_code = ""
                     self._print_qr_code(scan_url)
@@ -893,8 +898,8 @@ class WeixinChannel(BaseChannel):
         self._client = self._new_http_client(httpx.Timeout(60, connect=30))
         self._running = True
 
-    async def connect_fetch_qr_code(self) -> tuple[str, str]:
+    async def connect_fetch_qr_code(self, *, force: bool = False) -> tuple[str, str]:
-        return await self._fetch_qr_code()
+        return await self._fetch_qr_code(force=force)
 
     async def connect_poll_qr_code(
         self,
@@ -954,7 +959,7 @@ class WeixinChannel(BaseChannel):
         self._client = self._new_http_client(httpx.Timeout(60, connect=30))
         self._running = True  # Enable polling loop in _qr_login()
         try:
-            return await self._qr_login()
+            return await self._qr_login(force=force)
         finally:
             self._running = False
             if self._client:

nanobot/channels/weixin/tests/test_connect.py

@@ -25,7 +25,9 @@ async def test_weixin_connect_store_saves_confirmed_qr_login(
     )
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    async def fake_fetch_qr_code(
+        self: WeixinChannel, **_kwargs: Any
+    ) -> tuple[str, str]:
         return "qr-1", "https://qr.example/1"
 
     async def fake_api_get_with_base(
@@ -86,14 +88,31 @@ async def test_weixin_reconnect_keeps_existing_account_until_scan_succeeds(
     )
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    observed_force: list[bool] = []
-        return "qr-reconnect", "https://qr.example/reconnect"
+
+    async def fake_fetch_qr_code(
+        self: WeixinChannel,
+        *,
+        force: bool = False,
+    ) -> tuple[str, str]:
+        observed_force.append(force)
+        return f"qr-reconnect-{len(observed_force)}", "https://qr.example/reconnect"
+
+    async def fake_api_get_with_base(
+        self: WeixinChannel,
+        **_kwargs: Any,
+    ) -> dict[str, str]:
+        return {"status": "expired"}
 
     monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code)
+    monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
 
     store = WeixinConnectStore()
     started = await store.start(force=True)
+    refreshed = await store.poll(started["session_id"])
 
+    assert refreshed["status"] == "pending"
+    assert observed_force == [True, True]
     assert json.loads(state_file.read_text(encoding="utf-8")) == existing
     cancelled = await store.cancel(started["session_id"])
     assert cancelled["status"] == "cancelled"
@@ -116,7 +135,9 @@ async def test_weixin_cancel_wins_over_inflight_confirmation(
     poll_started = asyncio.Event()
     release_poll = asyncio.Event()
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    async def fake_fetch_qr_code(
+        self: WeixinChannel, **_kwargs: Any
+    ) -> tuple[str, str]:
         return "qr-cancel", "https://qr.example/cancel"
 
     async def fake_api_get_with_base(
@@ -162,7 +183,9 @@ async def test_weixin_connect_store_handles_verification_code(
     )
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    async def fake_fetch_qr_code(
+        self: WeixinChannel, **_kwargs: Any
+    ) -> tuple[str, str]:
         return "qr-verify", "https://qr.example/verify"
 
     responses = [
@@ -204,7 +227,7 @@ async def test_weixin_connect_store_handles_verification_code(
 
 
 @pytest.mark.asyncio
-async def test_weixin_connect_store_treats_existing_binding_as_success(
+async def test_weixin_connect_store_rejects_existing_binding_during_forced_login(
     tmp_path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -221,7 +244,12 @@ async def test_weixin_connect_store_treats_existing_binding_as_success(
     )
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    async def fake_fetch_qr_code(
+        self: WeixinChannel,
+        *,
+        force: bool = False,
+    ) -> tuple[str, str]:
+        assert force is True
         return "qr-existing", "https://qr.example/existing"
 
     async def fake_api_get_with_base(
@@ -237,8 +265,8 @@ async def test_weixin_connect_store_treats_existing_binding_as_success(
     started = await store.start(force=True)
     completed = await store.poll(started["session_id"])
 
-    assert completed["status"] == "succeeded"
+    assert completed["status"] == "failed"
-    assert "already connected" in completed["message"]
+    assert "new WeChat login" in completed["message"]
     assert json.loads((state_dir / "account.json").read_text())["token"] == "working-token"
 
 
@@ -255,7 +283,9 @@ async def test_weixin_connect_store_rejects_existing_binding_without_local_crede
     )
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
 
-    async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]:
+    async def fake_fetch_qr_code(
+        self: WeixinChannel, **_kwargs: Any
+    ) -> tuple[str, str]:
         return "qr-missing", "https://qr.example/missing"
 
     async def fake_api_get_with_base(
@@ -268,7 +298,7 @@ async def test_weixin_connect_store_rejects_existing_binding_without_local_crede
     monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base)
 
     store = WeixinConnectStore()
-    started = await store.start(force=True)
+    started = await store.start(force=False)
     completed = await store.poll(started["session_id"])
 
     assert completed["status"] == "failed"

nanobot/channels/weixin/tests/test_weixin_channel.py

@@ -197,34 +197,54 @@ def test_save_state_with_empty_runtime_token_preserves_persisted_account(tmp_pat
 
 
 @pytest.mark.asyncio
-async def test_login_force_does_not_short_circuit_on_persisted_account(tmp_path) -> None:
+async def test_login_force_ignores_persisted_account_through_qr_flow(tmp_path) -> None:
-    channel = WeixinChannel(
+    persisted = {
-        WeixinConfig(enabled=True, allow_from=["*"], state_dir=str(tmp_path)),
-        MessageBus(),
-    )
-    (tmp_path / "account.json").write_text(
-        json.dumps(
-            {
         "token": "persisted-token",
         "get_updates_buf": "persisted-cursor",
         "context_tokens": {"wx-user": "ctx-persisted"},
         "typing_tickets": {"wx-user": {"ticket": "ticket-persisted"}},
         "base_url": "https://persisted.example",
     }
+    channel = WeixinChannel(
+        WeixinConfig(
+            enabled=True,
+            allow_from=["*"],
+            token="configured-token",
+            state_dir=str(tmp_path),
         ),
+        MessageBus(),
+    )
+    (tmp_path / "account.json").write_text(
+        json.dumps(persisted),
         encoding="utf-8",
     )
-    channel._qr_login = AsyncMock(return_value=False)
+    channel._print_qr_code = lambda _url: None
+    channel._api_post = AsyncMock(
+        side_effect=[
+            {"qrcode": "qr-1", "qrcode_img_content": "url-1"},
+            {"qrcode": "qr-2", "qrcode_img_content": "url-2"},
+        ]
+    )
+    channel._api_get_with_base = AsyncMock(
+        side_effect=[
+            {"status": "expired"},
+            {"status": "binded_redirect"},
+        ]
+    )
 
     ok = await channel.login(force=True)
 
     assert ok is False
-    channel._qr_login.assert_awaited_once()
+    assert [call.args[1]["local_token_list"] for call in channel._api_post.await_args_list] == [
+        [],
+        [],
+    ]
     assert channel._token == ""
     assert channel._get_updates_buf == ""
     assert channel._context_tokens == {}
     assert channel._typing_tickets == {}
     assert channel.config.base_url == "https://ilinkai.weixin.qq.com"
+    assert json.loads((tmp_path / "account.json").read_text()) == persisted
 
 
 @pytest.mark.asyncio