@samitouri / QOSami_Agents_Zero-2
Code Commits Issues Pulls Sessions CI/CD
main
main
py 283 lines 8.4 KB
Raw
1 from __future__ import annotations
2
3 import secrets
4 import webbrowser
5 from typing import Any
6
7 from plugins._oauth.helpers.providers.base import (
8 CODEX_PROVIDER_ID,
9 DUMMY_API_KEY,
10 CallbackResult,
11 LoginPollResult,
12 LoginStartResult,
13 OAuthProviderMetadata,
14 )
15 from plugins._oauth.helpers.state import (
16 get_device_attempt,
17 pop_device_attempt,
18 put_attempt,
19 put_device_attempt,
20 )
21
22
23 CODEX_DEFAULT_MODEL = "gpt-5.5"
24 CODEX_FALLBACK_CONFIG = {
25 "enabled": True,
26 "models": [],
27 "proxy_base_path": "/oauth/codex",
28 "callback_path": "/auth/callback",
29 "open_browser_from_server": False,
30 }
31
32
33 class CodexOAuthProvider:
34 provider_id = CODEX_PROVIDER_ID
35
36 def metadata(self) -> OAuthProviderMetadata:
37 cfg = _codex_config()
38 models = list(cfg["models"])
39 return OAuthProviderMetadata(
40 provider_id=CODEX_PROVIDER_ID,
41 display_name="Codex/ChatGPT",
42 short_name="Codex",
43 model_provider_id=CODEX_PROVIDER_ID,
44 icon="openai",
45 auth_flow="device_code",
46 default_model=models[0] if models else CODEX_DEFAULT_MODEL,
47 default_models=models,
48 proxy_base_path=cfg["proxy_base_path"],
49 callback_path=cfg["callback_path"],
50 )
51
52 def status(self) -> dict[str, Any]:
53 from plugins._oauth.helpers import codex
54
55 cfg = _codex_config()
56 return {
57 **self.metadata().to_dict(),
58 **codex.status(),
59 "enabled": cfg["enabled"],
60 "proxy_base_path": cfg["proxy_base_path"],
61 "callback_path": cfg["callback_path"],
62 "v1_base_path": f'{cfg["proxy_base_path"]}/v1',
63 }
64
65 def start_login(self, input: dict[str, Any] | None = None, request: Any = None) -> LoginStartResult:
66 del input, request
67 from plugins._oauth.helpers import codex
68
69 cfg = _codex_config()
70 if not cfg["enabled"]:
71 return LoginStartResult(
72 ok=False,
73 provider_id=CODEX_PROVIDER_ID,
74 flow="device_code",
75 error="Codex/ChatGPT account connection is disabled.",
76 )
77
78 try:
79 device = codex.request_device_code()
80 attempt_id = secrets.token_urlsafe(24)
81 attempt = put_device_attempt(
82 attempt_id,
83 device["device_auth_id"],
84 device["user_code"],
85 device["interval"],
86 device["expires_at"],
87 provider_id=CODEX_PROVIDER_ID,
88 )
89 except Exception as exc:
90 return LoginStartResult(
91 ok=False,
92 provider_id=CODEX_PROVIDER_ID,
93 flow="device_code",
94 error=str(exc),
95 )
96
97 return LoginStartResult(
98 ok=True,
99 provider_id=CODEX_PROVIDER_ID,
100 flow="device_code",
101 attempt_id=attempt.attempt_id,
102 verification_url=device["verification_url"],
103 user_code=attempt.user_code,
104 interval=attempt.interval,
105 expires_at=attempt.expires_at,
106 )
107
108 def start_browser_login(
109 self,
110 input: dict[str, Any] | None = None,
111 request: Any = None,
112 ) -> LoginStartResult:
113 del input
114 cfg = _codex_config()
115 if not cfg["enabled"]:
116 return LoginStartResult(
117 ok=False,
118 provider_id=CODEX_PROVIDER_ID,
119 flow="browser_pkce",
120 error="Codex/ChatGPT account connection is disabled.",
121 )
122
123 from plugins._oauth.helpers import codex
124
125 redirect_uri = _redirect_uri(request, cfg["callback_path"])
126 pkce = codex.generate_pkce()
127 state = codex.generate_state()
128 attempt = put_attempt(
129 state,
130 pkce.verifier,
131 redirect_uri,
132 provider_id=CODEX_PROVIDER_ID,
133 )
134 auth_url = codex.build_authorize_url(redirect_uri, state, pkce)
135
136 if cfg["open_browser_from_server"]:
137 try:
138 webbrowser.open(auth_url)
139 except Exception:
140 pass
141
142 return LoginStartResult(
143 ok=True,
144 provider_id=CODEX_PROVIDER_ID,
145 flow="browser_pkce",
146 auth_url=auth_url,
147 redirect_uri=redirect_uri,
148 expires_at=attempt.expires_at,
149 )
150
151 def poll_login(self, input: dict[str, Any] | None = None, request: Any = None) -> LoginPollResult:
152 del request
153 data = input or {}
154 attempt_id = str(data.get("attempt_id") or "").strip()
155 if not attempt_id:
156 return LoginPollResult(
157 ok=False,
158 provider_id=CODEX_PROVIDER_ID,
159 error="Missing device authorization attempt.",
160 )
161
162 attempt = get_device_attempt(attempt_id)
163 if attempt is None:
164 return LoginPollResult(
165 ok=False,
166 provider_id=CODEX_PROVIDER_ID,
167 expired=True,
168 error="Device authorization expired.",
169 )
170 if attempt.provider_id != CODEX_PROVIDER_ID:
171 return LoginPollResult(
172 ok=False,
173 provider_id=CODEX_PROVIDER_ID,
174 error="Device authorization provider mismatch.",
175 )
176
177 try:
178 from plugins._oauth.helpers import codex
179
180 result = codex.poll_device_authorization(
181 attempt.device_auth_id,
182 attempt.user_code,
183 )
184 except Exception as exc:
185 return LoginPollResult(ok=False, provider_id=CODEX_PROVIDER_ID, error=str(exc))
186
187 if result.get("completed"):
188 pop_device_attempt(attempt_id)
189 account_id = str(result.get("account_id") or "")
190 return LoginPollResult(
191 ok=True,
192 provider_id=CODEX_PROVIDER_ID,
193 completed=True,
194 account_label=str(result.get("account_label") or account_id),
195 account_id=account_id,
196 )
197
198 return LoginPollResult(
199 ok=True,
200 provider_id=CODEX_PROVIDER_ID,
201 completed=False,
202 interval=attempt.interval,
203 expires_at=attempt.expires_at,
204 )
205
206 def complete_callback(
207 self,
208 args: dict[str, Any],
209 request: Any = None,
210 ) -> CallbackResult:
211 del args, request
212 return CallbackResult(
213 ok=False,
214 provider_id=CODEX_PROVIDER_ID,
215 error="Codex OAuth callback is handled by compatibility routes.",
216 )
217
218 def manual_callback(
219 self,
220 input: dict[str, Any],
221 request: Any = None,
222 ) -> LoginPollResult:
223 del input, request
224 return LoginPollResult(
225 ok=False,
226 provider_id=CODEX_PROVIDER_ID,
227 error="Codex uses device-code login in this UI.",
228 )
229
230 def models(self) -> list[str]:
231 from plugins._oauth.helpers import codex
232
233 return codex.fetch_models()
234
235 def model_catalog(self) -> list[dict[str, Any]]:
236 from plugins._oauth.helpers import codex
237
238 return codex.fetch_model_catalog()
239
240 def disconnect(self) -> dict[str, Any]:
241 from plugins._oauth.helpers import codex
242
243 return codex.disconnect_auth()
244
245 def api_key(self) -> str:
246 return DUMMY_API_KEY
247
248 def register_routes(self, app: Any) -> None:
249 del app
250 return None
251
252
253 def _redirect_uri(request: Any, callback_path: str) -> str:
254 origin = ""
255 if request is not None:
256 origin = (getattr(request, "headers", {}).get("Origin") or "").rstrip("/")
257 if not _is_local_origin(origin):
258 origin = getattr(request, "url_root", "").rstrip("/")
259 return f"{origin}{callback_path}"
260
261
262 def _codex_config() -> dict[str, Any]:
263 try:
264 from plugins._oauth.helpers.config import codex_config
265
266 return codex_config()
267 except ModuleNotFoundError as exc:
268 if exc.name and exc.name.startswith("plugins._oauth"):
269 raise
270 return dict(CODEX_FALLBACK_CONFIG)
271
272
273 def _is_local_origin(origin: str) -> bool:
274 if not origin:
275 return False
276 return (
277 origin.startswith("http://localhost:")
278 or origin == "http://localhost"
279 or origin.startswith("http://127.0.0.1:")
280 or origin == "http://127.0.0.1"
281 or origin.startswith("http://[::1]:")
282 or origin == "http://[::1]"
283 )