| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import subprocess |
| 6 | import time |
| 7 | from datetime import datetime, timezone |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | import requests |
| 12 | |
| 13 | from plugins._orchestrator.helpers.adapters.base import TerminalAgentAdapter |
| 14 | |
| 15 | # Codex CLI's public OAuth client (same constants the official CLI uses; |
| 16 | # see also plugins/_oauth for the reference implementation). |
| 17 | CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann" |
| 18 | ISSUER = "https://auth.openai.com" |
| 19 | TOKEN_URL = "https://auth.openai.com/oauth/token" |
| 20 | AUTH_FILENAME = "auth.json" |
| 21 | DEVICE_CODE_TIMEOUT_SECONDS = 15 * 60 |
| 22 | |
| 23 | |
| 24 | class CodexAdapter(TerminalAgentAdapter): |
| 25 | id = "codex" |
| 26 | title = "OpenAI Codex CLI" |
| 27 | binary = "codex" |
| 28 | install_hint = "npm install -g @openai/codex" |
| 29 | description = "OpenAI Codex CLI for autonomous coding tasks in a workdir." |
| 30 | |
| 31 | # --- authentication ---------------------------------------------------- |
| 32 | |
| 33 | def _plugin_auth_path(self) -> Path: |
| 34 | return self.data_dir() / AUTH_FILENAME |
| 35 | |
| 36 | def _external_auth_path(self) -> Path: |
| 37 | codex_home = os.environ.get("CODEX_HOME", "").strip() |
| 38 | if codex_home: |
| 39 | return Path(codex_home).expanduser() / AUTH_FILENAME |
| 40 | return Path.home() / ".codex" / AUTH_FILENAME |
| 41 | |
| 42 | def auth_status(self, config: dict[str, Any] | None = None) -> dict[str, Any]: |
| 43 | plugin_path = self._plugin_auth_path() |
| 44 | if _has_chatgpt_tokens(plugin_path): |
| 45 | return { |
| 46 | "connected": True, |
| 47 | "mode": "plugin", |
| 48 | "auth_path": str(plugin_path), |
| 49 | } |
| 50 | external_path = self._external_auth_path() |
| 51 | if _has_chatgpt_tokens(external_path): |
| 52 | return { |
| 53 | "connected": True, |
| 54 | "mode": "external", |
| 55 | "auth_path": str(external_path), |
| 56 | } |
| 57 | return {"connected": False, "mode": "", "auth_path": ""} |
| 58 | |
| 59 | def supports_device_login(self) -> bool: |
| 60 | return True |
| 61 | |
| 62 | def start_device_login(self) -> dict[str, Any]: |
| 63 | response = requests.post( |
| 64 | f"{ISSUER}/api/accounts/deviceauth/usercode", |
| 65 | headers={"Content-Type": "application/json"}, |
| 66 | json={"client_id": CLIENT_ID}, |
| 67 | timeout=30, |
| 68 | ) |
| 69 | if not response.ok: |
| 70 | raise RuntimeError(_oauth_error(response)) |
| 71 | payload = response.json() if isinstance(response.json(), dict) else {} |
| 72 | device_auth_id = str(payload.get("device_auth_id") or "") |
| 73 | user_code = str(payload.get("user_code") or payload.get("usercode") or "") |
| 74 | if not device_auth_id or not user_code: |
| 75 | raise RuntimeError("Device authorization response did not include a code.") |
| 76 | return { |
| 77 | "device_auth_id": device_auth_id, |
| 78 | "user_code": user_code, |
| 79 | "interval": _safe_int(payload.get("interval"), 5), |
| 80 | "expires_at": time.time() + DEVICE_CODE_TIMEOUT_SECONDS, |
| 81 | "verification_url": f"{ISSUER}/codex/device", |
| 82 | } |
| 83 | |
| 84 | def poll_device_login(self, payload: dict[str, Any]) -> dict[str, Any]: |
| 85 | device_auth_id = str(payload.get("device_auth_id") or "") |
| 86 | user_code = str(payload.get("user_code") or "") |
| 87 | if not device_auth_id or not user_code: |
| 88 | raise RuntimeError("Missing device_auth_id or user_code.") |
| 89 | response = requests.post( |
| 90 | f"{ISSUER}/api/accounts/deviceauth/token", |
| 91 | headers={"Content-Type": "application/json"}, |
| 92 | json={"device_auth_id": device_auth_id, "user_code": user_code}, |
| 93 | timeout=30, |
| 94 | ) |
| 95 | if response.status_code in {403, 404}: |
| 96 | return {"completed": False} |
| 97 | if not response.ok: |
| 98 | raise RuntimeError(_oauth_error(response)) |
| 99 | data = response.json() if isinstance(response.json(), dict) else {} |
| 100 | authorization_code = str(data.get("authorization_code") or "") |
| 101 | verifier = str(data.get("code_verifier") or "") |
| 102 | if not authorization_code or not verifier: |
| 103 | raise RuntimeError("Device authorization response missing exchange data.") |
| 104 | tokens = _exchange_code( |
| 105 | authorization_code, f"{ISSUER}/deviceauth/callback", verifier |
| 106 | ) |
| 107 | account_id = _derive_account_id(tokens["id_token"]) |
| 108 | self._write_auth(tokens, account_id) |
| 109 | return {"completed": True, "account_id": account_id} |
| 110 | |
| 111 | def can_disconnect(self, config: dict[str, Any] | None = None) -> bool: |
| 112 | status = self.auth_status(config) |
| 113 | return bool(status.get("connected")) |
| 114 | |
| 115 | def disconnect(self, config: dict[str, Any] | None = None) -> dict[str, Any]: |
| 116 | status = self.auth_status(config) |
| 117 | path = self._plugin_auth_path() |
| 118 | if status.get("mode") == "plugin" and path.exists(): |
| 119 | path.unlink() |
| 120 | return {"removed": True, "mode": "plugin"} |
| 121 | if status.get("mode") == "external": |
| 122 | result = subprocess.run( |
| 123 | [self.resolve_binary(config), "logout"], |
| 124 | capture_output=True, |
| 125 | text=True, |
| 126 | timeout=30, |
| 127 | ) |
| 128 | if result.returncode != 0: |
| 129 | raise RuntimeError((result.stderr or result.stdout).strip()) |
| 130 | return {"removed": True, "mode": "external"} |
| 131 | return {"removed": False, "message": "No Codex credentials found."} |
| 132 | |
| 133 | def _write_auth(self, tokens: dict[str, str], account_id: str) -> None: |
| 134 | auth_data = { |
| 135 | "auth_mode": "chatgpt", |
| 136 | "OPENAI_API_KEY": None, |
| 137 | "tokens": { |
| 138 | "id_token": tokens["id_token"], |
| 139 | "access_token": tokens["access_token"], |
| 140 | "refresh_token": tokens["refresh_token"], |
| 141 | "account_id": account_id, |
| 142 | }, |
| 143 | "last_refresh": _utc_now_iso(), |
| 144 | } |
| 145 | path = self._plugin_auth_path() |
| 146 | path.parent.mkdir(parents=True, exist_ok=True) |
| 147 | tmp = path.with_suffix(".tmp") |
| 148 | tmp.write_text(json.dumps(auth_data, indent=2)) |
| 149 | os.chmod(tmp, 0o600) |
| 150 | tmp.replace(path) |
| 151 | |
| 152 | |
| 153 | def _has_chatgpt_tokens(path: Path) -> bool: |
| 154 | try: |
| 155 | data = json.loads(path.read_text()) |
| 156 | except (OSError, ValueError): |
| 157 | return False |
| 158 | tokens = data.get("tokens") if isinstance(data, dict) else None |
| 159 | if not isinstance(tokens, dict): |
| 160 | return False |
| 161 | return bool(tokens.get("access_token") and tokens.get("refresh_token")) |
| 162 | |
| 163 | |
| 164 | def _exchange_code(code: str, redirect_uri: str, verifier: str) -> dict[str, str]: |
| 165 | response = requests.post( |
| 166 | TOKEN_URL, |
| 167 | headers={"Content-Type": "application/x-www-form-urlencoded"}, |
| 168 | data={ |
| 169 | "grant_type": "authorization_code", |
| 170 | "code": code, |
| 171 | "redirect_uri": redirect_uri, |
| 172 | "client_id": CLIENT_ID, |
| 173 | "code_verifier": verifier, |
| 174 | }, |
| 175 | timeout=30, |
| 176 | ) |
| 177 | if not response.ok: |
| 178 | raise RuntimeError(_oauth_error(response)) |
| 179 | payload = response.json() if isinstance(response.json(), dict) else {} |
| 180 | tokens = { |
| 181 | "id_token": str(payload.get("id_token") or ""), |
| 182 | "access_token": str(payload.get("access_token") or ""), |
| 183 | "refresh_token": str(payload.get("refresh_token") or ""), |
| 184 | } |
| 185 | missing = [key for key, value in tokens.items() if not value] |
| 186 | if missing: |
| 187 | raise RuntimeError(f"OAuth token response is missing: {', '.join(missing)}") |
| 188 | return tokens |
| 189 | |
| 190 | |
| 191 | def _derive_account_id(id_token: str) -> str: |
| 192 | claims = _parse_jwt_claims(id_token) |
| 193 | auth_claims = claims.get("https://api.openai.com/auth") |
| 194 | if isinstance(auth_claims, dict): |
| 195 | value = auth_claims.get("chatgpt_account_id") |
| 196 | if isinstance(value, str): |
| 197 | return value |
| 198 | return "" |
| 199 | |
| 200 | |
| 201 | def _parse_jwt_claims(token: str) -> dict[str, Any]: |
| 202 | import base64 |
| 203 | |
| 204 | try: |
| 205 | payload_part = token.split(".")[1] |
| 206 | padded = payload_part + "=" * (-len(payload_part) % 4) |
| 207 | value = json.loads(base64.urlsafe_b64decode(padded)) |
| 208 | return value if isinstance(value, dict) else {} |
| 209 | except Exception: |
| 210 | return {} |
| 211 | |
| 212 | |
| 213 | def _oauth_error(response: requests.Response) -> str: |
| 214 | try: |
| 215 | payload = response.json() |
| 216 | except Exception: |
| 217 | payload = None |
| 218 | if isinstance(payload, dict): |
| 219 | for key in ("error_description", "error"): |
| 220 | value = payload.get(key) |
| 221 | if isinstance(value, str) and value: |
| 222 | return value |
| 223 | return f"OAuth endpoint returned status {response.status_code}." |
| 224 | |
| 225 | |
| 226 | def _safe_int(value: Any, default: int) -> int: |
| 227 | try: |
| 228 | return int(value) |
| 229 | except (TypeError, ValueError): |
| 230 | return default |
| 231 | |
| 232 | |
| 233 | def _utc_now_iso() -> str: |
| 234 | return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") |