agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(remote-control): add pairing start (#25675)

Browse Source 
## 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`
This commit is contained in:
Anton Panasenko
2026-06-01 18:05:50 -07:00
committed by GitHub
Unverified
parent 1ad0d7aa4b
commit 0002316687
15 changed files with 1302 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/export.rs
+19
View File
@@ -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(())
}
}
codex-rs/app-server-protocol/src/protocol/common.rs
+17
View File
@@ -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]
codex-rs/app-server-protocol/src/protocol/v2/remote_control.rs
+18
View File
@@ -44,6 +44,24 @@ pub struct RemoteControlStatusReadResponse {
pub environment_id: Option<String>,
}
#[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<String>,
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/")]
codex-rs/app-server-transport/src/transport/remote_control/enroll.rs
+121 -7
View File
@@ -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<RemoteControlPairingStartResponse> {
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::<StartRemoteControlPairingResponse>(&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("<redacted>".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("<redacted>".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::<serde_json::Value>(&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": "<redacted>",
"pairing_code": "<redacted>",
"manual_pairing_code": "<redacted>",
})
);
}
@@ -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(),
codex-rs/app-server-transport/src/transport/remote_control/mod.rs
+221
View File
@@ -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<watch::Sender<bool>>,
status_tx: Arc<watch::Sender<RemoteControlStatusChangedNotification>>,
state_db_available: bool,
current_enrollment: CurrentRemoteControlEnrollment,
auth_manager: Arc<AuthManager>,
}
type CurrentRemoteControlEnrollment = Arc<StdMutex<Option<RemoteControlEnrollment>>>;
#[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<RemoteControlPairingStartResponse> {
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<AuthManager>,
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<Arc<StateRuntime>>,
@@ -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,
},
))
}
codex-rs/app-server-transport/src/transport/remote_control/protocol.rs
+28
View File
@@ -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<String>,
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(),
}
);
}
codex-rs/app-server-transport/src/transport/remote_control/tests.rs
+49
View File
@@ -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<AuthManager>,
) -> 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(),
codex-rs/app-server-transport/src/transport/remote_control/tests/pairing_tests.rs
+494
View File
@@ -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::<serde_json::Value>(&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::<serde_json::Value>(&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: <none>"));
assert!(err.contains("cf-ray: <none>"));
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"
);
}
codex-rs/app-server-transport/src/transport/remote_control/websocket.rs
+63 -25
View File
@@ -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<RemoteControlEnrollment>,
auth_recovery: UnauthorizedRecovery,
auth_change_rx: watch::Receiver<u64>,
current_enrollment: CurrentRemoteControlEnrollment,
client_tracker: Arc<Mutex<ClientTracker>>,
state: Arc<Mutex<WebsocketState>>,
server_event_rx: Arc<Mutex<mpsc::Receiver<super::QueuedServerEnvelope>>>,
@@ -288,6 +291,7 @@ enum ConnectionEndReason {
pub(super) struct RemoteControlChannels {
pub(super) transport_event_tx: mpsc::Sender<TransportEvent>,
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<RemoteControlEnrollment>,
connect_options: RemoteControlConnectOptions<'_>,
status_publisher: &RemoteControlStatusPublisher,
current_enrollment: &CurrentRemoteControlEnrollment,
) -> io::Result<(
WebSocketStream<MaybeTlsStream<TcpStream>>,
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<RemoteControlEnrollment>,
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<u64>,
) -> 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,
&current_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,
&current_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,
codex-rs/app-server/README.md
+1
View File
@@ -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**).
codex-rs/app-server/src/message_processor.rs
+5
View File
@@ -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()
codex-rs/app-server/src/request_processors/remote_control_processor.rs
+24
View File
@@ -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<RemoteControlPairingStartResponse, JSONRPCErrorError> {
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;
codex-rs/app-server/src/request_processors/remote_control_processor/remote_control_processor_tests.rs
+48
View File
@@ -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(),
}
);
}
codex-rs/app-server/tests/common/test_app_server.rs
+11
View File
@@ -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<i64> {
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<i64> {
let params = Some(serde_json::to_value(params)?);
codex-rs/app-server/tests/suite/v2/remote_control.rs
+183 -16
View File
@@ -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<oneshot::Receiver<Result<String>>>,
server_task: JoinHandle<()>,
@@ -132,20 +194,7 @@ struct BlockingRemoteControlBackend {
impl BlockingRemoteControlBackend {
async fn start(codex_home: &std::path::Path) -> Result<Self> {
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<oneshot::Receiver<Result<String>>>,
server_task: JoinHandle<()>,
}
impl PairingRemoteControlBackend {
async fn start(codex_home: &std::path::Path) -> Result<Self> {
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<String> {
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<TcpStream>,
}
async fn configured_remote_control_listener(codex_home: &std::path::Path) -> Result<TcpListener> {
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<TcpStream>)> {
let request = read_http_request(&listener).await?;
Ok((request.request_line, request.reader))
}
async fn read_http_request(listener: &TcpListener) -> Result<HttpRequest> {
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(())
}