diff --git a/packages/application/auth/wechat_oauth_service.py b/packages/application/auth/wechat_oauth_service.py index 16cdf5d47..d8cd3d125 100755 --- a/packages/application/auth/wechat_oauth_service.py +++ b/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 diff --git a/tests/unit/test_wechat_oauth_service.py b/tests/unit/test_wechat_oauth_service.py index 7a2ed3397..759732c62 100755 --- a/tests/unit/test_wechat_oauth_service.py +++ b/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