packages/application/auth/wechat_oauth_service.py

@@ -200,7 +200,15 @@ class WechatOAuthService:
             return None, "微信登录处理失败"
 
 
+# 模块级单例：state 存储必须跨请求共享，否则 /wechat/url 生成的 state
+# 与 /wechat/callback 校验时不在同一个 MemoryStateStore，回调必然 400。
+# 多实例部署时应替换为 Redis state store（单容器多 worker 也需如此）。
+_oauth_service_singleton: WechatOAuthService | None = None
+
+
 def get_wechat_oauth_service() -> WechatOAuthService:
-    """获取微信 OAuth 服务单例"""
-    # TODO: 可替换为 Redis state store
-    return WechatOAuthService()
+    """获取微信 OAuth 服务单例（state store 跨请求共享）"""
+    global _oauth_service_singleton
+    if _oauth_service_singleton is None:
+        _oauth_service_singleton = WechatOAuthService()
+    return _oauth_service_singleton

tests/unit/test_wechat_oauth_service.py

@@ -388,3 +388,52 @@ class TestGetWechatOAuthService:
         """返回 WechatOAuthService 实例"""
         service = get_wechat_oauth_service()
         assert isinstance(service, WechatOAuthService)
+
+    def test_singleton_same_instance_across_calls(self, monkeypatch):
+        """#1718 回归：工厂必须返回同一实例，否则 state store 不共享"""
+        import packages.application.auth.wechat_oauth_service as mod
+
+        monkeypatch.setattr(mod, "_oauth_service_singleton", None)
+        s1 = get_wechat_oauth_service()
+        s2 = get_wechat_oauth_service()
+        assert s1 is s2
+
+    def test_state_survives_across_factory_calls(self, monkeypatch):
+        """#1718 回归：/wechat/url 与 /wechat/callback 经工厂拿到同一 state store
+
+        模拟两次请求各自调用工厂：第一个实例生成 state，第二个实例（同一单例）
+        必须能校验通过。修复前工厂每次 new 一个实例，回调必现 400「无效的 state」。
+        """
+        import packages.application.auth.wechat_oauth_service as mod
+
+        monkeypatch.setattr(mod, "_oauth_service_singleton", None)
+        monkeypatch.setenv("WECHAT_OPEN_APP_ID", "wx-test")
+        monkeypatch.setenv("WECHAT_OPEN_APP_SECRET", "secret-test")
+        monkeypatch.setenv("WECHAT_OPEN_REDIRECT_URI", "https://example.com/cb")
+
+        # 请求1：生成授权链接（state 写入单例 store）
+        _, state = get_wechat_oauth_service().generate_auth_url()
+
+        # 请求2：回调校验（应命中同一个 store；微信 API 用 mock 避免外网）
+        with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get:
+            mock_get.return_value = MagicMock(
+                json=MagicMock(
+                    return_value={
+                        "access_token": "at",
+                        "openid": "oid",
+                        "unionid": "uid",
+                        "nickname": "n",
+                        "headimgurl": "http://x/a.png",
+                    }
+                )
+            )
+            user_info, error = get_wechat_oauth_service().handle_callback("code-x", state)
+
+        assert error is None, f"state 应跨请求共享，实际报错: {error}"
+        assert user_info is not None
+        assert user_info.openid == "oid"
+
+        # state 一次性消费，重放必须失败
+        user_info2, error2 = get_wechat_oauth_service().handle_callback("code-y", state)
+        assert user_info2 is None
+        assert "state" in error2