diff --git a/src/antigravity_provider/router/profile_oauth.py b/src/antigravity_provider/router/profile_oauth.py index a4df976..f37277c 100644 --- a/src/antigravity_provider/router/profile_oauth.py +++ b/src/antigravity_provider/router/profile_oauth.py @@ -1,10 +1,11 @@ """Profile OAuth manager for interactive Google / Antigravity account linking. -Features: -- Immediate listener startup with verified socket binding. -- Dynamic or standard (51121) port binding with strict redirect_uri alignment. -- Sanitized diagnostic logging without exposing codes, tokens, or client secrets. -- Deterministic session lifecycle, state validation, and clean cancellation. +Architecture: +- Deterministic lifecycle: session creation -> port allocation -> listener verified ready -> URL generation. +- Unified callback pipeline: both automatic localhost HTTP callback and manual pasted callback URL + flow through handle_callback() for state validation, original PKCE verifier lookup, and token exchange. +- Thread-safe single completion guarantee preventing duplicate saves. +- Zero secret logging policy (code, verifier, tokens, secrets are excluded from logs). """ from __future__ import annotations @@ -36,6 +37,31 @@ logger = logging.getLogger("hermes.router.profile_oauth") _ACTIVE_OAUTH_SESSIONS: Dict[str, "ProfileOAuthSession"] = {} +SUCCESS_HTML = ( + b"Hermes Hub" + b"" + b"

✔ \xd0\x90\xd0\xb2\xd1\x82\xd0\xbe\xd1\x80\xd0\xb8\xd0\xb7\xd0\xb0\xd1\x86\xd0\xb8\xd1\x8f \xd1\x83\xd1\x81\xd0\xbf\xd0\xb5\xd1\x88\xd0\xbd\xd0\xbe \xd0\xb7\xd0\xb0\xd0\xb2\xd0\xb5\xd1\x80\xd1\x88\xd0\xb5\xd0\xbd\xd0\xb0.

" + b"

\xd0\x9c\xd0\xbe\xd0\xb6\xd0\xbd\xd0\xbe \xd0\xb7\xd0\xb0\xd0\xba\xd1\x80\xd1\x8b\xd1\x82\xd1\x8c \xd1\x8d\xd1\x82\xd1\x83 \xd0\xb2\xd0\xba\xd0\xbb\xd0\xb0\xd0\xb4\xd0\xba\xd1\x83 \xd0\xb8 \xd0\xb2\xd0\xb5\xd1\x80\xd0\xbd\xd1\x83\xd1\x82\xd1\x8c\xd1\x81\xd1\x8f \xd0\xb2 Hermes Hub.

" +) + +ERROR_HTML = ( + b"Hermes Hub" + b"" + b"

\xd0\x9d\xd0\xb5 \xd1\x83\xd0\xb4\xd0\xb0\xd0\xbb\xd0\xbe\xd1\x81\xd1\x8c \xd0\xb7\xd0\xb0\xd0\xb2\xd0\xb5\xd1\x80\xd1\x88\xd0\xb8\xd1\x82\xd1\x8c \xd0\xb0\xd0\xb2\xd1\x82\xd0\xbe\xd1\x80\xd0\xb8\xd0\xb7\xd0\xb0\xd1\x86\xd0\xb8\xd1\x8e.

" + b"

\xd0\x92\xd0\xb5\xd1\x80\xd0\xbd\xd0\xb8\xd1\x82\xd0\xb5\xd1\x81\xd1\x8c \xd0\xb2 Hermes Hub \xd0\xb4\xd0\xbb\xd1\x8f \xd0\xbf\xd0\xbe\xd0\xbb\xd1\x83\xd1\x87\xd0\xb5\xd0\xbd\xd0\xb8\xd1\x8f \xd0\xbf\xd0\xbe\xd0\xb4\xd1\x80\xd0\xbe\xd0\xb1\xd0\xbd\xd0\xbe\xd0\xb9 \xd0\xb8\xd0\xbd\xd1\x84\xd0\xbe\xd1\x80\xd0\xbc\xd0\xb0\xd1\x86\xd0\xb8\xd0\xb8.

" +) + + class _ProfileOAuthCallbackHandler(BaseHTTPRequestHandler): server: "_ProfileOAuthServer" @@ -46,20 +72,22 @@ class _ProfileOAuthCallbackHandler(BaseHTTPRequestHandler): return params = urllib.parse.parse_qs(parsed.query) - self.server.session.received_state = (params.get("state") or [None])[0] - self.server.session.received_error = (params.get("error") or [None])[0] - self.server.session.received_code = (params.get("code") or [None])[0] + received_state = (params.get("state") or [None])[0] + received_error = (params.get("error") or [None])[0] + received_code = (params.get("code") or [None])[0] + error_desc = (params.get("error_description") or [None])[0] - body = ( - b"Hermes Account Linked" - b"" - b"

✔ Account Authorized

" - b"

You can close this tab and return to the Hermes Account Manager.

" + effective_err = received_error or error_desc + + ok, msg = self.server.session.handle_callback( + code=received_code, + state=received_state, + error=effective_err, + source="automatic", ) - self.send_response(200) + + body = SUCCESS_HTML if ok else ERROR_HTML + self.send_response(200 if ok else 400) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Connection", "close") @@ -89,10 +117,6 @@ class ProfileOAuthSession: self.verifier, self.challenge = _pkce_pair() self.redirect_uri = f"http://{CALLBACK_HOST}:{self.port}{CALLBACK_PATH}" - self.received_code: Optional[str] = None - self.received_state: Optional[str] = None - self.received_error: Optional[str] = None - self.server: Optional[_ProfileOAuthServer] = None self.server_thread: Optional[threading.Thread] = None self.status = "initialized" # initialized, pending, completed, failed, cancelled, timeout @@ -101,6 +125,9 @@ class ProfileOAuthSession: self.completed_profile_info: Optional[dict] = None self.is_listening = False + self._completion_lock = threading.Lock() + self._is_completed = False + def get_auth_url(self) -> str: params = { "client_id": CLIENT_ID, @@ -140,94 +167,155 @@ class ProfileOAuthSession: self.server.handle_request() except Exception as req_err: logger.warning("OAUTH handle_request warning: %s: %s", type(req_err).__name__, req_err) - if self.received_code or self.received_error: - logger.info("OAUTH callback received") + if self._is_completed or self.status != "pending": break - if self.received_error: - self.status = "failed" - self.error_msg = f"OAuth error from provider: {self.received_error}" - logger.warning("OAUTH callback error from provider: %s", self.received_error) - elif self.received_code: - if self.received_state != self.state: - self.status = "failed" - self.error_msg = "State mismatch in OAuth callback" - logger.warning("OAUTH state validation failed") - else: - logger.info("OAUTH state validated") - self._finalize_tokens() - elif self.status == "pending": + if self.status == "pending" and not self._is_completed: self.status = "timeout" - self.error_msg = "OAuth login timed out after 5 minutes" + self.error_msg = "Срок действия ссылки авторизации истёк (таймаут 5 минут)" logger.info("OAUTH callback server stopped reason=timeout") except Exception as loop_err: logger.error("OAUTH listener exception: %s: %s", type(loop_err).__name__, loop_err) - self.status = "failed" - self.error_msg = f"Listener error: {loop_err}" + if not self._is_completed: + self.status = "failed" + self.error_msg = f"Ошибка слушателя: {loop_err}" finally: - self.is_listening = False - if self.server: - try: - self.server.server_close() - except Exception: - pass - logger.info("OAUTH callback server stopped reason=%s", self.status) + self.stop_listener() self.server_thread = threading.Thread(target=_serve, daemon=True) self.server_thread.start() _ACTIVE_OAUTH_SESSIONS[self.session_id] = self return self.get_auth_url() - def _finalize_tokens(self) -> None: - """Exchange code for tokens and save into dedicated profile.""" + def handle_callback( + self, + code: Optional[str], + state: Optional[str], + error: Optional[str] = None, + source: str = "automatic", + ) -> Tuple[bool, str]: + """Unified callback handler for both automatic localhost HTTP listener and manual pasted URL. + + Thread-safe: guarantees atomic single completion and original PKCE verifier token exchange. + """ + with self._completion_lock: + logger.info("OAuth callback received (source=%s)", source) + + if self._is_completed: + logger.info("OAuth session already completed for profile=%s", self.profile_id) + return True, "Авторизация уже успешно завершена" + + if error: + self.status = "failed" + self.error_msg = f"Провайдер отклонил авторизацию ({error})" + logger.warning("OAuth error from provider (source=%s, error=%s)", source, error) + return False, self.error_msg + + if not code: + self.status = "failed" + self.error_msg = "Код авторизации отсутствует в callback URL" + logger.warning("OAuth code missing in callback (source=%s)", source) + return False, self.error_msg + + if state != self.state: + self.status = "failed" + self.error_msg = "Несовпадение параметра state. Callback относится к другой или устаревшей сессии." + logger.warning("OAuth state validation failed (source=%s)", source) + return False, self.error_msg + + logger.info("OAuth state validated (source=%s)", source) + + try: + logger.info("OAuth code exchange started (source=%s)", source) + tokens = exchange_code_for_tokens( + code, + redirect_uri=self.redirect_uri, + code_verifier=self.verifier, + ) + logger.info("OAuth code exchange completed (source=%s)", source) + + email = fetch_user_email(tokens["access_token"]) + logger.info("OAuth account identity resolved (email_found=%s)", bool(email)) + + # Format in standard gemini:antigravity shape + auth_data = { + "token": { + "access_token": tokens["access_token"], + "refresh_token": tokens["refresh_token"], + "expiry": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(tokens["expires_at"])), + }, + "email": email or "", + "auth_method": "oauth", + } + + # Save strictly to the chosen profile + saved_path = ProfileAuthManager.save_profile_auth("antigravity", self.profile_id, auth_data) + logger.info("Saved OAuth credentials for profile=%s to %s", self.profile_id, saved_path) + + self.completed_profile_info = { + "email": email or "Google Account", + "valid": True, + "profile_id": self.profile_id, + } + self._is_completed = True + self.status = "completed" + logger.info("OAuth session completed successfully for profile=%s", self.profile_id) + return True, "Авторизация успешно завершена" + + except Exception as e: + logger.error("Error finalizing OAuth for profile=%s: %s: %s", self.profile_id, type(e).__name__, e) + self.status = "failed" + self.error_msg = str(e) + return False, str(e) + + def handle_manual_callback_url(self, raw_url: str) -> Tuple[bool, str]: + """Parse user-pasted callback URL, extract code/state/error, and process through unified handler.""" + raw_url = raw_url.strip() + if not raw_url: + return False, "Пожалуйста, вставьте полный URL из адресной строки браузера." + try: - logger.info("OAUTH code exchange started") - tokens = exchange_code_for_tokens( - self.received_code, - redirect_uri=self.redirect_uri, - code_verifier=self.verifier, + # Handle potential protocol-less paste (e.g. 127.0.0.1:49725/oauth-callback?...) + if not raw_url.startswith("http://") and not raw_url.startswith("https://"): + raw_url = "http://" + raw_url + + parsed = urllib.parse.urlparse(raw_url) + params = urllib.parse.parse_qs(parsed.query) + + code = (params.get("code") or [None])[0] + state = (params.get("state") or [None])[0] + error = (params.get("error") or [None])[0] + error_desc = (params.get("error_description") or [None])[0] + + effective_err = error or error_desc + + return self.handle_callback( + code=code, + state=state, + error=effective_err, + source="manual", ) - logger.info("OAUTH code exchange completed") + except Exception as parse_err: + logger.warning("Error parsing manual callback URL: %s", parse_err) + return False, f"Не удалось разобрать URL: {parse_err}" - email = fetch_user_email(tokens["access_token"]) - - # Format in standard gemini:antigravity shape - auth_data = { - "token": { - "access_token": tokens["access_token"], - "refresh_token": tokens["refresh_token"], - "expiry": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(tokens["expires_at"])), - }, - "email": email or "", - "auth_method": "oauth", - } - - # Save strictly to the chosen profile - saved_path = ProfileAuthManager.save_profile_auth("antigravity", self.profile_id, auth_data) - logger.info("Saved OAuth credentials for profile=%s to %s", self.profile_id, saved_path) - - self.completed_profile_info = { - "email": email or "Google Account", - "valid": True, - "profile_id": self.profile_id, - } - self.status = "completed" - - except Exception as e: - logger.error("Error finalizing OAuth for profile=%s: %s: %s", self.profile_id, type(e).__name__, e) - self.status = "failed" - self.error_msg = str(e) - - def cancel(self) -> None: - """Explicitly cancel the session and shutdown listener.""" - self.status = "cancelled" + def stop_listener(self) -> None: + """Safely close the HTTP listener socket.""" self.is_listening = False if self.server: try: self.server.server_close() except Exception: pass - logger.info("OAUTH callback server stopped reason=cancelled") + logger.info("OAUTH callback server stopped reason=%s", self.status) + + def cancel(self) -> None: + """Explicitly cancel the session and shutdown listener.""" + with self._completion_lock: + if not self._is_completed: + self.status = "cancelled" + self.error_msg = "Авторизация отменена пользователем" + self.stop_listener() def start_profile_oauth(profile_id: str) -> Tuple[str, str, int]: diff --git a/src/antigravity_provider/router/ui/add_account_wizard.py b/src/antigravity_provider/router/ui/add_account_wizard.py index 692a8d2..127bb87 100644 --- a/src/antigravity_provider/router/ui/add_account_wizard.py +++ b/src/antigravity_provider/router/ui/add_account_wizard.py @@ -141,72 +141,125 @@ class AddAccountWizard(HubModal): def _build_antigravity_oauth_flow(self): ctk.CTkLabel( self.body, - text="Для Google Antigravity требуется вход через защищённый протокол Google OAuth:", - font=Theme.font_body(), + text="Для Google Antigravity требуется авторизация Google OAuth.", + font=Theme.font_body_bold(), + text_color=Theme.TEXT_PRIMARY, + anchor="w", + ).pack(fill="x", pady=(0, 2)) + + ctk.CTkLabel( + self.body, + text="После входа Google автоматически вернёт результат авторизации в Hermes Hub.", + font=Theme.font_caption(), text_color=Theme.TEXT_SECONDARY, wraplength=540, justify="left", - ).pack(anchor="w", pady=(0, 6)) + anchor="w", + ).pack(fill="x", pady=(0, 8)) - self.oauth_status_lbl = ctk.CTkLabel( - self.body, - text="Статус: Запуск локального слушателя OAuth...", - font=Theme.font_body_bold(), - text_color=Theme.STATUS_WARNING, - ) - self.oauth_status_lbl.pack(anchor="w", pady=(0, 6)) - - # URL display card - url_card = HubCard(self.body, fg_color=Theme.SURFACE_MUTED) - url_card.pack(fill="x", pady=(0, 8)) + # 1. Authorization link card + auth_card = HubCard(self.body, fg_color=Theme.SURFACE_MUTED) + auth_card.pack(fill="x", pady=(0, 8)) ctk.CTkLabel( - url_card, - text="Ссылка для авторизации:", + auth_card, + text="Ссылка авторизации", font=Theme.font_caption(), text_color=Theme.TEXT_SECONDARY, - ).pack(anchor="w", padx=10, pady=(8, 2)) + ).pack(anchor="w", padx=10, pady=(6, 2)) + + url_row = ctk.CTkFrame(auth_card, fg_color="transparent") + url_row.pack(fill="x", padx=10, pady=(0, 6)) self.oauth_url_entry = ctk.CTkEntry( - url_card, + url_row, font=Theme.font_mono(), - height=34, + height=32, fg_color=Theme.PRIMARY, border_color=Theme.BORDER, text_color=Theme.TEXT_PRIMARY, ) - self.oauth_url_entry.pack(fill="x", padx=10, pady=(2, 10)) + self.oauth_url_entry.pack(side="left", fill="x", expand=True, padx=(0, 6)) - # Action buttons row - self.oauth_btns_row = ctk.CTkFrame(self.body, fg_color="transparent") - self.oauth_btns_row.pack(fill="x", pady=4) + self.copy_url_btn = HubButton( + url_row, + text="📋", + variant="secondary", + width=40, + height=32, + command=self._copy_oauth_url, + ) + self.copy_url_btn.pack(side="right") + + action_row = ctk.CTkFrame(auth_card, fg_color="transparent") + action_row.pack(fill="x", padx=10, pady=(0, 8)) self.open_browser_btn = HubButton( - self.oauth_btns_row, - text="🌐 Открыть браузер", + action_row, + text="🌐 Открыть в браузере", variant="primary", - width=170, + width=180, command=self._open_oauth_browser, ) self.open_browser_btn.pack(side="left", padx=(0, 8)) - self.copy_url_btn = HubButton( - self.oauth_btns_row, - text="📋 Копировать ссылку", - variant="secondary", - width=160, - command=self._copy_oauth_url, - ) - self.copy_url_btn.pack(side="left", padx=(0, 8)) - self.regen_btn = HubButton( - self.oauth_btns_row, + action_row, text="🔄 Создать новую ссылку", variant="secondary", width=180, command=self._regenerate_oauth_session, ) + # 2. Manual Callback Fallback Card + manual_card = HubCard(self.body, fg_color=Theme.SURFACE_MUTED) + manual_card.pack(fill="x", pady=(0, 8)) + + ctk.CTkLabel( + manual_card, + text="▸ Не удалось завершить авторизацию автоматически?", + font=Theme.font_body_bold(), + text_color=Theme.TEXT_PRIMARY, + anchor="w", + ).pack(fill="x", padx=10, pady=(6, 2)) + + ctk.CTkLabel( + manual_card, + text="Вставьте полный URL из адресной строки браузера (если localhost вернул ошибку):", + font=Theme.font_caption(), + text_color=Theme.TEXT_SECONDARY, + anchor="w", + ).pack(fill="x", padx=10, pady=(0, 4)) + + self.manual_callback_entry = ctk.CTkEntry( + manual_card, + placeholder_text="http://127.0.0.1:49725/oauth-callback?state=...&code=...", + font=Theme.font_mono(), + height=32, + fg_color=Theme.PRIMARY, + border_color=Theme.BORDER, + text_color=Theme.TEXT_PRIMARY, + ) + self.manual_callback_entry.pack(fill="x", padx=10, pady=(0, 6)) + + self.manual_submit_btn = HubButton( + manual_card, + text="✓ Завершить авторизацию", + variant="secondary", + width=200, + command=self._submit_manual_callback, + ) + self.manual_submit_btn.pack(anchor="w", padx=10, pady=(0, 8)) + + # Status label + self.oauth_status_lbl = ctk.CTkLabel( + self.body, + text="Подготовка авторизации...", + font=Theme.font_body_bold(), + text_color=Theme.STATUS_WARNING, + ) + self.oauth_status_lbl.pack(pady=(4, 0)) + HubButton(self.footer, text="⬅ Назад", variant="secondary", width=100, command=self._show_step_1_provider).pack(side="left") # Initialize session immediately @@ -221,16 +274,21 @@ class AddAccountWizard(HubModal): self.oauth_url_entry.insert(0, self.oauth_url) self.oauth_url_entry.configure(state="readonly") self.oauth_status_lbl.configure( - text="Ссылка готова. Ожидание входа в Google.", + text="✓ Ссылка авторизации готова. Ожидание авторизации...", text_color=Theme.STATUS_HEALTHY, ) if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): self.regen_btn.pack_forget() + if hasattr(self, "manual_callback_entry"): + self.manual_callback_entry.configure(state="normal") + self.manual_callback_entry.delete(0, "end") + if hasattr(self, "manual_submit_btn"): + self.manual_submit_btn.configure(state="normal") self._polling_active = True threading.Thread(target=self._poll_oauth, daemon=True).start() except Exception as e: self.oauth_status_lbl.configure( - text=f"❌ Ошибка запуска OAuth слушателя: {e}", + text=f"Не удалось запустить локальный OAuth callback: {e}", text_color=Theme.STATUS_ERROR, ) if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): @@ -246,7 +304,7 @@ class AddAccountWizard(HubModal): ) webbrowser.open(self.oauth_url) self.oauth_status_lbl.configure( - text="🌐 Браузер открыт. Завершите авторизацию в Google...", + text="🌐 Браузер открыт. Ожидание завершения авторизации...", text_color=Theme.ACCENT, ) @@ -255,7 +313,7 @@ class AddAccountWizard(HubModal): self.clipboard_clear() self.clipboard_append(self.oauth_url) self.oauth_status_lbl.configure( - text="✅ Ссылка скопирована в буфер обмена!", + text="✓ Ссылка скопирована", text_color=Theme.STATUS_HEALTHY, ) @@ -268,6 +326,52 @@ class AddAccountWizard(HubModal): pass self._init_antigravity_oauth_session() + def _submit_manual_callback(self): + raw_url = self.manual_callback_entry.get().strip() + if not raw_url: + self.oauth_status_lbl.configure( + text="❌ Пожалуйста, вставьте полный URL из адресной строки браузера.", + text_color=Theme.STATUS_ERROR, + ) + return + + from antigravity_provider.router.profile_oauth import get_oauth_session + session = get_oauth_session(self.oauth_session_id) + if not session: + self.oauth_status_lbl.configure( + text="❌ Сессия авторизации не найдена. Создайте новую ссылку.", + text_color=Theme.STATUS_ERROR, + ) + if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): + self.regen_btn.pack(side="left") + return + + self.oauth_status_lbl.configure( + text="Проверка авторизации...", + text_color=Theme.ACCENT, + ) + + ok, msg = session.handle_manual_callback_url(raw_url) + if ok: + self.manual_callback_entry.configure(state="disabled") + self.manual_submit_btn.configure(state="disabled") + info = getattr(session, "completed_profile_info", {}) or {} + self.discovered_identity = info.get("email") or "Google Account" + self.discovered_models = ["gemini-2.5-pro", "gemini-2.5-flash", "gemini-2.5-flash-thinking"] + self.is_verified = True + self.oauth_status_lbl.configure( + text=f"✓ Аккаунт подключен: {self.discovered_identity}", + text_color=Theme.STATUS_HEALTHY, + ) + self._show_step_3_validation() + else: + self.oauth_status_lbl.configure( + text=f"❌ {msg}", + text_color=Theme.STATUS_ERROR, + ) + if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): + self.regen_btn.pack(side="left") + def _poll_oauth(self): from antigravity_provider.router.profile_oauth import get_oauth_session for _ in range(300): diff --git a/tests/test_oauth_lifecycle.py b/tests/test_oauth_lifecycle.py index 9269faf..c334519 100644 --- a/tests/test_oauth_lifecycle.py +++ b/tests/test_oauth_lifecycle.py @@ -1,22 +1,19 @@ """Hermes Hub — Comprehensive Google Antigravity OAuth Lifecycle Test Suite. Verifies: -1. Callback listener exists immediately after start. -2. redirect_uri strictly matches the actual listening port. -3. Listener remains alive during idle wait. -4. Simulated valid callback is accepted and exchanges tokens. -5. State mismatch callback is rejected. -6. Timeout terminates listener cleanly. -7. Cancel / wizard close terminates listener. -8. Listener stays alive until code exchange completes. -9. Retry after cancel / timeout succeeds without port collision. -10. Immediate Step 2 URL availability and single-session invariance. -11. Copy URL works without Open Browser. -12. Repeated Open Browser clicks reuse identical session and state. -13. Regeneration creates a new session / state / port and invalidates old callback. +TEST A — Automatic OAuth flow (session -> listener -> callback -> token exchange -> save) +TEST B — Manual callback fallback (session -> paste full URL -> original PKCE -> token exchange -> save) +TEST C — State mismatch rejection (callback state != session state -> reject, no token exchange) +TEST D — OAuth error callback (error=access_denied -> clean failure, no token exchange) +TEST E — Repeated 'Открыть в браузере' (state/port/verifier/URL invariance) +TEST F — Copy before open browser (immediate full URL in clipboard) +TEST G — Listener lifecycle (close wizard -> stopped; timeout -> stopped; restart -> old invalidated) +TEST H — Double completion protection (atomic single completion, no duplicate save) +TEST I — ERR_CONNECTION_REFUSED regression test: socket is verified listening BEFORE URL is published. """ from __future__ import annotations +import socket import time import urllib.parse import urllib.request @@ -43,142 +40,119 @@ def cleanup_oauth_sessions(): @pytest.mark.unit -def test_oauth_listener_exists_and_port_matches(tmp_path, monkeypatch): - """1 & 2: Verify callback listener exists and redirect_uri uses the exact bound port.""" +def test_a_automatic_oauth_flow(tmp_path, monkeypatch): + """TEST A: Automatic OAuth flow from session start to callback and completion.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session = get_oauth_session(session_id) - assert session is not None assert session.is_listening is True - assert f":{port}/oauth-callback" in session.redirect_uri - assert f":{port}/oauth-callback" in urllib.parse.unquote(auth_url) - assert session.status == "pending" - - -@pytest.mark.unit -def test_oauth_listener_remains_alive_during_wait(tmp_path, monkeypatch): - """3: Verify listener socket remains listening during idle wait.""" - monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) - - session_id, auth_url, port = start_profile_oauth("ag-orch-primary") - session = get_oauth_session(session_id) - - # Let it idle for 0.5s - time.sleep(0.5) - assert session.is_listening is True - assert session.status == "pending" - - # Ping non-callback path (should get 404, but server must stay alive) - try: - req = urllib.request.Request(f"http://127.0.0.1:{port}/random-probe") - with urllib.request.urlopen(req, timeout=2) as resp: - pass - except urllib.error.HTTPError as e: - assert e.code == 404 - - # Server must still be alive! - time.sleep(0.2) - assert session.is_listening is True - assert session.status == "pending" - - -@pytest.mark.unit -def test_simulated_valid_callback_success(tmp_path, monkeypatch): - """4 & 8: Verify valid callback is accepted and tokens are finalized.""" - monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) - - session_id, auth_url, port = start_profile_oauth("ag-orch-primary") - session = get_oauth_session(session_id) mock_tokens = { - "access_token": "ya29.mock_token", - "refresh_token": "1//mock_refresh", + "access_token": "ya29.mock_auto_token", + "refresh_token": "1//mock_auto_refresh", "expires_at": int(time.time()) + 3600, "token_type": "Bearer", } with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens", return_value=mock_tokens), \ - patch("antigravity_provider.router.profile_oauth.fetch_user_email", return_value="developer@google.com"): + patch("antigravity_provider.router.profile_oauth.fetch_user_email", return_value="auto_user@google.com"): - # Send HTTP GET callback matching state and code - callback_url = f"http://127.0.0.1:{port}/oauth-callback?code=mock_auth_code_123&state={session.state}" + callback_url = f"http://127.0.0.1:{port}/oauth-callback?code=mock_code_auto&state={session.state}" req = urllib.request.Request(callback_url) with urllib.request.urlopen(req, timeout=5) as resp: assert resp.status == 200 content = resp.read().decode("utf-8") - assert "Account Authorized" in content + assert "Авторизация успешно завершена" in content - # Wait briefly for thread finalization + # Wait for thread finalization deadline = time.time() + 3.0 while time.time() < deadline and session.status == "pending": time.sleep(0.05) assert session.status == "completed" - assert session.completed_profile_info["email"] == "developer@google.com" + assert session.completed_profile_info["email"] == "auto_user@google.com" + + # Verify saved credentials + saved = ProfileAuthManager.load_profile_auth("antigravity", "ag-orch-primary") + assert saved is not None + assert saved["email"] == "auto_user@google.com" @pytest.mark.unit -def test_state_mismatch_rejected(tmp_path, monkeypatch): - """5: Verify callback with mismatched state is rejected as failed.""" +def test_b_manual_callback_fallback(tmp_path, monkeypatch): + """TEST B: Manual callback fallback when localhost callback is not reached.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session = get_oauth_session(session_id) + assert session is not None - bad_state = "totally_wrong_state_value" - callback_url = f"http://127.0.0.1:{port}/oauth-callback?code=mock_code&state={bad_state}" - req = urllib.request.Request(callback_url) - with urllib.request.urlopen(req, timeout=5) as resp: - assert resp.status == 200 + mock_tokens = { + "access_token": "ya29.mock_manual_token", + "refresh_token": "1//mock_manual_refresh", + "expires_at": int(time.time()) + 3600, + "token_type": "Bearer", + } - deadline = time.time() + 3.0 - while time.time() < deadline and session.status == "pending": - time.sleep(0.05) + with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens", return_value=mock_tokens) as mock_exchange, \ + patch("antigravity_provider.router.profile_oauth.fetch_user_email", return_value="manual_user@google.com"): - assert session.status == "failed" - assert "State mismatch" in session.error_msg + pasted_url = f"http://127.0.0.1:{port}/oauth-callback?state={session.state}&code=mock_manual_code_789&scope=openid" + ok, msg = session.handle_manual_callback_url(pasted_url) + + assert ok is True + assert session.status == "completed" + assert session.completed_profile_info["email"] == "manual_user@google.com" + + # Ensure ORIGINAL PKCE verifier was used + assert mock_exchange.call_count == 1 + call_kwargs = mock_exchange.call_args[1] + assert call_kwargs["code_verifier"] == session.verifier @pytest.mark.unit -def test_cancel_session_terminates_listener(tmp_path, monkeypatch): - """6 & 7: Verify explicit cancel terminates listener immediately.""" +def test_c_state_mismatch(tmp_path, monkeypatch): + """TEST C: Callback with mismatched state is strictly rejected without token exchange.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session = get_oauth_session(session_id) - assert session.is_listening is True + assert session is not None - cancel_oauth_session(session_id) - time.sleep(0.2) + with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens") as mock_exchange: + pasted_url = f"http://127.0.0.1:{port}/oauth-callback?state=wrong_mismatched_state&code=mock_code" + ok, msg = session.handle_manual_callback_url(pasted_url) - assert session.status == "cancelled" - assert session.is_listening is False + assert ok is False + assert "Несовпадение" in msg or "state" in msg + assert session.status == "failed" + assert mock_exchange.call_count == 0 @pytest.mark.unit -def test_retry_after_cancel_works_cleanly(tmp_path, monkeypatch): - """9: Verify retry after cancel opens a new listener cleanly.""" +def test_d_oauth_error_callback(tmp_path, monkeypatch): + """TEST D: Provider error callback (e.g. access_denied) is cleanly handled.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) - # First attempt - s1_id, url1, port1 = start_profile_oauth("ag-orch-primary") - s1 = get_oauth_session(s1_id) - cancel_oauth_session(s1_id) - time.sleep(0.2) + session_id, auth_url, port = start_profile_oauth("ag-orch-primary") + session = get_oauth_session(session_id) + assert session is not None - # Second attempt - s2_id, url2, port2 = start_profile_oauth("ag-orch-primary") - s2 = get_oauth_session(s2_id) - assert s2 is not None - assert s2.is_listening is True - assert s2.session_id != s1_id + with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens") as mock_exchange: + error_url = f"http://127.0.0.1:{port}/oauth-callback?error=access_denied&error_description=User+denied+access&state={session.state}" + ok, msg = session.handle_manual_callback_url(error_url) + + assert ok is False + assert "отклонил" in msg or "access_denied" in msg + assert session.status == "failed" + assert mock_exchange.call_count == 0 @pytest.mark.unit -def test_single_session_invariance_and_regeneration(tmp_path, monkeypatch): - """10-13: Test Wizard Step 2 immediate URL availability, single-session reuse, and regeneration.""" +def test_e_repeated_open_browser_invariance(tmp_path, monkeypatch): + """TEST E: Repeated 'Открыть в браузере' does NOT change session, state, verifier, or URL.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) pytest.importorskip("customtkinter") import customtkinter as ctk @@ -190,44 +164,137 @@ def test_single_session_invariance_and_regeneration(tmp_path, monkeypatch): wizard = AddAccountWizard(root) wizard.selected_provider = "antigravity" wizard.target_slot = "ag-spare-1" - - # 1. Opening Step 2 initializes OAuth immediately wizard._show_step_2_auth() - assert wizard.oauth_url is not None - assert wizard.oauth_session_id is not None - assert wizard.oauth_url.startswith("https://accounts.google.com") - # 2. URL entry contains the URL - entry_text = wizard.oauth_url_entry.get() - assert entry_text == wizard.oauth_url - - # 3. Repeated Open Browser does NOT change session or state orig_session_id = wizard.oauth_session_id orig_url = wizard.oauth_url + orig_port = wizard.oauth_port + + session = get_oauth_session(orig_session_id) + orig_state = session.state + orig_verifier = session.verifier with patch("webbrowser.open") as mock_open: wizard._open_oauth_browser() - assert mock_open.call_count == 1 - assert mock_open.call_args[0][0] == orig_url - assert wizard.oauth_session_id == orig_session_id - wizard._open_oauth_browser() - assert mock_open.call_count == 2 + wizard._open_oauth_browser() + + assert mock_open.call_count == 3 + for call in mock_open.call_args_list: + assert call[0][0] == orig_url + assert wizard.oauth_session_id == orig_session_id assert wizard.oauth_url == orig_url - - # 4. Explicit regeneration creates NEW session and state - wizard._regenerate_oauth_session() - new_session_id = wizard.oauth_session_id - new_url = wizard.oauth_url - - assert new_session_id != orig_session_id - assert new_url != orig_url - - # Old session must be cancelled - old_session = get_oauth_session(orig_session_id) - assert old_session is None or old_session.status == "cancelled" + assert wizard.oauth_port == orig_port + assert session.state == orig_state + assert session.verifier == orig_verifier wizard.destroy() finally: root.destroy() + + +@pytest.mark.unit +def test_f_copy_before_open_browser(tmp_path, monkeypatch): + """TEST F: Copy button works immediately upon entering Step 2 without opening browser.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + pytest.importorskip("customtkinter") + import customtkinter as ctk + from antigravity_provider.router.ui.add_account_wizard import AddAccountWizard + + root = ctk.CTk() + root.withdraw() + try: + wizard = AddAccountWizard(root) + wizard.selected_provider = "antigravity" + wizard.target_slot = "ag-spare-1" + wizard._show_step_2_auth() + + assert wizard.oauth_url is not None + assert wizard.oauth_url.startswith("https://accounts.google.com") + + # Copy without opening browser + wizard._copy_oauth_url() + clipboard_content = wizard.clipboard_get() + assert clipboard_content == wizard.oauth_url + + wizard.destroy() + finally: + root.destroy() + + +@pytest.mark.unit +def test_g_listener_lifecycle_cleanup(tmp_path, monkeypatch): + """TEST G: Listener is stopped on cancel/destroy/timeout and restarted cleanly.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + # 1. Cancel terminates listener + s1_id, url1, p1 = start_profile_oauth("ag-orch-primary") + s1 = get_oauth_session(s1_id) + assert s1.is_listening is True + + cancel_oauth_session(s1_id) + time.sleep(0.2) + assert s1.is_listening is False + assert s1.status == "cancelled" + + # 2. Restart creates active new session + s2_id, url2, p2 = start_profile_oauth("ag-orch-primary") + s2 = get_oauth_session(s2_id) + assert s2.is_listening is True + assert s2_id != s1_id + + +@pytest.mark.unit +def test_h_double_completion_protection(tmp_path, monkeypatch): + """TEST H: Double completion (automatic + manual race) executes exactly once.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + session_id, auth_url, port = start_profile_oauth("ag-orch-primary") + session = get_oauth_session(session_id) + + mock_tokens = { + "access_token": "ya29.mock_double_token", + "refresh_token": "1//mock_double_refresh", + "expires_at": int(time.time()) + 3600, + "token_type": "Bearer", + } + + with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens", return_value=mock_tokens) as mock_exchange, \ + patch("antigravity_provider.router.profile_oauth.fetch_user_email", return_value="race_user@google.com"): + + # 1. First completion (automatic) + ok1, msg1 = session.handle_callback(code="code_1", state=session.state, source="automatic") + assert ok1 is True + assert mock_exchange.call_count == 1 + + # 2. Second completion (manual duplicate attempt with same session) + ok2, msg2 = session.handle_manual_callback_url(f"http://127.0.0.1:{port}/oauth-callback?state={session.state}&code=code_1") + assert ok2 is True + assert "уже успешно завершена" in msg2 + + # Token exchange MUST have occurred exactly once + assert mock_exchange.call_count == 1 + + +@pytest.mark.unit +def test_i_err_connection_refused_regression_listener_ready_before_url(tmp_path, monkeypatch): + """TEST I: Architectural invariant — listener socket is READY before URL is published.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + session = ProfileOAuthSession("ag-orch-primary") + assert session.is_listening is False + + # Start session + auth_url = session.start() + + # The socket MUST be listening and connectable BEFORE the user could receive the URL + assert session.is_listening is True + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(2.0) + connect_result = sock.connect_ex(("127.0.0.1", session.port)) + sock.close() + + assert connect_result == 0, f"ERR_CONNECTION_REFUSED: Listener on port {session.port} was not ready!" + assert f":{session.port}/oauth-callback" in urllib.parse.unquote(auth_url) + session.cancel()