ochenstarik/hermes-hub
1
0
Fork 0
Code Issues Pull requests Projects Releases 13 Packages Wiki Activity Actions

feat(oauth): adopt Cockpit Tools model with immediate URL readiness, manual callback fallback, and race protection

Browse source
This commit is contained in:
Hermes Team 2026-08-20 20:17:37 +07:00
parent 0d9005f55b
commit 249a88867c
3 changed files with 510 additions and 251 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

262
src/antigravity_provider/router/profile_oauth.py
View file

@ -1,10 +1,11 @@
"""Profile OAuth manager for interactive Google / Antigravity account linking. """Profile OAuth manager for interactive Google / Antigravity account linking.
Features: Architecture:
- Immediate listener startup with verified socket binding. - Deterministic lifecycle: session creation -> port allocation -> listener verified ready -> URL generation.
- Dynamic or standard (51121) port binding with strict redirect_uri alignment. - Unified callback pipeline: both automatic localhost HTTP callback and manual pasted callback URL
- Sanitized diagnostic logging without exposing codes, tokens, or client secrets. flow through handle_callback() for state validation, original PKCE verifier lookup, and token exchange.
- Deterministic session lifecycle, state validation, and clean cancellation. - Thread-safe single completion guarantee preventing duplicate saves.
- Zero secret logging policy (code, verifier, tokens, secrets are excluded from logs).
""" """
from __future__ import annotations from __future__ import annotations
@ -36,6 +37,31 @@ logger = logging.getLogger("hermes.router.profile_oauth")
_ACTIVE_OAUTH_SESSIONS: Dict[str, "ProfileOAuthSession"] = {} _ACTIVE_OAUTH_SESSIONS: Dict[str, "ProfileOAuthSession"] = {}
SUCCESS_HTML = (
b"<!DOCTYPE html><html><head><meta charset='utf-8'><title>Hermes Hub</title>"
b"<style>body{background:#0f172a;color:#f8fafc;font-family:-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,sans-serif;"
b"display:flex;align-items:center;justify-content:center;height:100vh;margin:0;}"
b".card{background:#1e293b;padding:36px 48px;border-radius:16px;border:1px solid #334155;text-align:center;"
b"box-shadow:0 20px 35px rgba(0,0,0,0.4);max-width:440px;}"
b"h1{color:#10b981;font-size:22px;margin:0 0 12px 0;}"
b"p{color:#94a3b8;font-size:14px;line-height:1.5;margin:0;}</style></head>"
b"<body><div class='card'><h1>&#10004; \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.</h1>"
b"<p>\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.</p></div></body></html>"
)
ERROR_HTML = (
b"<!DOCTYPE html><html><head><meta charset='utf-8'><title>Hermes Hub</title>"
b"<style>body{background:#0f172a;color:#f8fafc;font-family:-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,sans-serif;"
b"display:flex;align-items:center;justify-content:center;height:100vh;margin:0;}"
b".card{background:#1e293b;padding:36px 48px;border-radius:16px;border:1px solid #ef4444;text-align:center;"
b"box-shadow:0 20px 35px rgba(0,0,0,0.4);max-width:440px;}"
b"h1{color:#ef4444;font-size:22px;margin:0 0 12px 0;}"
b"p{color:#94a3b8;font-size:14px;line-height:1.5;margin:0;}</style></head>"
b"<body><div class='card'><h1>\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.</h1>"
b"<p>\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.</p></div></body></html>"
)
class _ProfileOAuthCallbackHandler(BaseHTTPRequestHandler): class _ProfileOAuthCallbackHandler(BaseHTTPRequestHandler):
server: "_ProfileOAuthServer" server: "_ProfileOAuthServer"
@ -46,20 +72,22 @@ class _ProfileOAuthCallbackHandler(BaseHTTPRequestHandler):
return return
params = urllib.parse.parse_qs(parsed.query) params = urllib.parse.parse_qs(parsed.query)
self.server.session.received_state = (params.get("state") or [None])[0] received_state = (params.get("state") or [None])[0]
self.server.session.received_error = (params.get("error") or [None])[0] received_error = (params.get("error") or [None])[0]
self.server.session.received_code = (params.get("code") or [None])[0] received_code = (params.get("code") or [None])[0]
error_desc = (params.get("error_description") or [None])[0]
body = ( effective_err = received_error or error_desc
b"<!DOCTYPE html><html><head><meta charset='utf-8'><title>Hermes Account Linked</title>"
b"<style>body{background:#0f172a;color:#f8fafc;font-family:sans-serif;display:flex;align-items:center;" ok, msg = self.server.session.handle_callback(
b"justify-content:center;height:100vh;margin:0;}.card{background:#1e293b;padding:32px;border-radius:12px;" code=received_code,
b"border:1px solid #334155;text-align:center;box-shadow:0 10px 25px rgba(0,0,0,0.5);}h1{color:#10b981;font-size:24px;}" state=received_state,
b"p{color:#94a3b8;margin-top:12px;}</style></head><body>" error=effective_err,
b"<div class='card'><h1>&#10004; Account Authorized</h1>" source="automatic",
b"<p>You can close this tab and return to the Hermes Account Manager.</p></div></body></html>"
) )
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-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close") self.send_header("Connection", "close")
@ -89,10 +117,6 @@ class ProfileOAuthSession:
self.verifier, self.challenge = _pkce_pair() self.verifier, self.challenge = _pkce_pair()
self.redirect_uri = f"http://{CALLBACK_HOST}:{self.port}{CALLBACK_PATH}" 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: Optional[_ProfileOAuthServer] = None
self.server_thread: Optional[threading.Thread] = None self.server_thread: Optional[threading.Thread] = None
self.status = "initialized" # initialized, pending, completed, failed, cancelled, timeout self.status = "initialized" # initialized, pending, completed, failed, cancelled, timeout
@ -101,6 +125,9 @@ class ProfileOAuthSession:
self.completed_profile_info: Optional[dict] = None self.completed_profile_info: Optional[dict] = None
self.is_listening = False self.is_listening = False
self._completion_lock = threading.Lock()
self._is_completed = False
def get_auth_url(self) -> str: def get_auth_url(self) -> str:
params = { params = {
"client_id": CLIENT_ID, "client_id": CLIENT_ID,
@ -140,94 +167,155 @@ class ProfileOAuthSession:
self.server.handle_request() self.server.handle_request()
except Exception as req_err: except Exception as req_err:
logger.warning("OAUTH handle_request warning: %s: %s", type(req_err).__name__, req_err) logger.warning("OAUTH handle_request warning: %s: %s", type(req_err).__name__, req_err)
if self.received_code or self.received_error: if self._is_completed or self.status != "pending":
logger.info("OAUTH callback received")
break break
if self.received_error: if self.status == "pending" and not self._is_completed:
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":
self.status = "timeout" 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") logger.info("OAUTH callback server stopped reason=timeout")
except Exception as loop_err: except Exception as loop_err:
logger.error("OAUTH listener exception: %s: %s", type(loop_err).__name__, loop_err) logger.error("OAUTH listener exception: %s: %s", type(loop_err).__name__, loop_err)
self.status = "failed" if not self._is_completed:
self.error_msg = f"Listener error: {loop_err}" self.status = "failed"
self.error_msg = f"Ошибка слушателя: {loop_err}"
finally: finally:
self.is_listening = False self.stop_listener()
if self.server:
try:
self.server.server_close()
except Exception:
pass
logger.info("OAUTH callback server stopped reason=%s", self.status)
self.server_thread = threading.Thread(target=_serve, daemon=True) self.server_thread = threading.Thread(target=_serve, daemon=True)
self.server_thread.start() self.server_thread.start()
_ACTIVE_OAUTH_SESSIONS[self.session_id] = self _ACTIVE_OAUTH_SESSIONS[self.session_id] = self
return self.get_auth_url() return self.get_auth_url()
def _finalize_tokens(self) -> None: def handle_callback(
"""Exchange code for tokens and save into dedicated profile.""" 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: try:
logger.info("OAUTH code exchange started") # Handle potential protocol-less paste (e.g. 127.0.0.1:49725/oauth-callback?...)
tokens = exchange_code_for_tokens( if not raw_url.startswith("http://") and not raw_url.startswith("https://"):
self.received_code, raw_url = "http://" + raw_url
redirect_uri=self.redirect_uri,
code_verifier=self.verifier, 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"]) def stop_listener(self) -> None:
"""Safely close the HTTP listener socket."""
# 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"
self.is_listening = False self.is_listening = False
if self.server: if self.server:
try: try:
self.server.server_close() self.server.server_close()
except Exception: except Exception:
pass 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]: def start_profile_oauth(profile_id: str) -> Tuple[str, str, int]:

184
src/antigravity_provider/router/ui/add_account_wizard.py
View file

@ -141,72 +141,125 @@ class AddAccountWizard(HubModal):
def _build_antigravity_oauth_flow(self): def _build_antigravity_oauth_flow(self):
ctk.CTkLabel( ctk.CTkLabel(
self.body, self.body,
text="Для Google Antigravity требуется вход через защищённый протокол Google OAuth:", text="Для Google Antigravity требуется авторизация Google OAuth.",
font=Theme.font_body(), 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, text_color=Theme.TEXT_SECONDARY,
wraplength=540, wraplength=540,
justify="left", justify="left",
).pack(anchor="w", pady=(0, 6)) anchor="w",
).pack(fill="x", pady=(0, 8))
self.oauth_status_lbl = ctk.CTkLabel( # 1. Authorization link card
self.body, auth_card = HubCard(self.body, fg_color=Theme.SURFACE_MUTED)
text="Статус: Запуск локального слушателя OAuth...", auth_card.pack(fill="x", pady=(0, 8))
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))
ctk.CTkLabel( ctk.CTkLabel(
url_card, auth_card,
text="Ссылка для авторизации:", text="Ссылка авторизации",
font=Theme.font_caption(), font=Theme.font_caption(),
text_color=Theme.TEXT_SECONDARY, 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( self.oauth_url_entry = ctk.CTkEntry(
url_card, url_row,
font=Theme.font_mono(), font=Theme.font_mono(),
height=34, height=32,
fg_color=Theme.PRIMARY, fg_color=Theme.PRIMARY,
border_color=Theme.BORDER, border_color=Theme.BORDER,
text_color=Theme.TEXT_PRIMARY, 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.copy_url_btn = HubButton(
self.oauth_btns_row = ctk.CTkFrame(self.body, fg_color="transparent") url_row,
self.oauth_btns_row.pack(fill="x", pady=4) 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.open_browser_btn = HubButton(
self.oauth_btns_row, action_row,
text="🌐 Открыть браузер", text="🌐 Открыть в браузере",
variant="primary", variant="primary",
width=170, width=180,
command=self._open_oauth_browser, command=self._open_oauth_browser,
) )
self.open_browser_btn.pack(side="left", padx=(0, 8)) 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.regen_btn = HubButton(
self.oauth_btns_row, action_row,
text="🔄 Создать новую ссылку", text="🔄 Создать новую ссылку",
variant="secondary", variant="secondary",
width=180, width=180,
command=self._regenerate_oauth_session, 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") HubButton(self.footer, text="⬅ Назад", variant="secondary", width=100, command=self._show_step_1_provider).pack(side="left")
# Initialize session immediately # Initialize session immediately
@ -221,16 +274,21 @@ class AddAccountWizard(HubModal):
self.oauth_url_entry.insert(0, self.oauth_url) self.oauth_url_entry.insert(0, self.oauth_url)
self.oauth_url_entry.configure(state="readonly") self.oauth_url_entry.configure(state="readonly")
self.oauth_status_lbl.configure( self.oauth_status_lbl.configure(
text="Ссылка готова. Ожидание входа в Google.", text="✓ Ссылка авторизации готова. Ожидание авторизации...",
text_color=Theme.STATUS_HEALTHY, text_color=Theme.STATUS_HEALTHY,
) )
if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists():
self.regen_btn.pack_forget() 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 self._polling_active = True
threading.Thread(target=self._poll_oauth, daemon=True).start() threading.Thread(target=self._poll_oauth, daemon=True).start()
except Exception as e: except Exception as e:
self.oauth_status_lbl.configure( self.oauth_status_lbl.configure(
text=f"❌ Ошибка запуска OAuth слушателя: {e}", text=f"Не удалось запустить локальный OAuth callback: {e}",
text_color=Theme.STATUS_ERROR, text_color=Theme.STATUS_ERROR,
) )
if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists(): if hasattr(self, "regen_btn") and self.regen_btn.winfo_exists():
@ -246,7 +304,7 @@ class AddAccountWizard(HubModal):
) )
webbrowser.open(self.oauth_url) webbrowser.open(self.oauth_url)
self.oauth_status_lbl.configure( self.oauth_status_lbl.configure(
text="🌐 Браузер открыт. Завершите авторизацию в Google...", text="🌐 Браузер открыт. Ожидание завершения авторизации...",
text_color=Theme.ACCENT, text_color=Theme.ACCENT,
) )
@ -255,7 +313,7 @@ class AddAccountWizard(HubModal):
self.clipboard_clear() self.clipboard_clear()
self.clipboard_append(self.oauth_url) self.clipboard_append(self.oauth_url)
self.oauth_status_lbl.configure( self.oauth_status_lbl.configure(
text="✅ Ссылка скопирована в буфер обмена!", text="✓ Ссылка скопирована",
text_color=Theme.STATUS_HEALTHY, text_color=Theme.STATUS_HEALTHY,
) )
@ -268,6 +326,52 @@ class AddAccountWizard(HubModal):
pass pass
self._init_antigravity_oauth_session() 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): def _poll_oauth(self):
from antigravity_provider.router.profile_oauth import get_oauth_session from antigravity_provider.router.profile_oauth import get_oauth_session
for _ in range(300): for _ in range(300):

315
tests/test_oauth_lifecycle.py
View file

@ -1,22 +1,19 @@
"""Hermes Hub — Comprehensive Google Antigravity OAuth Lifecycle Test Suite. """Hermes Hub — Comprehensive Google Antigravity OAuth Lifecycle Test Suite.
Verifies: Verifies:
1. Callback listener exists immediately after start. TEST A — Automatic OAuth flow (session -> listener -> callback -> token exchange -> save)
2. redirect_uri strictly matches the actual listening port. TEST B — Manual callback fallback (session -> paste full URL -> original PKCE -> token exchange -> save)
3. Listener remains alive during idle wait. TEST C — State mismatch rejection (callback state != session state -> reject, no token exchange)
4. Simulated valid callback is accepted and exchanges tokens. TEST D — OAuth error callback (error=access_denied -> clean failure, no token exchange)
5. State mismatch callback is rejected. TEST E — Repeated 'Открыть в браузере' (state/port/verifier/URL invariance)
6. Timeout terminates listener cleanly. TEST F — Copy before open browser (immediate full URL in clipboard)
7. Cancel / wizard close terminates listener. TEST G — Listener lifecycle (close wizard -> stopped; timeout -> stopped; restart -> old invalidated)
8. Listener stays alive until code exchange completes. TEST H — Double completion protection (atomic single completion, no duplicate save)
9. Retry after cancel / timeout succeeds without port collision. TEST I — ERR_CONNECTION_REFUSED regression test: socket is verified listening BEFORE URL is published.
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.
""" """
from __future__ import annotations from __future__ import annotations
import socket
import time import time
import urllib.parse import urllib.parse
import urllib.request import urllib.request
@ -43,142 +40,119 @@ def cleanup_oauth_sessions():
@pytest.mark.unit @pytest.mark.unit
def test_oauth_listener_exists_and_port_matches(tmp_path, monkeypatch): def test_a_automatic_oauth_flow(tmp_path, monkeypatch):
"""1 & 2: Verify callback listener exists and redirect_uri uses the exact bound port.""" """TEST A: Automatic OAuth flow from session start to callback and completion."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session_id, auth_url, port = start_profile_oauth("ag-orch-primary")
session = get_oauth_session(session_id) session = get_oauth_session(session_id)
assert session is not None assert session is not None
assert session.is_listening is True 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 = { mock_tokens = {
"access_token": "ya29.mock_token", "access_token": "ya29.mock_auto_token",
"refresh_token": "1//mock_refresh", "refresh_token": "1//mock_auto_refresh",
"expires_at": int(time.time()) + 3600, "expires_at": int(time.time()) + 3600,
"token_type": "Bearer", "token_type": "Bearer",
} }
with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens", return_value=mock_tokens), \ 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_code_auto&state={session.state}"
callback_url = f"http://127.0.0.1:{port}/oauth-callback?code=mock_auth_code_123&state={session.state}"
req = urllib.request.Request(callback_url) req = urllib.request.Request(callback_url)
with urllib.request.urlopen(req, timeout=5) as resp: with urllib.request.urlopen(req, timeout=5) as resp:
assert resp.status == 200 assert resp.status == 200
content = resp.read().decode("utf-8") 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 deadline = time.time() + 3.0
while time.time() < deadline and session.status == "pending": while time.time() < deadline and session.status == "pending":
time.sleep(0.05) time.sleep(0.05)
assert session.status == "completed" 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 @pytest.mark.unit
def test_state_mismatch_rejected(tmp_path, monkeypatch): def test_b_manual_callback_fallback(tmp_path, monkeypatch):
"""5: Verify callback with mismatched state is rejected as failed.""" """TEST B: Manual callback fallback when localhost callback is not reached."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session_id, auth_url, port = start_profile_oauth("ag-orch-primary")
session = get_oauth_session(session_id) session = get_oauth_session(session_id)
assert session is not None
bad_state = "totally_wrong_state_value" mock_tokens = {
callback_url = f"http://127.0.0.1:{port}/oauth-callback?code=mock_code&state={bad_state}" "access_token": "ya29.mock_manual_token",
req = urllib.request.Request(callback_url) "refresh_token": "1//mock_manual_refresh",
with urllib.request.urlopen(req, timeout=5) as resp: "expires_at": int(time.time()) + 3600,
assert resp.status == 200 "token_type": "Bearer",
}
deadline = time.time() + 3.0 with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens", return_value=mock_tokens) as mock_exchange, \
while time.time() < deadline and session.status == "pending": patch("antigravity_provider.router.profile_oauth.fetch_user_email", return_value="manual_user@google.com"):
time.sleep(0.05)
assert session.status == "failed" pasted_url = f"http://127.0.0.1:{port}/oauth-callback?state={session.state}&code=mock_manual_code_789&scope=openid"
assert "State mismatch" in session.error_msg 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 @pytest.mark.unit
def test_cancel_session_terminates_listener(tmp_path, monkeypatch): def test_c_state_mismatch(tmp_path, monkeypatch):
"""6 & 7: Verify explicit cancel terminates listener immediately.""" """TEST C: Callback with mismatched state is strictly rejected without token exchange."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
session_id, auth_url, port = start_profile_oauth("ag-orch-primary") session_id, auth_url, port = start_profile_oauth("ag-orch-primary")
session = get_oauth_session(session_id) session = get_oauth_session(session_id)
assert session.is_listening is True assert session is not None
cancel_oauth_session(session_id) with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens") as mock_exchange:
time.sleep(0.2) 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 ok is False
assert session.is_listening is False assert "Несовпадение" in msg or "state" in msg
assert session.status == "failed"
assert mock_exchange.call_count == 0
@pytest.mark.unit @pytest.mark.unit
def test_retry_after_cancel_works_cleanly(tmp_path, monkeypatch): def test_d_oauth_error_callback(tmp_path, monkeypatch):
"""9: Verify retry after cancel opens a new listener cleanly.""" """TEST D: Provider error callback (e.g. access_denied) is cleanly handled."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
# First attempt session_id, auth_url, port = start_profile_oauth("ag-orch-primary")
s1_id, url1, port1 = start_profile_oauth("ag-orch-primary") session = get_oauth_session(session_id)
s1 = get_oauth_session(s1_id) assert session is not None
cancel_oauth_session(s1_id)
time.sleep(0.2)
# Second attempt with patch("antigravity_provider.router.profile_oauth.exchange_code_for_tokens") as mock_exchange:
s2_id, url2, port2 = start_profile_oauth("ag-orch-primary") error_url = f"http://127.0.0.1:{port}/oauth-callback?error=access_denied&error_description=User+denied+access&state={session.state}"
s2 = get_oauth_session(s2_id) ok, msg = session.handle_manual_callback_url(error_url)
assert s2 is not None
assert s2.is_listening is True assert ok is False
assert s2.session_id != s1_id assert "отклонил" in msg or "access_denied" in msg
assert session.status == "failed"
assert mock_exchange.call_count == 0
@pytest.mark.unit @pytest.mark.unit
def test_single_session_invariance_and_regeneration(tmp_path, monkeypatch): def test_e_repeated_open_browser_invariance(tmp_path, monkeypatch):
"""10-13: Test Wizard Step 2 immediate URL availability, single-session reuse, and regeneration.""" """TEST E: Repeated 'Открыть в браузере' does NOT change session, state, verifier, or URL."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
pytest.importorskip("customtkinter") pytest.importorskip("customtkinter")
import customtkinter as ctk import customtkinter as ctk
@ -190,44 +164,137 @@ def test_single_session_invariance_and_regeneration(tmp_path, monkeypatch):
wizard = AddAccountWizard(root) wizard = AddAccountWizard(root)
wizard.selected_provider = "antigravity" wizard.selected_provider = "antigravity"
wizard.target_slot = "ag-spare-1" wizard.target_slot = "ag-spare-1"
# 1. Opening Step 2 initializes OAuth immediately
wizard._show_step_2_auth() 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_session_id = wizard.oauth_session_id
orig_url = wizard.oauth_url 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: with patch("webbrowser.open") as mock_open:
wizard._open_oauth_browser() 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() 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_session_id == orig_session_id
assert wizard.oauth_url == orig_url assert wizard.oauth_url == orig_url
assert wizard.oauth_port == orig_port
# 4. Explicit regeneration creates NEW session and state assert session.state == orig_state
wizard._regenerate_oauth_session() assert session.verifier == orig_verifier
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"
wizard.destroy() wizard.destroy()
finally: finally:
root.destroy() 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()