From ffa992b5fe46c3e95f38784fb36b2a1105d4d11a Mon Sep 17 00:00:00 2001 From: Ramshouriesh R Date: Sun, 23 Aug 2026 23:07:28 +0530 Subject: [PATCH 1/3] fix: keep interactive controls live without WebSockets --- server/src/main.rs | 836 +++++++++++++----- server/src/rbac.rs | 26 +- tests/journey/fleet-read-plane.spec.ts | 24 + .../providers/WebSocketProvider.tsx | 254 +++++- .../__tests__/WebSocketProvider.test.tsx | 70 +- 5 files changed, 936 insertions(+), 274 deletions(-) diff --git a/server/src/main.rs b/server/src/main.rs index 89d5b387..e1c6e25e 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -56,14 +56,14 @@ mod webhook; pub const CE_USER_LIMIT: usize = 3; use axum::{ - Router, + Json, Router, extract::{ DefaultBodyLimit, Query, State, ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}, }, - http::StatusCode, + http::{StatusCode, header}, response::IntoResponse, - routing::get, + routing::{get, post}, }; use axum_extra::extract::cookie::CookieJar; use futures_util::{SinkExt, StreamExt}; @@ -85,12 +85,18 @@ use tracing_subscriber::{EnvFilter, fmt, prelude::*}; type AgentTx = mpsc::UnboundedSender; type UiTx = mpsc::UnboundedSender; +type UiCommandTx = mpsc::UnboundedSender; +type UiHttpRx = Arc>>; pub struct UiClient { tx: UiTx, + command_tx: UiCommandTx, login: String, client_ip: String, agent_access: AgentAccess, + http_receiver: Option, + http_token: Option, + last_seen: Arc, } #[derive(Debug, Clone, Deserialize)] @@ -586,6 +592,14 @@ async fn main() { .nest("/metrics", metrics::routes()) .nest("/telemetry", telemetry::routes()) .nest("/core/v1", core::routes()) + // HTTP control tunnel used only when the browser cannot establish a + // WebSocket through its network/edge path. It carries the exact same + // UiMessage protocol and is protected by cookie auth, CSRF, RBAC, ACLs, + // approval gates, and operation ownership below. + .route("/ui/connect", post(ui_http_connect_handler)) + .route("/ui/disconnect", post(ui_http_disconnect_handler)) + .route("/ui/poll", post(ui_http_poll_handler)) + .route("/ui/send", post(ui_http_send_handler)) .route("/me", get(me_handler)) .route("/healthz", get(healthz)) .route("/audit", get(audit_handler)) @@ -1759,17 +1773,50 @@ async fn handle_agent_socket(socket: WebSocket, state: Arc, token: Str } } -async fn handle_ui_socket( - socket: WebSocket, +const UI_HTTP_POLL_WAIT: std::time::Duration = std::time::Duration::from_secs(20); +const UI_HTTP_IDLE_TTL_SECS: u64 = 90; +const UI_HTTP_MAX_BATCH: usize = 128; +const UI_HTTP_MAX_SESSIONS_PER_USER: usize = 8; + +struct RegisteredUiClient { + client_id: u64, + command_tx: UiCommandTx, + outbound_rx: UiHttpRx, +} + +#[derive(Debug, Deserialize)] +struct UiHttpClientRequest { + client_id: u64, + client_token: String, +} + +#[derive(Debug, Deserialize)] +struct UiHttpSendRequest { + client_id: u64, + client_token: String, + messages: Vec, +} + +#[derive(Serialize)] +struct UiHttpConnectResponse { + client_id: u64, + client_token: String, + messages: Vec, +} + +#[derive(Serialize)] +struct UiHttpMessagesResponse { + messages: Vec, +} + +async fn register_ui_client( state: Arc, login: String, initial_role: auth::Role, token_iat: i64, client_ip: String, -) { - tracing::info!(%login, "new ui websocket connection"); - let (mut sender, mut receiver) = socket.split(); - + http_token: Option, +) -> Result { let (initial_agents, initial_capabilities) = { let map = state.agents.lock().await; let agents = map.keys().cloned().collect::>(); @@ -1779,26 +1826,42 @@ async fn handle_ui_socket( .collect::>(); (agents, capabilities) }; - let mut agent_access = if initial_role == auth::Role::Admin { + let agent_access = if initial_role == auth::Role::Admin { AgentAccess::Unrestricted } else { ee_fetch_agent_access(&login, &client_ip, &initial_agents).await }; - let (tx, mut rx) = mpsc::unbounded_channel::(); + let (tx, rx) = mpsc::unbounded_channel::(); + let outbound_rx = Arc::new(Mutex::new(rx)); + let (command_tx, command_rx) = mpsc::unbounded_channel::(); let client_id = state.ui_id_counter.fetch_add(1, Ordering::Relaxed); - state.ui_clients.lock().await.insert( + let last_seen = Arc::new(AtomicU64::new(now_unix().max(0) as u64)); + let mut clients = state.ui_clients.lock().await; + // Admission and insertion share one lock so concurrent connect requests + // cannot race past the per-user bound. + if http_token.is_some() + && clients + .values() + .filter(|client| client.login == login && client.http_token.is_some()) + .count() + >= UI_HTTP_MAX_SESSIONS_PER_USER + { + return Err(()); + } + clients.insert( client_id, UiClient { tx: tx.clone(), + command_tx: command_tx.clone(), login: login.clone(), client_ip: client_ip.clone(), agent_access: agent_access.clone(), + http_receiver: http_token.as_ref().map(|_| outbound_rx.clone()), + http_token: http_token.clone(), + last_seen: last_seen.clone(), }, ); - - // Terminal sessions this client opened, so recordings are stopped if the - // socket drops without an explicit StopTerminalRequest. - let mut rec_sessions: std::collections::HashSet = std::collections::HashSet::new(); + drop(clients); let (initial_agents, initial_capabilities) = ee_filter_agent_list(initial_agents, initial_capabilities, &agent_access); @@ -1807,7 +1870,248 @@ async fn handle_ui_socket( capabilities: initial_capabilities, }); + let process_state = state.clone(); + tokio::spawn(async move { + process_ui_messages( + command_rx, + process_state, + login, + token_iat, + client_ip, + tx, + client_id, + ) + .await; + }); + + if http_token.is_some() { + let reap_state = state.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(30)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + tick.tick().await; + loop { + tick.tick().await; + let seen = last_seen.load(Ordering::Relaxed); + let now = now_unix().max(0) as u64; + if now.saturating_sub(seen) <= UI_HTTP_IDLE_TTL_SECS { + continue; + } + let removed = reap_state + .ui_clients + .lock() + .await + .remove(&client_id) + .is_some(); + if removed { + tracing::info!(client_id, "idle HTTP UI client expired"); + } + break; + } + }); + } + + Ok(RegisteredUiClient { + client_id, + command_tx, + outbound_rx, + }) +} + +fn no_store_json(value: T) -> axum::response::Response { + ([(header::CACHE_CONTROL, "private, no-store")], Json(value)).into_response() +} + +fn http_control_token_matches(expected: &str, actual: &str) -> bool { + subtle::ConstantTimeEq::ct_eq(expected.as_bytes(), actual.as_bytes()).into() +} + +fn is_http_client_message(message: &UiMessage) -> bool { + matches!( + message, + UiMessage::ListAgentsRequest | UiMessage::SendToAgent { .. } + ) +} + +async fn receive_http_messages( + receiver: &mut mpsc::UnboundedReceiver, +) -> Vec { + let mut messages = Vec::new(); + if let Ok(Some(first)) = tokio::time::timeout(UI_HTTP_POLL_WAIT, receiver.recv()).await { + messages.push(first); + } + while messages.len() < UI_HTTP_MAX_BATCH { + match receiver.try_recv() { + Ok(message) => messages.push(message), + Err(_) => break, + } + } + messages +} + +async fn authenticate_http_ui_client( + state: &AppState, + jar: &CookieJar, + request: &UiHttpClientRequest, +) -> Result<(UiCommandTx, UiHttpRx), axum::response::Response> { + let claims = auth::current_user(jar, &state.db) + .await + .map_err(|(status, reason)| (status, reason).into_response())?; + let clients = state.ui_clients.lock().await; + let Some(client) = clients.get(&request.client_id) else { + return Err((StatusCode::GONE, "control session expired").into_response()); + }; + let Some(expected_token) = client.http_token.as_deref() else { + return Err((StatusCode::GONE, "not an HTTP control session").into_response()); + }; + let token_matches = http_control_token_matches(expected_token, &request.client_token); + if client.login != claims.sub || !token_matches { + return Err((StatusCode::FORBIDDEN, "control session mismatch").into_response()); + } + let Some(receiver) = client.http_receiver.clone() else { + return Err((StatusCode::GONE, "control session unavailable").into_response()); + }; + client + .last_seen + .store(now_unix().max(0) as u64, Ordering::Relaxed); + Ok((client.command_tx.clone(), receiver)) +} + +async fn ui_http_connect_handler( + jar: CookieJar, + headers: axum::http::HeaderMap, + axum::extract::ConnectInfo(peer): axum::extract::ConnectInfo, + State(state): State>, +) -> axum::response::Response { + let claims = match auth::current_user(&jar, &state.db).await { + Ok(claims) => claims, + Err((status, reason)) => return (status, reason).into_response(), + }; + let client_ip = throttle::real_client_ip(&headers, Some(peer.ip())); + let client_token = uuid::Uuid::new_v4().to_string(); + let registered = match register_ui_client( + state, + claims.sub, + auth::Role::parse(&claims.role), + claims.iat, + client_ip, + Some(client_token.clone()), + ) + .await + { + Ok(registered) => registered, + Err(()) => { + return ( + StatusCode::TOO_MANY_REQUESTS, + "too many HTTP control sessions", + ) + .into_response(); + } + }; + let messages = { + let mut receiver = registered.outbound_rx.lock().await; + receive_http_messages(&mut receiver).await + }; + tracing::info!( + client_id = registered.client_id, + "new HTTP UI control connection" + ); + no_store_json(UiHttpConnectResponse { + client_id: registered.client_id, + client_token, + messages, + }) +} + +async fn ui_http_poll_handler( + jar: CookieJar, + State(state): State>, + Json(request): Json, +) -> axum::response::Response { + let (_, receiver) = match authenticate_http_ui_client(&state, &jar, &request).await { + Ok(client) => client, + Err(response) => return response, + }; + // Only one long poll may wait on a control session at once. Returning a + // conflict instead of queuing arbitrary concurrent requests bounds origin + // tasks even if a client or extension retries too aggressively. + let Ok(mut receiver) = receiver.try_lock() else { + return (StatusCode::CONFLICT, "control poll already active").into_response(); + }; + no_store_json(UiHttpMessagesResponse { + messages: receive_http_messages(&mut receiver).await, + }) +} + +async fn ui_http_disconnect_handler( + jar: CookieJar, + State(state): State>, + Json(request): Json, +) -> axum::response::Response { + if let Err(response) = authenticate_http_ui_client(&state, &jar, &request).await { + return response; + } + state.ui_clients.lock().await.remove(&request.client_id); + no_store_json(serde_json::json!({ "ok": true })) +} + +async fn ui_http_send_handler( + jar: CookieJar, + State(state): State>, + Json(request): Json, +) -> axum::response::Response { + if request.messages.is_empty() || request.messages.len() > UI_HTTP_MAX_BATCH { + return (StatusCode::BAD_REQUEST, "invalid control message batch").into_response(); + } + if request + .messages + .iter() + .any(|message| !is_http_client_message(message)) + { + return (StatusCode::BAD_REQUEST, "invalid client UI message").into_response(); + } + let client_request = UiHttpClientRequest { + client_id: request.client_id, + client_token: request.client_token, + }; + let (command_tx, _) = match authenticate_http_ui_client(&state, &jar, &client_request).await { + Ok(client) => client, + Err(response) => return response, + }; + for message in request.messages { + if command_tx.send(message).is_err() { + return (StatusCode::GONE, "control session closed").into_response(); + } + } + no_store_json(serde_json::json!({ "ok": true })) +} + +async fn handle_ui_socket( + socket: WebSocket, + state: Arc, + login: String, + initial_role: auth::Role, + token_iat: i64, + client_ip: String, +) { + tracing::info!(%login, "new ui websocket connection"); + let (mut sender, mut receiver) = socket.split(); + let registered = register_ui_client( + state.clone(), + login, + initial_role, + token_iat, + client_ip, + None, + ) + .await + .expect("WebSocket UI clients are not subject to the HTTP session cap"); + let client_id = registered.client_id; + let command_tx = registered.command_tx; + let outbound_rx = registered.outbound_rx; + let send_task = tokio::spawn(async move { + let mut rx = outbound_rx.lock().await; let mut hb = tokio::time::interval(std::time::Duration::from_secs(25)); hb.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); hb.tick().await; @@ -1853,281 +2157,297 @@ async fn handle_ui_socket( tracing::warn!(error = %e, "dropped un-parseable UI message"); } if let Ok(parsed_msg) = parsed { - match parsed_msg { - UiMessage::ListAgentsRequest => { - let (agents, capabilities) = { - let map = state.agents.lock().await; - let agents = map.keys().cloned().collect::>(); - let capabilities = map - .iter() - .map(|(id, entry)| (id.clone(), entry.capabilities.clone())) - .collect::>(); - (agents, capabilities) - }; - agent_access = ui_agent_access(&state, &login, &client_ip, &agents).await; - if let Some(client) = state.ui_clients.lock().await.get_mut(&client_id) { - client.agent_access = agent_access.clone(); - } - let (agents, capabilities) = - ee_filter_agent_list(agents, capabilities, &agent_access); - let _ = tx.send(UiMessage::ListAgentsResponse { - agents, - capabilities, + if command_tx.send(parsed_msg).is_err() { + break; + } + } + } + + send_task.abort(); + // Dropping the map's command sender lets process_ui_messages perform the + // shared ownership/recording cleanup for both transports. + state.ui_clients.lock().await.remove(&client_id); +} + +async fn process_ui_messages( + mut command_rx: mpsc::UnboundedReceiver, + state: Arc, + login: String, + token_iat: i64, + client_ip: String, + tx: UiTx, + client_id: u64, +) { + // Terminal sessions this client opened, so recordings are stopped if the + // control transport disappears without an explicit StopTerminalRequest. + let mut rec_sessions: std::collections::HashSet = std::collections::HashSet::new(); + while let Some(parsed_msg) = command_rx.recv().await { + match parsed_msg { + UiMessage::ListAgentsRequest => { + let (agents, capabilities) = { + let map = state.agents.lock().await; + let agents = map.keys().cloned().collect::>(); + let capabilities = map + .iter() + .map(|(id, entry)| (id.clone(), entry.capabilities.clone())) + .collect::>(); + (agents, capabilities) + }; + let agent_access = ui_agent_access(&state, &login, &client_ip, &agents).await; + if let Some(client) = state.ui_clients.lock().await.get_mut(&client_id) { + client.agent_access = agent_access.clone(); + } + let (agents, capabilities) = + ee_filter_agent_list(agents, capabilities, &agent_access); + let _ = tx.send(UiMessage::ListAgentsResponse { + agents, + capabilities, + }); + } + UiMessage::SendToAgent { agent_id, message } => { + let target_access = + ui_agent_access(&state, &login, &client_ip, std::slice::from_ref(&agent_id)) + .await; + if let Some(client) = state.ui_clients.lock().await.get_mut(&client_id) { + merge_exact_agent_access(&mut client.agent_access, &agent_id, &target_access); + } + if !agent_allowed_by_access(&agent_id, &target_access) { + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type: "agent_access".to_string(), + reason: "not in your allowed agents".to_string(), }); + continue; } - UiMessage::SendToAgent { agent_id, message } => { - let target_access = ui_agent_access( - &state, - &login, - &client_ip, - std::slice::from_ref(&agent_id), - ) - .await; - if let Some(client) = state.ui_clients.lock().await.get_mut(&client_id) { - merge_exact_agent_access( - &mut client.agent_access, - &agent_id, - &target_access, + let variant_type = serde_json::to_value(&message) + .ok() + .and_then(|v| v.get("type").and_then(|t| t.as_str().map(String::from))) + .unwrap_or_else(|| "unknown".into()); + // Session-epoch guard: a revoked session (logout / + // role-change / MFA-disable) must not keep issuing + // mutating commands over an already-open control session. + // The HTTP tunnel is authenticated per request too, but this + // shared check keeps WebSocket and HTTP processing identical. + if !auth::is_dev_mode() { + let session_reason = match crate::db::get_user(&state.db, &login).await { + Ok(Some(row)) if token_iat >= row.session_epoch => None, + Ok(Some(_)) => Some("session revoked — please sign in again"), + Ok(None) => Some("session no longer exists"), + Err(error) => { + tracing::error!(%error, %login, "ui control: session verification failed"); + Some("session verification unavailable") + } + }; + if let Some(reason) = session_reason { + tracing::warn!( + %login, %agent_id, %variant_type, %reason, + "ui control session failed session validation" ); + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type: "session_revoked".to_string(), + reason: reason.to_string(), + }); + break; } - if !agent_allowed_by_access(&agent_id, &target_access) { + } + let security = match message.ui_request_security() { + Ok(security) => security, + Err(error) => { + tracing::warn!( + %login, + %agent_id, + variant = %variant_type, + ?error, + "ui control: rejected invalid or non-request message" + ); let _ = tx.send(UiMessage::PermissionDenied { agent_id: agent_id.clone(), - variant_type: "agent_access".to_string(), - reason: "not in your allowed agents".to_string(), + variant_type, + reason: "message is not an allowed UI request".to_string(), }); continue; } - let variant_type = serde_json::to_value(&message) - .ok() - .and_then(|v| v.get("type").and_then(|t| t.as_str().map(String::from))) - .unwrap_or_else(|| "unknown".into()); - // Session-epoch guard: a revoked session (logout / - // role-change / MFA-disable) must not keep issuing - // mutating commands over an already-open WebSocket. - // The HTTP rbac middleware checks this on every - // request; the WS plane bypasses that middleware, so - // re-check here. A mismatch closes the socket. - if !auth::is_dev_mode() { - let session_reason = match crate::db::get_user(&state.db, &login).await { - Ok(Some(row)) if token_iat >= row.session_epoch => None, - Ok(Some(_)) => Some("session revoked — please sign in again"), - Ok(None) => Some("session no longer exists"), - Err(error) => { - tracing::error!(%error, %login, "ui ws: session verification failed"); - Some("session verification unavailable") + }; + // EE ACL enforcement (skip for admins) + if ee::ee_active() && !auth::is_dev_mode() { + let is_admin = matches!( + crate::db::get_user(&state.db, &login).await, + Ok(Some(row)) if row.role == "admin" + ); + if !is_admin { + if let Some(action) = security.action { + if !ee_check_permission( + &login, + action, + &agent_id, + Some(client_ip.as_str()), + ) + .await + { + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type: action.to_string(), + reason: format!("denied: {action}"), + }); + continue; } - }; - if let Some(reason) = session_reason { - tracing::warn!( - %login, %agent_id, %variant_type, %reason, - "ui ws: closing socket after failed session validation" - ); - let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type: "session_revoked".to_string(), - reason: reason.to_string(), - }); - break; } } - let security = match message.ui_request_security() { - Ok(security) => security, - Err(error) => { - tracing::warn!( - %login, - %agent_id, - variant = %variant_type, - ?error, - "ui ws: rejected invalid or non-request message" - ); - let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type, - reason: "message is not an allowed UI request".to_string(), - }); - continue; - } + } + // CE RBAC over both interactive transports. Without this gate + // a viewer with a verified session could + // ControlServiceRequest, AptUpgradeRequest, + // DockerContainerActionRequest, send terminal + // keystrokes, etc., bypassing the entire role + // model. Re-resolve the role from the DB on every + // mutating message so a freshly-demoted admin is + // blocked immediately. + if !auth::is_dev_mode() && security.class.requires_admin() { + let role_str = match crate::db::get_user(&state.db, &login).await { + Ok(Some(row)) => row.role, + _ => "viewer".to_string(), }; - // EE ACL enforcement (skip for admins) - if ee::ee_active() && !auth::is_dev_mode() { - let is_admin = matches!( - crate::db::get_user(&state.db, &login).await, - Ok(Some(row)) if row.role == "admin" + if auth::Role::parse(&role_str) != auth::Role::Admin { + tracing::warn!( + %login, %agent_id, variant = %variant_type, + "ui control: rejected mutating message from non-admin" ); - if !is_admin { - if let Some(action) = security.action { - if !ee_check_permission( - &login, - action, - &agent_id, - Some(client_ip.as_str()), - ) - .await - { - let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type: action.to_string(), - reason: format!("denied: {action}"), - }); - continue; - } - } - } + crate::db::record_audit( + &state.db, + now_unix(), + Some(&login), + Some(&agent_id), + "ws.send_to_agent.denied", + false, + Some(&format!("role={role_str} variant={variant_type}")), + ) + .await; + // Tell the UI the request was rejected so the + // calling panel doesn't sit in "waiting for + // output…" forever. Best-effort; if the send + // fails the client is already gone. + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type, + reason: "admin only".to_string(), + }); + continue; } - // CE RBAC over the WebSocket plane. The HTTP rbac - // middleware doesn't run here — without this gate - // a viewer with a verified session could - // ControlServiceRequest, AptUpgradeRequest, - // DockerContainerActionRequest, send terminal - // keystrokes, etc., bypassing the entire role - // model. Re-resolve the role from the DB on every - // mutating message so a freshly-demoted admin is - // blocked immediately. - if !auth::is_dev_mode() && security.class.requires_admin() { - let role_str = match crate::db::get_user(&state.db, &login).await { - Ok(Some(row)) => row.role, - _ => "viewer".to_string(), - }; - if auth::Role::parse(&role_str) != auth::Role::Admin { - tracing::warn!( - %login, %agent_id, variant = %variant_type, - "ui ws: rejected mutating message from non-admin" - ); - crate::db::record_audit( - &state.db, - now_unix(), - Some(&login), - Some(&agent_id), - "ws.send_to_agent.denied", - false, - Some(&format!("role={role_str} variant={variant_type}")), - ) - .await; - // Tell the UI the request was rejected so the - // calling panel doesn't sit in "waiting for - // output…" forever. Best-effort; if the send - // fails the client is already gone. + } + // EE command-approval gate (dual control). A discrete action + // that matches an approval rule is HELD: we stash the + // serialized message in EE and tell the UI it's pending — a + // second admin approves in the Approvals tab, then EE calls + // back to /internal/execute-approved to run it. Interactive + // Interactive streams and reads are never approval-held. + if ee::ee_active() && !auth::is_dev_mode() { + if security.class.requires_approval() { + let Some(action) = security.action else { let _ = tx.send(UiMessage::PermissionDenied { agent_id: agent_id.clone(), - variant_type, - reason: "admin only".to_string(), + variant_type: variant_type.clone(), + reason: "request has no approval action mapping".to_string(), }); continue; - } - } - // EE command-approval gate (dual control). A discrete action - // that matches an approval rule is HELD: we stash the - // serialized message in EE and tell the UI it's pending — a - // second admin approves in the Approvals tab, then EE calls - // back to /internal/execute-approved to run it. Interactive - // Interactive streams and reads are never approval-held. - if ee::ee_active() && !auth::is_dev_mode() { - if security.class.requires_approval() { - let Some(action) = security.action else { + }; + let payload = serde_json::to_string(&message).unwrap_or_default(); + match ee_check_approval(&login, action, &agent_id, &payload).await { + Ok(None) => { /* no rule matched — run it now */ } + Ok(Some(req_id)) => { + crate::db::record_audit( + &state.db, + now_unix(), + Some(&login), + Some(&agent_id), + "ws.approval.held", + true, + Some(&format!("action={action} request={req_id}")), + ) + .await; let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type: variant_type.clone(), - reason: "request has no approval action mapping".to_string(), - }); - continue; - }; - let payload = serde_json::to_string(&message).unwrap_or_default(); - match ee_check_approval(&login, action, &agent_id, &payload).await { - Ok(None) => { /* no rule matched — run it now */ } - Ok(Some(req_id)) => { - crate::db::record_audit( - &state.db, - now_unix(), - Some(&login), - Some(&agent_id), - "ws.approval.held", - true, - Some(&format!("action={action} request={req_id}")), - ) - .await; - let _ = tx.send(UiMessage::PermissionDenied { agent_id: agent_id.clone(), variant_type: "approval_pending".to_string(), reason: format!( "held for approval — request #{req_id}; a second admin must approve it in the Approvals tab" ), }); - continue; - } - Err(()) => { - let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type: "approval_unavailable".to_string(), - reason: "approval system unavailable — action blocked (fail-closed)".to_string(), - }); - continue; - } + continue; + } + Err(()) => { + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type: "approval_unavailable".to_string(), + reason: + "approval system unavailable — action blocked (fail-closed)" + .to_string(), + }); + continue; } } } - if let Some(operation) = operation_routing::ui_operation(&message) { - use operation_routing::UiOperation; - let allowed = { - let mut owners = state.operation_owners.lock().await; - match &operation { - UiOperation::Start(key) => { - owners.claim(&agent_id, key.clone(), client_id) - } - UiOperation::Use(key) => { - owners.owner(&agent_id, key) == Some(client_id) - } - UiOperation::Stop(key) => owners.release(&agent_id, key, client_id), + } + if let Some(operation) = operation_routing::ui_operation(&message) { + use operation_routing::UiOperation; + let allowed = { + let mut owners = state.operation_owners.lock().await; + match &operation { + UiOperation::Start(key) => { + owners.claim(&agent_id, key.clone(), client_id) } - }; - if !allowed { - let _ = tx.send(UiMessage::PermissionDenied { - agent_id: agent_id.clone(), - variant_type: variant_type.clone(), - reason: "operation is owned by another client".to_string(), - }); - continue; + UiOperation::Use(key) => { + owners.owner(&agent_id, key) == Some(client_id) + } + UiOperation::Stop(key) => owners.release(&agent_id, key, client_id), } + }; + if !allowed { + let _ = tx.send(UiMessage::PermissionDenied { + agent_id: agent_id.clone(), + variant_type: variant_type.clone(), + reason: "operation is owned by another client".to_string(), + }); + continue; } - // Recording tap: terminal session lifecycle + INPUT (user→agent). - match &message { - Message::StartTerminalRequest { session_id } => { - state - .recorder - .start(session_id, &agent_id, &login, "host") - .await; - rec_sessions.insert(session_id.clone()); - } - Message::TerminalData { session_id, data } => { - state.recorder.frame(session_id, "i", data).await; - } - Message::StopTerminalRequest { session_id } => { - state.recorder.stop(session_id).await; - rec_sessions.remove(session_id); - } - _ => {} + } + // Recording tap: terminal session lifecycle + INPUT (user→agent). + match &message { + Message::StartTerminalRequest { session_id } => { + state + .recorder + .start(session_id, &agent_id, &login, "host") + .await; + rec_sessions.insert(session_id.clone()); + } + Message::TerminalData { session_id, data } => { + state.recorder.frame(session_id, "i", data).await; } - if let Some(entry) = state.agents.lock().await.get(&agent_id) { - let _ = entry.tx.send(message); + Message::StopTerminalRequest { session_id } => { + state.recorder.stop(session_id).await; + rec_sessions.remove(session_id); } + _ => {} + } + if let Some(entry) = state.agents.lock().await.get(&agent_id) { + let _ = entry.tx.send(message); } - _ => {} } + _ => {} } } - send_task.abort(); + // Removing an already-removed entry is harmless. WebSocket teardown and + // the HTTP idle reaper deliberately race this cleanup path. state.ui_clients.lock().await.remove(&client_id); state .operation_owners .lock() .await .release_client(client_id); - // Close any recordings still open for this client (abrupt disconnect). for sid in rec_sessions { state.recorder.stop(&sid).await; } - tracing::info!(client_id, "ui client disconnected"); + tracing::info!(client_id, "ui control client disconnected"); } pub(crate) async fn ee_fetch_agent_access( @@ -2402,4 +2722,34 @@ mod tests { assert!(!ui_websocket_session_allowed(true, true)); assert!(!ui_websocket_session_allowed(false, false)); } + + #[test] + fn http_control_tokens_require_an_exact_match() { + assert!(http_control_token_matches( + "random-session-token", + "random-session-token" + )); + assert!(!http_control_token_matches( + "random-session-token", + "other-session-token" + )); + assert!(!http_control_token_matches("random-session-token", "")); + } + + #[test] + fn http_control_accepts_only_client_originated_ui_variants() { + assert!(is_http_client_message(&UiMessage::ListAgentsRequest)); + assert!(is_http_client_message(&UiMessage::SendToAgent { + agent_id: "node-a-id".into(), + message: Message::SystemStatsRequest, + })); + assert!(!is_http_client_message(&UiMessage::ListAgentsResponse { + agents: Vec::new(), + capabilities: HashMap::new(), + })); + assert!(!is_http_client_message(&UiMessage::AgentMessage { + agent_id: "node-a-id".into(), + message: Message::SystemStatsRequest, + })); + } } diff --git a/server/src/rbac.rs b/server/src/rbac.rs index 79ad7064..95057a83 100644 --- a/server/src/rbac.rs +++ b/server/src/rbac.rs @@ -26,6 +26,13 @@ fn is_api_keys_path(path: &str) -> bool { path == "/ee/keys" || path.starts_with("/ee/keys/") } +fn is_ui_transport_path(path: &str) -> bool { + matches!( + path, + "/ui/connect" | "/ui/disconnect" | "/ui/poll" | "/ui/send" + ) +} + fn is_mutating(method: &Method) -> bool { matches!( *method, @@ -140,7 +147,12 @@ pub async fn middleware( // keys. EE scopes every mutation by the CE-injected login, so this only // lets a viewer manage their own keys — never escalation. All other guards // (auth, session-epoch, MFA above) still applied. - if is_mutating(&method) && !is_api_keys_path(&path) { + // The HTTP UI transport carries both reads and writes in a UiMessage + // envelope. Its handler applies the same per-message security class, + // ACL, approval, and operation-ownership checks as the WebSocket path. + // Do not reject the whole batch merely because the tunnel itself is POST; + // current_user above still authenticates every request. + if is_mutating(&method) && !is_api_keys_path(&path) && !is_ui_transport_path(&path) { if auth::Role::parse(&claims.role) != auth::Role::Admin { return forbidden("viewer role: read-only"); } @@ -151,7 +163,7 @@ pub async fn middleware( #[cfg(test)] mod tests { - use super::{is_api_keys_path, is_cli_read}; + use super::{is_api_keys_path, is_cli_read, is_ui_transport_path}; use axum::http::Method; #[test] @@ -163,6 +175,16 @@ mod tests { assert!(!is_api_keys_path("/ee/metrics/panels")); } + #[test] + fn matches_only_exact_ui_transport_routes() { + assert!(is_ui_transport_path("/ui/connect")); + assert!(is_ui_transport_path("/ui/disconnect")); + assert!(is_ui_transport_path("/ui/poll")); + assert!(is_ui_transport_path("/ui/send")); + assert!(!is_ui_transport_path("/ui/send/extra")); + assert!(!is_ui_transport_path("/ui/send-anything")); + } + #[test] fn cli_http_scope_is_read_only_and_segment_bounded() { assert!(is_cli_read(&Method::GET, "/core/v1/fleet")); diff --git a/tests/journey/fleet-read-plane.spec.ts b/tests/journey/fleet-read-plane.spec.ts index 61204851..51ba0ab0 100644 --- a/tests/journey/fleet-read-plane.spec.ts +++ b/tests/journey/fleet-read-plane.spec.ts @@ -67,6 +67,30 @@ test('Fleet remains durable across reload, disconnect, SSE loss, and reconnect', expect(finalPayload.hosts).toHaveLength(1); expect(finalPayload.hosts[0].agent_id).toBe('journey-agent-id'); + // A browser/network that blocks WebSocket upgrades must retain the full + // interactive control plane over the authenticated HTTPS tunnel. + await page.routeWebSocket('**/ui/ws', (webSocket) => webSocket.close()); + const fallbackConnected = page.waitForResponse((response) => + response.url().endsWith('/api/ui/connect') && response.ok(), + ); + await page.goto('/?agent=journey-agent'); + await fallbackConnected; + const terminal = page.locator('.xterm'); + await expect(terminal).toBeVisible({ timeout: 30_000 }); + await terminal.click({ position: { x: 24, y: 24 } }); + const fallbackMarker = 'SHELLFLEET_HTTP_FALLBACK_OK'; + const octalMarker = [...fallbackMarker] + .map((character) => `\\${character.charCodeAt(0).toString(8).padStart(3, '0')}`) + .join(''); + // Type an encoded command so the assertion can only match command output, + // not the terminal's local echo of what Playwright typed. + await page.keyboard.type(`printf '${octalMarker}\\n'`); + await page.keyboard.press('Enter'); + await expect(page.locator('.xterm-accessibility-tree')).toContainText( + fallbackMarker, + { timeout: 30_000 }, + ); + // Android browsers can expose a desktop-class CSS viewport near 930px. // Keep navigation off-canvas and stack the selected-host panes throughout // that compact range instead of squeezing two dashboards side-by-side. diff --git a/web/src/components/providers/WebSocketProvider.tsx b/web/src/components/providers/WebSocketProvider.tsx index f5ab5307..32c4f457 100644 --- a/web/src/components/providers/WebSocketProvider.tsx +++ b/web/src/components/providers/WebSocketProvider.tsx @@ -12,6 +12,7 @@ import { import { AgentMessagePayload, UiMessage } from '@/lib/types'; import { effectiveAgentDirectory } from '@/lib/agentDirectory'; import { reconnectDelay } from '@/lib/backoff'; +import { apiFetch } from '@/lib/api'; import { useSession } from './SessionProvider'; import { useCoreFleet } from './CoreFleetProvider'; import { useUi } from './UiProvider'; @@ -39,6 +40,17 @@ const WebSocketContext = createContext(null); const CONNECT_TIMEOUT_MS = 12_000; const DIRECTORY_SYNC_INTERVAL_MS = 15_000; const DIRECTORY_STALE_AFTER_MS = 45_000; +const HTTP_SEND_BATCH_DELAY_MS = 12; +const HTTP_SEND_MAX_BATCH = 128; + +type HttpControlSession = { + clientId: number; + clientToken: string; +}; + +type HttpControlResponse = { + messages: UiMessage[]; +}; // Resolve at connection time, in the browser, from the page's current // origin. NEXT_PUBLIC_* values are frozen into Next.js client bundles at build @@ -68,6 +80,9 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { const connectTimeout = useRef | null>(null); const directorySyncTimer = useRef | null>(null); const reconnectAttempt = useRef(0); + // Installed by the connection effect while the HTTPS control fallback is + // active. sendMessage stays stable for every consumer and selects WS first. + const httpSendRef = useRef<((message: UiMessage) => void) | null>(null); // Subscribers stored in a ref so message dispatch never races with React's // render cycle. The previous implementation kept the "last message" in // useState, which dropped events when several messages arrived in the @@ -105,6 +120,23 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { [], ); + const handleIncomingMessage = useCallback((msg: UiMessage) => { + if (msg.type === 'ListAgentsResponse') { + setIsConnected(true); + setSocketAgents(msg.payload.agents); + setSocketCapabilities(msg.payload.capabilities ?? {}); + } else if (msg.type === 'AgentMessage') { + dispatch(msg.payload.agent_id, msg.payload.message); + } else if (msg.type === 'PermissionDenied') { + const { variant_type, reason } = msg.payload; + if (variant_type === 'approval_pending') { + toastRef.current('info', reason); + } else { + toastRef.current('error', `${variant_type} denied: ${reason}`); + } + } + }, [dispatch]); + useEffect(() => { // Only open the WS once the session is fully authed. Connecting // earlier (during /login, /mfa, /security with a pending-MFA @@ -118,6 +150,13 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { } let disposed = false; let lastDirectoryResponseAt = 0; + let httpSession: HttpControlSession | null = null; + let httpStarting = false; + let httpSending = false; + let httpQueue: UiMessage[] = []; + let httpFlushTimer: ReturnType | null = null; + let httpPollAbort: AbortController | null = null; + let httpDirectoryTimer: ReturnType | null = null; reconnectAttempt.current = 0; const clearReconnectTimer = () => { @@ -136,6 +175,10 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { clearInterval(directorySyncTimer.current); directorySyncTimer.current = null; } + if (httpDirectoryTimer !== null) { + clearInterval(httpDirectoryTimer); + httpDirectoryTimer = null; + } }; const resetSocketState = () => { @@ -144,6 +187,175 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { setSocketCapabilities({}); }; + const receiveUiMessage = (message: UiMessage) => { + if (message.type === 'ListAgentsResponse') { + lastDirectoryResponseAt = Date.now(); + } + handleIncomingMessage(message); + }; + + const disconnectHttpSession = (session: HttpControlSession) => { + // `keepalive` gives browsers a chance to release the server-side slot + // during reload/navigation. The origin's idle reaper remains the final + // cleanup path if the network is already unavailable. + void apiFetch('/api/ui/disconnect', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + client_id: session.clientId, + client_token: session.clientToken, + }), + keepalive: true, + }).catch(() => {}); + }; + + const failHttpTransport = () => { + if (!httpSession && !httpStarting) return; + const failedSession = httpSession; + httpSession = null; + httpStarting = false; + httpSendRef.current = null; + httpPollAbort?.abort(); + httpPollAbort = null; + if (httpFlushTimer !== null) { + clearTimeout(httpFlushTimer); + httpFlushTimer = null; + } + if (httpDirectoryTimer !== null) { + clearInterval(httpDirectoryTimer); + httpDirectoryTimer = null; + } + // A failed POST may have reached the origin even if its response did + // not reach us. Never replay queued control actions automatically. + httpQueue = []; + if (failedSession) disconnectHttpSession(failedSession); + resetSocketState(); + scheduleReconnect(); + }; + + const flushHttpQueue = async () => { + if (disposed || httpSending || !httpSession || httpQueue.length === 0) return; + httpSending = true; + const session = httpSession; + const messages = httpQueue.splice(0, HTTP_SEND_MAX_BATCH); + try { + const response = await apiFetch('/api/ui/send', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + client_id: session.clientId, + client_token: session.clientToken, + messages, + }), + }); + if (!response.ok) throw new Error(`HTTP control send returned ${response.status}`); + } catch (error) { + if (!disposed && httpSession === session) { + console.warn('[shellfleet] HTTPS control send failed:', error); + failHttpTransport(); + } + } finally { + httpSending = false; + if (!disposed && httpSession && httpQueue.length > 0 && httpFlushTimer === null) { + httpFlushTimer = setTimeout(() => { + httpFlushTimer = null; + void flushHttpQueue(); + }, HTTP_SEND_BATCH_DELAY_MS); + } + } + }; + + const enqueueHttpMessage = (message: UiMessage) => { + if (!httpSession || disposed) return; + httpQueue.push(message); + if (httpFlushTimer === null) { + httpFlushTimer = setTimeout(() => { + httpFlushTimer = null; + void flushHttpQueue(); + }, HTTP_SEND_BATCH_DELAY_MS); + } + }; + + const pollHttp = async (session: HttpControlSession) => { + while (!disposed && httpSession === session) { + const controller = new AbortController(); + httpPollAbort = controller; + try { + const response = await apiFetch('/api/ui/poll', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + client_id: session.clientId, + client_token: session.clientToken, + }), + signal: controller.signal, + }); + if (!response.ok) throw new Error(`HTTP control poll returned ${response.status}`); + const body = (await response.json()) as HttpControlResponse; + for (const message of body.messages ?? []) receiveUiMessage(message); + } catch (error) { + if (!disposed && !controller.signal.aborted && httpSession === session) { + console.warn('[shellfleet] HTTPS control poll failed:', error); + failHttpTransport(); + } + return; + } finally { + if (httpPollAbort === controller) httpPollAbort = null; + } + } + }; + + async function startHttpFallback() { + if (disposed || httpStarting || httpSession) return; + httpStarting = true; + clearReconnectTimer(); + const current = wsRef.current; + wsRef.current = null; + try { + current?.close(); + } catch { + /* ignore the failed WebSocket while switching transports */ + } + try { + const response = await apiFetch('/api/ui/connect', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: '{}', + }); + if (!response.ok) throw new Error(`HTTP control connect returned ${response.status}`); + const body = (await response.json()) as HttpControlResponse & { + client_id: number; + client_token: string; + }; + if (!Number.isSafeInteger(body.client_id) || !body.client_token) { + throw new Error('HTTP control connect returned an invalid session'); + } + if (disposed) return; + const session = { clientId: body.client_id, clientToken: body.client_token }; + httpSession = session; + httpSendRef.current = enqueueHttpMessage; + reconnectAttempt.current = 0; + for (const message of body.messages ?? []) receiveUiMessage(message); + httpDirectoryTimer = setInterval(() => { + if (Date.now() - lastDirectoryResponseAt >= DIRECTORY_STALE_AFTER_MS) { + failHttpTransport(); + return; + } + enqueueHttpMessage({ type: 'ListAgentsRequest' }); + }, DIRECTORY_SYNC_INTERVAL_MS); + console.info('[shellfleet] interactive controls using HTTPS fallback'); + void pollHttp(session); + } catch (error) { + if (!disposed) { + console.warn('[shellfleet] HTTPS control fallback unavailable:', error); + httpStarting = false; + scheduleReconnect(); + } + } finally { + httpStarting = false; + } + } + const requestDirectory = (ws: WebSocket) => { if (ws.readyState !== WebSocket.OPEN) return false; try { @@ -156,7 +368,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { }; const scheduleReconnect = () => { - if (disposed || reconnectTimer.current !== null) { + if (disposed || httpSession || httpStarting || reconnectTimer.current !== null) { return; } const delay = reconnectDelay(reconnectAttempt.current); @@ -179,11 +391,11 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { /* the browser can throw while a socket is still being created */ } } - scheduleReconnect(); + void startHttpFallback(); }; const connect = () => { - if (disposed) { + if (disposed || httpSession || httpStarting) { return; } const current = wsRef.current; @@ -199,7 +411,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { ws = new WebSocket(resolveWsUrl()); } catch (error) { console.error('[shellfleet] failed to create UI WebSocket:', error); - scheduleReconnect(); + void startHttpFallback(); return; } wsRef.current = ws; @@ -261,23 +473,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { if (disposed || wsRef.current !== ws) return; try { const msg = JSON.parse(event.data) as UiMessage; - if (msg.type === 'ListAgentsResponse') { - lastDirectoryResponseAt = Date.now(); - setIsConnected(true); - setSocketAgents(msg.payload.agents); - setSocketCapabilities(msg.payload.capabilities ?? {}); - } else if (msg.type === 'AgentMessage') { - dispatch(msg.payload.agent_id, msg.payload.message); - } else if (msg.type === 'PermissionDenied') { - const { variant_type, reason } = msg.payload; - // approval_pending isn't a denial — the action is held awaiting a - // second admin's sign-off. Show it as info, not an error. - if (variant_type === 'approval_pending') { - toastRef.current('info', reason); - } else { - toastRef.current('error', `${variant_type} denied: ${reason}`); - } - } + receiveUiMessage(msg); } catch (e) { console.error('failed to parse WS message:', e); } @@ -286,6 +482,10 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { const recoverNow = () => { if (disposed) return; + if (httpSession) { + enqueueHttpMessage({ type: 'ListAgentsRequest' }); + return; + } clearReconnectTimer(); const ws = wsRef.current; if (!ws || ws.readyState === WebSocket.CLOSED || ws.readyState === WebSocket.CLOSING) { @@ -321,6 +521,14 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { document.removeEventListener('visibilitychange', handleVisibility); clearReconnectTimer(); clearConnectionTimers(); + httpSendRef.current = null; + const activeHttpSession = httpSession; + httpSession = null; + httpStarting = false; + httpPollAbort?.abort(); + if (httpFlushTimer !== null) clearTimeout(httpFlushTimer); + httpQueue = []; + if (activeHttpSession) disconnectHttpSession(activeHttpSession); const ws = wsRef.current; wsRef.current = null; try { @@ -329,7 +537,7 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { /* ignore cleanup errors */ } }; - }, [dispatch, status]); + }, [handleIncomingMessage, status]); const sendMessage = useCallback((msg: UiMessage) => { if (wsRef.current && wsRef.current.readyState === WebSocket.OPEN) { @@ -344,7 +552,9 @@ export function WebSocketProvider({ children }: { children: React.ReactNode }) { /* ignore */ } } + return; } + httpSendRef.current?.(msg); }, []); const sendToAgent = useCallback( diff --git a/web/src/components/providers/__tests__/WebSocketProvider.test.tsx b/web/src/components/providers/__tests__/WebSocketProvider.test.tsx index 25db00d6..7d01f675 100644 --- a/web/src/components/providers/__tests__/WebSocketProvider.test.tsx +++ b/web/src/components/providers/__tests__/WebSocketProvider.test.tsx @@ -1,5 +1,5 @@ import '@testing-library/jest-dom/vitest'; -import { act, cleanup, render, screen } from '@testing-library/react'; +import { act, cleanup, fireEvent, render, screen } from '@testing-library/react'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { WebSocketProvider, useWebSocket } from '../WebSocketProvider'; @@ -64,12 +64,13 @@ class MockWebSocket { } function Probe() { - const { agents, isConnected, liveAgents } = useWebSocket(); + const { agents, isConnected, liveAgents, sendMessage } = useWebSocket(); return (
{isConnected ? 'connected' : 'disconnected'} {`agents:${agents.join(',')}`} {`live:${liveAgents.join(',')}`} +
); } @@ -89,6 +90,7 @@ describe('WebSocketProvider', () => { MockWebSocket.instances = []; vi.useFakeTimers(); vi.stubGlobal('WebSocket', MockWebSocket); + vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('HTTP fallback unavailable'))); }); afterEach(() => { @@ -121,7 +123,7 @@ describe('WebSocketProvider', () => { expect(JSON.parse(socket.sent[1])).toEqual({ type: 'ListAgentsRequest' }); }); - it('retires a half-open socket and reconnects when directory responses stop', () => { + it('retires a half-open socket and reconnects when directory responses stop', async () => { renderProvider(); const socket = MockWebSocket.instances[0]; act(() => { @@ -137,23 +139,25 @@ describe('WebSocketProvider', () => { expect(screen.getByText('disconnected')).toBeInTheDocument(); expect(screen.getByText('agents:')).toBeInTheDocument(); + await act(async () => {}); act(() => vi.advanceTimersByTime(1_000)); expect(MockWebSocket.instances).toHaveLength(2); expect(MockWebSocket.instances[1].readyState).toBe(MockWebSocket.CONNECTING); }); - it('does not wait forever for the opening handshake', () => { + it('does not wait forever for the opening handshake', async () => { renderProvider(); const socket = MockWebSocket.instances[0]; act(() => vi.advanceTimersByTime(12_000)); expect(socket.readyState).toBe(MockWebSocket.CLOSED); + await act(async () => {}); act(() => vi.advanceTimersByTime(1_000)); expect(MockWebSocket.instances).toHaveLength(2); }); - it('reconnects immediately when the browser reports that the network returned', () => { + it('reconnects immediately when the browser reports that the network returned', async () => { renderProvider(); const socket = MockWebSocket.instances[0]; act(() => { @@ -162,6 +166,7 @@ describe('WebSocketProvider', () => { }); expect(MockWebSocket.instances).toHaveLength(1); + await act(async () => {}); act(() => window.dispatchEvent(new Event('online'))); expect(MockWebSocket.instances).toHaveLength(2); }); @@ -184,14 +189,15 @@ describe('WebSocketProvider', () => { expect(MockWebSocket.instances[0].url).toBe('ws://localhost:3000/ui/ws'); }); - it('ignores late messages from a retired socket generation', () => { + it('ignores late messages from a retired socket generation', async () => { renderProvider(); const first = MockWebSocket.instances[0]; act(() => { first.open(); first.serverClose(); - window.dispatchEvent(new Event('online')); }); + await act(async () => {}); + act(() => window.dispatchEvent(new Event('online'))); const second = MockWebSocket.instances[1]; act(() => { second.open(); @@ -224,4 +230,54 @@ describe('WebSocketProvider', () => { act(() => vi.advanceTimersByTime(60_000)); expect(MockWebSocket.instances).toHaveLength(1); }); + + it('keeps interactive controls live over HTTPS when WebSockets are blocked', async () => { + const requests: { url: string; body: unknown }[] = []; + const fetchMock = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + const body = init?.body ? JSON.parse(String(init.body)) : null; + requests.push({ url, body }); + if (url === '/api/ui/connect') { + return Promise.resolve(new Response(JSON.stringify({ + client_id: 7, + client_token: 'control-token', + messages: [{ + type: 'ListAgentsResponse', + payload: { agents: ['fallback-id'], capabilities: { 'fallback-id': ['systemd'] } }, + }], + }), { status: 200, headers: { 'Content-Type': 'application/json' } })); + } + if (url === '/api/ui/poll') { + return new Promise(() => {}); + } + if (url === '/api/ui/send') { + return Promise.resolve(new Response('{"ok":true}', { status: 200 })); + } + if (url === '/api/ui/disconnect') { + return Promise.resolve(new Response('{"ok":true}', { status: 200 })); + } + return Promise.reject(new Error(`unexpected URL: ${url}`)); + }); + vi.stubGlobal('fetch', fetchMock); + + renderProvider(); + act(() => MockWebSocket.instances[0].serverClose()); + await act(async () => {}); + + expect(screen.getByText('connected')).toBeInTheDocument(); + expect(screen.getByText('live:fallback-id')).toBeInTheDocument(); + + fireEvent.click(screen.getByRole('button', { name: 'refresh agents' })); + await act(async () => { + vi.advanceTimersByTime(HTTP_SEND_BATCH_DELAY_MS_FOR_TEST); + }); + const send = requests.find((request) => request.url === '/api/ui/send'); + expect(send?.body).toEqual({ + client_id: 7, + client_token: 'control-token', + messages: [{ type: 'ListAgentsRequest' }], + }); + }); }); + +const HTTP_SEND_BATCH_DELAY_MS_FOR_TEST = 12; From 918f037b6d8e534676a4dd4623e4870a92872407 Mon Sep 17 00:00:00 2001 From: Ramshouriesh R Date: Sun, 23 Aug 2026 23:10:53 +0530 Subject: [PATCH 2/3] fix: keep HTTP control auth errors compact --- server/src/main.rs | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/server/src/main.rs b/server/src/main.rs index e1c6e25e..b17216cc 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -1953,23 +1953,21 @@ async fn authenticate_http_ui_client( state: &AppState, jar: &CookieJar, request: &UiHttpClientRequest, -) -> Result<(UiCommandTx, UiHttpRx), axum::response::Response> { - let claims = auth::current_user(jar, &state.db) - .await - .map_err(|(status, reason)| (status, reason).into_response())?; +) -> Result<(UiCommandTx, UiHttpRx), (StatusCode, &'static str)> { + let claims = auth::current_user(jar, &state.db).await?; let clients = state.ui_clients.lock().await; let Some(client) = clients.get(&request.client_id) else { - return Err((StatusCode::GONE, "control session expired").into_response()); + return Err((StatusCode::GONE, "control session expired")); }; let Some(expected_token) = client.http_token.as_deref() else { - return Err((StatusCode::GONE, "not an HTTP control session").into_response()); + return Err((StatusCode::GONE, "not an HTTP control session")); }; let token_matches = http_control_token_matches(expected_token, &request.client_token); if client.login != claims.sub || !token_matches { - return Err((StatusCode::FORBIDDEN, "control session mismatch").into_response()); + return Err((StatusCode::FORBIDDEN, "control session mismatch")); } let Some(receiver) = client.http_receiver.clone() else { - return Err((StatusCode::GONE, "control session unavailable").into_response()); + return Err((StatusCode::GONE, "control session unavailable")); }; client .last_seen @@ -2030,7 +2028,7 @@ async fn ui_http_poll_handler( ) -> axum::response::Response { let (_, receiver) = match authenticate_http_ui_client(&state, &jar, &request).await { Ok(client) => client, - Err(response) => return response, + Err(error) => return error.into_response(), }; // Only one long poll may wait on a control session at once. Returning a // conflict instead of queuing arbitrary concurrent requests bounds origin @@ -2048,8 +2046,8 @@ async fn ui_http_disconnect_handler( State(state): State>, Json(request): Json, ) -> axum::response::Response { - if let Err(response) = authenticate_http_ui_client(&state, &jar, &request).await { - return response; + if let Err(error) = authenticate_http_ui_client(&state, &jar, &request).await { + return error.into_response(); } state.ui_clients.lock().await.remove(&request.client_id); no_store_json(serde_json::json!({ "ok": true })) @@ -2076,7 +2074,7 @@ async fn ui_http_send_handler( }; let (command_tx, _) = match authenticate_http_ui_client(&state, &jar, &client_request).await { Ok(client) => client, - Err(response) => return response, + Err(error) => return error.into_response(), }; for message in request.messages { if command_tx.send(message).is_err() { From 13eda02220f869f740f7e34b1cf0f1f44120c33f Mon Sep 17 00:00:00 2001 From: Ramshouriesh R Date: Sun, 23 Aug 2026 23:25:28 +0530 Subject: [PATCH 3/3] test: exercise fallback through root terminal broker --- Dockerfile.agent | 4 ++++ tests/journey/docker-compose.yml | 24 ++++++++++++++++++++++++ tests/journey/fleet-read-plane.spec.ts | 21 +++++++++++++++++---- 3 files changed, 45 insertions(+), 4 deletions(-) diff --git a/Dockerfile.agent b/Dockerfile.agent index 3bee4827..eb102778 100644 --- a/Dockerfile.agent +++ b/Dockerfile.agent @@ -12,6 +12,10 @@ RUN apt-get update && \ apt-get install -y --no-install-recommends systemd dbus bash ca-certificates && \ rm -rf /var/lib/apt/lists/* COPY --from=builder /usr/src/app/target/release/shellfleet-agent /app/agent +# The browser journey runs the local root broker as a separate root container +# and shares only its Unix socket with this restricted agent container. Native +# installs still run both binaries under their hardened systemd units. +COPY --from=builder /usr/src/app/target/release/shellfleet-approval-gate /app/gate # This image is the restricted local end-to-end test target (see # docker-compose.yml). It explicitly selects the restricted runtime contract # and therefore refuses root and effective capabilities. diff --git a/tests/journey/docker-compose.yml b/tests/journey/docker-compose.yml index f7aee135..3b25e238 100644 --- a/tests/journey/docker-compose.yml +++ b/tests/journey/docker-compose.yml @@ -62,9 +62,32 @@ services: SERVER_WS_URL: ws://server:8080/agent/ws volumes: - ./agent-token.txt:/var/lib/shellfleet-agent/agent-token.txt:ro + - gate-run:/run/shellfleet depends_on: server: condition: service_healthy + gate: + condition: service_healthy + networks: + - journey + + gate: + build: + context: ../.. + dockerfile: Dockerfile.agent + user: "0:10001" + command: ["/app/gate"] + restart: unless-stopped + environment: + SHELLFLEET_AGENT_UID: "10001" + volumes: + - gate-run:/run/shellfleet + healthcheck: + test: ["CMD-SHELL", "test -S /run/shellfleet/approval-gate.sock"] + interval: 1s + timeout: 1s + retries: 30 + start_period: 1s networks: - journey @@ -96,4 +119,5 @@ networks: driver: bridge volumes: + gate-run: server-data: diff --git a/tests/journey/fleet-read-plane.spec.ts b/tests/journey/fleet-read-plane.spec.ts index 51ba0ab0..43705541 100644 --- a/tests/journey/fleet-read-plane.spec.ts +++ b/tests/journey/fleet-read-plane.spec.ts @@ -82,14 +82,27 @@ test('Fleet remains durable across reload, disconnect, SSE loss, and reconnect', const octalMarker = [...fallbackMarker] .map((character) => `\\${character.charCodeAt(0).toString(8).padStart(3, '0')}`) .join(''); + let fallbackOutput = ''; + const terminalOutput = page.waitForResponse(async (response) => { + if (!response.url().endsWith('/api/ui/poll') || !response.ok()) return false; + const body = await response.json().catch(() => null) as { + messages?: Array<{ + payload?: { message?: { payload?: { data?: unknown } } }; + }>; + } | null; + for (const message of body?.messages ?? []) { + const data = message.payload?.message?.payload?.data; + if (Array.isArray(data) && data.every((byte) => Number.isInteger(byte))) { + fallbackOutput += Buffer.from(data).toString('utf8'); + } + } + return fallbackOutput.includes(fallbackMarker); + }, { timeout: 30_000 }); // Type an encoded command so the assertion can only match command output, // not the terminal's local echo of what Playwright typed. await page.keyboard.type(`printf '${octalMarker}\\n'`); await page.keyboard.press('Enter'); - await expect(page.locator('.xterm-accessibility-tree')).toContainText( - fallbackMarker, - { timeout: 30_000 }, - ); + await terminalOutput; // Android browsers can expose a desktop-class CSS viewport near 930px. // Keep navigation off-canvas and stack the selected-host panes throughout