"""Browser-based OAuth authentication for Cua CLI.""" import asyncio import os import platform import subprocess import urllib.parse from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread from typing import Optional # Default URLs DEFAULT_WEBSITE_URL = "https://cua.ai" # Timeout for browser authentication (2 minutes) AUTH_TIMEOUT_SECONDS = 120 @dataclass class AuthResult: """Result from browser authentication containing token and workspace metadata.""" token: str workspace_slug: str workspace_name: str org_name: str class CallbackHandler(BaseHTTPRequestHandler): """HTTP request handler for OAuth callback.""" token: Optional[str] = None workspace_slug: Optional[str] = None workspace_name: Optional[str] = None org_name: Optional[str] = None error: Optional[str] = None def do_GET(self) -> None: """Handle GET request from OAuth callback.""" parsed = urllib.parse.urlparse(self.path) params = urllib.parse.parse_qs(parsed.query) if "token" in params: CallbackHandler.token = params["token"][0] CallbackHandler.workspace_slug = params.get("workspace_slug", [""])[0] CallbackHandler.workspace_name = params.get("workspace_name", [""])[0] CallbackHandler.org_name = params.get("org_name", [""])[0] self._send_response( 200, "

Authentication successful!

" "

You can close this window and return to the terminal.

", ) elif "error" in params: CallbackHandler.error = params.get("error_description", params["error"])[0] self._send_response( 400, f"

Authentication failed

" f"

{CallbackHandler.error}

", ) else: self._send_response( 400, "

Invalid callback

" "

Missing token parameter.

", ) def _send_response(self, status: int, body: str) -> None: """Send an HTTP response.""" self.send_response(status) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write(body.encode()) def log_message(self, format: str, *args) -> None: """Suppress default logging.""" pass def open_browser(url: str) -> bool: """Open a URL in the default browser. Args: url: The URL to open Returns: True if the browser was opened successfully """ system = platform.system() try: if system == "Darwin": subprocess.run(["open", url], check=True) elif system == "Windows": subprocess.run(["cmd", "/c", "start", url], check=True, shell=True) else: # Linux and others subprocess.run(["xdg-open", url], check=True) return True except (subprocess.SubprocessError, FileNotFoundError): return False async def authenticate_via_browser( website_url: Optional[str] = None, workspace_slug: Optional[str] = None, ) -> AuthResult: """Authenticate via browser OAuth flow. Starts a local HTTP server to receive the callback, opens the browser to the authentication page, and waits for the token. Args: website_url: Base URL for the authentication page. Defaults to CUA_WEBSITE_URL env var or https://cua.ai workspace_slug: Optional workspace slug to pre-select in the browser Returns: AuthResult with token and workspace metadata Raises: TimeoutError: If authentication times out RuntimeError: If authentication fails """ website_url = website_url or os.environ.get("CUA_WEBSITE_URL", DEFAULT_WEBSITE_URL) # Reset handler state CallbackHandler.token = None CallbackHandler.workspace_slug = None CallbackHandler.workspace_name = None CallbackHandler.org_name = None CallbackHandler.error = None # Start local server on dynamic port server = HTTPServer(("localhost", 0), CallbackHandler) port = server.server_address[1] callback_url = f"http://localhost:{port}" # Build auth URL encoded_callback = urllib.parse.quote(callback_url, safe="") auth_url = f"{website_url}/cli-auth?callback_url={encoded_callback}" if workspace_slug: auth_url += f"&workspace_slug={urllib.parse.quote(workspace_slug, safe='')}" # Start server in background thread server_thread = Thread(target=server.handle_request, daemon=True) server_thread.start() # Open browser print("Opening browser for authentication...") if not open_browser(auth_url): print("Could not open browser automatically.") print(f"Please visit: {auth_url}") # Wait for callback with timeout try: for _ in range(AUTH_TIMEOUT_SECONDS): if CallbackHandler.token or CallbackHandler.error: break await asyncio.sleep(1) else: raise TimeoutError("Authentication timed out. Please try again.") if CallbackHandler.error: raise RuntimeError(f"Authentication failed: {CallbackHandler.error}") if not CallbackHandler.token: raise RuntimeError("Authentication failed: No token received") return AuthResult( token=CallbackHandler.token, workspace_slug=CallbackHandler.workspace_slug or "", workspace_name=CallbackHandler.workspace_name or "", org_name=CallbackHandler.org_name or "", ) finally: server.server_close()