From 20dc19e90dedddb5a04d42176cd94ea7e1071ab4 Mon Sep 17 00:00:00 2001 From: cong <3135055939@qq.com> Date: Sat, 4 Apr 2026 00:55:32 +0800 Subject: [PATCH] fix(chatgpt): recover workspace after add_phone with one-shot oauth restart --- platforms/chatgpt/oauth_client.py | 81 +++++++++++++++++++++++-------- tests/test_chatgpt_register.py | 55 +++++++++++++++++++++ 2 files changed, 117 insertions(+), 19 deletions(-) diff --git a/platforms/chatgpt/oauth_client.py b/platforms/chatgpt/oauth_client.py index e5c20d3..e0ac0e9 100644 --- a/platforms/chatgpt/oauth_client.py +++ b/platforms/chatgpt/oauth_client.py @@ -716,6 +716,12 @@ class OAuthClient: self._set_error(f"触发 passwordless OTP 异常: {e}") return None + def _recreate_session(self): + """重建会话，确保恢复链路使用全新 cookie 容器。""" + self.session = curl_requests.Session() + if self.proxy: + self.session.proxies = build_requests_proxy_config(self.proxy) + def login_and_get_tokens( self, email, @@ -728,6 +734,7 @@ class OAuthClient: prefer_passwordless_login=False, allow_phone_verification=True, login_source="", + _recovery_depth=0, ): """ 完整的 OAuth 登录流程，获取 tokens @@ -914,24 +921,52 @@ class OAuthClient: if self._state_is_add_phone(state): if not allow_phone_verification: - self._set_error( - "passwordless 登录后仍停留在 add_phone，未获取到 workspace / callback" + if self._state_supports_workspace_resolution(state): + self._log( + "步骤5: add_phone 命中，但检测到 workspace 线索，继续尝试 workspace/org 选择" + ) + elif prefer_passwordless_login and _recovery_depth < 1: + self._log( + "步骤5: add_phone 仍无 workspace/callback，重启一次全新 OAuth session + 新 PKCE" + ) + self._recreate_session() + return self.login_and_get_tokens( + email, + password, + device_id, + user_agent=user_agent, + sec_ch_ua=sec_ch_ua, + impersonate=impersonate, + skymail_client=skymail_client, + prefer_passwordless_login=prefer_passwordless_login, + allow_phone_verification=allow_phone_verification, + login_source=( + f"{login_source}:add_phone_recovery" + if login_source + else "add_phone_recovery" + ), + _recovery_depth=_recovery_depth + 1, + ) + else: + self._set_error( + "passwordless 登录后仍停留在 add_phone，未获取到 workspace / callback" + ) + return None + else: + next_state = self._handle_add_phone_verification( + device_id, + user_agent, + sec_ch_ua, + impersonate, + state, ) - return None - next_state = self._handle_add_phone_verification( - device_id, - user_agent, - sec_ch_ua, - impersonate, - state, - ) - if not next_state: - if not self.last_error: - self._set_error("手机号验证后未进入下一步 OAuth 状态") - return None - referer = state.current_url or referer - state = next_state - continue + if not next_state: + if not self.last_error: + self._set_error("手机号验证后未进入下一步 OAuth 状态") + return None + referer = state.current_url or referer + state = next_state + continue if self._state_requires_navigation(state): code, next_state = self._follow_flow_state( @@ -958,10 +993,18 @@ class OAuthClient: if self._state_supports_workspace_resolution(state): self._log("步骤6: 执行 workspace/org 选择") - code, next_state = self._oauth_submit_workspace_and_org( + consent_entry = ( state.continue_url or state.current_url - or f"{self.oauth_issuer}/sign-in-with-chatgpt/codex/consent", + or f"{self.oauth_issuer}/sign-in-with-chatgpt/codex/consent" + ) + if self._state_is_add_phone(state): + consent_entry = ( + f"{self.oauth_issuer}/sign-in-with-chatgpt/codex/consent" + ) + self._log("步骤6: 当前处于 add_phone，改用 canonical consent URL 继续") + code, next_state = self._oauth_submit_workspace_and_org( + consent_entry, device_id, user_agent, impersonate, diff --git a/tests/test_chatgpt_register.py b/tests/test_chatgpt_register.py index 3fcd4ff..1b955a9 100644 --- a/tests/test_chatgpt_register.py +++ b/tests/test_chatgpt_register.py @@ -295,6 +295,61 @@ class OAuthClientPasswordlessTests(unittest.TestCase): follow_state.assert_called_once() handle_phone.assert_not_called() + def test_login_and_get_tokens_uses_canonical_consent_url_when_state_is_add_phone(self): + client = self._make_client() + add_phone_state = FlowState( + page_type="add_phone", + continue_url="https://auth.openai.com/add-phone", + current_url="https://auth.openai.com/add-phone", + ) + + with mock.patch.object(client, "_bootstrap_oauth_session", return_value="https://auth.openai.com/log-in"), \ + mock.patch.object(client, "_submit_authorize_continue", return_value=add_phone_state), \ + mock.patch.object(client, "_state_supports_workspace_resolution", return_value=True), \ + mock.patch.object(client, "_state_requires_navigation", return_value=False), \ + mock.patch.object(client, "_oauth_submit_workspace_and_org", return_value=("auth-code", None)) as submit_workspace, \ + mock.patch.object(client, "_exchange_code_for_tokens", return_value={"access_token": "at"}): + tokens = client.login_and_get_tokens( + "user@example.com", + "Secret123!", + "device-fixed", + prefer_passwordless_login=True, + allow_phone_verification=False, + skymail_client=mock.Mock(), + ) + + self.assertEqual(tokens["access_token"], "at") + self.assertEqual( + submit_workspace.call_args.args[0], + "https://auth.openai.com/sign-in-with-chatgpt/codex/consent", + ) + + def test_login_and_get_tokens_retries_once_when_add_phone_has_no_workspace(self): + client = self._make_client() + add_phone_state = FlowState( + page_type="add_phone", + continue_url="https://auth.openai.com/add-phone", + current_url="https://auth.openai.com/add-phone", + ) + + with mock.patch.object(client, "_bootstrap_oauth_session", return_value="https://auth.openai.com/log-in") as bootstrap, \ + mock.patch.object(client, "_submit_authorize_continue", return_value=add_phone_state) as submit_continue, \ + mock.patch.object(client, "_state_supports_workspace_resolution", return_value=False), \ + mock.patch.object(client, "_state_requires_navigation", return_value=False): + tokens = client.login_and_get_tokens( + "user@example.com", + "Secret123!", + "device-fixed", + prefer_passwordless_login=True, + allow_phone_verification=False, + skymail_client=mock.Mock(), + ) + + self.assertIsNone(tokens) + self.assertEqual(bootstrap.call_count, 2) + self.assertEqual(submit_continue.call_count, 2) + self.assertIn("未获取到 workspace / callback", client.last_error) + def test_send_passwordless_login_otp_does_not_send_email_field(self): client = self._make_client() response = mock.Mock()