From 0002316687f63f3b9b0155cb9272eca13aae63e3 Mon Sep 17 00:00:00 2001 From: Anton Panasenko Date: Mon, 1 Jun 2026 18:05:50 -0700 Subject: [PATCH] feat(remote-control): add pairing start (#25675) ## Why Remote control enrollment authorizes a desktop server, but app-server v2 did not expose the follow-up pairing operation needed to mint a short-lived controller pairing artifact from that enrolled server. Clients need a narrow RPC that starts pairing without exposing the backend `serverId` or conflating pairing with websocket connection state. Issue: N/A; internal remote-control pairing API change. ## What Changed Added experimental app-server v2 `remoteControl/pairing/start` with `manualCode` input and `pairingCode`, nullable `manualPairingCode`, `environmentId`, and Unix-seconds `expiresAt` output. The method serializes under its own `global("remote-control-pairing")` scope and is documented in `app-server/README.md`. Extended the remote-control transport with private `/server/pair` request/response types and normalized `pair_url` handling. Pairing uses the current enrolled server bearer, refreshes that bearer when needed, keeps backend `server_id` private, validates returned `server_id` and `environment_id` against the current enrollment, and preserves backend status/header/body context for failures and malformed responses. Wired the request through `RemoteControlRequestProcessor` and `MessageProcessor`, mapping unavailable/disabled pairing to `invalid_request` and backend failures to internal errors. ## Verification - `just test -p codex-app-server-transport` - `just test -p codex-app-server remote_control_pairing_start_returns_pairing_artifacts` --- codex-rs/app-server-protocol/src/export.rs | 19 + .../src/protocol/common.rs | 17 + .../src/protocol/v2/remote_control.rs | 18 + .../src/transport/remote_control/enroll.rs | 128 ++++- .../src/transport/remote_control/mod.rs | 221 ++++++++ .../src/transport/remote_control/protocol.rs | 28 + .../src/transport/remote_control/tests.rs | 49 ++ .../remote_control/tests/pairing_tests.rs | 494 ++++++++++++++++++ .../src/transport/remote_control/websocket.rs | 88 +++- codex-rs/app-server/README.md | 1 + codex-rs/app-server/src/message_processor.rs | 5 + .../remote_control_processor.rs | 24 + .../remote_control_processor_tests.rs | 48 ++ .../tests/common/test_app_server.rs | 11 + .../tests/suite/v2/remote_control.rs | 199 ++++++- 15 files changed, 1302 insertions(+), 48 deletions(-) create mode 100644 codex-rs/app-server-transport/src/transport/remote_control/tests/pairing_tests.rs create mode 100644 codex-rs/app-server/src/request_processors/remote_control_processor/remote_control_processor_tests.rs diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 56cf7038f..b9c44fe9a 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -2924,4 +2924,23 @@ permissionProfile?: string | null}; let _cleanup = fs::remove_dir_all(&output_dir); Ok(()) } + + #[test] + fn generate_json_includes_remote_control_pairing_start_with_experimental_api() -> Result<()> { + let output_dir = std::env::temp_dir().join(format!("codex_schema_{}", Uuid::now_v7())); + fs::create_dir(&output_dir)?; + generate_json_with_experimental(&output_dir, /*experimental_api*/ true)?; + + let client_request_json = fs::read_to_string(output_dir.join("ClientRequest.json"))?; + assert!(client_request_json.contains("remoteControl/pairing/start")); + for schema in [ + "RemoteControlPairingStartParams.json", + "RemoteControlPairingStartResponse.json", + ] { + assert!(output_dir.join("v2").join(schema).exists()); + } + + let _cleanup = fs::remove_dir_all(&output_dir); + Ok(()) + } } diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index d532952f6..961a4a227 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -843,6 +843,12 @@ client_request_definitions! { serialization: global_shared_read("remote-control"), response: v2::RemoteControlStatusReadResponse, }, + #[experimental("remoteControl/pairing/start")] + RemoteControlPairingStart => "remoteControl/pairing/start" { + params: v2::RemoteControlPairingStartParams, + serialization: global("remote-control-pairing"), + response: v2::RemoteControlPairingStartResponse, + }, #[experimental("collaborationMode/list")] /// Lists collaboration mode presets. CollaborationModeList => "collaborationMode/list" { @@ -1977,6 +1983,17 @@ mod tests { }, }; assert_eq!(mcp_resource_read.serialization_scope(), None); + + let remote_control_pairing_start = ClientRequest::RemoteControlPairingStart { + request_id: request_id(), + params: v2::RemoteControlPairingStartParams::default(), + }; + assert_eq!( + remote_control_pairing_start.serialization_scope(), + Some(ClientRequestSerializationScope::Global( + "remote-control-pairing" + )) + ); } #[test] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/remote_control.rs b/codex-rs/app-server-protocol/src/protocol/v2/remote_control.rs index c8ad617e7..e5c39a777 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/remote_control.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/remote_control.rs @@ -44,6 +44,24 @@ pub struct RemoteControlStatusReadResponse { pub environment_id: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RemoteControlPairingStartParams { + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub manual_code: bool, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RemoteControlPairingStartResponse { + pub pairing_code: String, + pub manual_pairing_code: Option, + pub environment_id: String, + pub expires_at: i64, +} + #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(rename_all = "camelCase", export_to = "v2/")] diff --git a/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs b/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs index 6037caa65..ac6772637 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs @@ -1,9 +1,13 @@ +use super::pairing_unavailable_error; use super::protocol::EnrollRemoteServerRequest; use super::protocol::EnrollRemoteServerResponse; use super::protocol::RefreshRemoteServerRequest; use super::protocol::RemoteControlTarget; +use super::protocol::StartRemoteControlPairingRequest; +use super::protocol::StartRemoteControlPairingResponse; use axum::http::HeaderMap; use codex_api::SharedAuthProvider; +use codex_app_server_protocol::RemoteControlPairingStartResponse; use codex_login::default_client::build_reqwest_client; use codex_state::RemoteControlEnrollmentRecord; use codex_state::StateRuntime; @@ -17,6 +21,7 @@ use tracing::info; use tracing::warn; const REMOTE_CONTROL_ENROLL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +const REMOTE_CONTROL_PAIRING_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); const REMOTE_CONTROL_RESPONSE_BODY_MAX_BYTES: usize = 4096; const REMOTE_CONTROL_SERVER_TOKEN_REFRESH_SKEW_SECS: i64 = 30; @@ -28,6 +33,7 @@ pub(super) const REMOTE_CONTROL_INSTALLATION_ID_HEADER: &str = "x-codex-installa #[derive(Debug, Clone, PartialEq, Eq)] pub(super) struct RemoteControlEnrollment { + pub(super) remote_control_target: RemoteControlTarget, pub(super) account_id: String, pub(super) environment_id: String, pub(super) server_id: String, @@ -37,6 +43,99 @@ pub(super) struct RemoteControlEnrollment { } impl RemoteControlEnrollment { + pub(super) async fn start_pairing( + &self, + request: StartRemoteControlPairingRequest, + ) -> io::Result { + if self.should_refresh_server_token() { + return Err(pairing_unavailable_error()); + } + let remote_control_token = self + .remote_control_token + .as_deref() + .ok_or_else(pairing_unavailable_error)?; + + let response = build_reqwest_client() + .post(&self.remote_control_target.pair_url) + .timeout(REMOTE_CONTROL_PAIRING_TIMEOUT) + .bearer_auth(remote_control_token) + .json(&request) + .send() + .await + .map_err(|err| { + io::Error::other(format!( + "failed to start remote control pairing at `{}`: {err}", + self.remote_control_target.pair_url + )) + })?; + let headers = response.headers().clone(); + let status = response.status(); + let body = response.bytes().await.map_err(|err| { + io::Error::other(format!( + "failed to read remote control pairing response from `{}`: {err}", + self.remote_control_target.pair_url + )) + })?; + let body_preview = preview_remote_control_response_body(&body); + if !status.is_success() { + let error_kind = match status.as_u16() { + 401 | 403 => ErrorKind::PermissionDenied, + 404 => ErrorKind::NotFound, + _ => ErrorKind::Other, + }; + return Err(io::Error::new( + error_kind, + format!( + "remote control pairing failed at `{}`: HTTP {status}, {}, body: {body_preview}", + self.remote_control_target.pair_url, + format_headers(&headers) + ), + )); + } + + let pairing = serde_json::from_slice::(&body).map_err( + |err| { + io::Error::other(format!( + "failed to parse remote control pairing response from `{}`: HTTP {status}, {}, body: {body_preview}, decode error: {err}", + self.remote_control_target.pair_url, + format_headers(&headers) + )) + }, + )?; + let StartRemoteControlPairingResponse { + pairing_code, + manual_pairing_code, + server_id, + environment_id, + expires_at, + } = pairing; + if server_id != self.server_id || environment_id != self.environment_id { + return Err(io::Error::other(format!( + "remote control pairing returned mismatched enrollment: expected server_id={}, environment_id={}; got server_id={}, environment_id={}", + self.server_id, self.environment_id, server_id, environment_id + ))); + } + let expires_at = OffsetDateTime::parse(&expires_at, &Rfc3339) + .map_err(|err| { + io::Error::new( + ErrorKind::InvalidData, + format!( + "failed to parse remote control pairing response from `{}`: HTTP {status}, {}, body: {body_preview}, expires_at parse error: {err}", + self.remote_control_target.pair_url, + format_headers(&headers) + ), + ) + })? + .unix_timestamp(); + + Ok(RemoteControlPairingStartResponse { + pairing_code, + manual_pairing_code, + environment_id, + expires_at, + }) + } + pub(super) fn should_refresh_server_token(&self) -> bool { self.remote_control_token.is_none() || self.expires_at.is_none_or(|expires_at| { @@ -101,6 +200,7 @@ pub(super) async fn load_persisted_remote_control_enrollment( enrollment.environment_id ); Ok(Some(RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: enrollment.account_id, environment_id: enrollment.environment_id, server_id: enrollment.server_id, @@ -211,8 +311,14 @@ fn redact_remote_control_response_body(body: &str) -> String { let Some(body_object) = body_json.as_object_mut() else { return body.to_string(); }; - if let Some(remote_control_token) = body_object.get_mut("remote_control_token") { - *remote_control_token = serde_json::Value::String("".to_string()); + for sensitive_field in [ + "remote_control_token", + "pairing_code", + "manual_pairing_code", + ] { + if let Some(value) = body_object.get_mut(sensitive_field) { + *value = serde_json::Value::String("".to_string()); + } } body_json.to_string() } @@ -254,6 +360,7 @@ pub(super) async fn enroll_remote_control_server( ) .await?; let mut enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: auth.account_id.clone(), environment_id: enrollment_response.environment_id, server_id: enrollment_response.server_id, @@ -271,18 +378,17 @@ pub(super) async fn enroll_remote_control_server( } pub(super) async fn refresh_remote_control_server( - remote_control_target: &RemoteControlTarget, auth: &RemoteControlConnectionAuth, installation_id: &str, enrollment: &mut RemoteControlEnrollment, ) -> io::Result<()> { - let refresh_url = &remote_control_target.refresh_url; + let refresh_url = enrollment.remote_control_target.refresh_url.clone(); let request = RefreshRemoteServerRequest { server_id: enrollment.server_id.clone(), installation_id: installation_id.to_string(), }; let refreshed = send_remote_control_server_request::<_, EnrollRemoteServerResponse>( - refresh_url, + &refresh_url, auth, installation_id, &request, @@ -304,7 +410,7 @@ pub(super) async fn refresh_remote_control_server( update_remote_control_server_token( enrollment, - refresh_url, + &refresh_url, refreshed.remote_control_token, refreshed.expires_at, ) @@ -412,6 +518,8 @@ mod tests { #[test] fn remote_control_enrollment_refreshes_server_token_before_expiry() { let expires_soon = RemoteControlEnrollment { + remote_control_target: normalize_remote_control_url("http://localhost/backend-api/") + .expect("target should normalize"), account_id: "account-a".to_string(), environment_id: "env_first".to_string(), server_id: "srv_e_first".to_string(), @@ -433,12 +541,14 @@ mod tests { fn preview_remote_control_response_body_redacts_server_token() { assert_eq!( serde_json::from_str::(&preview_remote_control_response_body( - br#"{"server_id":"srv_e_test","remote_control_token":"secret"}"# + br#"{"server_id":"srv_e_test","remote_control_token":"secret","pairing_code":"pairing-code","manual_pairing_code":"ABCD-EFGH"}"# )) .expect("redacted response preview should stay valid json"), json!({ "server_id": "srv_e_test", "remote_control_token": "", + "pairing_code": "", + "manual_pairing_code": "", }) ); } @@ -453,6 +563,7 @@ mod tests { normalize_remote_control_url("https://api.chatgpt-staging.com/other/control") .expect("second target should parse"); let first_enrollment = RemoteControlEnrollment { + remote_control_target: first_target.clone(), account_id: "account-a".to_string(), environment_id: "env_first".to_string(), server_id: "srv_e_first".to_string(), @@ -461,6 +572,7 @@ mod tests { expires_at: None, }; let second_enrollment = RemoteControlEnrollment { + remote_control_target: second_target.clone(), account_id: "account-a".to_string(), environment_id: "env_second".to_string(), server_id: "srv_e_second".to_string(), @@ -533,6 +645,7 @@ mod tests { normalize_remote_control_url("https://api.chatgpt-staging.com/other/control") .expect("second target should parse"); let first_enrollment = RemoteControlEnrollment { + remote_control_target: first_target.clone(), account_id: "account-a".to_string(), environment_id: "env_first".to_string(), server_id: "srv_e_first".to_string(), @@ -541,6 +654,7 @@ mod tests { expires_at: None, }; let second_enrollment = RemoteControlEnrollment { + remote_control_target: second_target.clone(), account_id: "account-a".to_string(), environment_id: "env_second".to_string(), server_id: "srv_e_second".to_string(), diff --git a/codex-rs/app-server-transport/src/transport/remote_control/mod.rs b/codex-rs/app-server-transport/src/transport/remote_control/mod.rs index 357a98a30..a55667886 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/mod.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/mod.rs @@ -4,6 +4,8 @@ mod protocol; mod segment; mod websocket; +use self::enroll::RemoteControlEnrollment; +use self::enroll::refresh_remote_control_server; use crate::transport::remote_control::websocket::RemoteControlChannels; use crate::transport::remote_control::websocket::RemoteControlStatusPublisher; use crate::transport::remote_control::websocket::RemoteControlWebsocket; @@ -16,6 +18,8 @@ use super::CHANNEL_CAPACITY; use super::TransportEvent; use super::next_connection_id; use codex_app_server_protocol::RemoteControlConnectionStatus; +use codex_app_server_protocol::RemoteControlPairingStartParams; +use codex_app_server_protocol::RemoteControlPairingStartResponse; use codex_app_server_protocol::RemoteControlStatusChangedNotification; use codex_login::AuthManager; use codex_state::StateRuntime; @@ -26,6 +30,7 @@ use std::fmt; use std::io; use std::panic::AssertUnwindSafe; use std::sync::Arc; +use std::sync::Mutex as StdMutex; use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::sync::watch; @@ -52,8 +57,12 @@ pub struct RemoteControlHandle { enabled_tx: Arc>, status_tx: Arc>, state_db_available: bool, + current_enrollment: CurrentRemoteControlEnrollment, + auth_manager: Arc, } +type CurrentRemoteControlEnrollment = Arc>>; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RemoteControlUnavailable; @@ -108,6 +117,7 @@ impl RemoteControlHandle { *state = false; changed }); + clear_current_enrollment(&self.current_enrollment); let status = self.status(); info!( @@ -129,6 +139,88 @@ impl RemoteControlHandle { self.status_tx.subscribe() } + pub async fn start_pairing( + &self, + params: RemoteControlPairingStartParams, + ) -> io::Result { + if !*self.enabled_tx.borrow() { + return Err(Self::pairing_disabled_error()); + } + let mut auth = websocket::load_remote_control_auth(&self.auth_manager) + .await + .map_err(|_| pairing_unavailable_error())?; + let mut enrollment = { + let current_enrollment = self + .current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + current_enrollment + .as_ref() + .filter(|enrollment| enrollment.account_id == auth.account_id) + .cloned() + } + .ok_or_else(pairing_unavailable_error)?; + let installation_id = self.status().installation_id; + if enrollment.should_refresh_server_token() { + refresh_pairing_enrollment( + &self.current_enrollment, + &self.auth_manager, + &mut auth, + &installation_id, + &mut enrollment, + ) + .await?; + } + let pairing_request = || protocol::StartRemoteControlPairingRequest { + manual_code: params.manual_code, + }; + let pairing_response = match enrollment.start_pairing(pairing_request()).await { + Err(err) if err.kind() == io::ErrorKind::PermissionDenied => { + clear_pairing_server_token(&self.current_enrollment, &mut enrollment)?; + refresh_pairing_enrollment( + &self.current_enrollment, + &self.auth_manager, + &mut auth, + &installation_id, + &mut enrollment, + ) + .await?; + enrollment.start_pairing(pairing_request()).await + } + pairing_response => pairing_response, + }; + if let Err(err) = &pairing_response { + match err.kind() { + io::ErrorKind::NotFound => { + clear_current_enrollment_if_matches(&self.current_enrollment, &enrollment); + return Err(pairing_unavailable_error()); + } + io::ErrorKind::PermissionDenied => { + clear_pairing_server_token(&self.current_enrollment, &mut enrollment)?; + return Err(pairing_unavailable_error()); + } + _ => {} + } + } + if !*self.enabled_tx.borrow() { + return Err(Self::pairing_disabled_error()); + } + let current_auth = websocket::load_remote_control_auth(&self.auth_manager) + .await + .map_err(|_| pairing_unavailable_error())?; + if current_auth.account_id != auth.account_id { + return Err(pairing_unavailable_error()); + } + pairing_response + } + + fn pairing_disabled_error() -> io::Error { + io::Error::new( + io::ErrorKind::InvalidInput, + "remote control pairing requires remote control to be enabled", + ) + } + fn publish_status( &self, connection_status: RemoteControlConnectionStatus, @@ -160,6 +252,71 @@ impl RemoteControlHandle { } } +async fn refresh_pairing_enrollment( + current_enrollment: &CurrentRemoteControlEnrollment, + auth_manager: &Arc, + auth: &mut enroll::RemoteControlConnectionAuth, + installation_id: &str, + enrollment: &mut RemoteControlEnrollment, +) -> io::Result<()> { + if let Err(err) = refresh_remote_control_server(auth, installation_id, enrollment).await { + if err.kind() != io::ErrorKind::PermissionDenied { + return handle_pairing_refresh_error(current_enrollment, enrollment, err); + } + let mut auth_recovery = auth_manager.unauthorized_recovery(); + let mut auth_change_rx = auth_manager.auth_change_receiver(); + if !websocket::recover_remote_control_auth(&mut auth_recovery, &mut auth_change_rx).await { + return Err(err); + } + *auth = websocket::load_remote_control_auth(auth_manager) + .await + .map_err(|_| pairing_unavailable_error())?; + if auth.account_id != enrollment.account_id { + return Err(pairing_unavailable_error()); + } + if let Err(err) = refresh_remote_control_server(auth, installation_id, enrollment).await { + return handle_pairing_refresh_error(current_enrollment, enrollment, err); + } + } + if replace_current_enrollment(current_enrollment, enrollment) { + Ok(()) + } else { + Err(pairing_unavailable_error()) + } +} + +fn handle_pairing_refresh_error( + current_enrollment: &CurrentRemoteControlEnrollment, + enrollment: &RemoteControlEnrollment, + err: io::Error, +) -> io::Result<()> { + if err.kind() == io::ErrorKind::NotFound { + clear_current_enrollment_if_matches(current_enrollment, enrollment); + Err(pairing_unavailable_error()) + } else { + Err(err) + } +} + +fn clear_pairing_server_token( + current_enrollment: &CurrentRemoteControlEnrollment, + enrollment: &mut RemoteControlEnrollment, +) -> io::Result<()> { + enrollment.clear_server_token(); + if replace_current_enrollment(current_enrollment, enrollment) { + Ok(()) + } else { + Err(pairing_unavailable_error()) + } +} + +fn pairing_unavailable_error() -> io::Error { + io::Error::new( + io::ErrorKind::InvalidInput, + "remote control pairing is unavailable until enrollment completes", + ) +} + fn remote_control_status_with_connection_status( status: &RemoteControlStatusChangedNotification, connection_status: RemoteControlConnectionStatus, @@ -176,6 +333,64 @@ fn remote_control_status_with_connection_status( } } +fn publish_current_enrollment( + current_enrollment: &CurrentRemoteControlEnrollment, + enrollment: &RemoteControlEnrollment, +) { + *current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(enrollment.clone()); +} + +fn clear_current_enrollment(current_enrollment: &CurrentRemoteControlEnrollment) { + *current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; +} + +fn replace_current_enrollment( + current_enrollment: &CurrentRemoteControlEnrollment, + enrollment: &RemoteControlEnrollment, +) -> bool { + let mut current_enrollment = current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !current_enrollment + .as_ref() + .is_some_and(|current| same_remote_control_enrollment(current, enrollment)) + { + return false; + } + *current_enrollment = Some(enrollment.clone()); + true +} + +fn clear_current_enrollment_if_matches( + current_enrollment: &CurrentRemoteControlEnrollment, + enrollment: &RemoteControlEnrollment, +) { + let mut current_enrollment = current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if current_enrollment + .as_ref() + .is_some_and(|current| same_remote_control_enrollment(current, enrollment)) + { + *current_enrollment = None; + } +} + +fn same_remote_control_enrollment( + left: &RemoteControlEnrollment, + right: &RemoteControlEnrollment, +) -> bool { + // A refresh rotates only the bearer. Pairing remains current while the same persisted server + // record is still selected for the current account. + left.account_id == right.account_id + && left.server_id == right.server_id + && left.environment_id == right.environment_id +} + pub async fn start_remote_control( config: RemoteControlStartConfig, state_db: Option>, @@ -198,6 +413,9 @@ pub async fn start_remote_control( }; let (enabled_tx, enabled_rx) = watch::channel(initial_enabled); + let current_enrollment = Arc::new(StdMutex::new(None)); + let websocket_current_enrollment = current_enrollment.clone(); + let handle_auth_manager = auth_manager.clone(); let server_name = gethostname().to_string_lossy().trim().to_string(); let remote_control_url = config.remote_control_url; let installation_id = config.installation_id; @@ -245,6 +463,7 @@ pub async fn start_remote_control( RemoteControlChannels { transport_event_tx, status_publisher, + current_enrollment: websocket_current_enrollment, }, shutdown_token, enabled_rx, @@ -289,6 +508,8 @@ pub async fn start_remote_control( enabled_tx: Arc::new(enabled_tx), status_tx: Arc::new(status_tx), state_db_available, + current_enrollment, + auth_manager: handle_auth_manager, }, )) } diff --git a/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs b/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs index b10af42e8..8eeb28e26 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs @@ -12,6 +12,7 @@ pub(super) struct RemoteControlTarget { pub(super) websocket_url: String, pub(super) enroll_url: String, pub(super) refresh_url: String, + pub(super) pair_url: String, } #[derive(Debug, Serialize)] @@ -37,6 +38,20 @@ pub(super) struct RefreshRemoteServerRequest { pub(super) installation_id: String, } +#[derive(Debug, Serialize)] +pub(super) struct StartRemoteControlPairingRequest { + pub(super) manual_code: bool, +} + +#[derive(Debug, Deserialize)] +pub(super) struct StartRemoteControlPairingResponse { + pub(super) pairing_code: String, + pub(super) manual_pairing_code: Option, + pub(super) server_id: String, + pub(super) environment_id: String, + pub(super) expires_at: String, +} + #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(transparent)] pub struct ClientId(pub String); @@ -189,6 +204,9 @@ pub(super) fn normalize_remote_control_url( let refresh_url = remote_control_url .join("wham/remote/control/server/refresh") .map_err(map_url_parse_error)?; + let pair_url = remote_control_url + .join("wham/remote/control/server/pair") + .map_err(map_url_parse_error)?; let mut websocket_url = remote_control_url .join("wham/remote/control/server") .map_err(map_url_parse_error)?; @@ -207,6 +225,7 @@ pub(super) fn normalize_remote_control_url( websocket_url: websocket_url.to_string(), enroll_url: enroll_url.to_string(), refresh_url: refresh_url.to_string(), + pair_url: pair_url.to_string(), }) } @@ -227,6 +246,8 @@ mod tests { .to_string(), refresh_url: "https://chatgpt.com/backend-api/wham/remote/control/server/refresh" .to_string(), + pair_url: "https://chatgpt.com/backend-api/wham/remote/control/server/pair" + .to_string(), } ); assert_eq!( @@ -242,6 +263,9 @@ mod tests { refresh_url: "https://api.chatgpt-staging.com/backend-api/wham/remote/control/server/refresh" .to_string(), + pair_url: + "https://api.chatgpt-staging.com/backend-api/wham/remote/control/server/pair" + .to_string(), } ); } @@ -258,6 +282,8 @@ mod tests { .to_string(), refresh_url: "http://localhost:8080/backend-api/wham/remote/control/server/refresh" .to_string(), + pair_url: "http://localhost:8080/backend-api/wham/remote/control/server/pair" + .to_string(), } ); assert_eq!( @@ -271,6 +297,8 @@ mod tests { refresh_url: "https://localhost:8443/backend-api/wham/remote/control/server/refresh" .to_string(), + pair_url: "https://localhost:8443/backend-api/wham/remote/control/server/pair" + .to_string(), } ); } diff --git a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs index 85bcf0922..5d96f89d8 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs @@ -20,6 +20,7 @@ use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::RemoteControlConnectionStatus; +use codex_app_server_protocol::RemoteControlPairingStartParams; use codex_app_server_protocol::RemoteControlStatusChangedNotification; use codex_app_server_protocol::ServerNotification; use codex_config::types::AuthCredentialsStoreMode; @@ -39,7 +40,9 @@ use pretty_assertions::assert_eq; use serde_json::json; use std::collections::BTreeMap; use std::sync::Arc; +use std::sync::Mutex as StdMutex; use tempfile::TempDir; +use time::OffsetDateTime; use tokio::io::AsyncBufReadExt; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; @@ -57,7 +60,10 @@ use tokio_tungstenite::accept_hdr_async; use tokio_tungstenite::tungstenite; use tokio_util::sync::CancellationToken; +mod pairing_tests; + const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111"; +const TEST_REMOTE_CONTROL_URL: &str = "http://127.0.0.1:1/backend-api/wham/remote/control"; const TEST_REMOTE_CONTROL_SERVER_TOKEN: &str = "Remote Control Token"; const TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN: &str = "Refreshed Remote Control Token"; const TEST_REMOTE_CONTROL_SERVER_TOKEN_EXPIRES_AT: &str = "2999-01-01T00:00:00Z"; @@ -128,6 +134,40 @@ fn test_server_name() -> String { gethostname().to_string_lossy().trim().to_string() } +fn remote_control_handle_with_current_enrollment( + remote_control_url: &str, + auth_manager: Arc, +) -> RemoteControlHandle { + let (enabled_tx, _enabled_rx) = watch::channel(/*init*/ true); + let (status_tx, _status_rx) = watch::channel(RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + server_name: test_server_name(), + installation_id: TEST_INSTALLATION_ID.to_string(), + environment_id: Some("env_test".to_string()), + }); + let remote_control_target = normalize_remote_control_url(remote_control_url) + .expect("remote control target should normalize"); + let current_enrollment = Arc::new(StdMutex::new(Some(RemoteControlEnrollment { + remote_control_target, + account_id: "account_id".to_string(), + environment_id: "env_test".to_string(), + server_id: "srv_e_test".to_string(), + server_name: test_server_name(), + remote_control_token: Some(TEST_REMOTE_CONTROL_SERVER_TOKEN.to_string()), + expires_at: Some( + OffsetDateTime::from_unix_timestamp(33_336_362_096) + .expect("future timestamp should parse"), + ), + }))); + RemoteControlHandle { + enabled_tx: Arc::new(enabled_tx), + status_tx: Arc::new(status_tx), + state_db_available: true, + current_enrollment, + auth_manager, + } +} + fn remote_control_server_token_response( server_id: &str, environment_id: &str, @@ -1313,6 +1353,7 @@ async fn remote_control_http_mode_refreshes_persisted_enrollment_before_connecti let remote_control_target = normalize_remote_control_url(&remote_control_url).expect("target should parse"); let persisted_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_persisted".to_string(), server_id: "srv_e_persisted".to_string(), @@ -1418,6 +1459,7 @@ async fn remote_control_stdio_mode_waits_for_client_name_before_connecting() { normalize_remote_control_url(&remote_control_url).expect("target should parse"); let app_server_client_name = "stdio-client"; let persisted_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_persisted".to_string(), server_id: "srv_e_persisted".to_string(), @@ -1505,7 +1547,10 @@ async fn remote_control_waits_for_account_id_before_enrolling() { ) .await; let expected_server_name = gethostname().to_string_lossy().trim().to_string(); + let expected_remote_control_target = normalize_remote_control_url(&remote_control_url) + .expect("remote control target should normalize"); let expected_enrollment = RemoteControlEnrollment { + remote_control_target: expected_remote_control_target, account_id: "account_id".to_string(), environment_id: "env_ready".to_string(), server_id: "srv_e_ready".to_string(), @@ -1583,6 +1628,7 @@ async fn remote_control_http_mode_reenrolls_when_refresh_reports_stale_enrollmen normalize_remote_control_url(&remote_control_url).expect("target should parse"); let expected_server_name = gethostname().to_string_lossy().trim().to_string(); let stale_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_stale".to_string(), server_id: "srv_e_stale".to_string(), @@ -1591,6 +1637,7 @@ async fn remote_control_http_mode_reenrolls_when_refresh_reports_stale_enrollmen expires_at: None, }; let refreshed_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_refreshed".to_string(), server_id: "srv_e_refreshed".to_string(), @@ -1700,6 +1747,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() normalize_remote_control_url(&remote_control_url).expect("target should parse"); let expected_server_name = gethostname().to_string_lossy().trim().to_string(); let stale_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_stale".to_string(), server_id: "srv_e_stale".to_string(), @@ -1708,6 +1756,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() expires_at: None, }; let refreshed_enrollment = RemoteControlEnrollment { + remote_control_target: remote_control_target.clone(), account_id: "account_id".to_string(), environment_id: "env_refreshed".to_string(), server_id: "srv_e_refreshed".to_string(), diff --git a/codex-rs/app-server-transport/src/transport/remote_control/tests/pairing_tests.rs b/codex-rs/app-server-transport/src/transport/remote_control/tests/pairing_tests.rs new file mode 100644 index 000000000..176aeb4ed --- /dev/null +++ b/codex-rs/app-server-transport/src/transport/remote_control/tests/pairing_tests.rs @@ -0,0 +1,494 @@ +use super::super::protocol::StartRemoteControlPairingRequest; +use super::*; +use pretty_assertions::assert_eq; + +fn remote_control_enrollment( + remote_control_url: &str, + remote_control_token: &str, +) -> RemoteControlEnrollment { + RemoteControlEnrollment { + remote_control_target: normalize_remote_control_url(remote_control_url) + .expect("target should normalize"), + account_id: "account-id".to_string(), + environment_id: "environment-id".to_string(), + server_id: "server-id".to_string(), + server_name: "server-name".to_string(), + remote_control_token: Some(remote_control_token.to_string()), + expires_at: Some( + OffsetDateTime::from_unix_timestamp(33_336_362_096) + .expect("future timestamp should parse"), + ), + } +} + +async fn pairing_error(status: &'static str, body: &'static str) -> (String, String) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let expected_pair_url = normalize_remote_control_url(&remote_control_url) + .expect("target should normalize") + .pair_url; + let server_task = tokio::spawn(async move { + let pairing_request = accept_http_request(&listener).await; + respond_with_status_and_headers( + pairing_request.stream, + status, + &[("x-request-id", "request-123"), ("cf-ray", "ray-123")], + body, + ) + .await; + }); + + let err = remote_control_enrollment(&remote_control_url, "remote-control-token") + .start_pairing(StartRemoteControlPairingRequest { manual_code: false }) + .await + .expect_err("pairing should fail"); + server_task.await.expect("server task should finish"); + (err.to_string(), expected_pair_url) +} + +async fn pairing_response_error(body: serde_json::Value) -> String { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let server_task = tokio::spawn(async move { + let pairing_request = accept_http_request(&listener).await; + respond_with_json(pairing_request.stream, body).await; + }); + + let err = remote_control_enrollment(&remote_control_url, "remote-control-token") + .start_pairing(StartRemoteControlPairingRequest { manual_code: false }) + .await + .expect_err("pairing should fail"); + server_task.await.expect("server task should finish"); + err.to_string() +} + +#[tokio::test] +async fn remote_control_handle_starts_pairing_before_websocket_connects() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let server_task = tokio::spawn(async move { + let refresh_request = accept_http_request(&listener).await; + assert_eq!( + refresh_request.request_line, + "POST /backend-api/wham/remote/control/server/refresh HTTP/1.1" + ); + assert_eq!( + serde_json::from_str::(&refresh_request.body) + .expect("refresh request body should deserialize"), + json!({ + "server_id": "srv_e_test", + "installation_id": TEST_INSTALLATION_ID, + }) + ); + respond_with_json( + refresh_request.stream, + remote_control_server_token_response( + "srv_e_test", + "env_test", + TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN, + ), + ) + .await; + + let pairing_request = accept_http_request(&listener).await; + assert_eq!( + pairing_request.request_line, + "POST /backend-api/wham/remote/control/server/pair HTTP/1.1" + ); + assert_eq!( + pairing_request.headers.get("authorization"), + Some(&format!( + "Bearer {TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN}" + )) + ); + assert_eq!( + serde_json::from_str::(&pairing_request.body) + .expect("pairing request body should deserialize"), + json!({ "manual_code": true }) + ); + respond_with_json( + pairing_request.stream, + json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "srv_e_test", + "environment_id": "env_test", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await; + }); + let remote_handle = remote_control_handle_with_current_enrollment( + &remote_control_url, + remote_control_auth_manager(), + ); + remote_handle + .current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_mut() + .expect("current enrollment should exist") + .expires_at = Some(OffsetDateTime::now_utc() + time::Duration::seconds(29)); + + let response = remote_handle + .start_pairing(RemoteControlPairingStartParams { manual_code: true }) + .await + .expect("pairing should use the current server before websocket connect"); + server_task.await.expect("server task should finish"); + + assert_eq!( + response, + RemoteControlPairingStartResponse { + pairing_code: "pairing-code".to_string(), + manual_pairing_code: Some("ABCD-EFGH".to_string()), + environment_id: "env_test".to_string(), + expires_at: 33_336_362_096, + } + ); +} + +#[tokio::test] +async fn remote_control_handle_refreshes_after_pairing_auth_failure() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let server_task = tokio::spawn(async move { + let stale_pairing_request = accept_http_request(&listener).await; + assert_eq!( + stale_pairing_request.request_line, + "POST /backend-api/wham/remote/control/server/pair HTTP/1.1" + ); + assert_eq!( + stale_pairing_request.headers.get("authorization"), + Some(&format!("Bearer {TEST_REMOTE_CONTROL_SERVER_TOKEN}")) + ); + respond_with_status(stale_pairing_request.stream, "401 Unauthorized", "").await; + + let refresh_request = accept_http_request(&listener).await; + assert_eq!( + refresh_request.request_line, + "POST /backend-api/wham/remote/control/server/refresh HTTP/1.1" + ); + assert_eq!( + refresh_request.headers.get("authorization"), + Some(&"Bearer Access Token".to_string()) + ); + respond_with_json( + refresh_request.stream, + remote_control_server_token_response( + "srv_e_test", + "env_test", + TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN, + ), + ) + .await; + + let refreshed_pairing_request = accept_http_request(&listener).await; + assert_eq!( + refreshed_pairing_request.request_line, + "POST /backend-api/wham/remote/control/server/pair HTTP/1.1" + ); + assert_eq!( + refreshed_pairing_request.headers.get("authorization"), + Some(&format!( + "Bearer {TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN}" + )) + ); + respond_with_json( + refreshed_pairing_request.stream, + json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "srv_e_test", + "environment_id": "env_test", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await; + }); + let remote_handle = remote_control_handle_with_current_enrollment( + &remote_control_url, + remote_control_auth_manager(), + ); + + let response = remote_handle + .start_pairing(RemoteControlPairingStartParams::default()) + .await + .expect("pairing should refresh after server token auth failure"); + server_task.await.expect("server task should finish"); + + assert_eq!( + response, + RemoteControlPairingStartResponse { + pairing_code: "pairing-code".to_string(), + manual_pairing_code: Some("ABCD-EFGH".to_string()), + environment_id: "env_test".to_string(), + expires_at: 33_336_362_096, + } + ); +} + +#[tokio::test] +async fn remote_control_handle_recovers_auth_before_refreshing_pairing() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let server_task = tokio::spawn(async move { + let stale_refresh_request = accept_http_request(&listener).await; + assert_eq!( + stale_refresh_request.request_line, + "POST /backend-api/wham/remote/control/server/refresh HTTP/1.1" + ); + assert_eq!( + stale_refresh_request.headers.get("authorization"), + Some(&"Bearer stale-token".to_string()) + ); + respond_with_status(stale_refresh_request.stream, "401 Unauthorized", "").await; + + let recovered_refresh_request = accept_http_request(&listener).await; + assert_eq!( + recovered_refresh_request.request_line, + "POST /backend-api/wham/remote/control/server/refresh HTTP/1.1" + ); + assert_eq!( + recovered_refresh_request.headers.get("authorization"), + Some(&"Bearer fresh-token".to_string()) + ); + respond_with_json( + recovered_refresh_request.stream, + remote_control_server_token_response( + "srv_e_test", + "env_test", + TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN, + ), + ) + .await; + + let pairing_request = accept_http_request(&listener).await; + assert_eq!( + pairing_request.request_line, + "POST /backend-api/wham/remote/control/server/pair HTTP/1.1" + ); + assert_eq!( + pairing_request.headers.get("authorization"), + Some(&format!( + "Bearer {TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN}" + )) + ); + respond_with_json( + pairing_request.stream, + json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "srv_e_test", + "environment_id": "env_test", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await; + }); + let codex_home = TempDir::new().expect("temp dir should create"); + let mut stale_auth = remote_control_auth_dot_json(Some("account_id")); + stale_auth + .tokens + .as_mut() + .expect("stale auth should include tokens") + .access_token = "stale-token".to_string(); + save_auth( + codex_home.path(), + &stale_auth, + AuthCredentialsStoreMode::File, + ) + .expect("stale auth should save"); + let auth_manager = AuthManager::shared( + codex_home.path().to_path_buf(), + /*enable_codex_api_key_env*/ false, + AuthCredentialsStoreMode::File, + /*chatgpt_base_url*/ None, + ) + .await; + let mut fresh_auth = remote_control_auth_dot_json(Some("account_id")); + fresh_auth + .tokens + .as_mut() + .expect("fresh auth should include tokens") + .access_token = "fresh-token".to_string(); + save_auth( + codex_home.path(), + &fresh_auth, + AuthCredentialsStoreMode::File, + ) + .expect("fresh auth should save"); + let remote_handle = + remote_control_handle_with_current_enrollment(&remote_control_url, auth_manager); + remote_handle + .current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_mut() + .expect("current enrollment should exist") + .expires_at = Some(OffsetDateTime::now_utc() + time::Duration::seconds(29)); + + let response = remote_handle + .start_pairing(RemoteControlPairingStartParams::default()) + .await + .expect("pairing should refresh after auth recovery"); + server_task.await.expect("server task should finish"); + + assert_eq!( + response, + RemoteControlPairingStartResponse { + pairing_code: "pairing-code".to_string(), + manual_pairing_code: Some("ABCD-EFGH".to_string()), + environment_id: "env_test".to_string(), + expires_at: 33_336_362_096, + } + ); +} + +#[tokio::test] +async fn start_remote_control_pairing_preserves_backend_error_context() { + let (err, expected_pair_url) = + pairing_error("503 Service Unavailable", "pairing unavailable").await; + + assert_eq!( + err, + format!( + "remote control pairing failed at `{expected_pair_url}`: HTTP 503 Service Unavailable, request-id: request-123, cf-ray: ray-123, body: pairing unavailable" + ) + ); +} + +#[tokio::test] +async fn start_remote_control_pairing_preserves_decode_error_context() { + let (err, expected_pair_url) = pairing_error("200 OK", "{").await; + assert!(err.contains(&format!( + "failed to parse remote control pairing response from `{expected_pair_url}`: HTTP 200 OK" + ))); + assert!(err.contains("request-id: request-123")); + assert!(err.contains("cf-ray: ray-123")); + assert!(err.contains("body: {")); + assert!(err.contains("decode error:")); +} + +#[tokio::test] +async fn start_remote_control_pairing_rejects_mismatched_backend_enrollment() { + assert_eq!( + pairing_response_error(json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "other-server-id", + "environment_id": "other-environment-id", + "expires_at": "3026-05-22T12:34:56Z", + })) + .await, + "remote control pairing returned mismatched enrollment: expected server_id=server-id, environment_id=environment-id; got server_id=other-server-id, environment_id=other-environment-id" + ); +} + +#[tokio::test] +async fn start_remote_control_pairing_preserves_expiry_parse_error_context() { + let err = pairing_response_error(json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "server-id", + "environment_id": "environment-id", + "expires_at": "not-a-timestamp", + })) + .await; + + assert!(err.contains("failed to parse remote control pairing response")); + assert!(err.contains("HTTP 200 OK")); + assert!(err.contains("request-id: ")); + assert!(err.contains("cf-ray: ")); + assert!(err.contains("\"expires_at\":\"not-a-timestamp\"")); + assert!(err.contains("expires_at parse error:")); +} + +#[tokio::test] +async fn remote_control_handle_disable_clears_current_enrollment() { + let remote_handle = remote_control_handle_with_current_enrollment( + TEST_REMOTE_CONTROL_URL, + remote_control_auth_manager(), + ); + + remote_handle.disable(); + remote_handle.enable().expect("enable should succeed"); + assert_eq!( + remote_handle + .start_pairing(RemoteControlPairingStartParams::default()) + .await + .expect_err("re-enabled remote control should wait for a current server") + .to_string(), + "remote control pairing is unavailable until enrollment completes" + ); +} + +#[tokio::test] +async fn remote_control_handle_discards_pairing_response_after_auth_change() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let codex_home = TempDir::new().expect("temp dir should create"); + save_auth( + codex_home.path(), + &remote_control_auth_dot_json(Some("account_id")), + AuthCredentialsStoreMode::File, + ) + .expect("initial auth should save"); + let auth_manager = AuthManager::shared( + codex_home.path().to_path_buf(), + /*enable_codex_api_key_env*/ false, + AuthCredentialsStoreMode::File, + /*chatgpt_base_url*/ None, + ) + .await; + let remote_handle = + remote_control_handle_with_current_enrollment(&remote_control_url, auth_manager.clone()); + let pairing_task = tokio::spawn({ + let remote_handle = remote_handle.clone(); + async move { + remote_handle + .start_pairing(RemoteControlPairingStartParams::default()) + .await + } + }); + + let pairing_request = accept_http_request(&listener).await; + save_auth( + codex_home.path(), + &remote_control_auth_dot_json(Some("next_account_id")), + AuthCredentialsStoreMode::File, + ) + .expect("next auth should save"); + auth_manager.reload().await; + respond_with_json( + pairing_request.stream, + json!({ + "pairing_code": "stale-pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "srv_e_test", + "environment_id": "env_test", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await; + + assert_eq!( + pairing_task + .await + .expect("pairing task should join") + .expect_err("stale pairing response should be discarded") + .to_string(), + "remote control pairing is unavailable until enrollment completes" + ); +} diff --git a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs index 21710b5e2..5ec4f3d04 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs @@ -1,3 +1,17 @@ +use super::CurrentRemoteControlEnrollment; +use super::clear_current_enrollment; +use super::protocol::ClientEnvelope; +use super::protocol::ClientEvent; +use super::protocol::ClientId; +use super::protocol::RemoteControlTarget; +use super::protocol::ServerEnvelope; +use super::protocol::StreamId; +use super::publish_current_enrollment; +use super::remote_control_status_with_connection_status; +use super::segment::ClientSegmentObservation; +use super::segment::ClientSegmentReassembler; +use super::segment::REMOTE_CONTROL_SEGMENT_MAX_BYTES; +use super::segment::split_server_envelope_for_transport; use crate::transport::TransportEvent; use crate::transport::remote_control::client_tracker::ClientTracker; use crate::transport::remote_control::client_tracker::REMOTE_CONTROL_IDLE_SWEEP_INTERVAL; @@ -9,18 +23,6 @@ use crate::transport::remote_control::enroll::load_persisted_remote_control_enro use crate::transport::remote_control::enroll::preview_remote_control_response_body; use crate::transport::remote_control::enroll::refresh_remote_control_server; use crate::transport::remote_control::enroll::update_persisted_remote_control_enrollment; - -use super::protocol::ClientEnvelope; -use super::protocol::ClientEvent; -use super::protocol::ClientId; -use super::protocol::RemoteControlTarget; -use super::protocol::ServerEnvelope; -use super::protocol::StreamId; -use super::remote_control_status_with_connection_status; -use super::segment::ClientSegmentObservation; -use super::segment::ClientSegmentReassembler; -use super::segment::REMOTE_CONTROL_SEGMENT_MAX_BYTES; -use super::segment::split_server_envelope_for_transport; use axum::http::HeaderValue; use base64::Engine; use codex_app_server_protocol::RemoteControlConnectionStatus; @@ -251,6 +253,7 @@ pub(crate) struct RemoteControlWebsocket { enrollment: Option, auth_recovery: UnauthorizedRecovery, auth_change_rx: watch::Receiver, + current_enrollment: CurrentRemoteControlEnrollment, client_tracker: Arc>, state: Arc>, server_event_rx: Arc>>, @@ -288,6 +291,7 @@ enum ConnectionEndReason { pub(super) struct RemoteControlChannels { pub(super) transport_event_tx: mpsc::Sender, pub(super) status_publisher: RemoteControlStatusPublisher, + pub(super) current_enrollment: CurrentRemoteControlEnrollment, } #[derive(Clone)] @@ -404,6 +408,7 @@ impl RemoteControlWebsocket { enrollment: None, auth_recovery, auth_change_rx, + current_enrollment: channels.current_enrollment, client_tracker: Arc::new(Mutex::new(client_tracker)), state: Arc::new(Mutex::new(WebsocketState { outbound_buffer, @@ -611,6 +616,7 @@ impl RemoteControlWebsocket { &mut self.enrollment, connect_options, &self.status_publisher, + &self.current_enrollment, ) => connect_result, }; @@ -1229,6 +1235,7 @@ pub(super) async fn connect_remote_control_websocket( enrollment: &mut Option, connect_options: RemoteControlConnectOptions<'_>, status_publisher: &RemoteControlStatusPublisher, + current_enrollment: &CurrentRemoteControlEnrollment, ) -> io::Result<( WebSocketStream>, tungstenite::http::Response<()>, @@ -1237,6 +1244,7 @@ pub(super) async fn connect_remote_control_websocket( let Some(state_db) = state_db else { *enrollment = None; + clear_current_enrollment(current_enrollment); return Err(io::Error::new( ErrorKind::NotFound, "remote control requires sqlite state db", @@ -1249,6 +1257,7 @@ pub(super) async fn connect_remote_control_websocket( if err.kind() == ErrorKind::PermissionDenied { *enrollment = None; status_publisher.publish_environment_id(/*environment_id*/ None); + clear_current_enrollment(current_enrollment); } return Err(err); } @@ -1265,6 +1274,10 @@ pub(super) async fn connect_remote_control_websocket( ); *enrollment = None; status_publisher.publish_environment_id(/*environment_id*/ None); + clear_current_enrollment(current_enrollment); + } + if let Some(enrollment) = enrollment.as_mut() { + enrollment.remote_control_target = remote_control_target.clone(); } if let Some(enrollment) = enrollment.as_ref() { @@ -1321,13 +1334,8 @@ pub(super) async fn connect_remote_control_websocket( let enrollment_ref = enrollment.as_mut().ok_or_else(|| { io::Error::other("missing remote control enrollment before server refresh") })?; - match refresh_remote_control_server( - remote_control_target, - &auth, - connect_options.installation_id, - enrollment_ref, - ) - .await + match refresh_remote_control_server(&auth, connect_options.installation_id, enrollment_ref) + .await { Ok(()) => {} Err(err) if err.kind() == ErrorKind::NotFound => { @@ -1342,6 +1350,7 @@ pub(super) async fn connect_remote_control_websocket( connect_options.app_server_client_name, enrollment, status_publisher, + current_enrollment, ) .await; enroll_remote_control_server_if_missing( @@ -1374,6 +1383,7 @@ pub(super) async fn connect_remote_control_websocket( let enrollment_ref = enrollment.as_ref().ok_or_else(|| { io::Error::other("missing remote control enrollment after enrollment step") })?; + publish_current_enrollment(current_enrollment, enrollment_ref); let request = build_remote_control_websocket_request( &remote_control_target.websocket_url, enrollment_ref, @@ -1415,6 +1425,7 @@ pub(super) async fn connect_remote_control_websocket( connect_options.app_server_client_name, enrollment, status_publisher, + current_enrollment, ) .await; } @@ -1429,6 +1440,7 @@ pub(super) async fn connect_remote_control_websocket( ) })? .clear_server_token(); + clear_current_enrollment(current_enrollment); return Err(io::Error::other(format!( "remote control websocket auth failed with HTTP {}; refreshing server token before reconnect", response.status() @@ -1453,6 +1465,7 @@ async fn clear_remote_control_enrollment( app_server_client_name: Option<&str>, enrollment: &mut Option, status_publisher: &RemoteControlStatusPublisher, + current_enrollment: &CurrentRemoteControlEnrollment, ) { if let Err(clear_err) = update_persisted_remote_control_enrollment( Some(state_db), @@ -1467,6 +1480,7 @@ async fn clear_remote_control_enrollment( } *enrollment = None; status_publisher.publish_environment_id(/*environment_id*/ None); + clear_current_enrollment(current_enrollment); } async fn enroll_remote_control_server_if_missing( @@ -1534,7 +1548,7 @@ async fn enroll_remote_control_server_if_missing( Ok(()) } -async fn recover_remote_control_auth( +pub(super) async fn recover_remote_control_auth( auth_recovery: &mut UnauthorizedRecovery, auth_change_rx: &mut watch::Receiver, ) -> bool { @@ -1647,6 +1661,8 @@ mod tests { fn remote_control_enrollment(remote_control_token: Option<&str>) -> RemoteControlEnrollment { RemoteControlEnrollment { + remote_control_target: normalize_remote_control_url("http://localhost/backend-api/") + .expect("target should normalize"), account_id: "account_id".to_string(), environment_id: "env_test".to_string(), server_id: "srv_e_test".to_string(), @@ -1657,6 +1673,10 @@ mod tests { } } + fn test_current_enrollment() -> CurrentRemoteControlEnrollment { + Arc::new(std::sync::Mutex::new(None)) + } + #[test] fn next_reconnect_delay_resets_after_cap() { let mut reconnect_attempt = 9; @@ -1810,6 +1830,7 @@ mod tests { let mut enrollment = Some(remote_control_enrollment(Some( TEST_REMOTE_CONTROL_SERVER_TOKEN, ))); + let current_enrollment = test_current_enrollment(); let (status_publisher, status_rx) = remote_control_status_channel(); let err = match connect_remote_control_websocket( @@ -1828,6 +1849,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + ¤t_enrollment, ) .await { @@ -1837,6 +1859,12 @@ mod tests { server_task.await.expect("server task should succeed"); assert_eq!(err.to_string(), expected_error); + assert!( + current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_some() + ); assert_eq!( status_rx.borrow().clone(), RemoteControlStatusChangedNotification { @@ -1864,6 +1892,7 @@ mod tests { let mut enrollment = Some(remote_control_enrollment(Some( TEST_REMOTE_CONTROL_SERVER_TOKEN, ))); + let current_enrollment = test_current_enrollment(); let (status_publisher, status_rx) = remote_control_status_channel(); let server_task = tokio::spawn(async move { @@ -1891,6 +1920,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + ¤t_enrollment, ) .await .expect_err("unauthorized response should fail the websocket connect"); @@ -1909,11 +1939,14 @@ mod tests { err.to_string(), "remote control websocket auth failed with HTTP 401 Unauthorized; refreshing server token before reconnect" ); - assert_eq!( - enrollment, - Some(remote_control_enrollment( - /*remote_control_token*/ None - )) + let mut expected_enrollment = remote_control_enrollment(/*remote_control_token*/ None); + expected_enrollment.remote_control_target = remote_control_target; + assert_eq!(enrollment, Some(expected_enrollment)); + assert!( + current_enrollment + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_none() ); } @@ -1976,6 +2009,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + &test_current_enrollment(), ) .await .expect_err("unauthorized enrollment should fail the websocket connect"); @@ -2070,6 +2104,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + &test_current_enrollment(), ) .await .expect_err("unauthorized refresh should fail the websocket connect"); @@ -2135,6 +2170,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + &test_current_enrollment(), ) .await .expect_err("missing sqlite state db should fail remote control"); @@ -2185,6 +2221,7 @@ mod tests { app_server_client_name: None, }, &status_publisher, + &test_current_enrollment(), ) .await .expect_err("missing auth should fail remote control"); @@ -2236,6 +2273,7 @@ mod tests { RemoteControlChannels { transport_event_tx, status_publisher, + current_enrollment: test_current_enrollment(), }, shutdown_token, enabled_rx, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 29acc5198..eb6e4afd8 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -210,6 +210,7 @@ Example with notification opt-out: - `remoteControl/enable` — experimental; enable remote control for the current app-server process and return the current remote-control status snapshot. The caller is responsible for persisting the desired setting outside app-server. - `remoteControl/disable` — experimental; disable remote control for the current app-server process and return the current remote-control status snapshot. This does not revoke already enrolled controller devices. - `remoteControl/status/read` — experimental; read the current remote-control status snapshot. `status` is one of `disabled`, `connecting`, `connected`, or `errored`; `serverName` is the local machine name used by this app-server process; `environmentId` is a string when the app-server has a current enrollment and `null` when that enrollment is cleared, invalidated, or remote control is disabled. +- `remoteControl/pairing/start` — experimental; start a short-lived remote-control pairing artifact for the current app-server process. Pass `manualCode: true` to also request a manual pairing code. Returns `pairingCode`, `manualPairingCode`, `environmentId`, and Unix-seconds `expiresAt`; app-server intentionally does not expose the backend `serverId`. - `remoteControl/status/changed` — notification emitted when the remote-control status or client-visible environment id changes. `status` is one of `disabled`, `connecting`, `connected`, or `errored`; `serverName` is the local machine name used by this app-server process; `environmentId` is a string when the app-server has a current enrollment and `null` when that enrollment is cleared, invalidated, or remote control is disabled. Newly initialized app-server clients always receive the current status snapshot. - `skills/config/write` — write user-level skill config by name or absolute path. - `plugin/install` — install a plugin from a discovered marketplace entry, rejecting marketplace entries marked unavailable for install, install MCPs if any, and return the effective plugin auth policy plus any apps that still need auth (**under development; do not call from production clients yet**). diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 2dbd17a3d..7aa93ec86 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -917,6 +917,11 @@ impl MessageProcessor { .remote_control_processor .status_read() .map(|response| Some(response.into())), + ClientRequest::RemoteControlPairingStart { params, .. } => self + .remote_control_processor + .pairing_start(params) + .await + .map(|response| Some(response.into())), ClientRequest::ConfigRequirementsRead { params: _, .. } => self .config_processor .config_requirements_read() diff --git a/codex-rs/app-server/src/request_processors/remote_control_processor.rs b/codex-rs/app-server/src/request_processors/remote_control_processor.rs index 2fceeb503..7c279585f 100644 --- a/codex-rs/app-server/src/request_processors/remote_control_processor.rs +++ b/codex-rs/app-server/src/request_processors/remote_control_processor.rs @@ -5,7 +5,10 @@ use crate::transport::RemoteControlUnavailable; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RemoteControlDisableResponse; use codex_app_server_protocol::RemoteControlEnableResponse; +use codex_app_server_protocol::RemoteControlPairingStartParams; +use codex_app_server_protocol::RemoteControlPairingStartResponse; use codex_app_server_protocol::RemoteControlStatusReadResponse; +use std::io; #[derive(Clone)] pub(crate) struct RemoteControlRequestProcessor { @@ -42,6 +45,16 @@ impl RemoteControlRequestProcessor { }) } + pub(crate) async fn pairing_start( + &self, + params: RemoteControlPairingStartParams, + ) -> Result { + self.handle()? + .start_pairing(params) + .await + .map_err(map_pairing_start_error) + } + fn handle(&self) -> Result<&RemoteControlHandle, JSONRPCErrorError> { self.remote_control_handle .as_ref() @@ -52,3 +65,14 @@ impl RemoteControlRequestProcessor { fn map_unavailable(err: RemoteControlUnavailable) -> JSONRPCErrorError { invalid_request(err.to_string()) } + +fn map_pairing_start_error(err: io::Error) -> JSONRPCErrorError { + if err.kind() == io::ErrorKind::InvalidInput { + invalid_request(err.to_string()) + } else { + internal_error(err.to_string()) + } +} + +#[cfg(test)] +mod remote_control_processor_tests; diff --git a/codex-rs/app-server/src/request_processors/remote_control_processor/remote_control_processor_tests.rs b/codex-rs/app-server/src/request_processors/remote_control_processor/remote_control_processor_tests.rs new file mode 100644 index 000000000..8a1db50a3 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/remote_control_processor/remote_control_processor_tests.rs @@ -0,0 +1,48 @@ +use super::*; +use crate::error_code::INTERNAL_ERROR_CODE; +use crate::error_code::INVALID_REQUEST_ERROR_CODE; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn pairing_start_returns_internal_error_when_remote_control_is_unavailable() { + let err = RemoteControlRequestProcessor::new(/*remote_control_handle*/ None) + .pairing_start(RemoteControlPairingStartParams::default()) + .await + .expect_err("missing remote control should fail pairing"); + + assert_eq!( + err, + JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + data: None, + message: "remote control is unavailable for this app-server".to_string(), + } + ); +} + +#[test] +fn pairing_start_maps_invalid_input_to_invalid_request() { + assert_eq!( + map_pairing_start_error(io::Error::new( + io::ErrorKind::InvalidInput, + "remote control pairing is unavailable", + )), + JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + data: None, + message: "remote control pairing is unavailable".to_string(), + } + ); +} + +#[test] +fn pairing_start_maps_backend_failures_to_internal_error() { + assert_eq!( + map_pairing_start_error(io::Error::other("remote control pairing failed")), + JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + data: None, + message: "remote control pairing failed".to_string(), + } + ); +} diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index 3a3787d23..88378322a 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -67,6 +67,7 @@ use codex_app_server_protocol::ProcessKillParams; use codex_app_server_protocol::ProcessResizePtyParams; use codex_app_server_protocol::ProcessSpawnParams; use codex_app_server_protocol::ProcessWriteStdinParams; +use codex_app_server_protocol::RemoteControlPairingStartParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ReviewStartParams; use codex_app_server_protocol::SendAddCreditsNudgeEmailParams; @@ -643,6 +644,16 @@ impl TestAppServer { .await } + /// Send a `remoteControl/pairing/start` JSON-RPC request. + pub async fn send_remote_control_pairing_start_request( + &mut self, + params: RemoteControlPairingStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("remoteControl/pairing/start", params) + .await + } + /// Send an `app/list` JSON-RPC request. pub async fn send_apps_list_request(&mut self, params: AppsListParams) -> anyhow::Result { let params = Some(serde_json::to_value(params)?); diff --git a/codex-rs/app-server/tests/suite/v2/remote_control.rs b/codex-rs/app-server/tests/suite/v2/remote_control.rs index d15f2d418..86187cfc4 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_control.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_control.rs @@ -11,12 +11,15 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RemoteControlConnectionStatus; use codex_app_server_protocol::RemoteControlDisableResponse; use codex_app_server_protocol::RemoteControlEnableResponse; +use codex_app_server_protocol::RemoteControlPairingStartParams; +use codex_app_server_protocol::RemoteControlPairingStartResponse; use codex_app_server_protocol::RemoteControlStatusReadResponse; use codex_app_server_protocol::RequestId; use codex_config::types::AuthCredentialsStoreMode; use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncWriteExt; use tokio::io::BufReader; use tokio::net::TcpListener; use tokio::net::TcpStream; @@ -125,6 +128,65 @@ async fn remote_control_status_read_returns_connecting_status_after_enable() -> Ok(()) } +#[tokio::test] +async fn remote_control_pairing_start_returns_pairing_artifacts() -> Result<()> { + let codex_home = TempDir::new()?; + let mut backend = PairingRemoteControlBackend::start(codex_home.path()).await?; + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp.send_remote_control_enable_request().await?; + let _: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!( + timeout(DEFAULT_TIMEOUT, backend.wait_for_enroll_request()).await??, + "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1" + ); + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_matching_notification( + "remoteControl/status/changed enrolled", + |notification| { + notification.method == "remoteControl/status/changed" + && notification + .params + .as_ref() + .and_then(|params| params.get("environmentId")) + .and_then(serde_json::Value::as_str) + == Some("environment-id") + }, + ), + ) + .await??; + + let request_id = mcp + .send_remote_control_pairing_start_request(RemoteControlPairingStartParams { + manual_code: true, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(response.result.get("serverId"), None); + let received: RemoteControlPairingStartResponse = to_response(response)?; + + assert_eq!( + received, + RemoteControlPairingStartResponse { + pairing_code: "pairing-code".to_string(), + manual_pairing_code: Some("ABCD-EFGH".to_string()), + environment_id: "environment-id".to_string(), + expires_at: 33_336_362_096, + } + ); + Ok(()) +} + struct BlockingRemoteControlBackend { enroll_request_rx: Option>>, server_task: JoinHandle<()>, @@ -132,20 +194,7 @@ struct BlockingRemoteControlBackend { impl BlockingRemoteControlBackend { async fn start(codex_home: &std::path::Path) -> Result { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let remote_control_url = format!("http://{}/backend-api/", listener.local_addr()?); - write_mock_responses_config_toml_with_chatgpt_base_url( - codex_home, - &remote_control_url, - &remote_control_url, - )?; - write_chatgpt_auth( - codex_home, - ChatGptAuthFixture::new("chatgpt-token") - .account_id("account_id") - .chatgpt_account_id("account_id"), - AuthCredentialsStoreMode::File, - )?; + let listener = configured_remote_control_listener(codex_home).await?; let (enroll_request_tx, enroll_request_rx) = oneshot::channel(); let server_task = tokio::spawn(async move { @@ -175,19 +224,119 @@ impl BlockingRemoteControlBackend { } } +struct PairingRemoteControlBackend { + enroll_request_rx: Option>>, + server_task: JoinHandle<()>, +} + +impl PairingRemoteControlBackend { + async fn start(codex_home: &std::path::Path) -> Result { + let listener = configured_remote_control_listener(codex_home).await?; + let (enroll_request_tx, enroll_request_rx) = oneshot::channel(); + let server_task = tokio::spawn(async move { + let mut enroll_request_tx = Some(enroll_request_tx); + let result = async { + let enroll_request = read_http_request(&listener).await?; + if let Some(enroll_request_tx) = enroll_request_tx.take() { + let _ = enroll_request_tx.send(Ok(enroll_request.request_line.clone())); + } + respond_with_json( + enroll_request.reader.into_inner(), + serde_json::json!({ + "server_id": "server-id", + "environment_id": "environment-id", + "remote_control_token": "remote-control-token", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await?; + + let _websocket_request = read_http_request(&listener).await?; + let pair_http_request = read_http_request(&listener).await?; + respond_with_json( + pair_http_request.reader.into_inner(), + serde_json::json!({ + "pairing_code": "pairing-code", + "manual_pairing_code": "ABCD-EFGH", + "server_id": "server-id", + "environment_id": "environment-id", + "expires_at": "3026-05-22T12:34:56Z", + }), + ) + .await?; + std::future::pending::<()>().await; + Ok::<(), anyhow::Error>(()) + } + .await; + + if let Err(err) = result { + let err = err.to_string(); + if let Some(enroll_request_tx) = enroll_request_tx { + let _ = enroll_request_tx.send(Err(anyhow::anyhow!(err))); + } + } + }); + + Ok(Self { + enroll_request_rx: Some(enroll_request_rx), + server_task, + }) + } + + async fn wait_for_enroll_request(&mut self) -> Result { + self.enroll_request_rx + .take() + .context("enroll request should only be awaited once")? + .await? + } +} + +impl Drop for PairingRemoteControlBackend { + fn drop(&mut self) { + self.server_task.abort(); + } +} + impl Drop for BlockingRemoteControlBackend { fn drop(&mut self) { self.server_task.abort(); } } +struct HttpRequest { + request_line: String, + reader: BufReader, +} + +async fn configured_remote_control_listener(codex_home: &std::path::Path) -> Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let remote_control_url = format!("http://{}/backend-api/", listener.local_addr()?); + write_mock_responses_config_toml_with_chatgpt_base_url( + codex_home, + &remote_control_url, + &remote_control_url, + )?; + write_chatgpt_auth( + codex_home, + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account_id") + .chatgpt_account_id("account_id"), + AuthCredentialsStoreMode::File, + )?; + Ok(listener) +} + async fn read_enroll_request(listener: TcpListener) -> Result<(String, BufReader)> { + let request = read_http_request(&listener).await?; + Ok((request.request_line, request.reader)) +} + +async fn read_http_request(listener: &TcpListener) -> Result { let (stream, _) = listener.accept().await?; let mut reader = BufReader::new(stream); let mut request_line = String::new(); reader.read_line(&mut request_line).await?; - loop { let mut line = String::new(); reader.read_line(&mut line).await?; @@ -196,5 +345,23 @@ async fn read_enroll_request(listener: TcpListener) -> Result<(String, BufReader } } - Ok((request_line.trim_end().to_string(), reader)) + Ok(HttpRequest { + request_line: request_line.trim_end().to_string(), + reader, + }) +} + +async fn respond_with_json(stream: TcpStream, body: serde_json::Value) -> Result<()> { + let body = body.to_string(); + let mut stream = stream; + stream + .write_all( + format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await?; + Ok(()) }