diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index bfb28a227d..728e815ccf 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -1,7 +1,9 @@ use anyhow::Context; use anyhow::Result; +use app_test_support::ChatGptIdTokenClaims; use app_test_support::McpProcess; use app_test_support::create_mock_responses_server_sequence_unchecked; +use app_test_support::encode_id_token; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; @@ -35,6 +37,10 @@ use std::path::Path; use std::time::Duration; use tempfile::TempDir; use tokio::time::timeout; +use wiremock::Mock; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10); const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex."; @@ -313,6 +319,100 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> { Ok(()) } +#[tokio::test] +async fn realtime_conversation_uses_client_secret_with_external_chatgpt_auth() -> Result<()> { + skip_if_no_network!(Ok(())); + + let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + let chatgpt_server = wiremock::MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/realtime/client_secrets")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "value": "ek-app-server" + }))) + .mount(&chatgpt_server) + .await; + let realtime_server = start_websocket_server(vec![vec![vec![json!({ + "type": "session.updated", + "session": { "id": "sess_external", "instructions": "backend prompt" } + })]]]) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &responses_server.uri(), + realtime_server.uri(), + true, + )?; + std::fs::write( + codex_home.path().join("config.toml"), + format!( + "{}chatgpt_base_url = \"{}\"\n", + std::fs::read_to_string(codex_home.path().join("config.toml"))?, + chatgpt_server.uri(), + ), + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + mcp.initialize().await?; + login_with_chatgpt_auth_tokens(&mut mcp).await?; + + let thread_start_request_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_request_id)), + ) + .await??; + let thread_start: ThreadStartResponse = to_response(thread_start_response)?; + + let start_request_id = mcp + .send_thread_realtime_start_request(ThreadRealtimeStartParams { + thread_id: thread_start.thread.id.clone(), + prompt: "backend prompt".to_string(), + session_id: None, + }) + .await?; + let start_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_request_id)), + ) + .await??; + let _: ThreadRealtimeStartResponse = to_response(start_response)?; + + let started = + read_notification::(&mut mcp, "thread/realtime/started") + .await?; + assert_eq!(started.thread_id, thread_start.thread.id); + assert_eq!(started.version, RealtimeConversationVersion::V2); + + let requests = chatgpt_server.received_requests().await.unwrap_or_default(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].url.path(), "/codex/realtime/client_secrets"); + assert_eq!( + requests[0] + .headers + .get("chatgpt-account-id") + .and_then(|value| value.to_str().ok()), + Some("org-siwc") + ); + let request_body: serde_json::Value = serde_json::from_slice(&requests[0].body)?; + assert_eq!(request_body["session"]["type"], json!("realtime")); + + assert_eq!( + realtime_server.handshakes()[0] + .header("authorization") + .as_deref(), + Some("Bearer ek-app-server") + ); + + realtime_server.shutdown().await; + chatgpt_server.shutdown().await; + Ok(()) +} + #[tokio::test] async fn realtime_conversation_requires_feature_flag() -> Result<()> { skip_if_no_network!(Ok(())); @@ -390,6 +490,35 @@ async fn login_with_api_key(mcp: &mut McpProcess, api_key: &str) -> Result<()> { Ok(()) } +async fn login_with_chatgpt_auth_tokens(mcp: &mut McpProcess) -> Result<()> { + let access_token = encode_id_token( + &ChatGptIdTokenClaims::new() + .email("siwc@example.com") + .plan_type("business") + .chatgpt_account_id("org-siwc"), + )?; + let request_id = mcp + .send_chatgpt_auth_tokens_login_request( + access_token, + "org-siwc".to_string(), + Some("business".to_string()), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let login: LoginAccountResponse = to_response(response)?; + assert_eq!(login, LoginAccountResponse::ChatgptAuthTokens {}); + let _updated = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_notification_message("account/updated"), + ) + .await??; + Ok(()) +} + fn create_config_toml( codex_home: &Path, responses_server_uri: &str, diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 10c72d72be..bd670bd5fa 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -20,6 +20,8 @@ use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; use http::HeaderValue; +use serde_json::Value; +use serde_json::json; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; @@ -509,6 +511,27 @@ impl RealtimeWebsocketClient { } } +pub fn realtime_client_secret_request_body( + config: &RealtimeSessionConfig, +) -> Result { + let session_mode = normalized_session_mode(config.event_parser, config.session_mode); + let mut session = serde_json::to_value(session_update_session( + config.event_parser, + config.instructions.clone(), + session_mode, + )) + .map_err(|err| ApiError::Stream(format!("failed to encode realtime session config: {err}")))?; + if let Some(model) = config.model.as_ref() + && let Some(session_object) = session.as_object_mut() + { + session_object.insert("model".to_string(), Value::String(model.clone())); + } + + Ok(json!({ + "session": session, + })) +} + fn merge_request_headers( provider_headers: &HeaderMap, extra_headers: HeaderMap, diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs index d13585034a..1217397268 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/mod.rs @@ -13,6 +13,7 @@ pub use methods::RealtimeWebsocketClient; pub use methods::RealtimeWebsocketConnection; pub use methods::RealtimeWebsocketEvents; pub use methods::RealtimeWebsocketWriter; +pub use methods::realtime_client_secret_request_body; pub use protocol::RealtimeEventParser; pub use protocol::RealtimeSessionConfig; pub use protocol::RealtimeSessionMode; diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 6d519f488f..a072554395 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -15,6 +15,7 @@ mod auth_env_telemetry; mod client; mod client_common; pub mod codex; +mod realtime_client_secrets; mod realtime_context; mod realtime_conversation; pub use codex::SteerInputError; diff --git a/codex-rs/core/src/realtime_client_secrets.rs b/codex-rs/core/src/realtime_client_secrets.rs new file mode 100644 index 0000000000..fa32827c20 --- /dev/null +++ b/codex-rs/core/src/realtime_client_secrets.rs @@ -0,0 +1,69 @@ +use crate::CodexAuth; +use crate::config::Config; +use crate::default_client::create_client; +use crate::error::CodexErr; +use crate::error::Result as CodexResult; +use codex_api::RealtimeSessionConfig; +use codex_api::realtime_client_secret_request_body; +use serde::Deserialize; + +#[derive(Deserialize)] +struct RealtimeClientSecretResponse { + value: String, +} + +pub(crate) async fn fetch_realtime_client_secret( + auth: &CodexAuth, + config: &Config, + session_config: &RealtimeSessionConfig, +) -> CodexResult { + let bearer_token = auth.get_token().map_err(|err| { + CodexErr::InvalidRequest(format!("failed to read ChatGPT auth token: {err}")) + })?; + let body = realtime_client_secret_request_body(session_config) + .map_err(|err| CodexErr::InvalidRequest(err.to_string()))?; + let endpoint = format!( + "{}/codex/realtime/client_secrets", + normalized_chatgpt_base_url(&config.chatgpt_base_url) + ); + + let mut request = create_client().post(&endpoint).bearer_auth(bearer_token); + if let Some(account_id) = auth.get_account_id() { + request = request.header("ChatGPT-Account-Id", account_id); + } + + let response = request.json(&body).send().await.map_err(|err| { + CodexErr::InvalidRequest(format!("failed to request realtime client secret: {err}")) + })?; + + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + return Err(CodexErr::InvalidRequest(format!( + "failed to request realtime client secret: {status} {body}" + ))); + } + + let payload: RealtimeClientSecretResponse = response.json().await.map_err(|err| { + CodexErr::InvalidRequest(format!( + "failed to parse realtime client secret response: {err}" + )) + })?; + if payload.value.trim().is_empty() { + return Err(CodexErr::InvalidRequest( + "realtime client secret response was missing a value".to_string(), + )); + } + Ok(payload.value) +} + +fn normalized_chatgpt_base_url(input: &str) -> String { + let mut base_url = input.trim_end_matches('/').to_string(); + if (base_url.starts_with("https://chatgpt.com") + || base_url.starts_with("https://chat.openai.com")) + && !base_url.contains("/backend-api") + { + base_url = format!("{base_url}/backend-api"); + } + base_url +} diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 1ddd72d0fd..61c143184b 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -1,12 +1,12 @@ use crate::CodexAuth; use crate::api_bridge::map_api_error; -use crate::auth::read_openai_api_key_from_env; use crate::codex::Session; use crate::config::RealtimeWsMode; use crate::config::RealtimeWsVersion; use crate::default_client::default_headers; use crate::error::CodexErr; use crate::error::Result as CodexResult; +use crate::realtime_client_secrets::fetch_realtime_client_secret; use crate::realtime_context::build_realtime_startup_context; use async_channel::Receiver; use async_channel::Sender; @@ -453,7 +453,6 @@ async fn prepare_realtime_start( ) -> CodexResult { let provider = sess.provider().await; let auth = sess.services.auth_manager.auth().await; - let realtime_api_key = realtime_api_key(auth.as_ref(), &provider)?; let mut api_provider = provider.to_api_provider(Some(crate::auth::AuthMode::ApiKey))?; let config = sess.get_config().await; if let Some(realtime_ws_base_url) = &config.experimental_realtime_ws_base_url { @@ -494,8 +493,12 @@ async fn prepare_realtime_start( event_parser, session_mode, }; - let extra_headers = - realtime_request_headers(requested_session_id.as_deref(), realtime_api_key.as_str())?; + let realtime_bearer_token = + realtime_bearer_token(auth.as_ref(), &provider, &config, &session_config).await?; + let extra_headers = realtime_request_headers( + requested_session_id.as_deref(), + realtime_bearer_token.as_str(), + )?; Ok(PreparedRealtimeConversationStart { api_provider, extra_headers, @@ -625,9 +628,11 @@ fn realtime_text_from_handoff_request(handoff: &RealtimeHandoffRequested) -> Opt .or((!handoff.input_transcript.is_empty()).then_some(handoff.input_transcript.clone())) } -fn realtime_api_key( +async fn realtime_bearer_token( auth: Option<&CodexAuth>, provider: &crate::ModelProviderInfo, + config: &crate::config::Config, + session_config: &RealtimeSessionConfig, ) -> CodexResult { if let Some(api_key) = provider.api_key()? { return Ok(api_key); @@ -637,26 +642,23 @@ fn realtime_api_key( return Ok(token); } - if let Some(api_key) = auth.and_then(CodexAuth::api_key) { - return Ok(api_key.to_string()); - } - - // TODO(aibrahim): Remove this temporary fallback once realtime auth no longer - // requires API key auth for ChatGPT/SIWC sessions. - if provider.is_openai() - && let Some(api_key) = read_openai_api_key_from_env() - { - return Ok(api_key); + if let Some(auth) = auth { + if auth.is_chatgpt_auth() { + return fetch_realtime_client_secret(auth, config, session_config).await; + } + if let Some(api_key) = auth.api_key() { + return Ok(api_key.to_string()); + } } Err(CodexErr::InvalidRequest( - "realtime conversation requires API key auth".to_string(), + "realtime conversation requires API key or ChatGPT auth".to_string(), )) } fn realtime_request_headers( session_id: Option<&str>, - api_key: &str, + bearer_token: &str, ) -> CodexResult> { let mut headers = HeaderMap::new(); @@ -666,8 +668,8 @@ fn realtime_request_headers( headers.insert("x-session-id", session_id); } - let auth_value = HeaderValue::from_str(&format!("Bearer {api_key}")).map_err(|err| { - CodexErr::InvalidRequest(format!("invalid realtime api key header: {err}")) + let auth_value = HeaderValue::from_str(&format!("Bearer {bearer_token}")).map_err(|err| { + CodexErr::InvalidRequest(format!("invalid realtime bearer token header: {err}")) })?; headers.insert(AUTHORIZATION, auth_value); diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 06cb31630a..07f49e68da 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -2,7 +2,6 @@ use anyhow::Context; use anyhow::Result; use chrono::Utc; use codex_core::CodexAuth; -use codex_core::auth::OPENAI_API_KEY_ENV_VAR; use codex_protocol::ThreadId; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ConversationAudioParams; @@ -31,16 +30,17 @@ use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use std::fs; -use std::process::Command; use std::time::Duration; use tokio::sync::oneshot; use tokio::time::timeout; +use wiremock::Mock; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex."; const MEMORY_PROMPT_PHRASE: &str = "You have access to a memory folder with guidance from prior runs."; -const REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR: &str = - "CODEX_REALTIME_CONVERSATION_TEST_SUBPROCESS"; fn websocket_request_text( request: &core_test_support::responses::WebSocketRequest, ) -> Option { @@ -85,32 +85,6 @@ where } } -fn run_realtime_conversation_test_in_subprocess( - test_name: &str, - openai_api_key: Option<&str>, -) -> Result<()> { - let mut command = Command::new(std::env::current_exe()?); - command - .arg("--exact") - .arg(test_name) - .env(REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR, "1"); - match openai_api_key { - Some(openai_api_key) => { - command.env(OPENAI_API_KEY_ENV_VAR, openai_api_key); - } - None => { - command.env_remove(OPENAI_API_KEY_ENV_VAR); - } - } - let output = command.output()?; - assert!( - output.status.success(), - "subprocess test `{test_name}` failed\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr), - ); - Ok(()) -} async fn seed_recent_thread( test: &TestCodex, title: &str, @@ -289,17 +263,19 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> { - if std::env::var_os(REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR).is_none() { - return run_realtime_conversation_test_in_subprocess( - "suite::realtime_conversation::conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth", - Some("env-realtime-key"), - ); - } - +async fn conversation_start_mints_client_secret_with_chatgpt_auth() -> Result<()> { skip_if_no_network!(Ok(())); - let server = start_websocket_server(vec![ + let chatgpt_server = start_mock_server().await; + Mock::given(method("POST")) + .and(path("/codex/realtime/client_secrets")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "value": "ek-test-secret" + }))) + .mount(&chatgpt_server) + .await; + + let realtime_server = start_websocket_server(vec![ vec![], vec![vec![json!({ "type": "session.updated", @@ -308,9 +284,22 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> ]) .await; - let mut builder = test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()); - let test = builder.build_with_websocket_server(&server).await?; - assert!(server.wait_for_handshakes(1, Duration::from_secs(2)).await); + let mut builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config({ + let chatgpt_base_url = chatgpt_server.uri(); + move |config| { + config.chatgpt_base_url = chatgpt_base_url; + } + }); + let test = builder + .build_with_websocket_server(&realtime_server) + .await?; + assert!( + realtime_server + .wait_for_handshakes(1, Duration::from_secs(2)) + .await + ); test.codex .submit(Op::RealtimeConversationStart(ConversationStartParams { @@ -338,9 +327,35 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> assert_eq!(session_updated, "sess_env"); assert_eq!( - server.handshakes()[1].header("authorization").as_deref(), - Some("Bearer env-realtime-key") + realtime_server.handshakes()[1] + .header("authorization") + .as_deref(), + Some("Bearer ek-test-secret") ); + let requests = chatgpt_server.received_requests().await.unwrap_or_default(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method.as_str(), "POST"); + assert_eq!(requests[0].url.path(), "/codex/realtime/client_secrets"); + assert_eq!( + requests[0] + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer Access Token") + ); + assert_eq!( + requests[0] + .headers + .get("chatgpt-account-id") + .and_then(|value| value.to_str().ok()), + Some("account_id") + ); + let request_body: Value = serde_json::from_slice(&requests[0].body)?; + assert_eq!( + request_body["session"]["model"], + json!("realtime-test-model") + ); + assert_eq!(request_body["session"]["type"], json!("quicksilver")); test.codex.submit(Op::RealtimeConversationClose).await?; let _closed = wait_for_event_match(&test.codex, |msg| match msg { @@ -349,7 +364,8 @@ async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> }) .await; - server.shutdown().await; + realtime_server.shutdown().await; + chatgpt_server.shutdown().await; Ok(()) } @@ -437,13 +453,6 @@ async fn conversation_audio_before_start_emits_error() -> Result<()> { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Result<()> { - if std::env::var_os(REALTIME_CONVERSATION_TEST_SUBPROCESS_ENV_VAR).is_none() { - return run_realtime_conversation_test_in_subprocess( - "suite::realtime_conversation::conversation_start_preflight_failure_emits_realtime_error_only", - /*openai_api_key*/ None, - ); - } - skip_if_no_network!(Ok(())); let server = start_websocket_server(vec![]).await; @@ -464,7 +473,10 @@ async fn conversation_start_preflight_failure_emits_realtime_error_only() -> Res _ => None, }) .await; - assert_eq!(err, "realtime conversation requires API key auth"); + assert_eq!( + err, + "realtime conversation requires API key or ChatGPT auth" + ); let closed = timeout(Duration::from_millis(200), async { wait_for_event_match(&test.codex, |msg| match msg {