diff --git a/crates/openshell-driver-mxc/src/driver.rs b/crates/openshell-driver-mxc/src/driver.rs index 4ec9613eb..5253a3ae7 100644 --- a/crates/openshell-driver-mxc/src/driver.rs +++ b/crates/openshell-driver-mxc/src/driver.rs @@ -611,6 +611,15 @@ async fn run_lifecycle_inner( context: &LifecycleContext, startup_guard: tokio::sync::OwnedMutexGuard<()>, ) -> Result<(), String> { + let main = mxc_main_process_config( + context.sandbox.spec.as_ref(), + &context.sandbox_config.command, + ); + let workload_binary = resolve_workload_binary( + main.command + .first() + .ok_or_else(|| "MXC workload command is empty".to_string())?, + )?; create_restricted_state_dir(&context.host_state_dir, "host")?; // This directory briefly contains the boundary TLS private key and direct // proxy credential. Restrict host access before writing either secret; @@ -684,7 +693,7 @@ async fn run_lifecycle_inner( proxy_addr, proxy_authorization: authorization, proxy_url, - workload_binary: resolve_workload_binary(&context.sandbox_config.command[0])?, + workload_binary, child_env: sandbox_environment(&context.sandbox), } .provision() @@ -720,7 +729,7 @@ async fn run_lifecycle_inner( // ProcessContainer launch so unrelated local processes cannot squat it // during the more expensive host-side setup. drop(proxy_reservation); - let mut supervisor = spawn_supervisor(context, &descriptor_path, &auth_bundle_path)?; + let mut supervisor = spawn_supervisor(context, &descriptor_path, &auth_bundle_path, &main)?; let sandbox_command = encode_windows_command_line(&[ context.config.sandbox_binary_path.clone(), "--bootstrap".to_string(), @@ -815,11 +824,9 @@ fn spawn_supervisor( context: &LifecycleContext, descriptor_path: &Path, auth_bundle_path: &Path, + main: &openshell_core::sandbox_env::MainProcessConfig, ) -> Result { - let main_process_spec = openshell_core::sandbox_env::MainProcessConfig::encode_driver_spec( - context.sandbox.spec.as_ref(), - ) - .map_err(|error| format!("encode MXC main process spec: {error}"))?; + let main_process_spec = mxc_main_process_spec(main)?; let mut command = Command::new(&context.config.supervisor_binary_path); command .kill_on_drop(true) @@ -864,6 +871,24 @@ fn spawn_supervisor( .map_err(|error| format!("start host openshell-supervisor: {error}")) } +fn mxc_main_process_spec( + main: &openshell_core::sandbox_env::MainProcessConfig, +) -> Result { + serde_json::to_string(main).map_err(|error| format!("encode MXC main process spec: {error}")) +} + +fn mxc_main_process_config( + spec: Option<&DriverSandboxSpec>, + driver_command: &[String], +) -> openshell_core::sandbox_env::MainProcessConfig { + let mut main = openshell_core::sandbox_env::MainProcessConfig::from_driver_spec(spec); + // Match windows: the MXC driver-config command remains a supported launch + // source when no canonical command was supplied by the CLI. + if main.command.is_empty() { + main.command = driver_command.to_vec(); + } + main +} enum RuntimePairExit { Shutdown, Boundary(std::process::ExitStatus), @@ -1232,6 +1257,54 @@ fn platform_event(sandbox_id: String, reason: &str, message: String) -> WatchSan // the driver. Keep lifecycle/security tests below during their migration. #[cfg(test)] mod tests { + #[test] + fn mxc_driver_command_reaches_supervisor_without_generic_command() { + let command = vec![ + "cmd.exe".into(), + "/c".into(), + "echo ok 1> C:/work/result.txt".into(), + ]; + let encoded = mxc_main_process_spec(&mxc_main_process_config(None, &command)).unwrap(); + let main = openshell_core::sandbox_env::MainProcessConfig::decode(&encoded).unwrap(); + assert_eq!(main.command, command); + assert!(!main.await_main_process_attachment); + let spec = DriverSandboxSpec { + command: vec!["explicit.exe".into()], + ..Default::default() + }; + let encoded = + mxc_main_process_spec(&mxc_main_process_config(Some(&spec), &command)).unwrap(); + assert_eq!( + openshell_core::sandbox_env::MainProcessConfig::decode(&encoded) + .unwrap() + .command, + spec.command + ); + } + + #[test] + fn proxy_identity_uses_the_canonical_workload_command() { + let canonical = std::env::var("COMSPEC").unwrap(); + let spec = DriverSandboxSpec { + command: vec![canonical.clone(), "/C".into(), "exit 0".into()], + ..Default::default() + }; + let fallback = vec!["not-the-workload.exe".into()]; + let main = mxc_main_process_config(Some(&spec), &fallback); + assert_eq!(main.command, spec.command); + assert_eq!( + resolve_workload_binary(&main.command[0]).unwrap(), + resolve_workload_binary(&canonical).unwrap() + ); + assert_eq!( + openshell_core::sandbox_env::MainProcessConfig::decode( + &mxc_main_process_spec(&main).unwrap() + ) + .unwrap() + .command, + main.command + ); + } use super::*; #[test] diff --git a/crates/openshell-driver-mxc/src/mxc.rs b/crates/openshell-driver-mxc/src/mxc.rs index 99a3e6fce..c7ed82195 100644 --- a/crates/openshell-driver-mxc/src/mxc.rs +++ b/crates/openshell-driver-mxc/src/mxc.rs @@ -266,6 +266,18 @@ impl WxcExecInvoker { self.mock } + fn real_command(&self) -> Command { + let mut command = Command::new(&self.exec_path); + command + // Both processes in the runtime pair must die if their monitor is + // cancelled, not only when it takes the explicit shutdown branch. + .kill_on_drop(true) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()); + command + } + /// Launch the real boundary runtime without OS isolation for wiring tests. /// Filesystem denial must be tested with real MXC, not this mock. fn mock_spawn_with_grants( @@ -368,13 +380,8 @@ impl WxcExecInvoker { } let b64 = base64::engine::general_purpose::STANDARD.encode(json.as_bytes()); - let mut cmd = Command::new(&self.exec_path); - cmd.arg("--config-base64") - .arg(&b64) - // Retain child output so the gateway can surface sandbox logs. - .stdin(std::process::Stdio::piped()) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); + let mut cmd = self.real_command(); + cmd.arg("--config-base64").arg(&b64); if self.debug { cmd.arg("--debug"); } @@ -391,6 +398,43 @@ impl WxcExecInvoker { mod tests { use super::*; + #[cfg(target_os = "windows")] + #[tokio::test] + #[allow(unsafe_code)] // Native synchronization handle verifies cancellation. + async fn real_launcher_command_terminates_on_drop() { + use windows::Win32::Foundation::{CloseHandle, WAIT_OBJECT_0, WAIT_TIMEOUT}; + use windows::Win32::System::Threading::{ + OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject, + }; + + // Exercise the real launcher's command construction with a native, + // long-lived process. This is not an MXC enforcement test or mock E2E. + let powershell = PathBuf::from(std::env::var("SystemRoot").unwrap()) + .join("System32/WindowsPowerShell/v1.0/powershell.exe"); + let invoker = WxcExecInvoker::new(powershell, false); + let child = invoker + .real_command() + .args([ + "-NoProfile", + "-NonInteractive", + "-Command", + "Start-Sleep -Seconds 60", + ]) + .spawn() + .expect("launch long-lived PowerShell process"); + // SAFETY: OpenProcess supplies a synchronization-only handle for the + // child, retained through termination and closed exactly once below. + unsafe { + let handle = OpenProcess(PROCESS_SYNCHRONIZE, false, child.id().unwrap()) + .expect("open child synchronization handle"); + assert_eq!(WaitForSingleObject(handle, 0), WAIT_TIMEOUT); + drop(child); + let exited = WaitForSingleObject(handle, 5_000); + CloseHandle(handle).expect("close child handle"); + assert_eq!(exited, WAIT_OBJECT_0, "cancelled launcher must terminate"); + } + } + #[test] fn oneshot_processcontainer_config_json_shape() { // Mirror the JSON `run_oneshot` builds for the one-shot processContainer diff --git a/crates/openshell-mxc-boundary/README.md b/crates/openshell-mxc-boundary/README.md index b768fec8c..58835da0e 100644 --- a/crates/openshell-mxc-boundary/README.md +++ b/crates/openshell-mxc-boundary/README.md @@ -25,6 +25,23 @@ and common lifecycle validation to the shared Sandbox Protocol implementation. The backend keeps evidence opaque and compares validated properties with the confirmation; its default Linux validator rejects Windows evidence. +MXC initiates the control connection to the host backend's loopback listener, +as the Windows branch relay does. The boundary remains the TLS server on that +outbound socket; the supervisor validates its pinned certificate and supplies +the authenticated Sandbox Protocol bearer. A backend-owned raw connector lets +discovery, attachment, and reconnection retain this direction without adding +MXC interpretation to shared transport or changing host firewall policy. + +Authenticated RPCs renew the connection's JWT expiry deadline. Expiry or +connection shutdown cancels open process and forwarding exchanges, including +idle streams that have not sent a request yet. + +MXC owns separate host-launch and one-use boundary-bootstrap envelopes. Host +proxy address, authorization, and workload identity stay in the MXC launch +payload; the authenticated workload proxy URL stays in the MXC bootstrap. +Windows supervisor composition turns these into ordinary shared CONNECT +listener options. Shared isolation traits and Sandbox Protocol descriptors +contain no MXC proxy fields, and other compute drivers need no placeholders. MXC adopts main's authenticated image-policy discovery and ordered provider environment publication. Command-based MXC has no rootfs image policy to discover. Publication generations prevent stale refreshes from replacing newer @@ -33,6 +50,12 @@ is unsupported and rejected before process launch or environment replacement. Structured shell exec intent and runtime helpers are also rejected explicitly; they must not be silently treated as raw executable launches. +Main launch and raw exec share the authenticated proxy and installed CA +environment. Exec combines these with the latest provider publication rather +than retaining startup credentials. Caller environment overrides cannot replace +the governed proxy or CA settings. Process attachments drain stdout and stderr +before emitting exit, with a bounded deadline for inherited pipes. + Dynamic service forwarding uses the shared protocol's `LoopbackConnect` request and bidirectional stream. The boundary connects to the requested loopback port; it does not launch a relay or connect back to a new gateway relay listener. diff --git a/crates/openshell-mxc-boundary/src/runtime.rs b/crates/openshell-mxc-boundary/src/runtime.rs index bdb16c81e..19f781e7b 100644 --- a/crates/openshell-mxc-boundary/src/runtime.rs +++ b/crates/openshell-mxc-boundary/src/runtime.rs @@ -46,6 +46,7 @@ const CONTROL_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10); const AUTHENTICATED_RECONNECT_TIMEOUT: Duration = Duration::from_secs(30); const OUTPUT_BUFFER_BYTES: usize = 1024 * 1024; const MAX_REPLAY_ENTRIES: usize = 4096; +const OUTPUT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); pub fn run_boundary(config_path: &Path) -> Result<(), String> { let bytes = std::fs::read(config_path) @@ -176,14 +177,15 @@ async fn serve_grpc( connection_id: SandboxConnectionId, ) -> Result<(), String> { let (connection_shutdown, mut connection_closed) = tokio::sync::watch::channel(()); + let connection_expiry = Arc::new(ConnectionExpiry::new(connection_shutdown.clone())); runtime.register_connection(connection_id, connection_shutdown.clone()); let incoming = tokio_stream::StreamExt::chain( tokio_stream::iter([Ok::<_, io::Error>(GrpcServerIo { stream, - _connection_alive: connection_shutdown, _disconnect: DisconnectGuard { runtime: Arc::downgrade(&runtime), connection_id, + connection_shutdown, }, })]), tokio_stream::pending(), @@ -194,6 +196,8 @@ async fn serve_grpc( .add_service(IsolationBoundaryServer::new(GrpcBoundaryService { runtime: runtime.clone(), connection_id, + connection_expiry, + connection_closed: connection_closed.clone(), })) .serve_with_incoming_shutdown(incoming, async move { let _ = connection_closed.changed().await; @@ -205,17 +209,21 @@ async fn serve_grpc( struct GrpcServerIo { stream: openshell_isolation_interface::contract::BoundaryDuplexStream, - _connection_alive: tokio::sync::watch::Sender<()>, _disconnect: DisconnectGuard, } struct DisconnectGuard { runtime: std::sync::Weak, connection_id: SandboxConnectionId, + connection_shutdown: tokio::sync::watch::Sender<()>, } impl Drop for DisconnectGuard { fn drop(&mut self) { + // Expiry owns a sender too, so dropping the physical socket alone no + // longer closes the watch channel. Wake the server and every exchange + // explicitly, including a discovery connection before attachment. + let _ = self.connection_shutdown.send(()); if let Some(runtime) = self.runtime.upgrade() { runtime.transport_disconnected(self.connection_id); } @@ -259,10 +267,65 @@ impl tonic::transport::server::Connected for GrpcServerIo { struct GrpcBoundaryService { runtime: Arc, connection_id: SandboxConnectionId, + connection_expiry: Arc, + connection_closed: tokio::sync::watch::Receiver<()>, } type GrpcResponseStream = ReceiverStream>; +// Match the Linux boundary: renewed RPC credentials update the physical +// connection deadline, and expiry cancels every open exchange on it. +struct ConnectionExpiry { + deadline: tokio::sync::watch::Sender>, + worker: tokio::task::AbortHandle, +} + +impl ConnectionExpiry { + fn new(connection_shutdown: tokio::sync::watch::Sender<()>) -> Self { + let (deadline, mut updates) = tokio::sync::watch::channel(None); + let worker = tokio::spawn(async move { + loop { + let deadline = *updates.borrow_and_update(); + if let Some(deadline) = deadline { + tokio::select! { + () = tokio::time::sleep_until(deadline) => { + let _ = connection_shutdown.send(()); + return; + } + result = updates.changed() => { + if result.is_err() { return; } + } + } + } else if updates.changed().await.is_err() { + return; + } + } + }) + .abort_handle(); + Self { deadline, worker } + } + + fn update(&self, expires_at: i64) { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()); + self.deadline.send_replace((expires_at != 0).then(|| { + tokio::time::Instant::now() + + Duration::from_secs( + u64::try_from(expires_at) + .unwrap_or_default() + .saturating_sub(now), + ) + })); + } +} + +impl Drop for ConnectionExpiry { + fn drop(&mut self) { + self.worker.abort(); + } +} + #[tonic::async_trait] impl IsolationBoundary for GrpcBoundaryService { type ExchangeStream = GrpcResponseStream; @@ -275,11 +338,21 @@ impl IsolationBoundary for GrpcBoundaryService { let principal = self .runtime .authenticate_request(self.connection_id, request.metadata())?; - let (stream, response) = bridge_grpc_stream(request.into_inner()); + self.connection_expiry + .update(principal.session().expires_at); + let (stream, response) = + bridge_grpc_stream(request.into_inner(), self.connection_closed.clone()); let runtime = self.runtime.clone(); + let mut connection_closed = self.connection_closed.clone(); tokio::spawn(async move { - if let Err(error) = serve_one(stream, runtime, principal).await { - tracing::warn!(%error, "MXC Sandbox Protocol exchange failed"); + tokio::select! { + biased; + _ = connection_closed.changed() => {} + result = serve_one(stream, runtime, principal) => { + if let Err(error) = result { + tracing::warn!(%error, "MXC Sandbox Protocol exchange failed"); + } + } } }); Ok(tonic::Response::new(response)) @@ -299,39 +372,57 @@ impl IsolationBoundary for GrpcBoundaryService { } fn bridge_grpc_stream( - mut inbound: tonic::Streaming, + mut inbound: impl tokio_stream::Stream> + + Send + + Unpin + + 'static, + connection_closed: tokio::sync::watch::Receiver<()>, ) -> (tokio::io::DuplexStream, GrpcResponseStream) { let (application, bridge) = tokio::io::duplex(256 * 1024); let (mut reader, mut writer) = tokio::io::split(bridge); let (outbound, outbound_rx) = tokio::sync::mpsc::channel(64); + let mut inbound_closed = connection_closed.clone(); tokio::spawn(async move { - loop { - match inbound.message().await { - Ok(Some(chunk)) if writer.write_all(&chunk.data).await.is_ok() => {} - Ok(Some(_)) | Err(_) => return, - Ok(None) => { - let _ = writer.shutdown().await; - return; + tokio::select! { + biased; + _ = inbound_closed.changed() => {} + () = async { + loop { + match tokio_stream::StreamExt::next(&mut inbound).await { + Some(Ok(chunk)) if writer.write_all(&chunk.data).await.is_ok() => {} + Some(Ok(_) | Err(_)) => return, + None => { + let _ = writer.shutdown().await; + return; + } + } } - } + } => {} } }); + let mut outbound_closed = connection_closed; tokio::spawn(async move { - let mut buffer = vec![0_u8; 16 * 1024]; - loop { - let Ok(read) = reader.read(&mut buffer).await else { - return; - }; - if read == 0 - || outbound - .send(Ok(BoundaryChunk { - data: buffer[..read].to_vec(), - })) - .await - .is_err() - { - return; - } + tokio::select! { + biased; + _ = outbound_closed.changed() => {} + () = async { + let mut buffer = vec![0_u8; 16 * 1024]; + loop { + let Ok(read) = reader.read(&mut buffer).await else { + return; + }; + if read == 0 + || outbound + .send(Ok(BoundaryChunk { + data: buffer[..read].to_vec(), + })) + .await + .is_err() + { + return; + } + } + } => {} } }); (application, ReceiverStream::new(outbound_rx)) @@ -363,6 +454,7 @@ struct BoundaryRuntime { processes: Mutex>>, main_process: Mutex>, provider_environment: Mutex<(u64, HashMap, u64)>, + ca_paths: Mutex>, replay: Mutex>, replay_order: Mutex>, exec_requests: Mutex>, @@ -406,6 +498,7 @@ impl BoundaryRuntime { processes: Mutex::new(HashMap::new()), main_process: Mutex::new(None), provider_environment: Mutex::new((0, HashMap::new(), 0)), + ca_paths: Mutex::new(None), replay: Mutex::new(HashMap::new()), replay_order: Mutex::new(VecDeque::new()), exec_requests: Mutex::new(HashSet::new()), @@ -736,26 +829,21 @@ impl BoundaryRuntime { return guest_error(BoundaryErrorKind::Denied, "MXC start policy changed"); } } - let mut environment = self.config.child_env.clone(); - environment.extend(provider_env.clone()); - if let Some(proxy_url) = &self.config.direct_proxy_url { - for key in ["HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"] { - environment.insert(key.to_string(), proxy_url.clone()); - } - environment.insert("NO_PROXY".to_string(), String::new()); - environment.insert("no_proxy".to_string(), String::new()); - } - match install_ca_material( + let ca_paths = match install_ca_material( &self.config.generation, ca_cert.as_deref().map(str::as_bytes), ca_bundle.as_deref().map(str::as_bytes), ) { - Ok(Some((certificate, bundle))) => { - install_ca_environment(&mut environment, &certificate, &bundle); - } - Ok(None) => {} + Ok(paths) => paths, Err(error) => return guest_error(BoundaryErrorKind::Process, error), - } + }; + let environment = workload_environment( + &self.config.child_env, + provider_env, + &[], + &self.proxy_url, + ca_paths.as_ref(), + ); let process_id = format!("{}:main:0", self.config.generation); let (program, args) = agent_command(spec.clone()); let process = match ManagedProcess::spawn( @@ -773,6 +861,7 @@ impl BoundaryRuntime { }; lock(&self.processes).insert(process_id.clone(), process); *lock(&self.main_process) = Some(process_id.clone()); + *lock(&self.ca_paths) = ca_paths; *lock(&self.provider_environment) = (*provider_env_revision, provider_env.clone(), 0); *lock(&self.lifecycle) = Lifecycle::Running; let response = Response::Started { @@ -811,9 +900,13 @@ impl BoundaryRuntime { self.config.generation, self.next_exec.fetch_add(1, Ordering::Relaxed) ); - let mut environment = self.config.child_env.clone(); - environment.extend(lock(&self.provider_environment).1.clone()); - environment.extend(spec.env.iter().cloned()); + let environment = workload_environment( + &self.config.child_env, + &lock(&self.provider_environment).1, + &spec.env, + &self.proxy_url, + lock(&self.ca_paths).as_ref(), + ); let process = match ManagedProcess::spawn( id.clone(), ProcessKindWire::Exec, @@ -1097,6 +1190,31 @@ fn install_ca_environment( } } +fn workload_environment( + child: &HashMap, + providers: &HashMap, + overrides: &[(String, String)], + proxy_url: &str, + ca_paths: Option<&(PathBuf, PathBuf)>, +) -> HashMap { + let mut environment = child.clone(); + environment.extend(providers.clone()); + environment.extend(overrides.iter().cloned()); + for (key, value) in [ + ("HTTP_PROXY", proxy_url), + ("HTTPS_PROXY", proxy_url), + ("NO_PROXY", ""), + ] { + environment.retain(|existing, _| !existing.eq_ignore_ascii_case(key)); + environment.insert(key.to_string(), value.to_string()); + environment.insert(key.to_ascii_lowercase(), value.to_string()); + } + if let Some((certificate, bundle)) = ca_paths { + install_ca_environment(&mut environment, certificate, bundle); + } + environment +} + fn install_ca_material( generation: &str, certificate: Option<&[u8]>, @@ -1127,6 +1245,27 @@ struct ManagedProcess { attached: Arc, } +async fn drain_output(readers: &mut [tokio::task::JoinHandle<()>; 2]) { + // A descendant can inherit the pipes after its parent exits. Bound the + // drain, then stop readers before publishing EXIT so no data follows it. + let drained = tokio::time::timeout(OUTPUT_DRAIN_TIMEOUT, async { + for reader in readers.iter_mut() { + let _ = reader.await; + } + }) + .await; + if drained.is_err() { + for reader in readers.iter() { + reader.abort(); + } + for reader in readers.iter_mut() { + if !reader.is_finished() { + let _ = reader.await; + } + } + } +} + impl ManagedProcess { async fn spawn( id: String, @@ -1162,8 +1301,10 @@ impl ManagedProcess { .take() .ok_or_else(|| "MXC workload stderr pipe is unavailable".to_string())?; let output = OutputLog::new(); - output.spawn_reader(stdout, STREAM_STDOUT); - output.spawn_reader(stderr, STREAM_STDERR); + let readers = [ + output.spawn_reader(stdout, STREAM_STDOUT), + output.spawn_reader(stderr, STREAM_STDERR), + ]; let (exit_tx, exit) = tokio::sync::watch::channel(None); let process = Self { id, @@ -1174,11 +1315,15 @@ impl ManagedProcess { exit, attached: Arc::new(AtomicBool::new(false)), }; - process.start_monitor(exit_tx); + process.start_monitor(exit_tx, readers); Ok(process) } - fn start_monitor(&self, exit_tx: tokio::sync::watch::Sender>) { + fn start_monitor( + &self, + exit_tx: tokio::sync::watch::Sender>, + mut readers: [tokio::task::JoinHandle<()>; 2], + ) { let child = self.child.clone(); let output = self.output.clone(); tokio::spawn(async move { @@ -1190,6 +1335,7 @@ impl ManagedProcess { match result { Ok(Some(status)) => { let status = ExitStatusWire::Exited(status.code().unwrap_or(1)); + drain_output(&mut readers).await; output.publish_exit(status); exit_tx.send_replace(Some(status)); return; @@ -1197,6 +1343,7 @@ impl ManagedProcess { Ok(None) => tokio::time::sleep(Duration::from_millis(25)).await, Err(_) => { let status = ExitStatusWire::Exited(1); + drain_output(&mut readers).await; output.publish_exit(status); exit_tx.send_replace(Some(status)); return; @@ -1295,7 +1442,7 @@ impl OutputLog { self: &Arc, mut reader: impl tokio::io::AsyncRead + Send + Unpin + 'static, channel: u8, - ) { + ) -> tokio::task::JoinHandle<()> { let output = self.clone(); tokio::spawn(async move { let mut buffer = vec![0_u8; 16 * 1024]; @@ -1305,7 +1452,7 @@ impl OutputLog { Ok(read) => output.publish(channel, buffer[..read].to_vec()), } } - }); + }) } fn publish_exit(&self, status: ExitStatusWire) { @@ -1405,7 +1552,7 @@ async fn bridge_process( let _guard = process.acquire_attachment()?; let (mut reader, mut writer) = tokio::io::split(stream); let input_process = process.clone(); - let mut input = tokio::spawn(async move { + let input = async move { while let Some((channel, payload)) = read_stream_frame(&mut reader).await? { match channel { STREAM_STDIN => { @@ -1428,14 +1575,13 @@ async fn bridge_process( } } Ok::<(), io::Error>(()) - }); + }; + tokio::pin!(input); let mut cursor = process.output.cursor(); loop { tokio::select! { result = &mut input => { - return result - .map_err(|error| format!("join MXC process input: {error}"))? - .map_err(|error| format!("read MXC process input: {error}")); + return result.map_err(|error| format!("read MXC process input: {error}")); } event = cursor.recv() => { let Some(event) = event else { return Ok(()) }; @@ -1486,6 +1632,117 @@ fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { mod tests { use super::*; + #[test] + fn exec_environment_keeps_proxy_and_ca_with_refreshed_providers() { + let ca_paths = ( + PathBuf::from(r"C:\trusted\ca.pem"), + PathBuf::from(r"C:\trusted\bundle.pem"), + ); + let child = HashMap::from([("BASE".into(), "bootstrap".into())]); + let providers = HashMap::from([("TOKEN".into(), "refreshed".into())]); + let overrides = vec![ + ("OTHER".into(), "exec".into()), + ("Https_Proxy".into(), "http://untrusted".into()), + ("ssl_cert_file".into(), "untrusted-ca".into()), + ]; + let proxy = "http://user:secret@127.0.0.1:3128"; + let environment = + workload_environment(&child, &providers, &overrides, proxy, Some(&ca_paths)); + for key in ["HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"] { + assert_eq!(environment[key], proxy); + } + assert_eq!(environment["NO_PROXY"], ""); + assert_eq!( + environment["SSL_CERT_FILE"], + ca_paths.1.display().to_string() + ); + assert_eq!( + environment["NODE_EXTRA_CA_CERTS"], + ca_paths.0.display().to_string() + ); + assert!(!environment.contains_key("Https_Proxy")); + assert!(!environment.contains_key("ssl_cert_file")); + assert_eq!(environment["TOKEN"], "refreshed"); + assert_eq!(environment["BASE"], "bootstrap"); + assert_eq!(environment["OTHER"], "exec"); + assert!( + !workload_environment(&child, &HashMap::new(), &[], proxy, Some(&ca_paths)) + .contains_key("TOKEN") + ); + } + + #[tokio::test] + async fn connection_expiry_closes_the_authenticated_connection() { + let (shutdown, mut closed) = tokio::sync::watch::channel(()); + let expiry = ConnectionExpiry::new(shutdown); + expiry.update(1); // Already expired: terminate immediately. + tokio::time::timeout(Duration::from_secs(2), closed.changed()) + .await + .expect("expiry fires") + .expect("shutdown notification"); + } + + #[tokio::test] + async fn expired_connection_cancels_idle_exchange_streams() { + let (shutdown, closed) = tokio::sync::watch::channel(()); + let expiry = ConnectionExpiry::new(shutdown); + let (inbound, inbound_rx) = tokio::sync::mpsc::channel(1); + let (mut application, mut response) = + bridge_grpc_stream(ReceiverStream::new(inbound_rx), closed); + expiry.update(1); + tokio::time::timeout(Duration::from_secs(2), async { + let mut byte = [0]; + assert_eq!(application.read(&mut byte).await.unwrap(), 0); + assert!(tokio_stream::StreamExt::next(&mut response).await.is_none()); + assert!( + inbound + .send(Ok(BoundaryChunk { data: vec![1] })) + .await + .is_err() + ); + }) + .await + .expect("expired exchanges must close without client activity"); + } + + #[tokio::test] + async fn physical_disconnect_closes_discovery_even_with_an_expiry_worker() { + let (shutdown, mut closed) = tokio::sync::watch::channel(()); + let _expiry = ConnectionExpiry::new(shutdown.clone()); + let guard = DisconnectGuard { + runtime: std::sync::Weak::new(), + connection_id: SandboxConnectionId::new(), + connection_shutdown: shutdown, + }; + drop(guard); + tokio::time::timeout(Duration::from_secs(2), closed.changed()) + .await + .expect("disconnected discovery must not wait for JWT expiry") + .expect("explicit disconnect notification"); + } + + #[tokio::test] + async fn renewed_credentials_replace_the_connection_deadline() { + let (shutdown, mut closed) = tokio::sync::watch::channel(()); + let expiry = ConnectionExpiry::new(shutdown); + expiry.deadline.send_replace(Some( + tokio::time::Instant::now() + Duration::from_millis(20), + )); + expiry + .deadline + .send_replace(Some(tokio::time::Instant::now() + Duration::from_mins(1))); + assert!( + tokio::time::timeout(Duration::from_millis(50), closed.changed()) + .await + .is_err() + ); + expiry.update(1); + tokio::time::timeout(Duration::from_secs(2), closed.changed()) + .await + .expect("new deadline fires") + .expect("shutdown notification"); + } + #[test] fn installed_ca_replaces_case_insensitive_trust_overrides() { let certificate = Path::new(r"C:\trusted\ca.pem"); @@ -1636,7 +1893,7 @@ mod tests { .await .expect("process exits"); assert!(matches!(status, ExitStatusWire::Exited(0))); - // Drain stdout independently: exit notification can precede the reader. + // Production attachments stop at EXIT; all output must precede it. let mut cursor = process.output.cursor(); let output = tokio::time::timeout(Duration::from_secs(10), async { let mut output = Vec::new(); @@ -1644,9 +1901,9 @@ mod tests { let event = cursor.recv().await.expect("retained output event"); if event.channel == STREAM_STDOUT { output.extend(event.payload); - if String::from_utf8_lossy(&output).contains("relay-free launch") { - return output; - } + } + if event.channel == STREAM_EXIT { + return output; } } }) @@ -1655,6 +1912,51 @@ mod tests { assert!(String::from_utf8_lossy(&output).contains("relay-free launch")); } + #[tokio::test] + async fn process_bridge_drains_stdout_and_stderr_before_exit() { + let process = Arc::new( + ManagedProcess::spawn( + "drain-test".into(), + ProcessKindWire::Exec, + std::env::var("COMSPEC").unwrap(), + vec![ + "/D".into(), + "/C".into(), + "(for /L %i in (1,1,2000) do @echo output-line) & echo last-error 1>&2".into(), + ], + None, + HashMap::new(), + ) + .await + .expect("launch real command"), + ); + let (mut client, server) = tokio::io::duplex(4096); + let bridge = tokio::spawn(bridge_process(server, process)); + let (stdout, stderr) = tokio::time::timeout(Duration::from_secs(10), async { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + while let Some((channel, payload)) = read_stream_frame(&mut client).await.unwrap() { + match channel { + STREAM_STDOUT => stdout.extend(payload), + STREAM_STDERR => stderr.extend(payload), + STREAM_EXIT => break, + _ => panic!("unexpected output channel"), + } + } + (stdout, stderr) + }) + .await + .expect("attachment finishes"); + bridge.await.unwrap().unwrap(); + assert_eq!( + String::from_utf8_lossy(&stdout) + .matches("output-line") + .count(), + 2000 + ); + assert!(String::from_utf8_lossy(&stderr).contains("last-error")); + } + #[test] fn output_retention_is_bounded_and_reports_truncation() { let output = OutputLog::new(); diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 69f3ff373..a358af315 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -1227,6 +1227,12 @@ The driver assigns distinct Sandbox Protocol and proxy listeners plus fresh cred Attached provider credentials use the ordinary live supervisor refresh path. Static values stay in the host supervisor's endpoint-bound resolver and are injected only for matching requests. Dynamic token grants remain request-time operations in the host proxy. +Main launch and raw exec receive the authenticated proxy and installed CA +environment. Exec uses the latest provider environment. The proxy identity +uses the canonical workload command when supplied, falling back to the MXC +driver command. Expired control credentials close open process and forwarding +streams; renewed credentials update the connection deadline. + ### MicroVM Each sandbox runs inside its own libkrun microVM managed by the standalone `openshell-driver-vm` subprocess. Use this driver when you want stronger isolation than container namespaces alone.