mirror of
https://github.com/openai/codex.git
synced 2026-11-13 06:45:07 +00:00
feat(app-server): add remote control client management RPCs (#25785)
## Why
Remote-control clients need to list and revoke controller-device grants
without enabling or enrolling the local relay. These are signed-in
account-management operations, so coupling them to websocket, pairing,
enrollment, or persisted relay state would prevent clients from managing
stale grants from the picker.
Related enhancement request: N/A. This adds the Codex app-server surface
for the planned upstream environment-scoped revoke endpoint.
## What Changed
- Added experimental app-server v2 RPCs:
- `remoteControl/client/list`
- `remoteControl/client/revoke`
- Added picker-oriented protocol types and standard generated schema
fixtures. The list response intentionally omits backend account id,
enrollment status, and location fields.
- Added `app-server-transport/src/transport/remote_control/clients.rs`
for environment-scoped GET and DELETE requests. It builds escaped URL
path segments, forwards optional pagination query fields, sends ChatGPT
auth plus `chatgpt-account-id`, converts RFC3339 `last_seen_at` values
to Unix seconds, accepts `204 No Content` revoke responses, and retries
once after a `401`.
- Extracted shared ChatGPT auth loading and recovery into
`app-server-transport/src/transport/remote_control/auth.rs` so
websocket, pairing, and client management use the same account-auth
boundary.
- Retained the configured remote-control base URL on
`RemoteControlHandle` and resolve management URLs lazily, preserving
deferred validation while relay startup is disabled.
- Registered list as `global_shared_read("remote-control-clients")` and
revoke as `global("remote-control-clients")`.
## Verification
- Added transport coverage proving list and revoke work while relay
state is disabled, IDs are escaped, picker-only fields are returned,
timestamps are converted, revoke accepts `204`, auth headers are
forwarded, `401` retries exactly once, `403` is not retried, and
malformed list payloads retain decode context.
- Added an app-server integration test proving both JSON-RPC methods
work before relay enablement and successful revoke returns `{}`.
- Regenerated and validated experimental and standard app-server schema
fixtures.
This commit is contained in:
1 parent
1fd2a6d328
commit
98a62a62ce
18 files changed
+1281
-130
No files matched your search
@@ -0,0 +1,105 @@
|
||||
use codex_api::SharedAuthProvider;
|
||||
use codex_login::AuthManager;
|
||||
use codex_login::UnauthorizedRecovery;
|
||||
use std::io;
|
||||
use std::io::ErrorKind;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::watch;
|
||||
use tracing::info;
|
||||
use tracing::warn;
|
||||
|
||||
pub(super) struct RemoteControlConnectionAuth {
|
||||
pub(super) auth_provider: SharedAuthProvider,
|
||||
pub(super) account_id: String,
|
||||
}
|
||||
|
||||
pub(super) async fn load_remote_control_auth(
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
) -> io::Result<RemoteControlConnectionAuth> {
|
||||
let mut reloaded = false;
|
||||
let auth = loop {
|
||||
let Some(auth) = auth_manager.auth().await else {
|
||||
if reloaded {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::PermissionDenied,
|
||||
"remote control requires ChatGPT authentication",
|
||||
));
|
||||
}
|
||||
auth_manager.reload().await;
|
||||
reloaded = true;
|
||||
continue;
|
||||
};
|
||||
if !auth.uses_codex_backend() {
|
||||
break auth;
|
||||
}
|
||||
if auth.get_account_id().is_none() && !reloaded {
|
||||
auth_manager.reload().await;
|
||||
reloaded = true;
|
||||
continue;
|
||||
}
|
||||
break auth;
|
||||
};
|
||||
|
||||
if !auth.uses_codex_backend() {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::PermissionDenied,
|
||||
"remote control requires ChatGPT authentication; API key auth is not supported",
|
||||
));
|
||||
}
|
||||
|
||||
Ok(RemoteControlConnectionAuth {
|
||||
auth_provider: codex_model_provider::auth_provider_from_auth(&auth),
|
||||
account_id: auth.get_account_id().ok_or_else(|| {
|
||||
io::Error::new(
|
||||
ErrorKind::WouldBlock,
|
||||
"remote control enrollment is waiting for a ChatGPT account id",
|
||||
)
|
||||
})?,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) async fn recover_remote_control_auth(
|
||||
auth_recovery: &mut UnauthorizedRecovery,
|
||||
auth_change_rx: &mut watch::Receiver<u64>,
|
||||
) -> bool {
|
||||
if !auth_recovery.has_next() {
|
||||
return false;
|
||||
}
|
||||
|
||||
let mode = auth_recovery.mode_name();
|
||||
let step = auth_recovery.step_name();
|
||||
let auth_change_revision_before_recovery = *auth_change_rx.borrow();
|
||||
match auth_recovery.next().await {
|
||||
Ok(step_result) => {
|
||||
if step_result.auth_state_changed() == Some(true) {
|
||||
mark_recovery_auth_change_seen(
|
||||
auth_change_rx,
|
||||
auth_change_revision_before_recovery,
|
||||
);
|
||||
}
|
||||
info!(
|
||||
"remote control auth recovery succeeded: mode={mode}, step={step}, auth_state_changed={:?}",
|
||||
step_result.auth_state_changed()
|
||||
);
|
||||
true
|
||||
}
|
||||
Err(err) => {
|
||||
warn!("remote control auth recovery failed: mode={mode}, step={step}: {err}");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn mark_recovery_auth_change_seen(
|
||||
auth_change_rx: &mut watch::Receiver<u64>,
|
||||
auth_change_revision_before_recovery: u64,
|
||||
) {
|
||||
let auth_change_revision_after_recovery = *auth_change_rx.borrow();
|
||||
if auth_change_revision_after_recovery == auth_change_revision_before_recovery.wrapping_add(1) {
|
||||
// Recovery updated the same watch that wakes the outer reconnect
|
||||
// loop. Mark only that single revision seen; if more revisions
|
||||
// arrived while recovery was in flight, leave them pending so the
|
||||
// reconnect loop still reacts to the later external auth change.
|
||||
auth_change_rx.borrow_and_update();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,307 @@
|
||||
use super::auth::RemoteControlConnectionAuth;
|
||||
use super::auth::load_remote_control_auth;
|
||||
use super::auth::recover_remote_control_auth;
|
||||
use super::enroll::REMOTE_CONTROL_ACCOUNT_ID_HEADER;
|
||||
use super::enroll::format_headers;
|
||||
use super::enroll::preview_remote_control_response_body;
|
||||
use super::protocol::normalize_remote_control_base_url;
|
||||
use axum::http::HeaderMap;
|
||||
use codex_app_server_protocol::RemoteControlClient;
|
||||
use codex_app_server_protocol::RemoteControlClientsListOrder;
|
||||
use codex_app_server_protocol::RemoteControlClientsListParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsListResponse;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeResponse;
|
||||
use codex_login::AuthManager;
|
||||
use codex_login::default_client::build_reqwest_client;
|
||||
use serde::Deserialize;
|
||||
use std::io;
|
||||
use std::io::ErrorKind;
|
||||
use std::sync::Arc;
|
||||
use time::OffsetDateTime;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use url::Url;
|
||||
|
||||
const REMOTE_CONTROL_CLIENT_MANAGEMENT_TIMEOUT: std::time::Duration =
|
||||
std::time::Duration::from_secs(30);
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ListRemoteControlClientsResponse {
|
||||
items: Vec<RemoteControlClientResponse>,
|
||||
#[serde(default)]
|
||||
cursor: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct RemoteControlClientResponse {
|
||||
client_id: String,
|
||||
#[serde(default)]
|
||||
display_name: Option<String>,
|
||||
#[serde(default)]
|
||||
device_type: Option<String>,
|
||||
#[serde(default)]
|
||||
platform: Option<String>,
|
||||
#[serde(default)]
|
||||
os_version: Option<String>,
|
||||
#[serde(default)]
|
||||
device_model: Option<String>,
|
||||
#[serde(default)]
|
||||
app_version: Option<String>,
|
||||
#[serde(default)]
|
||||
last_seen_at: Option<String>,
|
||||
}
|
||||
|
||||
enum ClientManagementRequest<'a> {
|
||||
List {
|
||||
url: &'a Url,
|
||||
params: &'a RemoteControlClientsListParams,
|
||||
},
|
||||
Revoke {
|
||||
url: &'a Url,
|
||||
},
|
||||
}
|
||||
|
||||
struct ClientManagementResponse {
|
||||
status: axum::http::StatusCode,
|
||||
headers: HeaderMap,
|
||||
body: Vec<u8>,
|
||||
}
|
||||
|
||||
pub(super) async fn list_remote_control_clients(
|
||||
remote_control_url: &str,
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
params: RemoteControlClientsListParams,
|
||||
) -> io::Result<RemoteControlClientsListResponse> {
|
||||
if params.environment_id.is_empty() {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control client list requires environmentId",
|
||||
));
|
||||
}
|
||||
if params
|
||||
.limit
|
||||
.is_some_and(|limit| !(1..=100).contains(&limit))
|
||||
{
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control client list limit must be between 1 and 100",
|
||||
));
|
||||
}
|
||||
let url = environment_clients_url(remote_control_url, ¶ms.environment_id)?;
|
||||
let response = send_client_management_request(
|
||||
auth_manager,
|
||||
ClientManagementRequest::List {
|
||||
url: &url,
|
||||
params: ¶ms,
|
||||
},
|
||||
"list remote control clients",
|
||||
)
|
||||
.await?;
|
||||
let ClientManagementResponse {
|
||||
status,
|
||||
headers,
|
||||
body,
|
||||
} = response;
|
||||
let body_preview = preview_remote_control_response_body(&body);
|
||||
ensure_success_response(status, &headers, &url, &body_preview, "client list")?;
|
||||
let response = serde_json::from_slice::<ListRemoteControlClientsResponse>(&body).map_err(
|
||||
|err| {
|
||||
io::Error::other(format!(
|
||||
"failed to parse remote control client list response from `{url}`: HTTP {status}, {}, body: {body_preview}, decode error: {err}",
|
||||
format_headers(&headers)
|
||||
))
|
||||
},
|
||||
)?;
|
||||
Ok(RemoteControlClientsListResponse {
|
||||
data: response
|
||||
.items
|
||||
.into_iter()
|
||||
.map(RemoteControlClient::try_from)
|
||||
.collect::<io::Result<_>>()?,
|
||||
next_cursor: response.cursor,
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) async fn revoke_remote_control_client(
|
||||
remote_control_url: &str,
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
params: RemoteControlClientsRevokeParams,
|
||||
) -> io::Result<RemoteControlClientsRevokeResponse> {
|
||||
if params.environment_id.is_empty() {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control client revoke requires environmentId",
|
||||
));
|
||||
}
|
||||
if params.client_id.is_empty() {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control client revoke requires clientId",
|
||||
));
|
||||
}
|
||||
let mut url = environment_clients_url(remote_control_url, ¶ms.environment_id)?;
|
||||
url.path_segments_mut()
|
||||
.map_err(|()| {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control URL cannot be a base",
|
||||
)
|
||||
})?
|
||||
.push(¶ms.client_id);
|
||||
let response = send_client_management_request(
|
||||
auth_manager,
|
||||
ClientManagementRequest::Revoke { url: &url },
|
||||
"revoke remote control client",
|
||||
)
|
||||
.await?;
|
||||
let ClientManagementResponse {
|
||||
status,
|
||||
headers,
|
||||
body,
|
||||
} = response;
|
||||
let body_preview = preview_remote_control_response_body(&body);
|
||||
ensure_success_response(status, &headers, &url, &body_preview, "client revoke")?;
|
||||
Ok(RemoteControlClientsRevokeResponse {})
|
||||
}
|
||||
|
||||
async fn send_client_management_request(
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
request: ClientManagementRequest<'_>,
|
||||
action: &str,
|
||||
) -> io::Result<ClientManagementResponse> {
|
||||
let mut auth_recovery = auth_manager.unauthorized_recovery();
|
||||
let mut auth_change_rx = auth_manager.auth_change_receiver();
|
||||
let auth = load_remote_control_auth(auth_manager).await?;
|
||||
let response = send_client_management_request_once(&auth, &request, action).await?;
|
||||
if response.status.as_u16() != 401
|
||||
|| !recover_remote_control_auth(&mut auth_recovery, &mut auth_change_rx).await
|
||||
{
|
||||
return Ok(response);
|
||||
}
|
||||
let auth = load_remote_control_auth(auth_manager).await?;
|
||||
send_client_management_request_once(&auth, &request, action).await
|
||||
}
|
||||
|
||||
async fn send_client_management_request_once(
|
||||
auth: &RemoteControlConnectionAuth,
|
||||
request: &ClientManagementRequest<'_>,
|
||||
action: &str,
|
||||
) -> io::Result<ClientManagementResponse> {
|
||||
let client = build_reqwest_client();
|
||||
let mut auth_headers = HeaderMap::new();
|
||||
auth.auth_provider.add_auth_headers(&mut auth_headers);
|
||||
let request = match request {
|
||||
ClientManagementRequest::List { url, params } => {
|
||||
let mut query = Vec::new();
|
||||
if let Some(cursor) = ¶ms.cursor {
|
||||
query.push(("cursor", cursor.clone()));
|
||||
}
|
||||
if let Some(limit) = params.limit {
|
||||
query.push(("limit", limit.to_string()));
|
||||
}
|
||||
if let Some(order) = params.order {
|
||||
query.push((
|
||||
"order",
|
||||
match order {
|
||||
RemoteControlClientsListOrder::Asc => "asc",
|
||||
RemoteControlClientsListOrder::Desc => "desc",
|
||||
}
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
client.get((*url).clone()).query(&query)
|
||||
}
|
||||
ClientManagementRequest::Revoke { url } => client.delete((*url).clone()),
|
||||
};
|
||||
let response = request
|
||||
.timeout(REMOTE_CONTROL_CLIENT_MANAGEMENT_TIMEOUT)
|
||||
.headers(auth_headers)
|
||||
.header(REMOTE_CONTROL_ACCOUNT_ID_HEADER, &auth.account_id)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| io::Error::other(format!("failed to {action}: {err}")))?;
|
||||
let headers = response.headers().clone();
|
||||
let status = response.status();
|
||||
let body = response
|
||||
.bytes()
|
||||
.await
|
||||
.map_err(|err| io::Error::other(format!("failed to read {action} response: {err}")))?
|
||||
.to_vec();
|
||||
Ok(ClientManagementResponse {
|
||||
status,
|
||||
headers,
|
||||
body,
|
||||
})
|
||||
}
|
||||
|
||||
fn ensure_success_response(
|
||||
status: axum::http::StatusCode,
|
||||
headers: &HeaderMap,
|
||||
url: &Url,
|
||||
body_preview: &str,
|
||||
response_kind: &str,
|
||||
) -> io::Result<()> {
|
||||
if status.is_success() {
|
||||
return Ok(());
|
||||
}
|
||||
let error_kind = match status.as_u16() {
|
||||
400 => ErrorKind::InvalidInput,
|
||||
401 | 403 => ErrorKind::PermissionDenied,
|
||||
404 => ErrorKind::NotFound,
|
||||
_ => ErrorKind::Other,
|
||||
};
|
||||
Err(io::Error::new(
|
||||
error_kind,
|
||||
format!(
|
||||
"remote control {response_kind} failed at `{url}`: HTTP {status}, {}, body: {body_preview}",
|
||||
format_headers(headers)
|
||||
),
|
||||
))
|
||||
}
|
||||
|
||||
fn environment_clients_url(remote_control_url: &str, environment_id: &str) -> io::Result<Url> {
|
||||
let mut url = normalize_remote_control_base_url(remote_control_url)?
|
||||
.join("wham/remote/control/environments")
|
||||
.map_err(io::Error::other)?;
|
||||
url.path_segments_mut()
|
||||
.map_err(|()| {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"remote control URL cannot be a base",
|
||||
)
|
||||
})?
|
||||
.push(environment_id)
|
||||
.push("clients");
|
||||
Ok(url)
|
||||
}
|
||||
|
||||
impl TryFrom<RemoteControlClientResponse> for RemoteControlClient {
|
||||
type Error = io::Error;
|
||||
|
||||
fn try_from(client: RemoteControlClientResponse) -> Result<Self, Self::Error> {
|
||||
Ok(Self {
|
||||
client_id: client.client_id,
|
||||
display_name: client.display_name,
|
||||
device_type: client.device_type,
|
||||
platform: client.platform,
|
||||
os_version: client.os_version,
|
||||
device_model: client.device_model,
|
||||
app_version: client.app_version,
|
||||
last_seen_at: client
|
||||
.last_seen_at
|
||||
.map(|last_seen_at| {
|
||||
OffsetDateTime::parse(&last_seen_at, &Rfc3339)
|
||||
.map(OffsetDateTime::unix_timestamp)
|
||||
.map_err(|err| {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidData,
|
||||
format!(
|
||||
"failed to parse remote control client last_seen_at `{last_seen_at}`: {err}"
|
||||
),
|
||||
)
|
||||
})
|
||||
})
|
||||
.transpose()?,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
use super::auth::RemoteControlConnectionAuth;
|
||||
use super::pairing_unavailable_error;
|
||||
use super::protocol::EnrollRemoteServerRequest;
|
||||
use super::protocol::EnrollRemoteServerResponse;
|
||||
@@ -6,7 +7,6 @@ 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;
|
||||
@@ -151,11 +151,6 @@ impl RemoteControlEnrollment {
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct RemoteControlConnectionAuth {
|
||||
pub(super) auth_provider: SharedAuthProvider,
|
||||
pub(super) account_id: String,
|
||||
}
|
||||
|
||||
pub(super) async fn load_persisted_remote_control_enrollment(
|
||||
state_db: Option<&StateRuntime>,
|
||||
remote_control_target: &RemoteControlTarget,
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
mod auth;
|
||||
mod client_tracker;
|
||||
mod clients;
|
||||
mod enroll;
|
||||
mod protocol;
|
||||
mod segment;
|
||||
mod websocket;
|
||||
|
||||
use self::auth::load_remote_control_auth;
|
||||
use self::auth::recover_remote_control_auth;
|
||||
use self::enroll::RemoteControlEnrollment;
|
||||
use self::enroll::refresh_remote_control_server;
|
||||
use crate::transport::remote_control::websocket::RemoteControlChannels;
|
||||
@@ -17,6 +21,10 @@ use self::protocol::normalize_remote_control_url;
|
||||
use super::CHANNEL_CAPACITY;
|
||||
use super::TransportEvent;
|
||||
use super::next_connection_id;
|
||||
use codex_app_server_protocol::RemoteControlClientsListParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsListResponse;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeResponse;
|
||||
use codex_app_server_protocol::RemoteControlConnectionStatus;
|
||||
use codex_app_server_protocol::RemoteControlPairingStartParams;
|
||||
use codex_app_server_protocol::RemoteControlPairingStartResponse;
|
||||
@@ -57,6 +65,7 @@ pub struct RemoteControlHandle {
|
||||
enabled_tx: Arc<watch::Sender<bool>>,
|
||||
status_tx: Arc<watch::Sender<RemoteControlStatusChangedNotification>>,
|
||||
state_db_available: bool,
|
||||
remote_control_url: String,
|
||||
current_enrollment: CurrentRemoteControlEnrollment,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
}
|
||||
@@ -146,7 +155,7 @@ impl RemoteControlHandle {
|
||||
if !*self.enabled_tx.borrow() {
|
||||
return Err(Self::pairing_disabled_error());
|
||||
}
|
||||
let mut auth = websocket::load_remote_control_auth(&self.auth_manager)
|
||||
let mut auth = load_remote_control_auth(&self.auth_manager)
|
||||
.await
|
||||
.map_err(|_| pairing_unavailable_error())?;
|
||||
let mut enrollment = {
|
||||
@@ -205,7 +214,7 @@ impl RemoteControlHandle {
|
||||
if !*self.enabled_tx.borrow() {
|
||||
return Err(Self::pairing_disabled_error());
|
||||
}
|
||||
let current_auth = websocket::load_remote_control_auth(&self.auth_manager)
|
||||
let current_auth = load_remote_control_auth(&self.auth_manager)
|
||||
.await
|
||||
.map_err(|_| pairing_unavailable_error())?;
|
||||
if current_auth.account_id != auth.account_id {
|
||||
@@ -214,6 +223,22 @@ impl RemoteControlHandle {
|
||||
pairing_response
|
||||
}
|
||||
|
||||
pub async fn list_clients(
|
||||
&self,
|
||||
params: RemoteControlClientsListParams,
|
||||
) -> io::Result<RemoteControlClientsListResponse> {
|
||||
clients::list_remote_control_clients(&self.remote_control_url, &self.auth_manager, params)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn revoke_client(
|
||||
&self,
|
||||
params: RemoteControlClientsRevokeParams,
|
||||
) -> io::Result<RemoteControlClientsRevokeResponse> {
|
||||
clients::revoke_remote_control_client(&self.remote_control_url, &self.auth_manager, params)
|
||||
.await
|
||||
}
|
||||
|
||||
fn pairing_disabled_error() -> io::Error {
|
||||
io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
@@ -255,7 +280,7 @@ impl RemoteControlHandle {
|
||||
async fn refresh_pairing_enrollment(
|
||||
current_enrollment: &CurrentRemoteControlEnrollment,
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
auth: &mut enroll::RemoteControlConnectionAuth,
|
||||
auth: &mut auth::RemoteControlConnectionAuth,
|
||||
installation_id: &str,
|
||||
enrollment: &mut RemoteControlEnrollment,
|
||||
) -> io::Result<()> {
|
||||
@@ -265,10 +290,10 @@ async fn refresh_pairing_enrollment(
|
||||
}
|
||||
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 {
|
||||
if !recover_remote_control_auth(&mut auth_recovery, &mut auth_change_rx).await {
|
||||
return Err(err);
|
||||
}
|
||||
*auth = websocket::load_remote_control_auth(auth_manager)
|
||||
*auth = load_remote_control_auth(auth_manager)
|
||||
.await
|
||||
.map_err(|_| pairing_unavailable_error())?;
|
||||
if auth.account_id != enrollment.account_id {
|
||||
@@ -440,6 +465,7 @@ pub async fn start_remote_control(
|
||||
"starting app-server remote control websocket task"
|
||||
);
|
||||
let remote_control_url_for_log = remote_control_url.clone();
|
||||
let handle_remote_control_url = remote_control_url.clone();
|
||||
let installation_id_for_log = installation_id.clone();
|
||||
let server_name_for_log = server_name.clone();
|
||||
let shutdown_token_for_log = shutdown_token.clone();
|
||||
@@ -508,6 +534,7 @@ pub async fn start_remote_control(
|
||||
enabled_tx: Arc::new(enabled_tx),
|
||||
status_tx: Arc::new(status_tx),
|
||||
state_db_available,
|
||||
remote_control_url: handle_remote_control_url,
|
||||
current_enrollment,
|
||||
auth_manager: handle_auth_manager,
|
||||
},
|
||||
|
||||
@@ -177,6 +177,48 @@ fn is_localhost(host: &Option<Host<&str>>) -> bool {
|
||||
pub(super) fn normalize_remote_control_url(
|
||||
remote_control_url: &str,
|
||||
) -> io::Result<RemoteControlTarget> {
|
||||
let remote_control_url = normalize_remote_control_base_url(remote_control_url)?;
|
||||
let map_url_parse_error = |err: url::ParseError| -> io::Error {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
format!("invalid remote control URL `{remote_control_url}`: {err}"),
|
||||
)
|
||||
};
|
||||
|
||||
let enroll_url = remote_control_url
|
||||
.join("wham/remote/control/server/enroll")
|
||||
.map_err(map_url_parse_error)?;
|
||||
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)?;
|
||||
websocket_url
|
||||
.set_scheme(if enroll_url.scheme() == "https" {
|
||||
"wss"
|
||||
} else {
|
||||
"ws"
|
||||
})
|
||||
.map_err(|()| {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
format!("invalid remote control URL `{remote_control_url}`"),
|
||||
)
|
||||
})?;
|
||||
|
||||
Ok(RemoteControlTarget {
|
||||
websocket_url: websocket_url.to_string(),
|
||||
enroll_url: enroll_url.to_string(),
|
||||
refresh_url: refresh_url.to_string(),
|
||||
pair_url: pair_url.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn normalize_remote_control_base_url(remote_control_url: &str) -> io::Result<Url> {
|
||||
let map_url_parse_error = |err: url::ParseError| -> io::Error {
|
||||
io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
@@ -198,35 +240,14 @@ pub(super) fn normalize_remote_control_url(
|
||||
remote_control_url.set_path(&normalized_path);
|
||||
}
|
||||
|
||||
let enroll_url = remote_control_url
|
||||
.join("wham/remote/control/server/enroll")
|
||||
.map_err(map_url_parse_error)?;
|
||||
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)?;
|
||||
let host = enroll_url.host();
|
||||
match enroll_url.scheme() {
|
||||
"https" if is_localhost(&host) || is_allowed_remote_control_chatgpt_host(&host) => {
|
||||
websocket_url.set_scheme("wss").map_err(map_scheme_error)?;
|
||||
}
|
||||
"http" if is_localhost(&host) => {
|
||||
websocket_url.set_scheme("ws").map_err(map_scheme_error)?;
|
||||
}
|
||||
let host = remote_control_url.host();
|
||||
match remote_control_url.scheme() {
|
||||
"https" if is_localhost(&host) || is_allowed_remote_control_chatgpt_host(&host) => {}
|
||||
"http" if is_localhost(&host) => {}
|
||||
_ => return Err(map_scheme_error(())),
|
||||
}
|
||||
|
||||
Ok(RemoteControlTarget {
|
||||
websocket_url: websocket_url.to_string(),
|
||||
enroll_url: enroll_url.to_string(),
|
||||
refresh_url: refresh_url.to_string(),
|
||||
pair_url: pair_url.to_string(),
|
||||
})
|
||||
Ok(remote_control_url)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -60,6 +60,7 @@ use tokio_tungstenite::accept_hdr_async;
|
||||
use tokio_tungstenite::tungstenite;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
mod clients_tests;
|
||||
mod pairing_tests;
|
||||
|
||||
const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
|
||||
@@ -163,6 +164,7 @@ fn remote_control_handle_with_current_enrollment(
|
||||
enabled_tx: Arc::new(enabled_tx),
|
||||
status_tx: Arc::new(status_tx),
|
||||
state_db_available: true,
|
||||
remote_control_url: remote_control_url.to_string(),
|
||||
current_enrollment,
|
||||
auth_manager,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
use super::super::clients::list_remote_control_clients;
|
||||
use super::super::clients::revoke_remote_control_client;
|
||||
use super::*;
|
||||
use codex_app_server_protocol::RemoteControlClient;
|
||||
use codex_app_server_protocol::RemoteControlClientsListOrder;
|
||||
use codex_app_server_protocol::RemoteControlClientsListParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsListResponse;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeParams;
|
||||
use codex_app_server_protocol::RemoteControlClientsRevokeResponse;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
fn client_management_handle(
|
||||
remote_control_url: String,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
) -> RemoteControlHandle {
|
||||
let (enabled_tx, _enabled_rx) = watch::channel(/*init*/ false);
|
||||
let (status_tx, _status_rx) = watch::channel(RemoteControlStatusChangedNotification {
|
||||
status: RemoteControlConnectionStatus::Disabled,
|
||||
server_name: test_server_name(),
|
||||
installation_id: TEST_INSTALLATION_ID.to_string(),
|
||||
environment_id: None,
|
||||
});
|
||||
RemoteControlHandle {
|
||||
enabled_tx: Arc::new(enabled_tx),
|
||||
status_tx: Arc::new(status_tx),
|
||||
state_db_available: false,
|
||||
remote_control_url,
|
||||
current_enrollment: Arc::new(StdMutex::new(None)),
|
||||
auth_manager,
|
||||
}
|
||||
}
|
||||
|
||||
fn empty_client_list() -> serde_json::Value {
|
||||
json!({
|
||||
"items": [],
|
||||
"cursor": null,
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn remote_control_handle_lists_clients_while_disabled() {
|
||||
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 request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
request.request_line,
|
||||
"GET /backend-api/wham/remote/control/environments/env%20%2F%3F/clients?cursor=cursor+%2F%3F&limit=10&order=asc HTTP/1.1"
|
||||
);
|
||||
assert_eq!(
|
||||
request.headers.get("authorization"),
|
||||
Some(&"Bearer Access Token".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
request.headers.get(REMOTE_CONTROL_ACCOUNT_ID_HEADER),
|
||||
Some(&"account_id".to_string())
|
||||
);
|
||||
respond_with_json(
|
||||
request.stream,
|
||||
json!({
|
||||
"items": [{
|
||||
"client_id": "client-123",
|
||||
"account_user_id": "user-123",
|
||||
"enrollment_status": "enrolled_device_key",
|
||||
"display_name": "Anton Phone",
|
||||
"device_type": "phone",
|
||||
"platform": "ios",
|
||||
"os_version": "19.0",
|
||||
"device_model": "iPhone",
|
||||
"app_version": "1.2.3",
|
||||
"last_seen_at": "2026-03-05T07:00:00Z",
|
||||
"last_seen_city": "San Francisco",
|
||||
}],
|
||||
"cursor": "next-cursor",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
});
|
||||
let handle = client_management_handle(remote_control_url, remote_control_auth_manager());
|
||||
|
||||
let response = handle
|
||||
.list_clients(RemoteControlClientsListParams {
|
||||
environment_id: "env /?".to_string(),
|
||||
cursor: Some("cursor /?".to_string()),
|
||||
limit: Some(10),
|
||||
order: Some(RemoteControlClientsListOrder::Asc),
|
||||
})
|
||||
.await
|
||||
.expect("client list should succeed while remote control is disabled");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert_eq!(
|
||||
response,
|
||||
RemoteControlClientsListResponse {
|
||||
data: vec![RemoteControlClient {
|
||||
client_id: "client-123".to_string(),
|
||||
display_name: Some("Anton Phone".to_string()),
|
||||
device_type: Some("phone".to_string()),
|
||||
platform: Some("ios".to_string()),
|
||||
os_version: Some("19.0".to_string()),
|
||||
device_model: Some("iPhone".to_string()),
|
||||
app_version: Some("1.2.3".to_string()),
|
||||
last_seen_at: Some(1_772_694_000),
|
||||
}],
|
||||
next_cursor: Some("next-cursor".to_string()),
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn remote_control_handle_revokes_client_while_disabled() {
|
||||
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 request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
request.request_line,
|
||||
"DELETE /backend-api/wham/remote/control/environments/env%20%2F%3F/clients/client%20%2F%3F HTTP/1.1"
|
||||
);
|
||||
respond_with_status(request.stream, "204 No Content", "").await;
|
||||
});
|
||||
let handle = client_management_handle(remote_control_url, remote_control_auth_manager());
|
||||
|
||||
let response = handle
|
||||
.revoke_client(RemoteControlClientsRevokeParams {
|
||||
environment_id: "env /?".to_string(),
|
||||
client_id: "client /?".to_string(),
|
||||
})
|
||||
.await
|
||||
.expect("client revoke should succeed while remote control is disabled");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert_eq!(response, RemoteControlClientsRevokeResponse {});
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_remote_control_clients_recovers_auth_after_unauthorized() {
|
||||
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_request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
stale_request.headers.get("authorization"),
|
||||
Some(&"Bearer stale-token".to_string())
|
||||
);
|
||||
respond_with_status(stale_request.stream, "401 Unauthorized", "").await;
|
||||
|
||||
let recovered_request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
recovered_request.headers.get("authorization"),
|
||||
Some(&"Bearer fresh-token".to_string())
|
||||
);
|
||||
respond_with_json(recovered_request.stream, empty_client_list()).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 response = list_remote_control_clients(
|
||||
&remote_control_url,
|
||||
&auth_manager,
|
||||
RemoteControlClientsListParams {
|
||||
environment_id: "env-123".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("client list should recover auth");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert_eq!(
|
||||
response,
|
||||
RemoteControlClientsListResponse {
|
||||
data: Vec::new(),
|
||||
next_cursor: None,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_remote_control_clients_retries_unauthorized_only_once() {
|
||||
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_request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
stale_request.headers.get("authorization"),
|
||||
Some(&"Bearer stale-token".to_string())
|
||||
);
|
||||
respond_with_status(stale_request.stream, "401 Unauthorized", "").await;
|
||||
|
||||
let recovered_request = accept_http_request(&listener).await;
|
||||
assert_eq!(
|
||||
recovered_request.headers.get("authorization"),
|
||||
Some(&"Bearer fresh-token".to_string())
|
||||
);
|
||||
respond_with_status(recovered_request.stream, "401 Unauthorized", "").await;
|
||||
|
||||
assert!(
|
||||
timeout(Duration::from_millis(100), accept_http_request(&listener))
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
});
|
||||
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 err = list_remote_control_clients(
|
||||
&remote_control_url,
|
||||
&auth_manager,
|
||||
RemoteControlClientsListParams {
|
||||
environment_id: "env-123".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("second unauthorized response should fail");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert_eq!(err.kind(), std::io::ErrorKind::PermissionDenied);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn revoke_remote_control_client_does_not_retry_forbidden() {
|
||||
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 request = accept_http_request(&listener).await;
|
||||
respond_with_status_and_headers(
|
||||
request.stream,
|
||||
"403 Forbidden",
|
||||
&[("x-request-id", "request-123"), ("cf-ray", "ray-123")],
|
||||
"forbidden",
|
||||
)
|
||||
.await;
|
||||
});
|
||||
|
||||
let err = revoke_remote_control_client(
|
||||
&remote_control_url,
|
||||
&remote_control_auth_manager(),
|
||||
RemoteControlClientsRevokeParams {
|
||||
environment_id: "env-123".to_string(),
|
||||
client_id: "client-123".to_string(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("forbidden revoke should fail");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert_eq!(err.kind(), std::io::ErrorKind::PermissionDenied);
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
format!(
|
||||
"remote control client revoke failed at `{remote_control_url}wham/remote/control/environments/env-123/clients/client-123`: HTTP 403 Forbidden, request-id: request-123, cf-ray: ray-123, body: forbidden"
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_remote_control_clients_preserves_decode_error_context() {
|
||||
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 request = accept_http_request(&listener).await;
|
||||
respond_with_status(request.stream, "200 OK", "{").await;
|
||||
});
|
||||
|
||||
let err = list_remote_control_clients(
|
||||
&remote_control_url,
|
||||
&remote_control_auth_manager(),
|
||||
RemoteControlClientsListParams {
|
||||
environment_id: "env-123".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("malformed client list should fail");
|
||||
server_task.await.expect("server task should finish");
|
||||
|
||||
assert!(
|
||||
err.to_string().contains(
|
||||
"failed to parse remote control client list response from `http://127.0.0.1:"
|
||||
)
|
||||
);
|
||||
assert!(err.to_string().contains("HTTP 200 OK"));
|
||||
assert!(err.to_string().contains("body: {"));
|
||||
assert!(err.to_string().contains("decode error:"));
|
||||
}
|
||||
@@ -13,9 +13,11 @@ 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::auth::RemoteControlConnectionAuth;
|
||||
use crate::transport::remote_control::auth::load_remote_control_auth;
|
||||
use crate::transport::remote_control::auth::recover_remote_control_auth;
|
||||
use crate::transport::remote_control::client_tracker::ClientTracker;
|
||||
use crate::transport::remote_control::client_tracker::REMOTE_CONTROL_IDLE_SWEEP_INTERVAL;
|
||||
use crate::transport::remote_control::enroll::RemoteControlConnectionAuth;
|
||||
use crate::transport::remote_control::enroll::RemoteControlEnrollment;
|
||||
use crate::transport::remote_control::enroll::enroll_remote_control_server;
|
||||
use crate::transport::remote_control::enroll::format_headers;
|
||||
@@ -1172,51 +1174,6 @@ fn build_remote_control_websocket_request(
|
||||
Ok(request)
|
||||
}
|
||||
|
||||
pub(crate) async fn load_remote_control_auth(
|
||||
auth_manager: &Arc<AuthManager>,
|
||||
) -> io::Result<RemoteControlConnectionAuth> {
|
||||
let mut reloaded = false;
|
||||
let auth = loop {
|
||||
let Some(auth) = auth_manager.auth().await else {
|
||||
if reloaded {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::PermissionDenied,
|
||||
"remote control requires ChatGPT authentication",
|
||||
));
|
||||
}
|
||||
auth_manager.reload().await;
|
||||
reloaded = true;
|
||||
continue;
|
||||
};
|
||||
if !auth.uses_codex_backend() {
|
||||
break auth;
|
||||
}
|
||||
if auth.get_account_id().is_none() && !reloaded {
|
||||
auth_manager.reload().await;
|
||||
reloaded = true;
|
||||
continue;
|
||||
}
|
||||
break auth;
|
||||
};
|
||||
|
||||
if !auth.uses_codex_backend() {
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::PermissionDenied,
|
||||
"remote control requires ChatGPT authentication; API key auth is not supported",
|
||||
));
|
||||
}
|
||||
|
||||
Ok(RemoteControlConnectionAuth {
|
||||
auth_provider: codex_model_provider::auth_provider_from_auth(&auth),
|
||||
account_id: auth.get_account_id().ok_or_else(|| {
|
||||
io::Error::new(
|
||||
ErrorKind::WouldBlock,
|
||||
"remote control enrollment is waiting for a ChatGPT account id",
|
||||
)
|
||||
})?,
|
||||
})
|
||||
}
|
||||
|
||||
fn next_reconnect_delay(reconnect_attempt: &mut u64) -> (std::time::Duration, bool) {
|
||||
let reconnect_delay = backoff(*reconnect_attempt).min(REMOTE_CONTROL_RECONNECT_BACKOFF_CAP);
|
||||
let reconnect_backoff_reset = reconnect_delay == REMOTE_CONTROL_RECONNECT_BACKOFF_CAP;
|
||||
@@ -1548,52 +1505,6 @@ async fn enroll_remote_control_server_if_missing(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn recover_remote_control_auth(
|
||||
auth_recovery: &mut UnauthorizedRecovery,
|
||||
auth_change_rx: &mut watch::Receiver<u64>,
|
||||
) -> bool {
|
||||
if !auth_recovery.has_next() {
|
||||
return false;
|
||||
}
|
||||
|
||||
let mode = auth_recovery.mode_name();
|
||||
let step = auth_recovery.step_name();
|
||||
let auth_change_revision_before_recovery = *auth_change_rx.borrow();
|
||||
match auth_recovery.next().await {
|
||||
Ok(step_result) => {
|
||||
if step_result.auth_state_changed() == Some(true) {
|
||||
mark_recovery_auth_change_seen(
|
||||
auth_change_rx,
|
||||
auth_change_revision_before_recovery,
|
||||
);
|
||||
}
|
||||
info!(
|
||||
"remote control websocket auth recovery succeeded: mode={mode}, step={step}, auth_state_changed={:?}",
|
||||
step_result.auth_state_changed()
|
||||
);
|
||||
true
|
||||
}
|
||||
Err(err) => {
|
||||
warn!("remote control websocket auth recovery failed: mode={mode}, step={step}: {err}");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn mark_recovery_auth_change_seen(
|
||||
auth_change_rx: &mut watch::Receiver<u64>,
|
||||
auth_change_revision_before_recovery: u64,
|
||||
) {
|
||||
let auth_change_revision_after_recovery = *auth_change_rx.borrow();
|
||||
if auth_change_revision_after_recovery == auth_change_revision_before_recovery.wrapping_add(1) {
|
||||
// Recovery updated the same watch that wakes the outer reconnect
|
||||
// loop. Mark only that single revision seen; if more revisions
|
||||
// arrived while recovery was in flight, leave them pending so the
|
||||
// reconnect loop still reacts to the later external auth change.
|
||||
auth_change_rx.borrow_and_update();
|
||||
}
|
||||
}
|
||||
|
||||
fn format_remote_control_websocket_connect_error(
|
||||
websocket_url: &str,
|
||||
err: &tungstenite::Error,
|
||||
@@ -1620,6 +1531,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::outgoing_message::OutgoingMessage;
|
||||
use crate::transport::remote_control::ServerEvent;
|
||||
use crate::transport::remote_control::auth::mark_recovery_auth_change_seen;
|
||||
use crate::transport::remote_control::protocol::StreamId;
|
||||
use crate::transport::remote_control::protocol::normalize_remote_control_url;
|
||||
use chrono::Utc;
|
||||
|
||||
Reference in new issue
Block a user