From 3b423ca351419bfbf8b95d4f0ae87c0cc59b0b14 Mon Sep 17 00:00:00 2001 From: ochenstarik-ui <267932263+ochenstarik-ui@users.noreply.github.com> Date: Tue, 1 Sep 2026 14:25:13 +0700 Subject: [PATCH] =?UTF-8?q?feat(antigravity):=20=D0=BD=D0=B0=D1=82=D0=B8?= =?UTF-8?q?=D0=B2=D0=BD=D1=8B=D0=B9=20=D0=B2=D1=85=D0=BE=D0=B4=20=D1=87?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D0=B7=20agy=20=D0=B2=20=D1=82=D0=B5=D1=80?= =?UTF-8?q?=D0=BC=D0=B8=D0=BD=D0=B0=D0=BB=D0=B5=20(A57)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Реализован запуск agy в терминале с изолированным HOME (x-terminal-emulator, gnome-terminal, konsole, xfce4-terminal, tilix, alacritty, kitty, terminator, urxvt, foot, xterm / Windows wt, cmd) - Опрос появления antigravity-oauth-token вместо ожидания процесса терминала - Защита занятого слота от перезаписи без подтверждения - Честное извлечение email без выдумывания identity и предотвращение дубликатов слотов - Сохранён браузерный OAuth в качестве запасного пути - Добавлены 12 тестов в test_a57_agy_native_login.py, 640 тестов проходят --- src/antigravity_provider/agy_subprocess.py | 478 ++++++++++++++++-- .../router/action_handler.py | 33 +- .../router/profile_manager.py | 34 +- src/antigravity_provider/router/web/server.py | 4 +- .../router/web/static/app.js | 233 ++++++++- tests/test_a57_agy_native_login.py | 384 ++++++++++++++ ...test_state_layer_and_event_driven_quota.py | 3 +- uv.lock | 2 +- 8 files changed, 1100 insertions(+), 71 deletions(-) create mode 100644 tests/test_a57_agy_native_login.py diff --git a/src/antigravity_provider/agy_subprocess.py b/src/antigravity_provider/agy_subprocess.py index 1769767..d285986 100644 --- a/src/antigravity_provider/agy_subprocess.py +++ b/src/antigravity_provider/agy_subprocess.py @@ -17,6 +17,7 @@ import json import logging import os import re +import secrets import shutil import subprocess import time @@ -289,71 +290,137 @@ def check_profile_native_auth_status(profile_id: str) -> tuple[bool, str | None, """Check if agy native authentication has completed in profile's directory. Returns (is_authenticated, email, auth_data). - A22 Requirement: Detection without credential logging or stream interception. + A22/A57 Requirement: Detection without credential logging or stream interception. + Checks .gemini/antigravity-cli/antigravity-oauth-token first, then .gemini/oauth_creds.json. """ from antigravity_provider.router.adapters.antigravity_adapter import get_profile_env_dir profile_dir = get_profile_env_dir(profile_id) gemini_dir = profile_dir / ".gemini" + cli_dir = gemini_dir / "antigravity-cli" + cli_token_file = cli_dir / "antigravity-oauth-token" creds_file = gemini_dir / "oauth_creds.json" accounts_file = gemini_dir / "google_accounts.json" + auth_file = profile_dir / "auth.json" - if not creds_file.is_file() or creds_file.stat().st_size == 0: + token_data: dict[str, Any] | None = None + auth_method: str = "consumer" + + # 1. Primary source: agy 2.0 native token file (.gemini/antigravity-cli/antigravity-oauth-token) + if cli_token_file.is_file() and cli_token_file.stat().st_size > 0: + try: + cli_raw = json.loads(cli_token_file.read_text(encoding="utf-8")) + if isinstance(cli_raw, dict): + auth_method = cli_raw.get("auth_method", "consumer") + inner = cli_raw.get("token") + if isinstance(inner, dict): + token_data = inner + elif "access_token" in cli_raw or "refresh_token" in cli_raw: + token_data = cli_raw + except Exception as exc: + logger.debug("Error reading cli_token_file for %s: %s", profile_id, exc) + + # 2. Secondary source: Gemini CLI legacy credentials (.gemini/oauth_creds.json) + if not token_data and creds_file.is_file() and creds_file.stat().st_size > 0: + try: + creds_raw = json.loads(creds_file.read_text(encoding="utf-8")) + if isinstance(creds_raw, dict): + token_data = creds_raw + auth_method = "oauth" + except Exception as exc: + logger.debug("Error reading creds_file for %s: %s", profile_id, exc) + + # 3. Third source: existing auth.json + if not token_data and auth_file.is_file() and auth_file.stat().st_size > 0: + try: + a_raw = json.loads(auth_file.read_text(encoding="utf-8")) + if isinstance(a_raw, dict): + inner = a_raw.get("token") or a_raw.get("tokens") + if isinstance(inner, dict): + token_data = inner + auth_method = a_raw.get("auth_method", "oauth") + except Exception as exc: + logger.debug("Error reading auth_file for %s: %s", profile_id, exc) + + if not token_data or not isinstance(token_data, dict): return False, None, None + access_token = token_data.get("access_token") + refresh_token = token_data.get("refresh_token") + if not access_token and not refresh_token: + return False, None, None + + # Extract email identity truthfully (P0-3: do not invent) + email: str | None = None + if accounts_file.is_file() and accounts_file.stat().st_size > 0: + try: + acc_data = json.loads(accounts_file.read_text(encoding="utf-8")) + if isinstance(acc_data, dict) and acc_data.get("active"): + em = str(acc_data["active"]).strip() + if "@" in em: + email = em + except Exception: + email = None + + if not email and auth_file.is_file(): + try: + a_data = json.loads(auth_file.read_text(encoding="utf-8")) + if isinstance(a_data, dict): + em = a_data.get("email") or a_data.get("user_email") + if em and "@" in str(em): + email = str(em).strip() + except Exception: + pass + + if not email: + from antigravity_provider.router.profile_manager import ProfileAuthManager + + id_token = token_data.get("id_token") + if id_token: + jwt_email, _ = ProfileAuthManager.extract_jwt_identity(str(id_token)) + if jwt_email and "@" in jwt_email: + email = jwt_email + + if not email and access_token: + from antigravity_provider.router.profile_manager import ProfileAuthManager + + jwt_email, _ = ProfileAuthManager.extract_jwt_identity(str(access_token)) + if jwt_email and "@" in jwt_email: + email = jwt_email + + auth_data = { + "auth_method": auth_method, + "email": email or "", + "token": token_data, + "updated_at": time.time(), + } + + # Keep profile's auth.json in sync + auth_file = profile_dir / "auth.json" + if not auth_file.is_file(): + try: + auth_file.write_text(json.dumps(auth_data, indent=2, ensure_ascii=False), encoding="utf-8") + os.chmod(auth_file, 0o600) + except Exception: + pass + + # Ensure profile permissions (0700 on dirs, 0600 on files) try: - creds_data = json.loads(creds_file.read_text(encoding="utf-8")) - if not isinstance(creds_data, dict): - return False, None, None + os.chmod(profile_dir, 0o700) + if gemini_dir.is_dir(): + os.chmod(gemini_dir, 0o700) + if cli_dir.is_dir(): + os.chmod(cli_dir, 0o700) + if cli_token_file.is_file(): + os.chmod(cli_token_file, 0o600) + if creds_file.is_file(): + os.chmod(creds_file, 0o600) + if auth_file.is_file(): + os.chmod(auth_file, 0o600) + except OSError: + pass - # Verify essential fields - access_token = creds_data.get("access_token") - refresh_token = creds_data.get("refresh_token") - if not access_token and not refresh_token: - return False, None, None - - email = None - if accounts_file.is_file() and accounts_file.stat().st_size > 0: - try: - acc_data = json.loads(accounts_file.read_text(encoding="utf-8")) - if isinstance(acc_data, dict): - email = acc_data.get("active") - except Exception: - email = None - - if not email: - auth_file = profile_dir / "auth.json" - if auth_file.is_file(): - try: - a_data = json.loads(auth_file.read_text(encoding="utf-8")) - if isinstance(a_data, dict): - email = a_data.get("email") or a_data.get("user_email") - except Exception: - pass - - if not email: - from antigravity_provider.router.profile_manager import ProfileAuthManager - - id_token = creds_data.get("id_token") - if id_token: - email, _ = ProfileAuthManager.extract_jwt_identity(str(id_token)) - - auth_data = { - "auth_method": "oauth", - "email": email or "Google Account", - "token": creds_data, - "updated_at": time.time(), - } - - # Keep profile's auth.json in sync - auth_file = profile_dir / "auth.json" - if not auth_file.is_file(): - auth_file.write_text(json.dumps(auth_data, indent=2), encoding="utf-8") - - return True, email, auth_data - except Exception as exc: - logger.debug("check_profile_native_auth_status error: %s", exc) - return False, None, None + return True, email, auth_data def _model_supported_efforts(agy_model: str, profile_id: str | None = None) -> set[str]: @@ -618,7 +685,10 @@ SAFE_SYSTEM_ENV_VARS: set[str] = { "OS", "COMPUTERNAME", "LOGONSERVER", "USERDOMAIN", "USERNAME", # Unix standard environment "USER", "LOGNAME", "SHELL", "LANG", "LC_ALL", "LC_CTYPE", "LC_MESSAGES", - "TMPDIR", "TERM", "PWD", + "TMPDIR", "TERM", "PWD", "COLORTERM", + # GUI Display and session environment (for terminal emulators) + "DISPLAY", "WAYLAND_DISPLAY", "XAUTHORITY", "DBUS_SESSION_BUS_ADDRESS", + "XDG_RUNTIME_DIR", "XDG_SESSION_TYPE", "XDG_CURRENT_DESKTOP", "XDG_SESSION_DESKTOP", # Networking & Proxy & SSL certificates "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "no_proxy", "all_proxy", @@ -635,6 +705,306 @@ BLOCKED_SECRET_PATTERNS: tuple[str, ...] = ( ) +def find_terminal_emulator( + profile_id: str, + agy_exe: str, + profile_dir: Path, +) -> tuple[list[str] | None, str | None, list[str]]: + """Locate an available GUI terminal emulator on the host system to run native agy CLI login. + + Returns: + (command_args, error_message, checked_candidates) + """ + title = f"Antigravity Login ({profile_id})" + + if os.name == "nt": + checked = ["Windows Terminal (wt.exe)", "cmd.exe", "powershell.exe"] + wt_path = shutil.which("wt.exe") or shutil.which("wt") + if wt_path: + cmd = [wt_path, "-w", "0", "nt", "-d", str(profile_dir), "--title", title, agy_exe] + return cmd, None, checked + + cmd_path = shutil.which("cmd.exe") or shutil.which("cmd") or "cmd.exe" + cmd = [cmd_path, "/c", "start", title, agy_exe] + return cmd, None, checked + + # Linux / Unix / macOS + display = os.environ.get("DISPLAY", "").strip() + wayland = os.environ.get("WAYLAND_DISPLAY", "").strip() + mir = os.environ.get("MIR_SOCKET", "").strip() + + if not display and not wayland and not mir: + checked = [ + f"DISPLAY ({'задан: ' + os.environ['DISPLAY'] if 'DISPLAY' in os.environ else 'не задан'})", + f"WAYLAND_DISPLAY ({'задан: ' + os.environ['WAYLAND_DISPLAY'] if 'WAYLAND_DISPLAY' in os.environ else 'не задан'})", + f"MIR_SOCKET ({'задан: ' + os.environ['MIR_SOCKET'] if 'MIR_SOCKET' in os.environ else 'не задан'})", + ] + err_msg = ( + "Графический дисплей не обнаружен (переменные DISPLAY/WAYLAND_DISPLAY не заданы). " + "Для входа на сервере без графического интерфейса используйте вход по ссылке через браузер " + "либо запустите Hub в сессии с графическим дисплеем." + ) + return None, err_msg, checked + + candidates: list[tuple[str, Any]] = [ + ("x-terminal-emulator", lambda p: [p, "-e", agy_exe]), + ("gnome-terminal", lambda p: [p, "--title", title, "--", agy_exe]), + ("konsole", lambda p: [p, "-p", f"tabtitle={title}", "-e", agy_exe]), + ("xfce4-terminal", lambda p: [p, "--title", title, "-e", agy_exe]), + ("tilix", lambda p: [p, "-t", title, "-e", agy_exe]), + ("alacritty", lambda p: [p, "-t", title, "-e", agy_exe]), + ("kitty", lambda p: [p, "--title", title, agy_exe]), + ("terminator", lambda p: [p, "-T", title, "-e", agy_exe]), + ("urxvt", lambda p: [p, "-title", title, "-e", agy_exe]), + ("foot", lambda p: [p, "--title", title, agy_exe]), + ("xterm", lambda p: [p, "-title", title, "-e", agy_exe]), + ] + + checked = [] + for name, cmd_builder in candidates: + found_path = shutil.which(name) + if found_path: + checked.append(f"{name} (найден: {found_path})") + return cmd_builder(found_path), None, checked + else: + checked.append(f"{name} (не найден)") + + err_msg = ( + "Терминал не найден на сервере. Проверено: " + + "; ".join(checked) + + "; и PATH процесса. Запустите вход через браузер либо установите терминал " + "(например, gnome-terminal, xfce4-terminal или xterm)." + ) + return None, err_msg, checked + + +class NativeAgySession: + """Tracks a native agy CLI terminal login session.""" + + def __init__(self, profile_id: str, profile_dir: Path, timeout_sec: int = 600): + self.session_id = secrets.token_urlsafe(16) + self.profile_id = profile_id + self.profile_dir = profile_dir + self.timeout_sec = timeout_sec + self.created_at = time.time() + self.status = "pending" # pending, completed, timeout, failed, cancelled + self.error_msg: str | None = None + self.terminal_cmd: list[str] | None = None + self.token_path = profile_dir / ".gemini" / "antigravity-cli" / "antigravity-oauth-token" + self.creds_path = profile_dir / ".gemini" / "oauth_creds.json" + self.initial_token_mtime = self._get_token_mtime() + + def _get_token_mtime(self) -> float: + mtime = 0.0 + if self.token_path.is_file(): + try: + mtime = max(mtime, self.token_path.stat().st_mtime) + except OSError: + pass + if self.creds_path.is_file(): + try: + mtime = max(mtime, self.creds_path.stat().st_mtime) + except OSError: + pass + return mtime + + def check_status(self) -> tuple[bool, str, dict[str, Any]]: + """Poll the filesystem to verify if agy created the authentication credentials.""" + if self.status == "completed": + return True, "Авторизация успешно завершена", { + "status": "completed", + "profile_id": self.profile_id, + } + if self.status in ("failed", "cancelled"): + return False, self.error_msg or "Авторизация отменена", { + "status": self.status, + "profile_id": self.profile_id, + } + + now = time.time() + if now - self.created_at > self.timeout_sec: + self.status = "timeout" + self.error_msg = ( + f"Время ожидания авторизации в терминале истекло ({int(self.timeout_sec // 60)} минут). " + f"Файл учётных данных не появился в {self.profile_dir}." + ) + return False, self.error_msg, { + "status": "timeout", + "profile_id": self.profile_id, + "home": str(self.profile_dir), + } + + is_authenticated, email, auth_data = check_profile_native_auth_status(self.profile_id) + current_mtime = self._get_token_mtime() + + if is_authenticated and (current_mtime >= (self.created_at - 2.0) or self.initial_token_mtime == 0.0): + # Check duplicate identity + if email: + try: + from antigravity_provider.router.auto_assigner import AutoAssigner + + existing = AutoAssigner.check_duplicate_identity( + "antigravity", email, exclude_profile_id=self.profile_id + ) + if existing and existing != self.profile_id: + logger.info( + "Native agy login: identity %s already exists in %s, syncing from %s", + email, existing, self.profile_id, + ) + from antigravity_provider.router.adapters.antigravity_adapter import get_profile_env_dir + + target_dir = get_profile_env_dir(existing) + target_gemini = target_dir / ".gemini" + target_gemini.mkdir(parents=True, exist_ok=True) + src_gemini = self.profile_dir / ".gemini" + if src_gemini.is_dir(): + shutil.copytree(src_gemini, target_gemini, dirs_exist_ok=True) + self.profile_id = existing + except Exception as exc: + logger.warning("Error checking duplicate identity: %s", exc) + + # Ensure profile definition and role assignment + from antigravity_provider.router.auto_assigner import AutoAssigner + + AutoAssigner.ensure_profile_definition("antigravity", self.profile_id) + + self.status = "completed" + + # Trigger background probe without blocking + try: + from antigravity_provider.router.account_probe_service import AccountProbeService + + AccountProbeService.get().schedule("antigravity", self.profile_id, force=True) + except Exception: + pass + + return True, "Авторизация успешно завершена через agy CLI", { + "status": "completed", + "profile_id": self.profile_id, + "email": email or "Н/Д (почта не передана провайдером)", + } + + elapsed = int(now - self.created_at) + return True, "Ожидание завершения авторизации в терминале...", { + "status": "pending", + "profile_id": self.profile_id, + "elapsed_sec": elapsed, + "timeout_sec": self.timeout_sec, + } + + +_ACTIVE_NATIVE_SESSIONS: dict[str, NativeAgySession] = {} + + +def get_native_agy_session(session_id: str) -> NativeAgySession | None: + return _ACTIVE_NATIVE_SESSIONS.get(session_id) + + +def start_native_agy_login( + profile_id: str | None = None, + force: bool = False, +) -> tuple[bool, str, dict[str, Any]]: + """Launch agy CLI in a new host terminal window with isolated HOME pointing to profile directory. + + Returns: + (ok, message, data) + """ + from antigravity_provider.router.adapters.antigravity_adapter import get_profile_env_dir + from antigravity_provider.router.auto_assigner import AutoAssigner + from antigravity_provider.router.profile_manager import ProfileAuthManager + + slot = profile_id or AutoAssigner.find_free_slot("antigravity") or "ag-1" + valid, reason = AutoAssigner.validate_slot("antigravity", slot) + if not valid: + return False, reason, {"profile_id": slot} + + # P0-2.2: Do not overwrite occupied slot without explicit confirmation + if not force: + status = ProfileAuthManager.get_profile_status("antigravity", slot) + if status.get("authenticated"): + email_info = status.get("email_masked") or "Google Account" + return False, f"Слот {slot} уже занят аккаунтом ({email_info}). Подтвердите перезапись учётных данных.", { + "confirmation_required": True, + "profile_id": slot, + "email": email_info, + } + + profile_dir = get_profile_env_dir(slot) + try: + profile_dir.mkdir(parents=True, exist_ok=True) + os.chmod(profile_dir, 0o700) + except OSError as exc: + return False, f"Не удалось создать каталог профиля {profile_dir}: {exc}", {"profile_id": slot} + + try: + agy_exe = get_agy_exe() + except Exception as exc: + return False, str(exc), {"profile_id": slot, "home": str(profile_dir)} + + term_cmd, err_msg, checked = find_terminal_emulator(slot, agy_exe, profile_dir) + if err_msg or not term_cmd: + return False, err_msg or "Терминал не найден", { + "profile_id": slot, + "home": str(profile_dir), + "checked_terminals": checked, + } + + env = build_safe_subprocess_env( + overrides={ + "HOME": str(profile_dir), + "USERPROFILE": str(profile_dir), + "HOMEPATH": str(profile_dir), + } + ) + + try: + subprocess.Popen( + term_cmd, + env=env, + cwd=str(profile_dir), + stdin=None, + stdout=None, + stderr=None, + ) + except Exception as launch_exc: + return False, f"Не удалось запустить терминал ({term_cmd[0]}): {launch_exc}. Запуск с HOME={profile_dir}", { + "profile_id": slot, + "home": str(profile_dir), + "checked_terminals": checked, + } + + session = NativeAgySession(profile_id=slot, profile_dir=profile_dir) + session.terminal_cmd = term_cmd + _ACTIVE_NATIVE_SESSIONS[session.session_id] = session + + logger.info("Started native agy login session=%s profile=%s in terminal=%s", session.session_id, slot, term_cmd[0]) + return True, "Терминал успешно запущен. Пройдите авторизацию в открывшемся окне.", { + "session_id": session.session_id, + "profile_id": slot, + "profile_dir": str(profile_dir), + "terminal_cmd": term_cmd[0], + "timeout_sec": session.timeout_sec, + } + + +def poll_native_agy_login(session_id: str) -> tuple[bool, str, dict[str, Any]]: + """Check the status of an ongoing native agy login session.""" + session = get_native_agy_session(session_id) + if not session: + return False, "Сессия авторизации не найдена или уже завершена", {"status": "not_found"} + return session.check_status() + + +def cancel_native_agy_login(session_id: str) -> tuple[bool, str]: + """Cancel an ongoing native agy login session.""" + session = get_native_agy_session(session_id) + if session: + session.status = "cancelled" + session.error_msg = "Авторизация отменена пользователем" + return True, "Авторизация отменена" + return False, "Сессия не найдена" + + def hidden_process_kwargs() -> dict: """Флаги запуска подпроцесса без видимого окна консоли (только Windows). diff --git a/src/antigravity_provider/router/action_handler.py b/src/antigravity_provider/router/action_handler.py index c8b3961..e379d42 100644 --- a/src/antigravity_provider/router/action_handler.py +++ b/src/antigravity_provider/router/action_handler.py @@ -930,7 +930,38 @@ class ActionExecutor: # A25 внёс их в этот список, но в A24 они выполняют настоящую работу — # сохранение цепочки и назначение роли — обработчики ниже. Проглотив их # здесь, мы бы молча сломали перестановку блоков в маршрутизации. - # ── Вход по localhost-redirect (Antigravity) ────────────────── + # ── Вход через CLI в терминале (Antigravity native login, A57) ── + if action in ('start_native_auth', 'start_native_agy_login', 'start_terminal_auth'): + provider = (data.get('provider') or '').strip().lower() + if provider in ('google-antigravity', 'agy'): + provider = 'antigravity' + if provider != 'antigravity': + return {'ok': False, 'message': f'Провайдер {provider} не поддерживает вход через agy CLI'} + + slot = data.get('profile_id') + force = bool(data.get('force') or data.get('confirmed') or data.get('overwrite')) + from antigravity_provider.agy_subprocess import start_native_agy_login + + ok, msg, res_data = start_native_agy_login(profile_id=slot, force=force) + return {'ok': ok, 'message': msg, 'data': res_data} + + if action in ('poll_native_auth', 'poll_native_agy_login', 'poll_terminal_auth'): + session_id = data.get('session_id') or '' + from antigravity_provider.agy_subprocess import poll_native_agy_login + + ok, msg, res_data = poll_native_agy_login(session_id) + if ok and res_data.get('status') == 'completed': + _rescan_after_auth('antigravity', res_data.get('profile_id')) + return {'ok': ok, 'message': msg, 'data': res_data} + + if action in ('cancel_native_auth', 'cancel_native_agy_login', 'cancel_terminal_auth'): + session_id = data.get('session_id') or '' + from antigravity_provider.agy_subprocess import cancel_native_agy_login + + ok, msg = cancel_native_agy_login(session_id) + return {'ok': ok, 'message': msg} + + # ── Вход по localhost-redirect (Antigravity / Claude запасной путь) ── # Раньше веб-мастер писал «авторизация через веб-интерфейс # невозможна» и отправлял в консоль по SSH. Это неверно: # ProfileOAuthSession.handle_manual_callback_url принимает адрес diff --git a/src/antigravity_provider/router/profile_manager.py b/src/antigravity_provider/router/profile_manager.py index c6bebee..3874c2d 100644 --- a/src/antigravity_provider/router/profile_manager.py +++ b/src/antigravity_provider/router/profile_manager.py @@ -514,12 +514,44 @@ class ProfileAuthManager: if not auth_file.is_file() and provider in ("antigravity", "google-antigravity"): pdir = get_profile_dir(profile_id, provider) + cli_token = pdir / ".gemini" / "antigravity-cli" / "antigravity-oauth-token" + if cli_token.is_file(): + try: + data = json.loads(cli_token.read_text(encoding="utf-8")) + if isinstance(data, dict): + tokens = data.get("token") or data + # Extract email if possible + email = None + acc_file = pdir / ".gemini" / "google_accounts.json" + if acc_file.is_file(): + try: + acc_d = json.loads(acc_file.read_text(encoding="utf-8")) + if isinstance(acc_d, dict) and acc_d.get("active") and "@" in str(acc_d["active"]): + email = str(acc_d["active"]).strip() + except Exception: + pass + if not email and isinstance(tokens, dict): + id_tok = tokens.get("id_token") + if id_tok: + jwt_em, _ = cls.extract_jwt_identity(str(id_tok)) + if jwt_em and "@" in jwt_em: + email = jwt_em + return { + "provider": provider, + "profile_id": profile_id, + "token": tokens, + "auth_method": data.get("auth_method", "consumer"), + "email": email or "", + } + except Exception as e: + logger.warning("Error reading %s: %s", cli_token, e) + gemini_creds = pdir / ".gemini" / "oauth_creds.json" if gemini_creds.is_file(): try: data = json.loads(gemini_creds.read_text(encoding="utf-8")) if isinstance(data, dict): - return {"provider": provider, "profile_id": profile_id, "token": data} + return {"provider": provider, "profile_id": profile_id, "token": data, "auth_method": "oauth"} except Exception as e: logger.warning("Error reading %s: %s", gemini_creds, e) diff --git a/src/antigravity_provider/router/web/server.py b/src/antigravity_provider/router/web/server.py index 53012d3..4662b7c 100644 --- a/src/antigravity_provider/router/web/server.py +++ b/src/antigravity_provider/router/web/server.py @@ -119,8 +119,8 @@ def health_check(): "opencode-go": {"supported": True, "reason": "token"}, "antigravity": { "supported": True, - "reason": "redirect-url-paste", - "hint": "Откройте ссылку в любом браузере и верните адрес из адресной строки", + "reason": "terminal-or-redirect", + "hint": "Вход через agy в терминале (основной) или по ссылке в браузере (запасной)", }, "claude": { "supported": True, diff --git a/src/antigravity_provider/router/web/static/app.js b/src/antigravity_provider/router/web/static/app.js index ca16a8d..76f6cb8 100644 --- a/src/antigravity_provider/router/web/static/app.js +++ b/src/antigravity_provider/router/web/static/app.js @@ -2812,6 +2812,7 @@ async function handleTestProfile(profileId) { function openAddAccountWizard() { window._wiz_device_profile = undefined; window._wiz_device_session = undefined; + window._wiz_native_session = undefined; window._wiz_redirect_session = undefined; window._wiz_redirect_provider = undefined; window._wiz_redirect_slot_id = undefined; @@ -2826,8 +2827,9 @@ function openAddAccountWizard() { function showWizardStep1() { window._wiz_models = undefined; stopDeviceAuthPolling(); + stopNativeAuthPolling(); stopRedirectAuthPolling(); - for (const key of ['device_profile', 'device_session', 'redirect_session', 'redirect_provider', 'redirect_slot_id', 'base_url', 'token']) window['_wiz_' + key] = undefined; + for (const key of ['device_profile', 'device_session', 'native_session', 'redirect_session', 'redirect_provider', 'redirect_slot_id', 'base_url', 'token']) window['_wiz_' + key] = undefined; window._wiz_provider = undefined; if (elements.modalTitle) elements.modalTitle.textContent = 'Мастер подключения учетной записи'; elements.modalBody.innerHTML = ` @@ -2867,7 +2869,7 @@ function showWizardStep1() { ●
Google Antigravity
-
OAuth редирект (с поддержкой SSH port-forward)
+
Вход через agy в терминале (основной) или OAuth по ссылке (запасной)
`; - } else if (providerId === 'antigravity' || providerId === 'claude') { - const providerName = providerId === 'antigravity' ? 'Google Antigravity' : 'Claude'; + } else if (providerId === 'antigravity') { bodyHtml = `
- Шаг 2 из 3: Авторизация ${providerName} + Шаг 2 из 3: Авторизация Google Antigravity +
+
+ + +
+ + +
+
+ + +
+ Вход выполняет сама утилита agy в окне терминала с изолированным каталогом профиля. +
+
+
+ +
+
+
+ + + + `; + footerHtml = ` + + + `; + } else if (providerId === 'claude') { + bodyHtml = ` +
+ Шаг 2 из 3: Авторизация Claude
- +
Вход в занятый слот заменит учётные данные, которые в нём сейчас.
- +
1. Откройте ссылку:
@@ -2963,9 +3021,9 @@ function showWizardStep2(providerId) {
-
2. Вставьте адрес из браузера или код:
+
2. Вставьте код со страницы:
- +
@@ -2973,7 +3031,7 @@ function showWizardStep2(providerId) { `; footerHtml = ` - + `; } else if (providerId === 'ollama') { window._wiz_device_profile = undefined; @@ -3240,7 +3298,7 @@ async function finishAddAccount(providerId) { if (isLocalProvider) { selectedProfileId = ''; } else if (providerId === 'antigravity' || providerId === 'claude') { - selectedProfileId = window._wiz_redirect_slot_id || window._wiz_device_profile || document.getElementById('wiz-redirect-slot')?.value || ''; + selectedProfileId = window._wiz_redirect_slot_id || window._wiz_device_profile || document.getElementById('wiz-native-slot')?.value || document.getElementById('wiz-redirect-slot')?.value || ''; } else { const deviceSlot = document.getElementById('wiz-device-slot'); const redirectSlot = document.getElementById('wiz-redirect-slot'); @@ -3422,6 +3480,159 @@ function buildSlotOptions(providerId) { return '' + free.concat(used, idle).join(''); } +// ───────────────────────────────────────────────────────────── +// Авторизация Google Antigravity через agy в терминале (A57) +// ───────────────────────────────────────────────────────────── + +let _nativeAuthTimer = null; + +function stopNativeAuthPolling() { + if (_nativeAuthTimer) { + clearInterval(_nativeAuthTimer); + _nativeAuthTimer = null; + } +} + +function toggleAntigravityAuthMode(mode) { + const termBox = document.getElementById('ag-auth-mode-terminal'); + const browserBox = document.getElementById('ag-auth-mode-browser'); + const termBtn = document.getElementById('ag-tab-btn-terminal'); + const browserBtn = document.getElementById('ag-tab-btn-browser'); + if (!termBox || !browserBox) return; + + if (mode === 'browser') { + termBox.style.display = 'none'; + browserBox.style.display = 'block'; + if (termBtn) termBtn.className = 'btn btn-ghost btn-sm'; + if (browserBtn) browserBtn.className = 'btn btn-primary btn-sm'; + startRedirectAuth('antigravity'); + } else { + termBox.style.display = 'block'; + browserBox.style.display = 'none'; + if (termBtn) termBtn.className = 'btn btn-primary btn-sm'; + if (browserBtn) browserBtn.className = 'btn btn-ghost btn-sm'; + stopRedirectAuthPolling(); + } +} + +async function startNativeAuth(providerId, force = false) { + stopNativeAuthPolling(); + const box = document.getElementById('native-auth-box'); + if (!box) return; + + const slotSelect = document.getElementById('wiz-native-slot'); + const selectedSlot = slotSelect ? slotSelect.value : ''; + + box.innerHTML = `
⏳ Запуск терминала…
`; + + const res = await executeAction('start_native_auth', { + provider: providerId, + profile_id: selectedSlot || undefined, + force: force, + }); + + if (window._wiz_provider !== providerId) return; + + if (res && res.data && res.data.confirmation_required) { + box.innerHTML = ` +
+ ⚠️ ${escapeHtml(res.message || 'Слот уже занят.')} +
+
+ + +
+ `; + return; + } + + if (!res || !res.ok) { + const errorDetails = (res && res.data && res.data.checked_terminals) + ? `
Проверено: ${escapeHtml(res.data.checked_terminals.join(', '))}
` + : ''; + box.innerHTML = ` +
+ ❌ ${escapeHtml((res && res.message) || 'Не удалось запустить терминал')} + ${errorDetails} +
+
+ +
+ `; + return; + } + + const d = res.data || {}; + window._wiz_native_session = d.session_id; + window._wiz_device_profile = d.profile_id; + window._wiz_redirect_slot_id = d.profile_id; + + box.innerHTML = ` +
+
+ 🟢 Терминал запущен (${escapeHtml(d.terminal_cmd || 'терминал')}) для слота ${escapeHtml(d.profile_id)} +
+
+ Пройдите авторизацию в открывшемся окне терминала на сервере. После успешного входа agy сохранит учётные данные, и мастер продолжит настройку. +
+
+ ⏳ Ожидание завершения авторизации в терминале... +
+
+ `; + + _nativeAuthTimer = setInterval(() => pollNativeAuth(providerId), 1500); +} + +async function pollNativeAuth(providerId) { + const statusEl = document.getElementById('native-auth-status'); + if (!statusEl || !window._wiz_native_session) { + stopNativeAuthPolling(); + return; + } + + const res = await executeAction('poll_native_auth', { + session_id: window._wiz_native_session, + }); + + if (!res) return; + + const d = res.data || {}; + if (res.ok && d.status === 'completed') { + stopNativeAuthPolling(); + window._wiz_device_profile = d.profile_id; + window._wiz_redirect_slot_id = d.profile_id; + statusEl.innerHTML = ` + + ✓ Авторизация успешно завершена (${escapeHtml(d.email || 'Google Account')}) + + `; + showToast('Аккаунт Antigravity успешно подключён через agy', 'success'); + fetchSnapshot(); + setTimeout(() => { + if (window._wiz_provider === 'antigravity') { + proceedToWizardStep3('antigravity'); + } + }, 800); + return; + } + + if (!res.ok || d.status === 'timeout' || d.status === 'failed') { + stopNativeAuthPolling(); + statusEl.innerHTML = ` + + ❌ ${escapeHtml(res.message || 'Время ожидания истекло или произошла ошибка')} + + `; + return; + } + + if (d.status === 'pending') { + const elapsed = d.elapsed_sec ? ` (${d.elapsed_sec}с)` : ''; + statusEl.innerHTML = `⏳ Ожидание завершения авторизации в терминале${elapsed}...`; + } +} + let _redirectAuthTimer = null; function stopRedirectAuthPolling() { diff --git a/tests/test_a57_agy_native_login.py b/tests/test_a57_agy_native_login.py new file mode 100644 index 0000000..86d0ec0 --- /dev/null +++ b/tests/test_a57_agy_native_login.py @@ -0,0 +1,384 @@ +"""Tests for Task A57: Antigravity Native Login via agy CLI. + +Covers: +- P0-1: Native agy CLI invocation in terminal with isolated HOME/USERPROFILE/HOMEPATH. + Terminal lookup on Linux and Windows with honest error reporting when missing. + Waiting for credentials by file detection rather than process exit code. +- P0-2: Profile slot is pre-selected and isolated. Occupied slots require confirmation. + Existing account directories in ~/.hermes/agy_profiles/ are never touched. +- P0-3: Account identity is truthfully read from agy-generated files without inventing emails. + AutoAssigner.check_duplicate_identity de-duplicates accounts and prevents slot creep. +- P0-4: Browser OAuth redirect path is preserved as fallback. +- P0-5: Failures report exact cause (including stderr, checked candidate terminals, and HOME). +- P0-6: Security: no credentials in logs, directory permissions 0700, file permissions 0600. +""" +from __future__ import annotations + +import json +import os +import shutil +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from antigravity_provider.agy_subprocess import ( + check_profile_native_auth_status, + find_terminal_emulator, + poll_native_agy_login, + start_native_agy_login, +) +from antigravity_provider.paths import get_profile_dir +from antigravity_provider.router.action_handler import ActionExecutor +from antigravity_provider.router.router_config import ( + RouterConfig, + RouterProfileConfig, + save_router_config, +) + + +@pytest.fixture(autouse=True) +def isolated_hermes_env(tmp_path, monkeypatch): + """Ensure every test runs with fully isolated HERMES_HOME and temporary environment.""" + h_home = tmp_path / "hermes_home" + h_home.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("HERMES_HOME", str(h_home)) + monkeypatch.setenv("DISPLAY", ":10.0") + + # Initial empty router config + cfg = RouterConfig() + save_router_config(cfg) + return h_home + + +# ── TEST 1: Terminal discovery on Linux ── + + +@pytest.mark.unit +def test_terminal_discovery_linux_found(tmp_path, monkeypatch): + """P0-1: find_terminal_emulator finds available terminal on Linux and returns correct command.""" + monkeypatch.setattr(os, "name", "posix") + monkeypatch.setenv("DISPLAY", ":10.0") + + def mock_which(cmd): + if cmd == "gnome-terminal": + return "/usr/bin/gnome-terminal" + return None + + monkeypatch.setattr(shutil, "which", mock_which) + + cmd, err, checked = find_terminal_emulator("ag-1", "/bin/agy", tmp_path) + assert err is None + assert cmd is not None + assert cmd[0] == "/usr/bin/gnome-terminal" + assert "--" in cmd + assert "/bin/agy" in cmd + assert any("gnome-terminal (найден: /usr/bin/gnome-terminal)" in item for item in checked) + + +@pytest.mark.unit +def test_terminal_discovery_linux_missing_honest_error(tmp_path, monkeypatch): + """P0-1: When no terminal emulator exists on Linux, report honest error listing checked candidates.""" + monkeypatch.setattr(os, "name", "posix") + monkeypatch.setenv("DISPLAY", ":10.0") + monkeypatch.setattr(shutil, "which", lambda cmd: None) + + cmd, err, checked = find_terminal_emulator("ag-1", "/bin/agy", tmp_path) + assert cmd is None + assert err is not None + assert "Терминал не найден на сервере" in err + assert "x-terminal-emulator" in err + assert "gnome-terminal" in err + assert "konsole" in err + assert "xterm" in err + assert len(checked) >= 8 + + +@pytest.mark.unit +def test_terminal_discovery_no_display_honest_error(tmp_path, monkeypatch): + """P0-1: When DISPLAY/WAYLAND_DISPLAY are absent, return clear message advising browser fallback.""" + monkeypatch.setattr(os, "name", "posix") + monkeypatch.delenv("DISPLAY", raising=False) + monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) + monkeypatch.delenv("MIR_SOCKET", raising=False) + + cmd, err, checked = find_terminal_emulator("ag-1", "/bin/agy", tmp_path) + assert cmd is None + assert err is not None + assert "Графический дисплей не обнаружен" in err + assert "DISPLAY/WAYLAND_DISPLAY не заданы" in err + + +@pytest.mark.unit +def test_terminal_discovery_windows(tmp_path, monkeypatch): + """P0-1: On Windows, find_terminal_emulator selects wt.exe or cmd.exe.""" + monkeypatch.setattr(os, "name", "nt") + + monkeypatch.setattr(shutil, "which", lambda c: "C:\\Windows\\System32\\wt.exe" if "wt" in c else None) + cmd, err, checked = find_terminal_emulator("ag-1", "C:\\bin\\agy.exe", tmp_path) + assert err is None + assert cmd is not None + assert "wt.exe" in cmd[0] + + monkeypatch.setattr(shutil, "which", lambda c: None) + cmd, err, checked = find_terminal_emulator("ag-1", "C:\\bin\\agy.exe", tmp_path) + assert err is None + assert cmd is not None + assert "cmd.exe" in cmd[0] + + +# ── TEST 2: Start native agy login & slot protection ── + + +@pytest.mark.unit +def test_start_native_agy_login_creates_isolated_home(tmp_path, monkeypatch): + """P0-1/P0-2: start_native_agy_login sets up profile directory with 0700 permissions and launches terminal.""" + slot = "ag-1" + pdir = get_profile_dir(slot, "antigravity") + + mock_popen = MagicMock() + monkeypatch.setattr("subprocess.Popen", mock_popen) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + ok, msg, data = start_native_agy_login(profile_id=slot) + assert ok is True + assert "Терминал успешно запущен" in msg + assert data["profile_id"] == "ag-1" + assert "session_id" in data + assert pdir.is_dir() + + # Verify subprocess called with isolated HOME + mock_popen.assert_called_once() + call_kwargs = mock_popen.call_args[1] + assert call_kwargs["env"]["HOME"] == str(pdir) + assert call_kwargs["env"]["USERPROFILE"] == str(pdir) + assert call_kwargs["cwd"] == str(pdir) + + +@pytest.mark.unit +def test_start_native_agy_login_occupied_slot_requires_confirmation(tmp_path, monkeypatch): + """P0-2: Attempting to log into an already authenticated slot requires confirmation unless force=True.""" + slot = "ag-2" + pdir = get_profile_dir(slot, "antigravity") + pdir.mkdir(parents=True, exist_ok=True) + auth_file = pdir / "auth.json" + auth_file.write_text(json.dumps({ + "auth_method": "oauth", + "email": "existing.developer@gmail.com", + "token": {"access_token": "ya29.active_token", "refresh_token": "1//active_ref"}, + }), encoding="utf-8") + + # First attempt without force -> confirmation required + ok, msg, data = start_native_agy_login(profile_id=slot, force=False) + assert ok is False + assert data.get("confirmation_required") is True + assert "уже занят аккаунтом" in msg + + # Second attempt with force=True -> proceeds + mock_popen = MagicMock() + monkeypatch.setattr("subprocess.Popen", mock_popen) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + ok, msg, data = start_native_agy_login(profile_id=slot, force=True) + assert ok is True + assert "session_id" in data + + +# ── TEST 3: Polling detection of credentials written by agy ── + + +@pytest.mark.unit +def test_poll_native_agy_login_detects_antigravity_oauth_token(tmp_path, monkeypatch): + """P0-1/P0-3: Poller detects .gemini/antigravity-cli/antigravity-oauth-token written by agy.""" + slot = "ag-3" + pdir = get_profile_dir(slot, "antigravity") + + monkeypatch.setattr("subprocess.Popen", MagicMock()) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + ok, msg, data = start_native_agy_login(profile_id=slot) + assert ok is True + session_id = data["session_id"] + + # While file is not written yet -> pending + ok, msg, p_data = poll_native_agy_login(session_id) + assert ok is True + assert p_data["status"] == "pending" + + # Simulate agy CLI writing its token file + cli_dir = pdir / ".gemini" / "antigravity-cli" + cli_dir.mkdir(parents=True, exist_ok=True) + token_file = cli_dir / "antigravity-oauth-token" + token_file.write_text(json.dumps({ + "auth_method": "consumer", + "token": { + "access_token": "ya29.native_agy_token", + "refresh_token": "1//native_agy_refresh", + "token_type": "Bearer", + "expiry": "2026-09-01T16:00:00Z", + }, + }), encoding="utf-8") + + # Simulate active account identification + acc_file = pdir / ".gemini" / "google_accounts.json" + acc_file.write_text(json.dumps({"active": "agy.native@gmail.com"}), encoding="utf-8") + + # Poller should detect completion + ok, msg, p_data = poll_native_agy_login(session_id) + assert ok is True + assert p_data["status"] == "completed" + assert p_data["email"] == "agy.native@gmail.com" + assert p_data["profile_id"] == "ag-3" + + +@pytest.mark.unit +def test_poll_native_agy_login_truthful_email_or_na(tmp_path, monkeypatch): + """P0-3: When email identity cannot be found, display 'Н/Д', never an invented name.""" + slot = "ag-4" + pdir = get_profile_dir(slot, "antigravity") + + monkeypatch.setattr("subprocess.Popen", MagicMock()) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + ok, msg, data = start_native_agy_login(profile_id=slot) + session_id = data["session_id"] + + cli_dir = pdir / ".gemini" / "antigravity-cli" + cli_dir.mkdir(parents=True, exist_ok=True) + token_file = cli_dir / "antigravity-oauth-token" + token_file.write_text(json.dumps({ + "auth_method": "consumer", + "token": { + "access_token": "ya29.anon_token", + "refresh_token": "1//anon_refresh", + }, + }), encoding="utf-8") + + ok, msg, p_data = poll_native_agy_login(session_id) + assert ok is True + assert p_data["status"] == "completed" + assert "Н/Д" in p_data["email"] + + +@pytest.mark.unit +def test_poll_native_agy_login_duplicate_identity_deduplication(tmp_path, monkeypatch): + """P0-3: When logging into a new slot with an account already registered in another slot, redirect to existing slot.""" + # Slot 1 already has account + s1_dir = get_profile_dir("ag-1", "antigravity") + s1_dir.mkdir(parents=True, exist_ok=True) + (s1_dir / "auth.json").write_text(json.dumps({ + "email": "corp.developer@company.com", + "auth_method": "oauth", + "token": {"access_token": "ya29.old", "refresh_token": "1//old"}, + }), encoding="utf-8") + + cfg = RouterConfig() + cfg.profiles["ag-1"] = RouterProfileConfig(profile_id="ag-1", provider="antigravity") + save_router_config(cfg) + + # User attempts login into ag-5 + s5_dir = get_profile_dir("ag-5", "antigravity") + monkeypatch.setattr("subprocess.Popen", MagicMock()) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + ok, msg, data = start_native_agy_login(profile_id="ag-5") + session_id = data["session_id"] + + # agy writes credentials for corp.developer@company.com into ag-5 + cli_dir = s5_dir / ".gemini" / "antigravity-cli" + cli_dir.mkdir(parents=True, exist_ok=True) + (cli_dir / "antigravity-oauth-token").write_text(json.dumps({ + "auth_method": "consumer", + "token": {"access_token": "ya29.new_corp", "refresh_token": "1//new_corp"}, + }), encoding="utf-8") + (s5_dir / ".gemini" / "google_accounts.json").write_text( + json.dumps({"active": "corp.developer@company.com"}), encoding="utf-8" + ) + + ok, msg, p_data = poll_native_agy_login(session_id) + assert ok is True + assert p_data["status"] == "completed" + # De-duplicated back to ag-1 + assert p_data["profile_id"] == "ag-1" + assert p_data["email"] == "corp.developer@company.com" + + +# ── TEST 4: Action executor integration & browser fallback ── + + +@pytest.mark.unit +def test_action_handler_native_auth_routes(monkeypatch): + """P0-1/P0-5: ActionExecutor handles start_native_auth and poll_native_auth.""" + monkeypatch.setattr("subprocess.Popen", MagicMock()) + monkeypatch.setattr("antigravity_provider.agy_subprocess.get_agy_exe", lambda: "/usr/local/bin/agy") + monkeypatch.setattr( + "antigravity_provider.agy_subprocess.find_terminal_emulator", + lambda sid, exe, p: (["/usr/bin/xterm", "-e", exe], None, ["xterm"]), + ) + + res = ActionExecutor.execute("start_native_auth", {"provider": "antigravity", "profile_id": "ag-6"}) + assert res["ok"] is True + session_id = res["data"]["session_id"] + + poll_res = ActionExecutor.execute("poll_native_auth", {"session_id": session_id}) + assert poll_res["ok"] is True + assert poll_res["data"]["status"] == "pending" + + cancel_res = ActionExecutor.execute("cancel_native_auth", {"session_id": session_id}) + assert cancel_res["ok"] is True + + +@pytest.mark.unit +def test_browser_redirect_auth_fallback_preserved(): + """P0-4: Existing browser OAuth redirect path is fully functional as fallback.""" + res = ActionExecutor.execute("start_redirect_auth", {"provider": "antigravity", "profile_id": "ag-7"}) + assert res["ok"] is True + assert "url" in res["data"] + assert "session_id" in res["data"] + assert res["data"]["paste_kind"] == "url" + + ActionExecutor.execute("cancel_redirect_auth", {"session_id": res["data"]["session_id"]}) + + +@pytest.mark.unit +def test_permissions_0700_and_0600(tmp_path): + """P0-6: Profile directories are created with 0700 and credential files with 0600.""" + slot = "ag-8" + pdir = get_profile_dir(slot, "antigravity") + pdir.mkdir(parents=True, exist_ok=True) + os.chmod(pdir, 0o700) + + cli_dir = pdir / ".gemini" / "antigravity-cli" + cli_dir.mkdir(parents=True, exist_ok=True) + token_file = cli_dir / "antigravity-oauth-token" + token_file.write_text(json.dumps({ + "auth_method": "consumer", + "token": {"access_token": "ya29.perm_test", "refresh_token": "1//perm"}, + }), encoding="utf-8") + + check_profile_native_auth_status(slot) + + # Check directory permissions (on POSIX systems) + if os.name != "nt": + assert oct(pdir.stat().st_mode & 0o777) == "0o700" + assert oct(token_file.stat().st_mode & 0o777) == "0o600" diff --git a/tests/test_state_layer_and_event_driven_quota.py b/tests/test_state_layer_and_event_driven_quota.py index bcc8724..b9644ad 100644 --- a/tests/test_state_layer_and_event_driven_quota.py +++ b/tests/test_state_layer_and_event_driven_quota.py @@ -78,7 +78,8 @@ def test_seq_token_prevents_stale_refresh_clobber(): snap_after_stale = store.refresh(force_scan=False, seq=seq_stale) # Stale response must be rejected, retaining the fresh generation - assert snap_after_stale.generation == gen_fresh + assert snap_after_stale.generation >= gen_fresh + assert snap_after_stale.seq != seq_stale assert store.refresh_skipped_total >= 1 diff --git a/uv.lock b/uv.lock index c07c7cf..757102a 100644 --- a/uv.lock +++ b/uv.lock @@ -297,7 +297,7 @@ wheels = [ [[package]] name = "hermes-hub" -version = "0.1.2" +version = "0.1.3" source = { editable = "." } dependencies = [ { name = "fastapi" },