diff --git a/moli-playground/chatgpt/chatgpt_auth_bridge.py b/moli-playground/chatgpt/chatgpt_auth_bridge.py new file mode 100644 index 0000000000..bf5031d94d --- /dev/null +++ b/moli-playground/chatgpt/chatgpt_auth_bridge.py @@ -0,0 +1,579 @@ +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import re +import shutil +import signal +import subprocess +import tempfile +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from chatgpt_cdp_demo import DemoError, read_json_url_no_proxy, reserve_local_port +from chatgpt_playwright_core import ( + PlaywrightChatGPTSession, + auth_blocking_reason_from_url, +) + +Reporter = Callable[[str], None] +PROFILE_METADATA_FILE = ".chatgpt-playground-auth.json" + + +@dataclass(frozen=True) +class ChromiumAuthState: + cookies: list[dict[str, object]] + local_storage: list[list[str]] + user_agent: str + + +def default_moli_chatgpt_profile_dir() -> Path: + data_home = os.environ.get("XDG_DATA_HOME") + root = ( + Path(data_home).expanduser() if data_home else Path.home() / ".local" / "share" + ) + return root / "moli" / "chatgpt-profile" + + +def prepare_profile_dir(profile_dir: str | os.PathLike[str]) -> Path: + path = Path(profile_dir).expanduser().resolve() + path.mkdir(mode=0o700, parents=True, exist_ok=True) + path.chmod(0o700) + return path + + +def load_profile_user_agent(profile_dir: str | os.PathLike[str]) -> str: + path = Path(profile_dir).expanduser().resolve() / PROFILE_METADATA_FILE + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError, TypeError): + return "" + if not isinstance(payload, dict): + return "" + user_agent = payload.get("user_agent") + return user_agent if isinstance(user_agent, str) else "" + + +def save_profile_user_agent( + profile_dir: str | os.PathLike[str], user_agent: str +) -> None: + directory = prepare_profile_dir(profile_dir) + target = directory / PROFILE_METADATA_FILE + descriptor, temporary_name = tempfile.mkstemp( + prefix=f"{PROFILE_METADATA_FILE}.", dir=directory + ) + temporary = Path(temporary_name) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump( + {"user_agent": user_agent}, handle, ensure_ascii=False, sort_keys=True + ) + handle.write("\n") + temporary.replace(target) + target.chmod(0o600) + except BaseException: + try: + os.close(descriptor) + except OSError: + pass + temporary.unlink(missing_ok=True) + raise + + +def chromium_cookie_params(cookies: list[dict[str, Any]]) -> list[dict[str, object]]: + result: list[dict[str, object]] = [] + for cookie in cookies: + domain = str(cookie.get("domain") or "") + if "chatgpt.com" not in domain and "openai.com" not in domain: + continue + item: dict[str, object] = { + "name": str(cookie.get("name") or ""), + "value": str(cookie.get("value") or ""), + "domain": domain, + "path": str(cookie.get("path") or "/"), + "secure": bool(cookie.get("secure")), + "httpOnly": bool(cookie.get("httpOnly")), + } + expires = cookie.get("expires") + if isinstance(expires, (int, float)) and expires > 0: + item["expires"] = expires + same_site = cookie.get("sameSite") + if same_site in {"Strict", "Lax", "None"}: + item["sameSite"] = same_site + result.append(item) + return result + + +def resolve_auth_chromium(args: argparse.Namespace) -> Path: + configured = str( + getattr(args, "auth_chromium_bin", "") + or getattr(args, "chromium_bin", "") + or os.environ.get("CHROMIUM_BIN") + or "" + ) + if configured: + candidate = Path(configured).expanduser().resolve() + if candidate.is_file(): + return candidate + raise DemoError(f"authentication Chromium binary does not exist: {candidate}") + for name in ( + "chromium", + "chromium-browser", + "google-chrome", + "google-chrome-stable", + ): + resolved = shutil.which(name) + if resolved: + return Path(resolved).resolve() + raise DemoError( + "Chromium is required for ChatGPT authentication; pass --auth-chromium-bin" + ) + + +def resolve_xvfb() -> Path: + configured = os.environ.get("XVFB_BIN") or "" + if configured: + candidate = Path(configured).expanduser().resolve() + if candidate.is_file(): + return candidate + raise DemoError(f"Xvfb binary does not exist: {candidate}") + resolved = shutil.which("Xvfb") + if resolved: + return Path(resolved).resolve() + raise DemoError("Xvfb is required for headful ChatGPT authentication") + + +def auth_proxy_options(args: argparse.Namespace) -> tuple[str, str]: + proxy = str( + getattr(args, "http_proxy", "") + or os.environ.get("HTTPS_PROXY") + or os.environ.get("https_proxy") + or os.environ.get("HTTP_PROXY") + or os.environ.get("http_proxy") + or "" + ) + no_proxy = str( + getattr(args, "http_no_proxy", "") + or os.environ.get("NO_PROXY") + or os.environ.get("no_proxy") + or "" + ) + return proxy, no_proxy + + +def terminate_process_group(process: subprocess.Popen[Any] | None) -> None: + if process is None or process.poll() is not None: + return + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + return + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=3) + + +def remove_auth_profile(path: Path | None) -> None: + if path is None or not path.exists(): + return + resolved = path.resolve() + temporary_root = Path(tempfile.gettempdir()).resolve() + if resolved.parent != temporary_root or not resolved.name.startswith( + "moli-chatgpt-auth-chromium." + ): + raise DemoError( + f"refusing to remove unexpected authentication profile: {resolved}" + ) + shutil.rmtree(resolved) + + +def start_xvfb() -> tuple[subprocess.Popen[str], str]: + process = subprocess.Popen( # noqa: S603 - executable is resolved from an explicit trusted path. + [ + str(resolve_xvfb()), + "-displayfd", + "1", + "-screen", + "0", + "1440x1000x24", + "-nolisten", + "tcp", + ], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + start_new_session=True, + ) + if process.stdout is None: + terminate_process_group(process) + raise DemoError("Xvfb stdout pipe is unavailable") + display_number = process.stdout.readline().strip() + if not display_number or process.poll() is not None: + terminate_process_group(process) + raise DemoError("Xvfb did not allocate a display for ChatGPT authentication") + return process, f":{display_number}" + + +async def wait_for_chromium_cdp( + endpoint: str, process: subprocess.Popen[bytes], timeout: float +) -> str: + deadline = time.monotonic() + timeout + last_error: BaseException | None = None + while time.monotonic() < deadline: + if process.poll() is not None: + return_code = process.returncode + raise DemoError( + f"authentication Chromium exited before CDP startup: rc={return_code}" + ) + try: + version = await asyncio.to_thread( + read_json_url_no_proxy, endpoint + "/json/version" + ) + websocket_url = version.get("webSocketDebuggerUrl") + if isinstance(websocket_url, str) and websocket_url: + return websocket_url + except BaseException as error: # noqa: BLE001 - report the final startup error. + last_error = error + await asyncio.sleep(0.1) + raise DemoError( + "timed out waiting for authentication Chromium CDP; " + f"last_error={type(last_error).__name__ if last_error else 'none'}" + ) + + +async def click_visible_text(page: Any, patterns: tuple[str, ...]) -> bool: + interactive = page.locator("button, [role='button'], a, label") + for pattern in patterns: + expression = re.compile(pattern, re.IGNORECASE) + for candidates in ( + interactive.filter(has_text=expression), + page.get_by_text(expression), + ): + for index in range(min(await candidates.count(), 12)): + candidate = candidates.nth(index) + try: + if await candidate.is_visible( + timeout=500 + ) and await candidate.is_enabled(timeout=500): + await candidate.click(timeout=5000) + return True + except Exception: # noqa: BLE001, S112 - probe the next visible candidate. + continue + return False + + +async def select_push_mfa( + session: PlaywrightChatGPTSession, reporter: Reporter +) -> None: + if session.page is None: + raise DemoError("MFA page is unavailable") + push_patterns = ( + r"push notification", + r"chatgpt app", + r"approve.*device", + r"push", + ) + clicked = await click_visible_text(session.page, push_patterns) + if not clicked: + expanded = await click_visible_text( + session.page, + ( + r"try another method", + r"use another method", + r"different method", + r"other (?:options|methods)", + ), + ) + if expanded: + reporter("expand MFA methods") + await asyncio.sleep(1) + clicked = await click_visible_text(session.page, push_patterns) + if not clicked: + raise DemoError("Push approval is not available on the ChatGPT MFA page") + reporter("select Push approval") + await asyncio.sleep(1) + await click_visible_text( + session.page, (r"^continue$", r"send push", r"send notification") + ) + + +async def submit_external_login_once( + session: PlaywrightChatGPTSession, + email: str, + password: str, +) -> None: + state = await session.navigate_to_initial_login_state() + if state.get("loggedIn"): + return + state = await session.navigate_to_login_form() + if state.get("blockingReason"): + raise DemoError( + f"login blocked before credentials: {state.get('blockingReason')}" + ) + if state.get("hasEmailInput"): + session.report("fill email in authentication Chromium") + await session.accept_cookie_consent() + await session.fill_email_native(email) + state = await session.wait_for_state_after_email_submit( + timeout=60, label="password form" + ) + if state.get("loggedIn"): + return + if not state.get("hasPasswordInput"): + raise DemoError( + f"password input did not appear; blocking={state.get('blockingReason')!r}" + ) + await session.wait_for_auth_password_runtime_ready(timeout=20) + session.report("submit password once in authentication Chromium") + await session.fill_password_native(password) + + +async def wait_for_external_login( + session: PlaywrightChatGPTSession, + args: argparse.Namespace, + reporter: Reporter, +) -> None: + deadline = time.monotonic() + max(1.0, float(args.login_timeout)) + selected_push = False + tried_email_verification = False + reported_device_approval = False + while time.monotonic() < deadline: + try: + state = await session.helper("loginState", timeout=15) + except Exception: # noqa: BLE001 - navigation can invalidate the helper realm. + await asyncio.sleep(0.5) + continue + if state.get("loggedIn"): + reporter("authentication approved") + return + reason = state.get("blockingReason") or auth_blocking_reason_from_url( + str(state.get("url") or "") + ) + if reason == "mfa" and not selected_push: + await select_push_mfa(session, reporter) + selected_push = True + await asyncio.sleep(1) + continue + if reason == "device-approval": + if ( + bool(getattr(args, "try_email_verification", False)) + and not tried_email_verification + ): + tried_email_verification = True + if await session.click_try_with_email_verification(): + await asyncio.sleep(1) + continue + if not reported_device_approval: + reporter("approve the new login in the ChatGPT app") + reported_device_approval = True + await asyncio.sleep(1) + continue + if reason in {"email-verification", "verification-code"}: + if not session.has_auth_code_source(): + raise DemoError( + "ChatGPT requested an email code but no auth-code input " + "is configured" + ) + code = await session.read_auth_code() + reporter("submit ChatGPT email verification code") + await session.fill_auth_code_native(code) + session._auth_code_cache = None + await asyncio.sleep(1.5) + continue + if reason: + raise DemoError( + f"ChatGPT authentication reached an unsupported step: {reason}" + ) + await asyncio.sleep(0.5) + if reported_device_approval: + raise DemoError("timed out waiting for ChatGPT device approval") + raise DemoError("timed out waiting for ChatGPT authentication") + + +async def export_auth_state(session: PlaywrightChatGPTSession) -> ChromiumAuthState: + if session.page is None or session.context is None: + raise DemoError("authentication Chromium page is unavailable") + if not session.page.url.startswith("https://chatgpt.com/"): + await session.page.goto( + "https://chatgpt.com/", wait_until="domcontentloaded", timeout=120_000 + ) + await session.wait_for_document_settle(timeout=10) + cookies = chromium_cookie_params(await session.context.cookies()) + local_storage = await session.page.evaluate("Object.entries(localStorage)") + user_agent = await session.page.evaluate("navigator.userAgent") + if not isinstance(local_storage, list) or not all( + isinstance(entry, list) + and len(entry) == 2 + and all(isinstance(value, str) for value in entry) + for entry in local_storage + ): + raise DemoError("authentication Chromium returned invalid localStorage state") + if not isinstance(user_agent, str) or not user_agent: + raise DemoError("authentication Chromium returned an empty user agent") + if not cookies: + raise DemoError("authentication Chromium returned no ChatGPT cookies") + return ChromiumAuthState( + cookies=cookies, local_storage=local_storage, user_agent=user_agent + ) + + +async def authenticate_in_external_chromium( + args: argparse.Namespace, + email: str, + password: str, + reporter: Reporter, +) -> ChromiumAuthState: + from playwright.async_api import async_playwright + + profile: Path | None = None + xvfb: subprocess.Popen[str] | None = None + chromium: subprocess.Popen[bytes] | None = None + session: PlaywrightChatGPTSession | None = None + try: + profile = Path(tempfile.mkdtemp(prefix="moli-chatgpt-auth-chromium.")) + profile.chmod(0o700) + xvfb, display = start_xvfb() + port = reserve_local_port() + endpoint = f"http://127.0.0.1:{port}" + command = [ + str(resolve_auth_chromium(args)), + f"--remote-debugging-port={port}", + "--remote-debugging-address=127.0.0.1", + "--remote-allow-origins=*", + f"--user-data-dir={profile}", + "--no-first-run", + "--no-default-browser-check", + "--password-store=basic", + "--window-size=1440,1000", + ] + proxy, no_proxy = auth_proxy_options(args) + if proxy: + command.append(f"--proxy-server={proxy}") + if no_proxy: + command.append(f"--proxy-bypass-list={no_proxy.replace(',', ';')}") + if hasattr(os, "geteuid") and os.geteuid() == 0: + command.append("--no-sandbox") + command.append("about:blank") + environment = os.environ.copy() + environment["DISPLAY"] = display + reporter("start authentication Chromium under Xvfb") + chromium = subprocess.Popen( # noqa: ASYNC220, S603 - trusted binary; explicit process-group cleanup. + command, + env=environment, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + websocket_url = await wait_for_chromium_cdp( + endpoint, chromium, float(args.startup_timeout) + ) + playwright = await async_playwright().start() + auth_args = argparse.Namespace(**vars(args)) + auth_args.backend = "chromium" + session = PlaywrightChatGPTSession(auth_args, reporter=reporter) + session.playwright = playwright + session.browser = await playwright.chromium.connect_over_cdp(websocket_url) + session.context = session.browser.contexts[0] + await session.finish_page_setup() + if session.page is None: + raise DemoError("authentication Chromium did not expose a page target") + webdriver = await session.page.evaluate("navigator.webdriver") + if webdriver: + raise DemoError( + "authentication Chromium unexpectedly exposes navigator.webdriver" + ) + await submit_external_login_once(session, email, password) + await wait_for_external_login(session, auth_args, reporter) + state = await export_auth_state(session) + reporter(f"captured ChatGPT session ({len(state.cookies)} cookies)") + return state + finally: + if session is not None: + await session.close() + terminate_process_group(chromium) + terminate_process_group(xvfb) + remove_auth_profile(profile) + + +async def import_auth_state( + session: PlaywrightChatGPTSession, + state: ChromiumAuthState, +) -> None: + if session.page is None or session.context is None: + raise DemoError("Moli page is unavailable for authentication import") + cdp = await session.context.new_cdp_session(session.page) + await cdp.send("Network.enable") + await cdp.send("Network.setCookies", {"cookies": state.cookies}) + await session.page.goto( + "https://chatgpt.com/", wait_until="domcontentloaded", timeout=180_000 + ) + await session.page.evaluate( + """entries => { + for (const [key, value] of entries) localStorage.setItem(key, value); + }""", + state.local_storage, + ) + await session.page.reload(wait_until="domcontentloaded", timeout=180_000) + await session.require_existing_session() + + +async def create_authenticated_moli_session( + args: argparse.Namespace, + email: str, + password: str, + reporter: Reporter = print, +) -> PlaywrightChatGPTSession: + profile_dir = str(getattr(args, "profile_dir", "") or "") + if profile_dir: + prepare_profile_dir(profile_dir) + if not getattr(args, "user_agent", None): + stored_user_agent = load_profile_user_agent(profile_dir) + if stored_user_agent: + args.user_agent = stored_user_agent + + existing = PlaywrightChatGPTSession(args, reporter=reporter) + try: + await existing.start() + except BaseException: + await existing.close() + raise + try: + await existing.require_existing_session() + except DemoError: + await existing.close() + reporter("Moli profile needs ChatGPT authentication") + else: + reporter("reuse logged-in Moli profile") + return existing + + if not password: + raise DemoError( + "password is required because the Moli profile is not logged in" + ) + + state = await authenticate_in_external_chromium(args, email, password, reporter) + if not getattr(args, "user_agent", None): + args.user_agent = state.user_agent + + authenticated = PlaywrightChatGPTSession(args, reporter=reporter) + try: + await authenticated.start() + await import_auth_state(authenticated, state) + if profile_dir: + save_profile_user_agent(profile_dir, state.user_agent) + reporter("ChatGPT authentication imported into Moli") + return authenticated + except BaseException: + await authenticated.close() + raise diff --git a/moli-playground/chatgpt/chatgpt_cdp_tui.py b/moli-playground/chatgpt/chatgpt_cdp_tui.py index e433aa0286..0a303278ce 100644 --- a/moli-playground/chatgpt/chatgpt_cdp_tui.py +++ b/moli-playground/chatgpt/chatgpt_cdp_tui.py @@ -73,6 +73,7 @@ class TextField: @dataclass class TuiState: fields: dict[str, TextField] + title: str = "Moli ChatGPT CDP TUI" focus: str = "email" logged_in: bool = False busy: bool = False @@ -290,7 +291,7 @@ def draw(stdscr: Any, state: TuiState, colors: dict[str, int]) -> None: stdscr.refresh() return - safe_add(stdscr, 0, 2, "Moli ChatGPT CDP TUI", colors["title"]) + safe_add(stdscr, 0, 2, state.title, colors["title"]) safe_add(stdscr, 1, 2, state.status[: width - 4], colors["dim"] if not state.busy else 0) safe_add(stdscr, 2, 0, "-" * (width - 1)) @@ -388,7 +389,7 @@ def submit_current(state: TuiState, backend: Backend) -> None: state.status = "email is required" add_message(state, "error", state.status) return - if not password: + if not password and bool(getattr(backend, "requires_password", True)): state.status = "password is required" add_message(state, "error", state.status) return @@ -424,7 +425,8 @@ def run_tui_with_backend(stdscr: Any, args: argparse.Namespace, backend_cls: Any "password": TextField("password", "Password", os.environ.get("CHATGPT_PASSWORD", ""), secret=True), "auth_code": TextField("auth_code", "Auth Code"), "prompt": TextField("prompt", "Prompt", args.prompt), - } + }, + title=str(getattr(args, "tui_title", "Moli ChatGPT CDP TUI")), ) backend = backend_cls(args, events) try: diff --git a/moli-playground/chatgpt/chatgpt_playwright_tui.py b/moli-playground/chatgpt/chatgpt_playwright_tui.py index 72ed586554..ce35440d59 100644 --- a/moli-playground/chatgpt/chatgpt_playwright_tui.py +++ b/moli-playground/chatgpt/chatgpt_playwright_tui.py @@ -7,12 +7,15 @@ import curses import os import queue import threading -from typing import Any +from chatgpt_auth_bridge import ( + create_authenticated_moli_session, + default_moli_chatgpt_profile_dir, +) from chatgpt_cdp_demo import ( DEFAULT_ANSWER_TIMEOUT, - DEFAULT_MOLI_HTTP_TIMEOUT_MS, DEFAULT_LOGIN_TIMEOUT, + DEFAULT_MOLI_HTTP_TIMEOUT_MS, DEFAULT_STARTUP_TIMEOUT, DEFAULT_URL, ) @@ -20,12 +23,19 @@ from chatgpt_cdp_tui import UiEvent, error_text, run_tui_with_backend from chatgpt_playwright_core import PlaywrightChatGPTSession TUI_LOGIN_TIMEOUT = max(300.0, DEFAULT_LOGIN_TIMEOUT) -TUI_TRY_EMAIL_VERIFICATION = True +TUI_TRY_EMAIL_VERIFICATION = False class PlaywrightBackend: - def __init__(self, args: argparse.Namespace, events: "queue.Queue[UiEvent]") -> None: + requires_password = False + + def __init__( + self, args: argparse.Namespace, events: "queue.Queue[UiEvent]" + ) -> None: self.args = args + self.requires_password = not ( + args.backend == "moli" and args.auth_backend == "chromium-bridge" + ) self.events = events self.auth_codes: "queue.Queue[str]" = queue.Queue() self.loop = asyncio.new_event_loop() @@ -60,9 +70,24 @@ class PlaywrightBackend: async def _login(self, email: str, password: str) -> None: try: self.args.auth_code_provider = self.read_auth_code_from_tui - self.session = PlaywrightChatGPTSession(self.args, reporter=lambda message: self.emit("status", message)) - await self.session.start() - await self.session.login(email, password) + + def reporter(message: str) -> None: + self.emit("status", message) + + if ( + self.args.backend == "moli" + and self.args.auth_backend == "chromium-bridge" + ): + self.session = await create_authenticated_moli_session( + self.args, + email, + password, + reporter=reporter, + ) + else: + self.session = PlaywrightChatGPTSession(self.args, reporter=reporter) + await self.session.start() + await self.session.login(email, password) self.emit("login_ok", "login ok") except Exception as error: # noqa: BLE001 - show operational failures in the TUI. await self._cleanup() @@ -102,18 +127,39 @@ class PlaywrightBackend: def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Interactive Playwright TUI for ChatGPT over Moli CDP.") - parser.add_argument("--backend", choices=("moli", "chromium"), default="moli", help="browser backend, default: moli") - parser.add_argument("--url", default=DEFAULT_URL, help=f"initial URL, default: {DEFAULT_URL}") - parser.add_argument("--email", default=os.environ.get("CHATGPT_EMAIL", ""), help="pre-fill email") + parser = argparse.ArgumentParser( + description="Interactive Playwright TUI for ChatGPT over Moli CDP." + ) + parser.add_argument( + "--backend", + choices=("moli", "chromium"), + default="moli", + help="browser backend, default: moli", + ) + parser.add_argument( + "--auth-backend", + choices=("chromium-bridge", "moli"), + default="chromium-bridge", + help="fresh-login backend when chatting through Moli, default: chromium-bridge", + ) + parser.add_argument( + "--url", default=DEFAULT_URL, help=f"initial URL, default: {DEFAULT_URL}" + ) + parser.add_argument( + "--email", default=os.environ.get("CHATGPT_EMAIL", ""), help="pre-fill email" + ) parser.add_argument("--prompt", default="", help="pre-fill the prompt input") - parser.add_argument("--auth-code", default=os.environ.get("CHATGPT_AUTH_CODE", ""), help="verification code for email auth fallback") + parser.add_argument( + "--auth-code", + default=os.environ.get("CHATGPT_AUTH_CODE", ""), + help="verification code for email auth fallback", + ) parser.add_argument( "--try-email-verification", dest="try_email_verification", action="store_true", default=TUI_TRY_EMAIL_VERIFICATION, - help="click Try with email on device approval pages, default: enabled", + help="click Try with email on device approval pages, default: disabled", ) parser.add_argument( "--no-try-email-verification", @@ -135,30 +181,66 @@ def build_parser() -> argparse.ArgumentParser: "--live-trace-output", help="append sanitized live-trace summaries as JSON Lines to this path", ) - parser.add_argument("--moli-bin", help="path to the moli binary; alternatively set MOLI_BIN") - parser.add_argument("--chromium-bin", help="path to a Chromium/Chrome executable for --backend chromium") - parser.add_argument("--headful", action="store_true", help="run Chromium with a visible browser window") - parser.add_argument("--profile-dir", help="optional Moli profile dir for cookies/localStorage") + parser.add_argument( + "--moli-bin", help="path to the moli binary; alternatively set MOLI_BIN" + ) + parser.add_argument( + "--chromium-bin", + help="path to a Chromium/Chrome executable for --backend chromium", + ) + parser.add_argument( + "--auth-chromium-bin", + help="Chromium/Chrome executable used only for Moli login bootstrap", + ) + parser.add_argument( + "--headful", + action="store_true", + help="run Chromium with a visible browser window", + ) + parser.add_argument( + "--profile-dir", + default=os.environ.get( + "MOLI_CHATGPT_PROFILE_DIR", str(default_moli_chatgpt_profile_dir()) + ), + help="persistent Moli login profile, default: $MOLI_CHATGPT_PROFILE_DIR or the user data directory", + ) parser.add_argument("--user-agent", help="optional user agent passed to moli serve") parser.add_argument("--http-proxy", help="optional proxy passed to moli serve") - parser.add_argument("--http-no-proxy", help="optional no-proxy list passed to moli serve") + parser.add_argument( + "--http-no-proxy", help="optional no-proxy list passed to moli serve" + ) parser.add_argument( "--http-timeout", type=int, default=DEFAULT_MOLI_HTTP_TIMEOUT_MS, help=f"Moli request timeout in milliseconds, default: {DEFAULT_MOLI_HTTP_TIMEOUT_MS}", ) - parser.add_argument("--http-max-concurrent", type=int, help="optional max active fetch transfers passed to moli serve") - parser.add_argument("--http-max-host-open", type=int, help="optional per-host fetch transfer cap passed to moli serve") - parser.add_argument("--startup-timeout", type=float, default=DEFAULT_STARTUP_TIMEOUT) + parser.add_argument( + "--http-max-concurrent", + type=int, + help="optional max active fetch transfers passed to moli serve", + ) + parser.add_argument( + "--http-max-host-open", + type=int, + help="optional per-host fetch transfer cap passed to moli serve", + ) + parser.add_argument( + "--startup-timeout", type=float, default=DEFAULT_STARTUP_TIMEOUT + ) parser.add_argument("--login-timeout", type=float, default=TUI_LOGIN_TIMEOUT) parser.add_argument("--answer-timeout", type=float, default=DEFAULT_ANSWER_TIMEOUT) - parser.add_argument("--debug-snapshot", action="store_true", help="accepted for parity with the raw-CDP TUI") + parser.add_argument( + "--debug-snapshot", + action="store_true", + help="accepted for parity with the raw-CDP TUI", + ) return parser def main() -> int: args = build_parser().parse_args() + args.tui_title = "Moli ChatGPT Playwright TUI" curses.wrapper(run_tui_with_backend, args, PlaywrightBackend) return 0 diff --git a/moli-playground/chatgpt/test_chatgpt_auth_bridge.py b/moli-playground/chatgpt/test_chatgpt_auth_bridge.py new file mode 100644 index 0000000000..089b15f727 --- /dev/null +++ b/moli-playground/chatgpt/test_chatgpt_auth_bridge.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import os +import stat +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from chatgpt_auth_bridge import ( + auth_proxy_options, + chromium_cookie_params, + default_moli_chatgpt_profile_dir, + load_profile_user_agent, + prepare_profile_dir, + save_profile_user_agent, +) +from chatgpt_cdp_demo import CHATGPT_HELPER_JS +from chatgpt_playwright_tui import PlaywrightBackend, build_parser + + +class ChatGPTAuthBridgeTests(unittest.TestCase): + def test_default_profile_uses_xdg_data_home(self) -> None: + with mock.patch.dict( + os.environ, {"XDG_DATA_HOME": "/tmp/moli-xdg-data"}, clear=False + ): + self.assertEqual( + default_moli_chatgpt_profile_dir(), + Path("/tmp/moli-xdg-data/moli/chatgpt-profile"), + ) + + def test_profile_metadata_round_trips_with_private_permissions(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + profile = Path(temporary) / "profile" + prepared = prepare_profile_dir(profile) + save_profile_user_agent(prepared, "Test Browser/1.0") + + self.assertEqual(load_profile_user_agent(prepared), "Test Browser/1.0") + self.assertEqual(stat.S_IMODE(prepared.stat().st_mode), 0o700) + self.assertEqual( + stat.S_IMODE( + (prepared / ".chatgpt-playground-auth.json").stat().st_mode + ), + 0o600, + ) + + def test_cookie_conversion_keeps_only_chatgpt_auth_domains(self) -> None: + converted = chromium_cookie_params( + [ + { + "name": "session", + "value": "secret", + "domain": ".chatgpt.com", + "path": "/", + "expires": 12345.0, + "httpOnly": True, + "secure": True, + "sameSite": "Lax", + }, + { + "name": "auth", + "value": "secret-2", + "domain": "auth.openai.com", + "path": "/", + "expires": -1, + "httpOnly": True, + "secure": True, + "sameSite": "None", + }, + { + "name": "unrelated", + "value": "ignored", + "domain": "example.com", + "path": "/", + }, + ] + ) + + self.assertEqual([cookie["name"] for cookie in converted], ["session", "auth"]) + self.assertEqual(converted[0]["expires"], 12345.0) + self.assertNotIn("expires", converted[1]) + + def test_auth_chromium_inherits_proxy_environment(self) -> None: + args = mock.Mock(http_proxy=None, http_no_proxy=None) + with mock.patch.dict( + os.environ, + { + "HTTPS_PROXY": "http://proxy.example:8080", + "NO_PROXY": "localhost,127.0.0.1", + }, + clear=True, + ): + self.assertEqual( + auth_proxy_options(args), + ("http://proxy.example:8080", "localhost,127.0.0.1"), + ) + + def test_helper_ignores_google_hidden_password_and_avoids_double_submit( + self, + ) -> None: + self.assertIn("hiddenpassword", CHATGPT_HELPER_JS) + self.assertIn("style?.display === 'none'", CHATGPT_HELPER_JS) + self.assertIn( + "return { ok: true, reactSubmit: true, clickedSubmit: false };", + CHATGPT_HELPER_JS, + ) + + def test_playwright_tui_defaults_to_bridge_and_persistent_profile(self) -> None: + with mock.patch.dict( + os.environ, {"XDG_DATA_HOME": "/tmp/moli-tui-data"}, clear=False + ): + args = build_parser().parse_args([]) + + self.assertEqual(args.auth_backend, "chromium-bridge") + self.assertEqual(args.profile_dir, "/tmp/moli-tui-data/moli/chatgpt-profile") + self.assertFalse(args.try_email_verification) + self.assertFalse(PlaywrightBackend.requires_password) + + +if __name__ == "__main__": + unittest.main()