diff --git a/docs/api.md b/docs/api.md index 99f26654d..fa4894d67 100644 --- a/docs/api.md +++ b/docs/api.md @@ -39,6 +39,12 @@ data: {"error":{"message":"Upstream failed","type":"server_error","param":null," Model name parsing supports configured aliases plus `-low`, `-medium`, `-high`, `-xhigh`, `-fast`, and `-flex` suffixes. +### `POST /v1/responses` + +Uses imported Codex accounts to call `POST /codex/responses` on the configured Codex backend. The upstream request is sent with Codex Desktop headers, account bearer token, optional account id, request id, and encrypted account-scoped Cookie replay. + +The current Rust route collects HTTP SSE until `response.completed` and returns the completed OpenAI-compatible response JSON. `previous_response_id` and explicit WebSocket-only requests are rejected until the WebSocket transport is implemented and verified. + ## `/admin/*` Admin endpoints are authenticated only by HttpOnly admin session cookies. diff --git a/docs/implementation-status.md b/docs/implementation-status.md index b8a914c18..df9aa7e1a 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -13,14 +13,14 @@ Update this file before each feature commit. | Logging and pagination | Completed | `eb0341e` + `fc16c48` + `b0c10f2` | `cargo test event_logs_are_cursor_paginated`; `cargo test admin_logs_are_cursor_paginated_and_include_request_id`; `cargo test rotating_writer_splits_files_when_size_limit_is_reached`; full `fmt/test/clippy` before commit | Adds cursor pagination, `Page` camelCase serialization, admin `/admin/logs` query API, event log repository, runtime tracing initialization, HTTP trace layer, and date/size-based file rotation. | | TLS headers and fingerprint | Completed | `1bdd08c` | `cargo test codex_headers_include_desktop_identity_and_turn_state`; `cargo tree \| rg 'reqwest\|rustls'`; full `fmt/test/clippy` before commit | Adds Codex Desktop fingerprint model, exact identity/request headers, and pinned reqwest/rustls client builder. | | Cookie persistence | Completed | `4710926` + `5d3cc6e` | `cargo test cookie_jar_captures_and_replays_account_scoped_cookies`; `cargo test cookie_repository_encrypts_and_replays_account_scoped_cookies`; full `fmt/test/clippy` before commit | Adds account-scoped in-memory Cookie jar and encrypted SQLite Cookie repository; Cookies are replayed only for the same account/domain. | -| Account pool and refresh | Partial | `0f97c5c` + `2f25db8` + `4dc0c9a` | `cargo test account_pool_skips_expired_disabled_banned_and_quota_exhausted_accounts`; `cargo test --test account_pool_scheduling_test`; `cargo test --test account_refresh_test`; full `fmt/test/clippy` before commit | Implements base account status filtering, refresh scheduler semantics, max concurrent slots, stale slot cleanup, least-used/round-robin/sticky strategies, tier priority, quota-cache filtering, Cloudflare cooldown skip, request staggering metadata, preferred account selection, exclude IDs, and capacity summary. Still missing settings-backed scheduling config, SQLite repository integration for pool snapshots/release usage, upstream retry/fallback wiring, and admin account CRUD. | +| Account pool and refresh | Partial | `0f97c5c` + `2f25db8` + `4dc0c9a` + `pending` | `cargo test account_pool_skips_expired_disabled_banned_and_quota_exhausted_accounts`; `cargo test --test account_pool_scheduling_test`; `cargo test --test account_refresh_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Implements base account status filtering, refresh scheduler semantics, max concurrent slots, stale slot cleanup, least-used/round-robin/sticky strategies, tier priority, quota-cache filtering, Cloudflare cooldown skip, request staggering metadata, preferred account selection, exclude IDs, capacity summary, settings-backed pool construction, and same-process import-to-pool sync. Still missing startup pool snapshots from SQLite, durable release-time slot state, upstream retry/fallback wiring, and admin account CRUD. | | Account repository | Completed | `14fe706` | `cargo test --test account_repository_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds encrypted SQLite account token storage, decrypting reads, cursor pagination, status/label updates, release-time usage counter persistence, and refresh-token-preserving token updates. | | Translation | Completed | `eff182e` | `cargo test chat_completion_translates_to_codex_response_request`; full `fmt/test/clippy` before commit | Adds OpenAI Chat request types, Codex Responses request type, translation, and OpenAI-compatible error body helper. | -| HTTP routes | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, admin API-key management routes, settings mutation/persistence, diagnostics, usage stats, backend model refresh, model cache persistence, and additional Responses routes. | +| HTTP routes | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` + `pending` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/v1/responses` HTTP SSE collection through imported Codex accounts, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, admin API-key management routes, settings mutation/persistence, diagnostics, usage stats, backend model refresh, model cache persistence, streaming passthrough, and WebSocket Responses routes. | | Admin/frontend response contract | Completed | `1c34ea7` + `b0c10f2` | `cargo test admin_envelope_serializes_lower_camel_case_request_id`; `cargo test admin_logs_reject_missing_admin_session_cookie`; `cargo test v1_response_has_request_id_header_without_admin_body_field`; full `fmt/test/clippy` before commit | Adds reusable admin envelopes with body codes, lower camelCase JSON, paginated list shape, requestId bodies, admin session cookie enforcement, and keeps `/v1/*` OpenAI-compatible. | | HTTP auth boundary | Completed | `cec4607` | `cargo test --test http_auth_test`; `cargo test --test routes_responses_test`; `cargo test --test admin_logs_route_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds a real `src/http/auth.rs` boundary for client API key and admin session cookie parsing. `/v1/*` accepts only `Bearer cpr_...`, admin session parsing ignores client API keys, and route handlers no longer own ad hoc credential parsing. | | Model catalog | Partial | `6341336` | `cargo test --test model_catalog_test`; `cargo test model_catalog_route_returns_codex_metadata_without_alias_entries`; full `fmt/test/clippy` before commit | Adds a native static Codex model catalog with `gpt-5.5` default, configured aliases, reasoning/service-tier suffix parsing, OpenAI-compatible `/v1/models`, extended catalog/detail/info routes, and debug summary. Still missing backend Codex model fetching per plan, cache persistence, `/admin/refresh-models`, plan allowlist integration from fetched models, and full model metadata parity. | -| Upstream lifecycle | Partial | `88cad87` + `225fbff` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; full `fmt/test/clippy` before commit | `/v1` currently enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, and includes reusable SSE/usage helpers. It does not yet acquire accounts, forward to Codex, replay/capture Cookies, stream/collect upstream responses, capture rate-limit headers, retry across accounts, record usage, or preserve previous-response/session affinity. | +| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `pending` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; `cargo test --test codex_client_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, replays/captures encrypted Cookies, collects `response.completed`, and records token usage. Still missing startup pool snapshot loading, rate-limit header capture, retry/fallback across accounts, 401 refresh integration, 429 quota cache, 402/403 classification, streaming passthrough, and previous-response/session affinity. | | Codex WebSocket boundary | Completed | `cc0516d` | `cargo test --test codex_websocket_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds `src/codex/websocket.rs` with tested transport classification for HTTP SSE versus WebSocket-only requests. `previous_response_id` and explicit WebSocket requests now fail through a typed boundary instead of being silently downgraded; real WebSocket transport remains an upstream lifecycle follow-up. | | Fingerprint updates | Partial | `dd26d31` + `e32220a` | `cargo test update_manifest_updates_app_version_and_build_number`; `cargo test fingerprint_updater_fetches_manifest_and_persists_history`; full `fmt/test/clippy` before commit | Parses a JSON update manifest and persists app version/build number history to SQLite. Still missing original appcast XML polling, update-state persistence, runtime header fingerprint selection from the latest stored fingerprint, and optional local Codex.app Chromium extraction. | | Runtime docs and packaging | Completed | `33b63f3` | `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test`; `cargo run`; `curl -s -i http://127.0.0.1:8080/health` | README scope, `.env.example`, Dockerfile, compose file, and health smoke test are complete. | diff --git a/docs/superpowers/plans/2026-06-11-codex-proxy-rs.md b/docs/superpowers/plans/2026-06-11-codex-proxy-rs.md index 59b09150b..ceedfe0a5 100644 --- a/docs/superpowers/plans/2026-06-11-codex-proxy-rs.md +++ b/docs/superpowers/plans/2026-06-11-codex-proxy-rs.md @@ -2963,6 +2963,7 @@ git commit -m "docs: document codex-proxy-rs scope" - [x] Add read-only `/admin/settings` for the retained Rust settings surface. The response is admin-session gated, lower camelCase, and excludes proxy URL, Ollama, third-party provider, OpenAI official API key, proxy API key, and Electron/self-update fields. Persistent settings mutation remains a separate task. - [x] Add a native static model catalog foundation. It exposes `gpt-5.5`, configured aliases, reasoning/service tier suffix parsing, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, and `/debug/models`. Backend model fetching/cache and `/admin/refresh-models` remain separate tasks. - [x] Add encrypted admin account import/list foundations. `/admin/accounts/import` stores access/refresh tokens through `SecretBox` and `/admin/accounts` returns paginated sanitized account metadata without tokens or token decryption. The runtime now loads or creates the master key file before constructing repository-backed routes. +- [x] Add the first native Codex upstream lifecycle slice. Imported accounts are synchronized into the process account pool, `/v1/responses` acquires an account, sends Codex Desktop HTTP SSE requests with encrypted Cookie replay/capture, collects `response.completed`, records usage, and releases the slot. Startup pool snapshots, streaming passthrough, retry/fallback, 401 refresh, rate-limit capture, and WebSocket remain separate tasks. ## 2026-06-11 Original Project Gap Audit @@ -2979,8 +2980,8 @@ The Rust project has no legacy compatibility requirement, but the in-scope behav - [ ] **Diagnostics:** expand health/admin diagnostics to include authenticated state, pool summary, capacity summary, transport/fingerprint status, paths, runtime metadata, and test-connection checks. Keep production-local gating for sensitive debug endpoints. - [ ] **Logs and usage stats:** add `/admin/logs/state`, clear/detail, error-log equivalents if retained, and `/admin/usage-stats/summary` plus paginated or bounded history queries. Keep log file rotation and SQLite event pagination. - [ ] **Model catalog:** replace the hard-coded `/v1/models` list with a model store that supports static defaults, configured aliases, `-low/-medium/-high/-xhigh/-fast/-flex` suffix parsing, backend model fetching per plan, cache persistence, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/admin/refresh-models`, and `/debug/models`. Static defaults, aliases, suffix parsing, read-only model routes, and debug summary are complete. Backend fetch/cache, plan snapshots, and admin refresh remain. -- [ ] **Account scheduling parity:** extend account acquisition beyond `active + least last_used`. Completed in the in-memory pool: max concurrent slots per account, stale slot cleanup, least-used/round-robin/sticky strategies, quota-cache skip, tier priority, model-plan filtering, exclude IDs, preferred account from session affinity, Cloudflare cooldown skip, request-staggering metadata, and capacity summary. Remaining: settings-backed scheduling config, pool snapshots from SQLite, release-time usage accounting through the repository, and upstream retry/fallback wiring. -- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect -> rate-limit capture -> usage release -> retry/fallback. Include Cookie replay/capture, Codex Desktop headers, `x-codex-turn-state`, `x-client-request-id`, upstream error classification, 401-triggered refresh, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, and OpenAI-compatible error mapping. +- [ ] **Account scheduling parity:** extend account acquisition beyond `active + least last_used`. Completed in the in-memory pool: max concurrent slots per account, stale slot cleanup, least-used/round-robin/sticky strategies, quota-cache skip, tier priority, model-plan filtering, exclude IDs, preferred account from session affinity, Cloudflare cooldown skip, request-staggering metadata, capacity summary, settings-backed construction, and import-to-pool sync. Remaining: startup pool snapshots from SQLite, durable release-time slot state, and upstream retry/fallback wiring. +- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect -> rate-limit capture -> usage release -> retry/fallback. Initial HTTP SSE collection is complete: `/v1/responses` acquires imported accounts, sends Codex Desktop headers, replays/captures Cookies, collects `response.completed`, records usage, and returns OpenAI-compatible JSON. Remaining: rate-limit capture, retry/fallback, 401-triggered refresh, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, streaming passthrough, and richer OpenAI-compatible error mapping. - [ ] **Session affinity and previous response:** persist an in-memory response-id to account mapping with TTL, prompt-cache/conversation identity, turn-state replay, and preferred account selection. `src/codex/websocket.rs` must either implement the WebSocket path with verified TLS/header behavior or the API must reject WebSocket-only cases explicitly with documented limitations until fingerprint parity is proven. - [ ] **Fingerprint auto-update parity:** poll the real Codex Desktop appcast/update source, parse app version/build number, persist update state, select the latest stored fingerprint for runtime headers, and optionally extract Chromium version from a local Codex.app path when configured. diff --git a/src/codex/client.rs b/src/codex/client.rs index 936474921..3ad6ab76d 100644 --- a/src/codex/client.rs +++ b/src/codex/client.rs @@ -1,4 +1,141 @@ -use reqwest::Client; +use reqwest::{ + header::{HeaderMap, HeaderName, HeaderValue, CONTENT_TYPE, COOKIE, SET_COOKIE}, + Client, StatusCode, +}; +use serde_json::Value; +use thiserror::Error; + +use crate::{ + codex::{ + headers::build_codex_headers, + sse::{parse_sse_events, SseError}, + types::CodexResponsesRequest, + usage::{extract_usage, TokenUsage}, + websocket::{ensure_http_sse_supported, WebSocketSupportError}, + }, + fingerprint::model::Fingerprint, +}; + +#[derive(Debug, Error)] +pub enum CodexClientError { + #[error("http transport error: {0}")] + Http(#[from] reqwest::Error), + #[error("invalid request header name: {0}")] + InvalidHeaderName(#[from] reqwest::header::InvalidHeaderName), + #[error("invalid request header value: {0}")] + InvalidHeaderValue(#[from] reqwest::header::InvalidHeaderValue), + #[error("unsupported transport: {0}")] + UnsupportedTransport(#[from] WebSocketSupportError), + #[error("invalid upstream SSE response: {0}")] + InvalidSse(#[from] SseError), + #[error("upstream returned status {status}: {body}")] + Upstream { status: StatusCode, body: String }, +} + +pub type CodexClientResult = Result; + +#[derive(Debug, Clone, Copy)] +pub struct CodexRequestContext<'a> { + pub access_token: &'a str, + pub account_id: Option<&'a str>, + pub request_id: &'a str, + pub turn_state: Option<&'a str>, + pub cookie_header: Option<&'a str>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CodexBackendResponse { + pub body: String, + pub usage: Option, + pub turn_state: Option, + pub set_cookie_headers: Vec, +} + +#[derive(Clone)] +pub struct CodexBackendClient { + client: Client, + base_url: String, + fingerprint: Fingerprint, +} + +impl CodexBackendClient { + pub fn new(client: Client, base_url: impl Into, fingerprint: Fingerprint) -> Self { + Self { + client, + base_url: base_url.into().trim_end_matches('/').to_string(), + fingerprint, + } + } + + pub async fn create_response( + &self, + request: &CodexResponsesRequest, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + ensure_http_sse_supported(request)?; + let url = format!("{}/codex/responses", self.base_url); + let response = self + .client + .post(url) + .headers(self.request_headers(context)?) + .json(request) + .send() + .await?; + let status = response.status(); + let turn_state = response + .headers() + .get("x-codex-turn-state") + .and_then(|value| value.to_str().ok()) + .map(ToString::to_string); + let set_cookie_headers = response + .headers() + .get_all(SET_COOKIE) + .iter() + .filter_map(|value| value.to_str().ok().map(ToString::to_string)) + .collect::>(); + let body = response.text().await?; + if !status.is_success() { + return Err(CodexClientError::Upstream { status, body }); + } + let usage = extract_sse_usage(&body)?; + + Ok(CodexBackendResponse { + body, + usage, + turn_state, + set_cookie_headers, + }) + } + + fn request_headers(&self, context: CodexRequestContext<'_>) -> CodexClientResult { + let mut headers = HeaderMap::new(); + for (name, value) in build_codex_headers( + &self.fingerprint, + context.access_token, + context.account_id, + context.turn_state, + context.request_id, + ) { + headers.insert( + HeaderName::from_bytes(name.as_bytes())?, + HeaderValue::from_str(&value)?, + ); + } + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + headers.insert( + HeaderName::from_static("openai-beta"), + HeaderValue::from_static("responses_websockets=2026-02-06"), + ); + headers.insert( + HeaderName::from_static("x-openai-internal-codex-residency"), + HeaderValue::from_static("us"), + ); + if let Some(cookie_header) = context.cookie_header { + headers.insert(COOKIE, HeaderValue::from_str(cookie_header)?); + } + Ok(headers) + } +} pub fn build_reqwest_client(force_http11: bool) -> Result { // Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。 @@ -16,3 +153,25 @@ pub fn build_reqwest_client(force_http11: bool) -> Result CodexClientResult> { + let events = parse_sse_events(body)?; + let mut usage: Option = None; + for event in events { + if event.data == "[DONE]" { + continue; + } + let Ok(value) = serde_json::from_str::(&event.data) else { + continue; + }; + let event_usage = + extract_usage(&value).or_else(|| value.get("response").and_then(extract_usage)); + if let Some(event_usage) = event_usage { + usage = Some(match usage { + Some(current) => current.merged(event_usage), + None => event_usage, + }); + } + } + Ok(usage) +} diff --git a/src/http/admin.rs b/src/http/admin.rs index 72b3677db..34e31a8e3 100644 --- a/src/http/admin.rs +++ b/src/http/admin.rs @@ -15,7 +15,7 @@ use uuid::Uuid; use crate::{ accounts::{ - model::AccountStatus, + model::{Account, AccountStatus}, repository::{NewAccount, StoredAccountMetadata}, }, auth::admin_session::verify_admin_password, @@ -541,16 +541,38 @@ pub async fn import_accounts( .into_response(); } }; + let email = empty_to_none(entry.email); + let account_id = empty_to_none(entry.account_id); + let user_id = empty_to_none(entry.user_id); + let label = empty_to_none(entry.label); + let plan_type = empty_to_none(entry.plan_type); + let refresh_token = empty_to_none(entry.refresh_token); + let access_token = token.to_string(); + let pool_account = Account { + id: id.clone(), + email: email.clone(), + account_id: account_id.clone(), + user_id: user_id.clone(), + label: label.clone(), + plan_type: plan_type.clone(), + access_token: access_token.clone(), + refresh_token: refresh_token.clone(), + access_token_expires_at: None, + status, + quota_limit_reached: false, + cloudflare_cooldown_until: None, + added_at: chrono::Utc::now().to_rfc3339(), + last_used_at: None, + }; let account = NewAccount { id, - email: empty_to_none(entry.email), - account_id: empty_to_none(entry.account_id), - user_id: empty_to_none(entry.user_id), - label: empty_to_none(entry.label), - plan_type: empty_to_none(entry.plan_type), - access_token: SecretString::new(token.to_string().into()), - refresh_token: empty_to_none(entry.refresh_token) - .map(|token| SecretString::new(token.into())), + email, + account_id, + user_id, + label, + plan_type, + access_token: SecretString::new(access_token.into()), + refresh_token: refresh_token.map(|token| SecretString::new(token.into())), access_token_expires_at: None, status, }; @@ -561,6 +583,7 @@ pub async fn import_accounts( ) .into_response(); } + state.account_pool().lock().await.insert(pool_account); imported += 1; } diff --git a/src/http/v1.rs b/src/http/v1.rs index fd03c920d..ea0ba1526 100644 --- a/src/http/v1.rs +++ b/src/http/v1.rs @@ -3,13 +3,24 @@ use axum::{ extract::{Path, State}, http::{HeaderMap, StatusCode}, response::IntoResponse, - Json, + Extension, Json, }; +use chrono::Utc; +use reqwest::Url; use serde::Deserialize; use serde_json::{json, Value}; use crate::{ - http::auth::client_api_key, models::catalog::ModelCatalog, state::AppState, + accounts::{pool::AccountAcquireRequest, repository::UsageDelta}, + codex::{ + client::{build_reqwest_client, CodexBackendClient, CodexClientError, CodexRequestContext}, + sse::parse_sse_events, + types::CodexResponsesRequest, + }, + fingerprint::model::Fingerprint, + http::{auth::client_api_key, middleware::RequestId}, + models::catalog::ModelCatalog, + state::AppState, translation::codex_to_openai::openai_error, }; @@ -18,10 +29,16 @@ const MODEL_CREATED_TIMESTAMP: i64 = 1_700_000_000; #[derive(Deserialize)] struct ResponsesBody { model: Option, + input: Option>, + instructions: Option, + reasoning: Option, + tools: Option>, + previous_response_id: Option, } pub async fn responses( State(state): State, + Extension(request_id): Extension, headers: HeaderMap, body: Bytes, ) -> impl IntoResponse { @@ -30,18 +47,106 @@ pub async fn responses( } let default_model = state.config().model.default_model.clone(); - let body = serde_json::from_slice::(&body).unwrap_or_else(|_| ResponsesBody { - model: Some(default_model.clone()), - }); - let model = body.model.unwrap_or(default_model); + let body = serde_json::from_slice::(&body) + .unwrap_or_else(|_| default_body(default_model.clone())); + let requested_model = body.model.clone().unwrap_or(default_model); let catalog = ModelCatalog::from_config(&state.config().model); - if !catalog.is_recognized_model_name(&model) { + if !catalog.is_recognized_model_name(&requested_model) { return ( StatusCode::NOT_FOUND, Json(openai_error("Model not found", "model_not_found")), ); } + let parsed_model = catalog.parse_model_name(&requested_model); + let codex_request = CodexResponsesRequest { + model: parsed_model.model_id.clone(), + instructions: body.instructions.unwrap_or_default(), + input: body.input.unwrap_or_default(), + stream: true, + store: false, + reasoning: body.reasoning, + tools: body.tools, + previous_response_id: body.previous_response_id, + use_websocket: false, + }; + let acquired = { + state + .account_pool() + .lock() + .await + .acquire_with(AccountAcquireRequest::new(&codex_request.model, Utc::now())) + }; + let Some(acquired) = acquired else { + return no_available_accounts_response(); + }; + let account = acquired.account; + let response = send_codex_request(&state, &codex_request, &account, request_id.as_str()).await; + state.account_pool().lock().await.release(&account.id); + let response = match response { + Ok(response) => response, + Err(error) => return codex_client_error_response(error), + }; + if let Some(cookie_repo) = state.cookie_repository() { + for cookie in &response.set_cookie_headers { + if cookie_repo + .capture_set_cookie(&account.id, cookie) + .await + .is_err() + { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(openai_error( + "Failed to persist upstream cookies", + "cookie_store_error", + )), + ); + } + } + } + if let (Some(repo), Some(usage)) = (state.account_repository(), response.usage) { + if repo + .record_usage( + &account.id, + UsageDelta { + input_tokens: u64_to_i64_saturating(usage.input_tokens), + output_tokens: u64_to_i64_saturating(usage.output_tokens), + cached_tokens: u64_to_i64_saturating(usage.cached_tokens), + }, + ) + .await + .is_err() + { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(openai_error( + "Failed to record account usage", + "usage_store_error", + )), + ); + } + } + + match completed_response_json(&response.body) { + Ok(Some(body)) => (StatusCode::OK, Json(body)), + Ok(None) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Codex response did not include response.completed", + "empty_upstream_response", + )), + ), + Err(_) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Invalid Codex SSE response", + "invalid_upstream_sse", + )), + ), + } +} + +fn no_available_accounts_response() -> (StatusCode, Json) { ( StatusCode::SERVICE_UNAVAILABLE, Json(openai_error( @@ -51,6 +156,102 @@ pub async fn responses( ) } +fn default_body(default_model: String) -> ResponsesBody { + ResponsesBody { + model: Some(default_model), + input: Some(Vec::new()), + instructions: Some(String::new()), + reasoning: None, + tools: None, + previous_response_id: None, + } +} + +async fn send_codex_request( + state: &AppState, + request: &CodexResponsesRequest, + account: &crate::accounts::model::Account, + request_id: &str, +) -> Result { + let request_domain = request_domain(&state.config().api.base_url); + let cookie_header = match (state.cookie_repository(), request_domain.as_deref()) { + (Some(repo), Some(domain)) => repo.cookie_header(&account.id, domain).await.ok().flatten(), + _ => None, + }; + let client = CodexBackendClient::new( + build_reqwest_client(state.config().tls.force_http11)?, + state.config().api.base_url.clone(), + Fingerprint::default_codex_desktop(), + ); + client + .create_response( + request, + CodexRequestContext { + access_token: &account.access_token, + account_id: account.account_id.as_deref(), + request_id, + turn_state: None, + cookie_header: cookie_header.as_deref(), + }, + ) + .await +} + +fn completed_response_json(body: &str) -> Result, crate::codex::sse::SseError> { + let events = parse_sse_events(body)?; + for event in events { + if event.event.as_deref() != Some("response.completed") { + continue; + } + let Ok(value) = serde_json::from_str::(&event.data) else { + continue; + }; + if let Some(response) = value.get("response") { + return Ok(Some(response.clone())); + } + } + Ok(None) +} + +fn codex_client_error_response(error: CodexClientError) -> (StatusCode, Json) { + match error { + CodexClientError::UnsupportedTransport(_) => ( + StatusCode::BAD_REQUEST, + Json(openai_error( + "previous_response_id requires Codex WebSocket transport", + "websocket_required", + )), + ), + CodexClientError::Upstream { status, body } => ( + status, + Json(openai_error( + &format!( + "Codex upstream error: {}", + body.chars().take(300).collect::() + ), + "upstream_error", + )), + ), + _ => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Codex upstream request failed", + "upstream_error", + )), + ), + } +} + +fn request_domain(base_url: &str) -> Option { + Url::parse(base_url) + .ok() + .and_then(|url| url.host_str().map(ToString::to_string)) +} + +fn u64_to_i64_saturating(value: u64) -> i64 { + value.min(i64::MAX as u64) as i64 +} + pub async fn models(State(state): State, headers: HeaderMap) -> impl IntoResponse { if !has_client_api_key(&headers) { return missing_client_api_key_response(); diff --git a/src/state.rs b/src/state.rs index 10eb2fa58..fcd5a7bbd 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,9 +1,14 @@ use std::sync::Arc; use sqlx::SqlitePool; +use tokio::sync::Mutex; -use crate::accounts::repository::AccountRepository; +use crate::accounts::{ + pool::{AccountPool, AccountPoolOptions, RotationStrategy}, + repository::AccountRepository, +}; use crate::config::AppConfig; +use crate::cookies::repository::CookieRepository; use crate::crypto::SecretBox; use crate::logs::repository::EventLogRepository; @@ -17,27 +22,32 @@ pub struct AppServices { pub db: Option, pub event_logs: Option, pub secret_box: Option, + pub account_pool: Arc>, } impl AppState { pub fn new(config: AppConfig) -> Self { + let account_pool = account_pool_from_config(&config); Self { services: Arc::new(AppServices { config, db: None, event_logs: None, secret_box: None, + account_pool, }), } } pub fn with_pool(config: AppConfig, pool: SqlitePool) -> Self { + let account_pool = account_pool_from_config(&config); Self { services: Arc::new(AppServices { config, db: Some(pool.clone()), event_logs: Some(EventLogRepository::new(pool)), secret_box: None, + account_pool, }), } } @@ -47,12 +57,14 @@ impl AppState { pool: SqlitePool, secret_box: SecretBox, ) -> Self { + let account_pool = account_pool_from_config(&config); Self { services: Arc::new(AppServices { config, db: Some(pool.clone()), event_logs: Some(EventLogRepository::new(pool)), secret_box: Some(secret_box), + account_pool, }), } } @@ -75,4 +87,30 @@ impl AppState { self.services.secret_box.clone()?, )) } + + pub fn cookie_repository(&self) -> Option { + Some(CookieRepository::new( + self.db()?.clone(), + self.services.secret_box.clone()?, + )) + } + + pub fn account_pool(&self) -> Arc> { + self.services.account_pool.clone() + } +} + +fn account_pool_from_config(config: &AppConfig) -> Arc> { + Arc::new(Mutex::new(AccountPool::with_options(AccountPoolOptions { + max_concurrent_per_account: config.auth.max_concurrent_per_account, + stale_slot_ttl: chrono::Duration::minutes(5), + rotation_strategy: match config.auth.rotation_strategy.as_str() { + "round_robin" => RotationStrategy::RoundRobin, + "sticky" => RotationStrategy::Sticky, + _ => RotationStrategy::LeastUsed, + }, + skip_quota_limited: config.quota.skip_exhausted, + tier_priority: config.auth.tier_priority.clone(), + model_plan_allowlist: std::collections::BTreeMap::new(), + }))) } diff --git a/tests/codex_client_test.rs b/tests/codex_client_test.rs new file mode 100644 index 000000000..cf5f73107 --- /dev/null +++ b/tests/codex_client_test.rs @@ -0,0 +1,95 @@ +use serde_json::json; +use wiremock::{ + matchers::{body_json, header, method, path}, + Mock, MockServer, ResponseTemplate, +}; + +use codex_proxy_rs::{ + codex::{ + client::{build_reqwest_client, CodexBackendClient, CodexRequestContext}, + types::CodexResponsesRequest, + usage::TokenUsage, + }, + fingerprint::model::Fingerprint, +}; + +#[tokio::test] +async fn codex_backend_client_should_send_desktop_headers_and_capture_response_metadata() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_1\",\"usage\":{\"input_tokens\":2,\"output_tokens\":3,\"input_tokens_details\":{\"cached_tokens\":1}}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-token")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(header("originator", "Codex Desktop")) + .and(header("x-client-request-id", "req_1")) + .and(header("x-codex-turn-state", "turn_1")) + .and(header("cookie", "cf_clearance=old")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "", + "input": [], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .insert_header( + "set-cookie", + "cf_clearance=new; Domain=.chatgpt.com; Path=/", + ) + .insert_header("x-codex-turn-state", "turn_2") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let client = CodexBackendClient::new( + build_reqwest_client(false).unwrap(), + server.uri(), + Fingerprint::default_for_tests(), + ); + + let response = client + .create_response( + &CodexResponsesRequest { + model: "gpt-5.5".to_string(), + instructions: String::new(), + input: Vec::new(), + stream: true, + store: false, + reasoning: None, + tools: None, + previous_response_id: None, + use_websocket: false, + }, + CodexRequestContext { + access_token: "access-token", + account_id: Some("chatgpt-account"), + request_id: "req_1", + turn_state: Some("turn_1"), + cookie_header: Some("cf_clearance=old"), + }, + ) + .await + .unwrap(); + + assert_eq!( + response.usage, + Some(TokenUsage { + input_tokens: 2, + output_tokens: 3, + cached_tokens: 1, + total_tokens: 5, + }) + ); + assert_eq!(response.turn_state.as_deref(), Some("turn_2")); + assert_eq!( + response.set_cookie_headers, + vec!["cf_clearance=new; Domain=.chatgpt.com; Path=/".to_string()] + ); +} diff --git a/tests/v1_upstream_route_test.rs b/tests/v1_upstream_route_test.rs new file mode 100644 index 000000000..0def4ccc8 --- /dev/null +++ b/tests/v1_upstream_route_test.rs @@ -0,0 +1,210 @@ +use std::collections::BTreeMap; + +use axum::{ + body::{to_bytes, Body}, + http::{Request, StatusCode}, +}; +use chrono::Utc; +use serde_json::{json, Value}; +use tower::ServiceExt; +use wiremock::{ + matchers::{body_json, header, method, path}, + Mock, MockServer, ResponseTemplate, +}; + +use codex_proxy_rs::{ + app::build_router, + config::{ + AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, LoggingConfig, ModelConfig, + QuotaConfig, QuotaWarningThresholds, SecurityConfig, ServerConfig, TlsConfig, + UsageStatsConfig, + }, + cookies::repository::CookieRepository, + crypto::SecretBox, + state::AppState, + storage::db::connect_sqlite, +}; + +fn test_config(database_url: String, base_url: String) -> AppConfig { + AppConfig { + server: ServerConfig { + host: "127.0.0.1".to_string(), + port: 0, + }, + api: ApiConfig { base_url }, + model: ModelConfig { + default_model: "gpt-5.5".to_string(), + default_reasoning_effort: None, + service_tier: None, + aliases: BTreeMap::new(), + }, + auth: AuthConfig { + refresh_margin_seconds: 300, + refresh_enabled: true, + refresh_concurrency: 2, + max_concurrent_per_account: 3, + request_interval_ms: 50, + rotation_strategy: "least_used".to_string(), + tier_priority: Vec::new(), + }, + quota: QuotaConfig { + refresh_interval_minutes: 5, + warning_thresholds: QuotaWarningThresholds { + primary: vec![80, 90], + secondary: vec![80, 90], + }, + skip_exhausted: true, + }, + usage_stats: UsageStatsConfig { + history_retention_days: None, + }, + database: DatabaseConfig { url: database_url }, + security: SecurityConfig { + master_key_file: "data/master.key".to_string(), + api_key_pepper_file: "data/api-key-pepper.key".to_string(), + }, + tls: TlsConfig { + force_http11: false, + }, + admin: AdminConfig { + session_ttl_minutes: 1440, + }, + logging: LoggingConfig { + directory: "logs".to_string(), + max_file_bytes: 10_485_760, + retention_days: 14, + enabled: false, + capacity: 2_000, + capture_body: false, + }, + } +} + +#[tokio::test] +async fn v1_responses_should_use_imported_account_and_record_usage() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"usage\":{\"input_tokens\":7,\"output_tokens\":4,\"input_tokens_details\":{\"cached_tokens\":2}}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "", + "input": [], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .insert_header( + "set-cookie", + "cf_clearance=new; Domain=.chatgpt.com; Path=/", + ) + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("v1-upstream.sqlite"); + let url = format!("sqlite://{}", db.display()); + let pool = connect_sqlite(&url).await.unwrap(); + seed_admin_session(&pool, "session_1").await; + let secret_box = SecretBox::new([21u8; 32]); + let app = build_router(AppState::with_pool_and_secret_box( + test_config(url, server.uri()), + pool.clone(), + secret_box.clone(), + )); + + let import_response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/admin/accounts/import") + .header("content-type", "application/json") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::from( + json!({ + "accounts": [{ + "id": "acct_imported", + "accountId": "chatgpt-account", + "token": "access-secret", + "refreshToken": "refresh-secret", + "status": "active" + }] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(import_response.status(), StatusCode::OK); + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header("authorization", "Bearer cpr_test") + .header("content-type", "application/json") + .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["id"], "resp_1"); + assert_eq!(body["usage"]["input_tokens"], 7); + let usage: (i64, i64, i64, i64) = sqlx::query_as( + "select request_count, input_tokens, output_tokens, cached_tokens from account_usage where account_id = ?", + ) + .bind("acct_imported") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(usage, (1, 7, 4, 2)); + let cookie_header = CookieRepository::new(pool, secret_box) + .cookie_header("acct_imported", "chatgpt.com") + .await + .unwrap(); + assert_eq!(cookie_header.as_deref(), Some("cf_clearance=new")); +} + +async fn seed_admin_session(pool: &sqlx::SqlitePool, session_id: &str) { + let now = Utc::now().to_rfc3339(); + sqlx::query( + "insert into admin_users (id, password_hash, created_at, updated_at) values (?, ?, ?, ?)", + ) + .bind("admin_1") + .bind("hash") + .bind(&now) + .bind(&now) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "insert into admin_sessions (id, user_id, expires_at, created_at) values (?, ?, ?, ?)", + ) + .bind(session_id) + .bind("admin_1") + .bind("2999-01-01T00:00:00Z") + .bind(now) + .execute(pool) + .await + .unwrap(); +} + +async fn response_json(response: axum::response::Response) -> Value { + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + serde_json::from_slice(&bytes).unwrap() +}