mirror of
https://github.com/openai/codex.git
synced 2026-10-19 17:28:46 +00:00
Use server authentication for explicit remote session commands (#49624)
## Why Session commands with `--remote` should use the selected server's authentication and execution environment, even when the caller has workload identity configured and cannot read its identity token. ## What changed Connect `queue`, `archive`, `unarchive`, and `delete` directly to the explicit remote app server after validating configuration. Skip caller credential and runtime initialization, use remote thread parameters, and preserve the caller's working-directory override. ## Testing Add end-to-end coverage with an unreadable caller identity token to verify queued turns use the server's API key and remote archive, unarchive, and deletion operate on the selected server. Add configuration-validation coverage for all four commands and `--no-daemon` rejection coverage for remote unarchive and delete. GitOrigin-RevId: 5da7abbb40506cd754a8863e1a9d8fad9b1f09a9
This commit is contained in:
cooper-oai
+290
-2
@@ -22,6 +22,16 @@ enum QueueResponse {
|
||||
UnknownMethodVariant,
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn socket_test_home() -> Result<TempDir> {
|
||||
// macOS temporary paths can exceed the Unix socket path limit.
|
||||
#[cfg(target_os = "macos")]
|
||||
let home = tempfile::tempdir_in("/tmp")?;
|
||||
#[cfg(not(target_os = "macos"))]
|
||||
let home = TempDir::new()?;
|
||||
Ok(home)
|
||||
}
|
||||
|
||||
async fn respond_to_queue_request<S>(
|
||||
stream: S,
|
||||
codex_home: &Path,
|
||||
@@ -188,10 +198,288 @@ async fn run_remote_queue_command(response: QueueResponse) -> Result<(Output, Va
|
||||
Ok((output, server.await??))
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn remote_session_commands_with_workload_identity_use_server_auth() -> Result<()> {
|
||||
use std::time::Duration;
|
||||
|
||||
use app_test_support::MockResponsesConfig;
|
||||
use app_test_support::TestAppServer;
|
||||
use app_test_support::create_final_assistant_message_sse_response;
|
||||
use app_test_support::create_mock_responses_server_sequence_unchecked;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use codex_app_server_client::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY;
|
||||
use codex_app_server_client::RemoteAppServerClient;
|
||||
use codex_app_server_client::RemoteAppServerConnectArgs;
|
||||
use codex_app_server_client::RemoteAppServerEndpoint;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ThreadListParams;
|
||||
use codex_app_server_protocol::ThreadListResponse;
|
||||
use codex_app_server_protocol::ThreadQueueListParams;
|
||||
use codex_app_server_protocol::ThreadQueueListResponse;
|
||||
use codex_app_server_protocol::ThreadSetNameParams;
|
||||
use codex_app_server_protocol::ThreadSetNameResponse;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::TurnStatus;
|
||||
use codex_protocol::shell_environment::OPENAI_FEDERATION_RULE_ID_ENV_VAR;
|
||||
use codex_protocol::shell_environment::OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR;
|
||||
use tokio::process::Command;
|
||||
use tokio::time::timeout;
|
||||
|
||||
let codex_home = socket_test_home()?;
|
||||
let codex = codex_utils_cargo_bin::cargo_bin("codex")?;
|
||||
let model = create_mock_responses_server_sequence_unchecked(vec![
|
||||
create_final_assistant_message_sse_response("queued message consumed")?,
|
||||
])
|
||||
.await;
|
||||
MockResponsesConfig::new(&model.uri())
|
||||
.with_root_config("features.plugins = false\nanalytics.enabled = false")
|
||||
.with_provider_config("env_key = \"CODEX_QUEUE_SERVER_API_KEY\"")
|
||||
.write(codex_home.path())?;
|
||||
let _server = TestAppServer::builder()
|
||||
.with_program(&codex)
|
||||
.with_codex_home(codex_home.path())
|
||||
.with_plugin_startup_tasks()
|
||||
.without_managed_config()
|
||||
.with_args(&["app-server", "--listen", "unix://"])
|
||||
.with_env_overrides(&[
|
||||
("OPENAI_API_KEY", None),
|
||||
("CODEX_API_KEY", None),
|
||||
("CODEX_ACCESS_TOKEN", None),
|
||||
(OPENAI_FEDERATION_RULE_ID_ENV_VAR, None),
|
||||
(OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR, None),
|
||||
("CODEX_QUEUE_SERVER_API_KEY", Some("server-only-test-key")),
|
||||
])
|
||||
.build()
|
||||
.await?;
|
||||
let socket_path = codex_app_server::app_server_control_socket_path(codex_home.path())?;
|
||||
timeout(Duration::from_secs(30), async {
|
||||
while !socket_path.as_path().try_exists()? {
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
Ok::<_, anyhow::Error>(())
|
||||
})
|
||||
.await??;
|
||||
|
||||
let endpoint = format!("unix://{}", socket_path.as_path().display());
|
||||
let mut app = RemoteAppServerClient::connect(RemoteAppServerConnectArgs {
|
||||
endpoint: RemoteAppServerEndpoint::UnixSocket { socket_path },
|
||||
client_name: "queue-e2e-test".to_string(),
|
||||
client_version: "0.1.0".to_string(),
|
||||
experimental_api: true,
|
||||
mcp_server_openai_form_elicitation: false,
|
||||
opt_out_notification_methods: Vec::new(),
|
||||
channel_capacity: DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
|
||||
})
|
||||
.await?;
|
||||
let started: ThreadStartResponse = app
|
||||
.request_typed(ClientRequest::ThreadStart {
|
||||
request_id: RequestId::Integer(1),
|
||||
params: ThreadStartParams::default(),
|
||||
})
|
||||
.await?;
|
||||
let thread_id = started.thread.id;
|
||||
let message = "synthetic WIF queue end-to-end message";
|
||||
let output = timeout(
|
||||
Duration::from_secs(30),
|
||||
Command::new(&codex)
|
||||
.env("CODEX_HOME", codex_home.path())
|
||||
.env(OPENAI_FEDERATION_RULE_ID_ENV_VAR, "rule-test")
|
||||
.env(
|
||||
OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR,
|
||||
codex_home.path().join("missing-identity-token"),
|
||||
)
|
||||
.env_remove("CODEX_QUEUE_SERVER_API_KEY")
|
||||
.kill_on_drop(true)
|
||||
.args(["queue", "--remote", "unix://"])
|
||||
.args(["--thread", &thread_id, "--message", message])
|
||||
.output(),
|
||||
)
|
||||
.await??;
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
let completed = timeout(Duration::from_secs(30), async {
|
||||
while let Some(event) = app.next_event().await {
|
||||
if let AppServerEvent::ServerNotification(notification) = event
|
||||
&& let ServerNotification::TurnCompleted(completed) = *notification
|
||||
{
|
||||
return Ok::<_, anyhow::Error>(completed);
|
||||
}
|
||||
}
|
||||
anyhow::bail!("app server disconnected before the queued turn completed")
|
||||
})
|
||||
.await??;
|
||||
assert_eq!(completed.thread_id, thread_id);
|
||||
assert_eq!(completed.turn.status, TurnStatus::Completed);
|
||||
let queue: ThreadQueueListResponse = app
|
||||
.request_typed(ClientRequest::ThreadQueueList {
|
||||
request_id: RequestId::Integer(2),
|
||||
params: ThreadQueueListParams {
|
||||
thread_id: thread_id.clone(),
|
||||
cursor: None,
|
||||
limit: None,
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
assert!(queue.data.is_empty());
|
||||
|
||||
let requests = model
|
||||
.received_requests()
|
||||
.await
|
||||
.context("model request capture unavailable")?;
|
||||
let request = requests
|
||||
.iter()
|
||||
.find(|request| request.url.path().ends_with("/responses"))
|
||||
.context("queued message did not reach the model")?;
|
||||
assert!(
|
||||
request.body_json::<Value>()?["input"]
|
||||
.to_string()
|
||||
.contains(message)
|
||||
);
|
||||
assert_eq!(
|
||||
request.headers["authorization"].to_str()?,
|
||||
"Bearer server-only-test-key"
|
||||
);
|
||||
let metadata: Value = serde_json::from_str(request.headers["x-codex-turn-metadata"].to_str()?)?;
|
||||
assert_eq!(metadata["turn_trigger"], "queue");
|
||||
|
||||
let name = "remote-session-lifecycle";
|
||||
let _: ThreadSetNameResponse = app
|
||||
.request_typed(ClientRequest::ThreadSetName {
|
||||
request_id: RequestId::Integer(3),
|
||||
params: ThreadSetNameParams {
|
||||
thread_id: thread_id.clone(),
|
||||
name: name.to_string(),
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
// A separate caller home makes name resolution rely on the selected server.
|
||||
let caller_home = TempDir::new()?;
|
||||
let remote_command = || {
|
||||
let mut command = Command::new(&codex);
|
||||
command
|
||||
.env("CODEX_HOME", caller_home.path())
|
||||
.current_dir(caller_home.path())
|
||||
.env(OPENAI_FEDERATION_RULE_ID_ENV_VAR, "rule-test")
|
||||
.env(
|
||||
OPENAI_IDENTITY_TOKEN_FILE_ENV_VAR,
|
||||
caller_home.path().join("missing-identity-token"),
|
||||
)
|
||||
.env_remove("CODEX_QUEUE_SERVER_API_KEY")
|
||||
.kill_on_drop(true)
|
||||
.args(["--remote", &endpoint]);
|
||||
command
|
||||
};
|
||||
let output = timeout(
|
||||
Duration::from_secs(30),
|
||||
remote_command().args(["delete", &thread_id]).output(),
|
||||
)
|
||||
.await??;
|
||||
assert!(!output.status.success());
|
||||
assert!(String::from_utf8(output.stderr)?.contains("cannot confirm session deletion"));
|
||||
|
||||
let cases: &[(&[&str], Option<bool>)] = &[
|
||||
(&["archive", name], Some(true)),
|
||||
(&["unarchive", name], Some(false)),
|
||||
(&["delete", "--force", &thread_id], None),
|
||||
];
|
||||
for &(args, expected_archived) in cases {
|
||||
let output = timeout(
|
||||
Duration::from_secs(30),
|
||||
remote_command().args(args).output(),
|
||||
)
|
||||
.await??;
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{args:?}: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
for archived in [false, true] {
|
||||
let threads: ThreadListResponse = app
|
||||
.request_typed(ClientRequest::ThreadList {
|
||||
request_id: RequestId::Integer(4),
|
||||
params: ThreadListParams {
|
||||
cursor: None,
|
||||
limit: None,
|
||||
sort_key: None,
|
||||
sort_direction: None,
|
||||
model_providers: None,
|
||||
source_kinds: None,
|
||||
originators: None,
|
||||
archived: Some(archived),
|
||||
section_id: None,
|
||||
project_id: None,
|
||||
cwd: None,
|
||||
use_state_db_only: false,
|
||||
search_term: None,
|
||||
parent_thread_id: None,
|
||||
ancestor_thread_id: None,
|
||||
},
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(
|
||||
threads.data.iter().any(|thread| thread.id == thread_id),
|
||||
expected_archived == Some(archived),
|
||||
"{args:?}: archived={archived}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_session_commands_validate_config() -> Result<()> {
|
||||
let cases: &[(&[&str], &str, &str)] = &[
|
||||
(
|
||||
&["--strict-config"],
|
||||
"config.toml",
|
||||
"unknown_queue_option = true\n",
|
||||
),
|
||||
(&["--profile", "queue"], "queue.config.toml", "model = 42\n"),
|
||||
(
|
||||
&[],
|
||||
"config.toml",
|
||||
"experimental_thread_store_endpoint = \"https://example.com\"\n",
|
||||
),
|
||||
];
|
||||
let commands: &[&[&str]] = &[
|
||||
&["queue", "--thread", THREAD_ID, "--message", "do the thing"],
|
||||
&["archive", THREAD_ID],
|
||||
&["unarchive", THREAD_ID],
|
||||
&["delete", "--force", THREAD_ID],
|
||||
];
|
||||
for &command in commands {
|
||||
for &(args, config_file, config) in cases {
|
||||
let codex_home = TempDir::new()?;
|
||||
std::fs::write(codex_home.path().join(config_file), config)?;
|
||||
let output = std::process::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?)
|
||||
.env("CODEX_HOME", codex_home.path())
|
||||
.current_dir(codex_home.path())
|
||||
.args(args)
|
||||
.args(["--remote", "ws://127.0.0.1:1"])
|
||||
.args(command)
|
||||
.output()?;
|
||||
let stderr = String::from_utf8(output.stderr)?;
|
||||
assert!(!output.status.success());
|
||||
assert!(
|
||||
stderr.contains("failed to load config.toml"),
|
||||
"{command:?}, {config:?}: {stderr}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn queue_rejects_local_daemon_that_does_not_support_queueing() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir_in("/tmp")?;
|
||||
let codex_home = socket_test_home()?;
|
||||
let socket_path = codex_app_server::app_server_control_socket_path(codex_home.path())?;
|
||||
std::fs::create_dir_all(
|
||||
socket_path
|
||||
@@ -232,7 +520,7 @@ async fn queue_rejects_local_daemon_that_does_not_support_queueing() -> Result<(
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn queue_rejects_overrides_that_bypass_local_daemon() -> Result<()> {
|
||||
let codex_home = tempfile::tempdir_in("/tmp")?;
|
||||
let codex_home = socket_test_home()?;
|
||||
let socket_path = codex_app_server::app_server_control_socket_path(codex_home.path())?;
|
||||
std::fs::create_dir_all(
|
||||
socket_path
|
||||
|
||||
Reference in New Issue
Block a user