From 1f931c1df1b8ecec678d4f7f426a2ff37b6cefab Mon Sep 17 00:00:00 2001 From: cnitlrt Date: Sat, 23 May 2026 12:55:03 +0800 Subject: [PATCH] fix: detect add-phone oauth blocks earlier --- src/autoteam/codex_auth.py | 100 ++++++++++++++++++++++---- tests/unit/test_codex_auth_session.py | 68 +++++++++++++++++- 2 files changed, 155 insertions(+), 13 deletions(-) diff --git a/src/autoteam/codex_auth.py b/src/autoteam/codex_auth.py index 419cc1f7..16bf3dff 100644 --- a/src/autoteam/codex_auth.py +++ b/src/autoteam/codex_auth.py @@ -35,6 +35,7 @@ CODEX_TOKEN_URL = "https://auth.openai.com/oauth/token" CODEX_CALLBACK_PORT = 1455 CODEX_REDIRECT_URI = f"http://localhost:{CODEX_CALLBACK_PORT}/auth/callback" +_EARLY_OAUTH_BLOCK_FAILURE_TYPES = {"add_phone", "human_verification", "site_unavailable"} def _generate_pkce(): @@ -93,6 +94,48 @@ def _classify_oauth_failure(url, body_excerpt=""): return "auth_code_missing", f"未获取到 auth code（停留在 {url or 'unknown'}）", True +def _build_oauth_failure_result(url, body_excerpt=""): + error_type, error_detail, retryable = _classify_oauth_failure(url, body_excerpt) + return { + "ok": False, + "bundle": None, + "error_type": error_type, + "error_detail": error_detail, + "retryable": retryable, + "current_url": url, + "body_excerpt": body_excerpt, + } + + +def _detect_early_oauth_block(page): + body_excerpt = _page_excerpt(page) + failure = _build_oauth_failure_result(getattr(page, "url", ""), body_excerpt) + if failure["error_type"] in _EARLY_OAUTH_BLOCK_FAILURE_TYPES: + return failure + return None + + +def _wait_for_oauth_page_progress(page, previous_url="", timeout=5): + previous_url = (previous_url or "").lower() + deadline = time.time() + timeout + + while time.time() < deadline: + current_url = (getattr(page, "url", "") or "").lower() + if f"localhost:{CODEX_CALLBACK_PORT}/auth/callback" in current_url: + return None + + failure = _detect_early_oauth_block(page) + if failure: + return failure + + if previous_url and current_url and current_url != previous_url: + return None + + time.sleep(0.5) + + return _detect_early_oauth_block(page) + + def _build_auth_url(code_challenge, state): params = { "client_id": CODEX_CLIENT_ID, @@ -1071,6 +1114,18 @@ def on_response(response): if auth_code: break + blocking_failure = _detect_early_oauth_block(page) + if blocking_failure: + logger.warning( + "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s", + step + 1, + blocking_failure["error_detail"], + blocking_failure["current_url"], + ) + _screenshot(page, f"codex_04_blocked_{step + 1}.png") + failure_result = blocking_failure + break + _screenshot(page, f"codex_04_step{step + 1}_before.png") try: @@ -1103,11 +1158,24 @@ def on_response(response): try: cont_btn = page.locator('button:has-text("继续"), button:has-text("Continue")').first if cont_btn.is_visible(timeout=3000): + previous_url = page.url cont_btn.click() - time.sleep(3) + failure = _wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=3) + if failure: + logger.warning( + "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s", + step + 1, + failure["error_detail"], + failure["current_url"], + ) + _screenshot(page, f"codex_04_workspace_blocked_{step + 1}.png") + failure_result = failure + break logger.info("[Codex] 已点击继续 (step %d)", step + 1) except Exception: pass + if failure_result: + break continue else: logger.warning("[Codex] 无法选择 workspace '%s' (step %d)", workspace_name, step + 1) @@ -1181,9 +1249,19 @@ def on_response(response): ).first if consent_btn.is_visible(timeout=5000): logger.info("[Codex] 点击同意/继续按钮 (step %d)...", step + 1) + previous_url = page.url consent_btn.click() - time.sleep(5) + failure = _wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=5) _screenshot(page, f"codex_04_consent_{step + 1}.png") + if failure: + logger.warning( + "[Codex] 提前识别到 OAuth 阻塞页 (step %d): %s | URL=%s", + step + 1, + failure["error_detail"], + failure["current_url"], + ) + failure_result = failure + break else: time.sleep(1) continue @@ -1213,16 +1291,7 @@ def on_response(response): _screenshot(page, "codex_05_no_callback.png") body_excerpt = _page_excerpt(page) logger.warning("[Codex] 未获取到 auth code，当前 URL: %s", page.url) - error_type, error_detail, retryable = _classify_oauth_failure(page.url, body_excerpt) - failure_result = { - "ok": False, - "bundle": None, - "error_type": error_type, - "error_detail": error_detail, - "retryable": retryable, - "current_url": page.url, - "body_excerpt": body_excerpt, - } + failure_result = _build_oauth_failure_result(page.url, body_excerpt) browser.close() @@ -1392,6 +1461,10 @@ def _detect_step(self): if self.auth_code: return "completed", None + blocking_failure = _detect_early_oauth_block(self.page) + if blocking_failure: + return blocking_failure["error_type"], blocking_failure["error_detail"] + if self._visible_locator(self.CODE_SELECTORS, timeout_ms=800): return "code_required", None if self._visible_locator(self.PASSWORD_SELECTORS, timeout_ms=800): @@ -1561,6 +1634,9 @@ def _advance(self, attempts=12): "detail": "主号 Codex 当前停留在密码页，且未找到一次性验证码入口", } + if step in _EARLY_OAUTH_BLOCK_FAILURE_TYPES: + return {"step": step, "detail": detail} + if step == "email_required": if self._auto_fill_email(): continue diff --git a/tests/unit/test_codex_auth_session.py b/tests/unit/test_codex_auth_session.py index 09a95bb0..aa00b9e7 100644 --- a/tests/unit/test_codex_auth_session.py +++ b/tests/unit/test_codex_auth_session.py @@ -84,10 +84,11 @@ def test_refresh_main_auth_file_saves_bundle_from_session_login(monkeypatch): class _FakeElement: - def __init__(self, text, *, visible=True): + def __init__(self, text, *, visible=True, on_click=None): self._text = text self.visible = visible self.clicked = False + self._on_click = on_click def is_visible(self, timeout=0): return self.visible @@ -97,6 +98,8 @@ def inner_text(self, timeout=0): def click(self, timeout=0, force=False): self.clicked = True + if self._on_click: + self._on_click() class _FakeCollection: @@ -185,6 +188,31 @@ def advance(self): self._body = "Codex wants access to your API organization Select a project Continue" +class _FakeConsentToAddPhonePage(_FakePage): + def __init__(self): + self._continue_button = _FakeElement("Continue", on_click=self._advance_to_add_phone) + super().__init__( + url="https://auth.openai.com/sign-in-with-chatgpt/codex/consent", + body="Codex wants access to your API organization Select a project Continue", + elements=[], + ) + + def _advance_to_add_phone(self): + self.url = "https://auth.openai.com/add-phone" + self._body = "Add a phone number to continue" + + def locator(self, selector): + if selector == "body": + return _FakeCollection(text=self._body) + if selector in { + 'button:has-text("Continue"), button:has-text("继续"), button:has-text("Allow")', + }: + return _FakeCollection(items=[self._continue_button]) + if selector in self._GENERIC_SELECTORS: + return _FakeCollection(items=self._elements) + return _FakeCollection(items=[]) + + def test_workspace_selection_detection_ignores_otp_pages(): page = _FakePage( url="https://auth.openai.com/email-verification", @@ -393,3 +421,41 @@ def test_team_workspace_selection_requires_exact_workspace_name(): assert codex_auth._workspace_label_candidates(page) == [("Personal account", items[1])] assert codex_auth._select_team_workspace(page, "Idapro") is False + + +def test_detect_early_oauth_block_detects_add_phone_page(): + page = _FakePage(url="https://auth.openai.com/add-phone", body="Add a phone number to continue") + + failure = codex_auth._detect_early_oauth_block(page) + + assert failure is not None + assert failure["error_type"] == "add_phone" + assert failure["error_detail"] == "需要手机号验证" + assert failure["retryable"] is False + + +def test_wait_for_oauth_page_progress_detects_add_phone_after_consent_click(): + page = _FakeConsentToAddPhonePage() + consent_btn = page.locator('button:has-text("Continue"), button:has-text("继续"), button:has-text("Allow")').first + + previous_url = page.url + consent_btn.click() + failure = codex_auth._wait_for_oauth_page_progress(page, previous_url=previous_url, timeout=1) + + assert failure is not None + assert failure["error_type"] == "add_phone" + assert failure["current_url"] == "https://auth.openai.com/add-phone" + + +def test_session_codex_flow_advance_returns_add_phone_immediately(): + flow = codex_auth.SessionCodexAuthFlow( + email="user@example.com", + session_token="session-token", + account_id="acc-1", + workspace_name="Idapro", + ) + flow.page = _FakePage(url="https://auth.openai.com/add-phone", body="Add a phone number to continue") + + result = flow._advance(attempts=1) + + assert result == {"step": "add_phone", "detail": "需要手机号验证"}