You can close this tab and return to Glossarion.
" "" ) self.wfile.write(html.encode("utf-8")) # Signal the server to stop (handled in a thread below) threading.Thread(target=self.server.shutdown, daemon=True).start() else: self.send_response(404) self.end_headers() def _find_available_port() -> int: """Return the callback port for OAuth. OpenAI's Codex client ID (app_EMoamEEZ73f0CkXaXp7hrann) has a fixed redirect URI registered as http://localhost:1455/auth/callback. We *must* use port 1455 or OpenAI will reject the request. """ import socket try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((CALLBACK_HOST, CALLBACK_PORT)) return CALLBACK_PORT except OSError: raise RuntimeError( f"Port {CALLBACK_PORT} is already in use. " "Please close any application using that port and try again. " "(OpenAI's OAuth requires this exact port.)" ) # =========================================================================== # OAuth flow orchestrator # =========================================================================== def run_oauth_flow(timeout: int = 300) -> Dict: """Run the full OAuth PKCE login flow. 1. Opens a browser for the user to authenticate with ChatGPT. 2. Captures the callback on a local server. 3. Exchanges the code for tokens. Returns the token dict (access_token, refresh_token, expires_at, …). Raises RuntimeError on failure or timeout. Parameters ---------- timeout : int Maximum seconds to wait for the user to complete the browser login. """ port = _find_available_port() redirect_uri = f"http://{CALLBACK_HOST}:{port}{CALLBACK_PATH}" code_verifier, code_challenge = generate_pkce() state = secrets.token_urlsafe(32) auth_url = build_auth_url(code_challenge, state, redirect_uri) # Start local callback server server = HTTPServer((CALLBACK_HOST, port), _OAuthCallbackHandler) server._auth_code = None server._returned_state = None server._error = None server.timeout = timeout print(f"🔐 Opening browser for ChatGPT login…") print(f" If the browser doesn't open, visit:\n {auth_url}") webbrowser.open(auth_url) # Serve until callback is received or timeout server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() server_thread.join(timeout=timeout) # Cleanup server.shutdown() if server._error: raise RuntimeError(f"OAuth error: {server._error}") if not server._auth_code: raise RuntimeError("OAuth login timed out – no callback received.") if server._returned_state != state: raise RuntimeError("OAuth state mismatch – possible CSRF attack.") # Exchange code for tokens print("🔑 Exchanging authorization code for tokens…") tokens = exchange_code_for_tokens(server._auth_code, code_verifier, redirect_uri) print("✅ ChatGPT OAuth authentication successful!") # Extract and log account info (non-sensitive) id_token = tokens.get("id_token", "") if id_token: info = extract_account_info(id_token) plan = info.get("plan_type", "unknown") email = info.get("email", "") if email: print(f" Account: {email} (plan: {plan})") return tokens # =========================================================================== # Token store (persistent, thread-safe) # =========================================================================== class AuthGPTTokenStore: """Thread-safe token store backed by a JSON file.""" def __init__(self, token_file: Optional[str] = None): self._token_file = ( token_file or os.environ.get("AUTHGPT_TOKEN_FILE") or _DEFAULT_TOKEN_FILE ) self._lock = threading.RLock() self._tokens: Optional[Dict] = None self._on_change_callbacks: List = [] # called after save/clear # Eagerly load cached tokens from disk (if any) self._load_from_disk() def on_token_change(self, callback): """Register *callback* to be called (no args) after tokens change.""" self._on_change_callbacks.append(callback) def _fire_change_callbacks(self): for cb in self._on_change_callbacks: try: cb() except Exception: pass # ------------------------------------------------------------------ # Persistence # ------------------------------------------------------------------ def _ensure_dir(self): d = os.path.dirname(self._token_file) if d: os.makedirs(d, exist_ok=True) def _load_from_disk(self): """Load tokens from the JSON file into memory.""" try: if os.path.isfile(self._token_file): with open(self._token_file, "r", encoding="utf-8") as f: self._tokens = json.load(f) logger.debug("AuthGPT tokens loaded from %s", self._token_file) except Exception as exc: logger.warning("Failed to load authgpt tokens: %s", exc) self._tokens = None def save_tokens(self, tokens: Dict): """Save tokens to disk and cache in memory.""" with self._lock: self._tokens = tokens try: self._ensure_dir() with open(self._token_file, "w", encoding="utf-8") as f: json.dump(tokens, f, indent=2) logger.debug("AuthGPT tokens saved to %s", self._token_file) except Exception as exc: logger.warning("Failed to save authgpt tokens: %s", exc) self._fire_change_callbacks() def load_tokens(self) -> Optional[Dict]: """Return cached tokens (loading from disk if needed).""" with self._lock: if self._tokens is None: self._load_from_disk() return self._tokens def clear_tokens(self): """Delete stored tokens (logout).""" with self._lock: self._tokens = None try: if os.path.isfile(self._token_file): os.remove(self._token_file) logger.info("AuthGPT tokens removed") except Exception as exc: logger.warning("Failed to remove token file: %s", exc) self._fire_change_callbacks() # ------------------------------------------------------------------ # Token access # ------------------------------------------------------------------ def _is_token_expired(self, tokens: Dict) -> bool: """Check if the access token has expired or is about to.""" expires_at = tokens.get("expires_at", 0) return time.time() >= (expires_at - TOKEN_REFRESH_MARGIN_SECONDS) def _try_refresh(self, tokens: Dict) -> Optional[Dict]: """Attempt to refresh the access token using the stored refresh token.""" rt = tokens.get("refresh_token") if not rt: return None try: new_tokens = refresh_access_token(rt) # Preserve fields from old tokens that aren't returned by refresh merged = {**tokens, **new_tokens} self.save_tokens(merged) logger.info("AuthGPT access token refreshed successfully") return merged except Exception as exc: logger.warning("AuthGPT token refresh failed: %s", exc) return None def get_valid_access_token(self, auto_login: bool = True) -> str: """Return a valid access token, refreshing or re-authenticating as needed. Parameters ---------- auto_login : bool If True and no valid token can be obtained, launch the browser OAuth flow interactively. Returns ------- str A valid Bearer access token. Raises ------ RuntimeError If a valid token cannot be obtained. """ with self._lock: tokens = self.load_tokens() # Happy path – have a valid token if tokens and tokens.get("access_token") and not self._is_token_expired(tokens): return tokens["access_token"] # Try refresh if tokens and tokens.get("refresh_token"): refreshed = self._try_refresh(tokens) if refreshed and refreshed.get("access_token"): return refreshed["access_token"] # No usable tokens – need interactive login if not auto_login: raise RuntimeError( "AuthGPT: No valid tokens and auto_login is disabled. " "Run the OAuth login flow first." ) # Detect headless environments (HF Spaces, Docker, etc.) where browser login is impossible is_headless = ( os.environ.get("SPACE_ID") is not None or os.environ.get("HF_SPACES") == "true" or os.environ.get("DOCKER_CONTAINER") == "true" or os.environ.get("KUBERNETES_SERVICE_HOST") is not None ) if is_headless: # Check for manually-provided tokens via environment variables env_access = os.environ.get("AUTHGPT_ACCESS_TOKEN", "").strip() env_refresh = os.environ.get("AUTHGPT_REFRESH_TOKEN", "").strip() if env_access: # User provided an access token directly — save and use it manual_tokens = { "access_token": env_access, "expires_at": time.time() + 3600, # assume 1h validity } if env_refresh: manual_tokens["refresh_token"] = env_refresh self.save_tokens(manual_tokens) logger.info("AuthGPT: Using access token from AUTHGPT_ACCESS_TOKEN env var") return env_access if env_refresh: # Try refreshing with the provided refresh token try: refreshed = refresh_access_token(env_refresh) self.save_tokens(refreshed) logger.info("AuthGPT: Obtained access token via AUTHGPT_REFRESH_TOKEN env var") return refreshed["access_token"] except Exception as ref_exc: raise RuntimeError( f"AuthGPT: AUTHGPT_REFRESH_TOKEN was set but refresh failed: {ref_exc}\n" "The refresh token may be expired. Please obtain a new one." ) raise RuntimeError( "AuthGPT: Browser-based OAuth login is not available in headless environments " "(e.g. Hugging Face Spaces, Docker containers).\n" "To use AuthGPT models, set one of these as environment secrets:\n" " • AUTHGPT_ACCESS_TOKEN — a valid ChatGPT OAuth access token\n" " • AUTHGPT_REFRESH_TOKEN — a ChatGPT OAuth refresh token (will auto-refresh)\n" "You can obtain these by running the OAuth flow locally first, then copying\n" "the tokens from ~/.glossarion/authgpt_tokens.json" ) print("🔄 AuthGPT: No valid token found – starting browser login…") new_tokens = run_oauth_flow() self.save_tokens(new_tokens) return new_tokens["access_token"] @property def has_tokens(self) -> bool: """Return True if any tokens are cached (may be expired).""" tokens = self.load_tokens() return bool(tokens and tokens.get("access_token")) @property def account_info(self) -> Dict: """Return account info extracted from cached id_token, or empty dict.""" tokens = self.load_tokens() if not tokens or not tokens.get("id_token"): return {} return extract_account_info(tokens["id_token"]) # Module-level singleton for convenience (lazy-initialized) _default_store: Optional[AuthGPTTokenStore] = None _default_store_lock = threading.Lock() def get_default_store() -> AuthGPTTokenStore: """Return the module-level default token store (singleton).""" global _default_store if _default_store is None: with _default_store_lock: if _default_store is None: _default_store = AuthGPTTokenStore() return _default_store # =========================================================================== # ChatGPT backend API adapter (Codex Responses API) # =========================================================================== # The Codex CLI uses /backend-api/codex/responses with the standard OpenAI # Responses API format. The /conversation endpoint is reserved for the # ChatGPT web UI and rejects third-party OAuth tokens with 403. # =========================================================================== def _build_responses_body( messages: List[Dict], model: str, temperature: Optional[float] = None, max_tokens: Optional[int] = None, ) -> Dict: """Build a Codex Responses API request body from standard OpenAI messages. The /backend-api/codex/responses endpoint expects: - "instructions": system-level instructions (required) - "input": list of {type, role, content:[{type, text}]} message objects - "model", "store", and optional temperature / max_output_tokens """ instructions = "" input_items: List[Dict[str, Any]] = [] def _convert_content_parts(content) -> List[Dict[str, Any]]: """Convert Chat Completions content to Responses API content parts. Handles both plain strings and multi-modal lists (text + image_url). """ if isinstance(content, str): return [{"type": "input_text", "text": content}] if not isinstance(content, list): return [{"type": "input_text", "text": str(content)}] parts: List[Dict[str, Any]] = [] for part in content: if isinstance(part, str): parts.append({"type": "input_text", "text": part}) continue if not isinstance(part, dict): continue ptype = part.get("type", "") if ptype == "text": parts.append({"type": "input_text", "text": part.get("text", "")}) elif ptype == "image_url": # Chat Completions: {"type": "image_url", "image_url": {"url": "..."}} # Responses API: {"type": "input_image", "image_url": "..."} img = part.get("image_url", {}) url = img.get("url", "") if isinstance(img, dict) else str(img) if url: parts.append({"type": "input_image", "image_url": url}) elif ptype == "input_text": # Already in Responses format parts.append(part) elif ptype == "input_image": parts.append(part) else: # Unknown part type — pass text if present text = part.get("text") if text: parts.append({"type": "input_text", "text": str(text)}) return parts or [{"type": "input_text", "text": ""}] for msg in messages: role = msg.get("role", "user") content = msg.get("content", "") # System messages become the instructions blob if role == "system": # instructions must be a string; extract text from multi-part content if isinstance(content, str): instructions = content elif isinstance(content, list): text_parts = [] for part in content: if isinstance(part, str): text_parts.append(part) elif isinstance(part, dict): t = part.get("text", "") if t: text_parts.append(str(t)) instructions = "\n".join(text_parts) else: instructions = str(content) continue # Map to Responses API structured message format input_items.append({ "type": "message", "role": "developer" if role == "system" else role, "content": _convert_content_parts(content), }) body: Dict[str, Any] = { "model": model, "instructions": instructions, "input": input_items, "store": False, "stream": True, } if max_tokens is not None: body["max_output_tokens"] = max_tokens return body def _parse_responses_result(data: Dict) -> Dict: """Extract content from a Responses API result.""" content = "" finish_reason = "stop" usage = None response_id = data.get("id") # The output field contains a list of items; find the message for item in data.get("output", []): if item.get("type") == "message": for part in item.get("content", []): if part.get("type") == "output_text": content += part.get("text", "") # Check status status = data.get("status", "") if status == "completed": finish_reason = "stop" elif status == "incomplete": incomplete = data.get("incomplete_details", {}) or {} reason = incomplete.get("reason", "") finish_reason = "length" if "tokens" in reason else reason or "incomplete" # Usage raw_usage = data.get("usage") if raw_usage: usage = { "prompt_tokens": raw_usage.get("input_tokens", 0), "completion_tokens": raw_usage.get("output_tokens", 0), "total_tokens": raw_usage.get("total_tokens", 0), } return { "content": content, "finish_reason": finish_reason, "conversation_id": response_id, "message_id": response_id, "usage": usage, } def _parse_sse_responses(raw_text: str) -> Dict: """Parse SSE stream from the Codex Responses endpoint. The endpoint may return a single JSON object or an SSE stream. """ # Try direct JSON first (non-streaming response) stripped = raw_text.strip() if stripped.startswith("{"): try: return _parse_responses_result(json.loads(stripped)) except json.JSONDecodeError: pass # SSE stream – look for response.completed event last_data = None content_parts: List[str] = [] for line in raw_text.splitlines(): line = line.strip() if not line.startswith("data: "): continue payload = line[6:] if payload == "[DONE]": break try: data = json.loads(payload) except json.JSONDecodeError: continue event_type = data.get("type", "") # Accumulate text deltas if event_type == "response.output_text.delta": content_parts.append(data.get("delta", "")) # Final completed event has the full response elif event_type == "response.completed": resp_obj = data.get("response", data) return _parse_responses_result(resp_obj) last_data = data # Fallback: if we accumulated deltas but no completed event if content_parts: return { "content": "".join(content_parts), "finish_reason": "stop", "conversation_id": None, "message_id": None, "usage": None, } # Last resort: try parsing the last data line if last_data: return _parse_responses_result(last_data) return { "content": "", "finish_reason": "error", "conversation_id": None, "message_id": None, "usage": None, } # --------------------------------------------------------------------------- # SSE stream processing helpers # --------------------------------------------------------------------------- def _process_sse_line( line: str, state: Dict, _log, log_stream: bool, t_start: float, ) -> bool: """Process a single SSE line, updating *state* in-place. Returns True when the stream should stop (saw [DONE] or response.completed). """ state["raw_lines"].append(line) if not state["got_first_data"] and line.startswith("data: "): state["got_first_data"] = True ttft = time.time() - t_start _log(f"📡 AuthGPT: First token in {ttft:.1f}s, streaming…") # Extract text deltas and display in real-time if line.startswith("data: ") and '"response.output_text.delta"' in line: try: delta_data = json.loads(line[6:]) delta_text = delta_data.get("delta", "") state["streamed_chars"] += len(delta_text) if log_stream and delta_text: log_buf = state["log_buf"] combined = "".join(log_buf) + delta_text for tag in ('', '', '', '', '', '', ''): combined = combined.replace(tag, tag + '\n') if "\n" in combined: parts = combined.split("\n") for part in parts[:-1]: _log(part) state["log_buf"] = [parts[-1]] else: log_buf.append(delta_text) if len("".join(log_buf)) > 150: _log("".join(log_buf)) state["log_buf"] = [] except (json.JSONDecodeError, KeyError): pass # Stop signals if line.strip() == "data: [DONE]": return True if '"type":"response.completed"' in line or '"type": "response.completed"' in line: return True return False def _finalize_stream(state: Dict, _log, log_stream: bool, t_start: float) -> Dict: """Flush log buffer, parse collected SSE lines, return result dict.""" if log_stream and state["log_buf"]: remainder = "".join(state["log_buf"]).strip() if remainder: _log(remainder) raw_text = "\n".join(state["raw_lines"]) t_total = time.time() - t_start _log(f"📡 AuthGPT: Stream finished in {t_total:.1f}s") result = _parse_sse_responses(raw_text) content = result.get("content", "") if not content: event_types = [] for rl in state["raw_lines"][:50]: if rl.startswith("data: ") and rl[6:] != "[DONE]": try: evt = json.loads(rl[6:]) t = evt.get("type", "(no type)") if t not in event_types: event_types.append(t) except Exception: pass _log(f"⚠️ AuthGPT: Empty content after parsing. Event types seen: {event_types}") return result def _new_stream_state() -> Dict: return { "raw_lines": [], "got_first_data": False, "streamed_chars": 0, "log_buf": [], } # --------------------------------------------------------------------------- # httpx-based SSE reader (preferred — real-time, no buffering) # --------------------------------------------------------------------------- def _stream_with_httpx( _httpx, url: str, body: Dict, headers: Dict, timeout: int, t_start: float, _log, log_stream: bool, connect_timeout: Optional[float] = None, ) -> Dict: """Stream SSE using httpx (same stack as the openai Python SDK).""" state = _new_stream_state() # httpx timeout: connect + read. When connect_timeout is None the connect # timeout falls back to the main ``timeout`` value (no separate limit). _timeout = _httpx.Timeout(timeout, connect=connect_timeout) with _httpx.stream( "POST", url, json=body, headers=headers, timeout=_timeout, ) as resp: if resp.status_code >= 400: error_body = resp.read().decode("utf-8", errors="replace") reason = getattr(resp, "reason_phrase", "") or "" detail = error_body try: detail = json.loads(error_body).get("detail", error_body) except Exception: pass if not detail: detail = "empty-body" summary = detail or reason or "Bad Request" try: suppress = (resp.status_code == 429) and ("usage_limit_reached" in str(error_body).lower()) except Exception: suppress = False if not suppress: _log(f"❌ AuthGPT HTTP {resp.status_code}. {summary}") raise RuntimeError( f"AuthGPT: {resp.status_code} – {summary} [reason={reason}]" ) # iter_lines() in httpx yields str lines as they arrive for line in resp.iter_lines(): if _cancel_event.is_set(): resp.close() raise RuntimeError("AuthGPT: stream cancelled by user") if _process_sse_line(line, state, _log, log_stream, t_start): break return _finalize_stream(state, _log, log_stream, t_start) # --------------------------------------------------------------------------- # requests-based SSE reader (fallback — may buffer due to urllib3/http.client) # --------------------------------------------------------------------------- def _stream_with_requests( url: str, body: Dict, headers: Dict, timeout: int, t_start: float, _log, log_stream: bool, ) -> Dict: """Stream SSE using requests (fallback when httpx is not available).""" state = _new_stream_state() resp = requests.post(url, json=body, headers=headers, timeout=timeout, stream=True) if resp.status_code >= 400: try: error_body = resp.text except Exception: error_body = "" try: reason = resp.reason or "" except Exception: reason = "" detail = error_body try: detail = resp.json().get("detail", error_body) except Exception: pass if not detail: detail = "empty-body" summary = detail or reason or "Bad Request" try: suppress = (resp.status_code == 429) and ("usage_limit_reached" in str(error_body).lower()) except Exception: suppress = False if not suppress: _log(f"❌ AuthGPT HTTP {resp.status_code}. {summary}") raise RuntimeError( f"AuthGPT: {resp.status_code} – {summary} [reason={reason}]" ) for raw_line in resp.iter_lines(chunk_size=1): if _cancel_event.is_set(): resp.close() raise RuntimeError("AuthGPT: stream cancelled by user") if raw_line is None: continue line = raw_line.decode("utf-8", errors="replace") if isinstance(raw_line, bytes) else raw_line if _process_sse_line(line, state, _log, log_stream, t_start): break return _finalize_stream(state, _log, log_stream, t_start) # --------------------------------------------------------------------------- # Public API – send chat completion # --------------------------------------------------------------------------- def send_chat_completion( access_token: str, messages: List[Dict], model: str = "gpt-5.2", temperature: Optional[float] = 0.7, max_tokens: Optional[int] = None, timeout: int = 600, base_url: Optional[str] = None, log_fn: Optional[Any] = None, connect_timeout: Optional[float] = None, ) -> Dict: """Send a chat completion request via the ChatGPT Codex Responses API. Parameters ---------- access_token : str OAuth access token (Bearer token). messages : list of dict Standard OpenAI-format messages (role + content). model : str Model name without the ``authgpt/`` prefix. temperature : float or None Sampling temperature. max_tokens : int or None Maximum response tokens. timeout : int Request timeout in seconds. base_url : str or None Override the ChatGPT backend base URL. Returns ------- dict ``{"content": str, "finish_reason": str, "usage": dict|None, "conversation_id": str|None, "message_id": str|None}`` Raises ------ requests.HTTPError On non-200 responses from the backend. RuntimeError On unexpected response format. """ effective_base = base_url or os.getenv("AUTHGPT_BASE_URL", CHATGPT_BASE_URL) url = f"{effective_base.rstrip('/')}{RESPONSES_ENDPOINT}" body = _build_responses_body( messages=messages, model=model, temperature=temperature, max_tokens=max_tokens, ) headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "Accept-Encoding": "identity", # Disable gzip so SSE streams in real-time } _log = log_fn or print logger.info("AuthGPT: POST %s model=%s", url, model) # AuthGPT always streams (the API requires it), so streaming log is on # by default. During batch translation, silence it unless the user # explicitly enabled the authgpt-specific batch log toggle. log_stream = os.getenv("LOG_STREAM_CHUNKS", "1").lower() not in ("0", "false") if os.getenv("BATCH_TRANSLATION", "0") == "1": log_stream = os.getenv("ALLOW_AUTHGPT_BATCH_STREAM_LOGS", "0").lower() not in ("0", "false") t_start = time.time() # Use httpx for SSE streaming — its h11-based HTTP parser yields data as # it arrives from the socket, unlike requests/urllib3 which buffers # entire SSL records through http.client's internal BufferedIOBase. # This is the same HTTP stack the official openai Python SDK uses. try: import httpx as _httpx return _stream_with_httpx( _httpx, url, body, headers, timeout, t_start, _log, log_stream, connect_timeout=connect_timeout, ) except ImportError: _log("⚠️ AuthGPT: httpx not installed, falling back to requests (streaming may be buffered)") return _stream_with_requests( url, body, headers, timeout, t_start, _log, log_stream, )