From e1e7c4031e3e28265ed4f65be573711a8392fc54 Mon Sep 17 00:00:00 2001 From: Emilien Macchi Date: Mon, 2 Nov 2026 22:24:17 -0500 Subject: [PATCH 1/2] fix(sandbox): keep serving through transient accept errors The boundary control listener treated any accept error beyond the benign kinds as fatal, so a transient EMFILE or ENFILE ended the boundary and the sandbox with it. The DNS TCP relay broke out of its accept loop on any error at all, including a connection that reset before accept, which silently disabled TCP DNS for the rest of the sandbox's life. Classify accept errors in a shared accept_errors module: EMFILE, ENFILE, ENOBUFS and ENOMEM are transient resource exhaustion, and ConnectionReset, ConnectionAborted and NotConnected mark a peer that disappeared before its connection could be served. The control listener now retries resource errors with a short backoff and a warning, and only unrecoverable errors end the boundary. The DNS TCP relay skips reset peers, retries resource errors with the same backoff, and logs before stopping on an unrecoverable error instead of stopping silently. Unit tests inject these errors into both classification paths, and a loop-level test drives the DNS TCP relay through an EMFILE and a reset-before-accept before serving a real query, then stops it on an unrecoverable error. Closes #5183 Signed-off-by: Emilien Macchi --- crates/openshell-sandbox/src/accept_errors.rs | 197 ++++++++++ .../openshell-sandbox/src/boundary_server.rs | 302 ++++++++++++++- crates/openshell-sandbox/src/lib.rs | 2 + .../openshell-sandbox/src/network_broker.rs | 354 ++++++++++++++++-- 4 files changed, 819 insertions(+), 36 deletions(-) create mode 100644 crates/openshell-sandbox/src/accept_errors.rs diff --git a/crates/openshell-sandbox/src/accept_errors.rs b/crates/openshell-sandbox/src/accept_errors.rs new file mode 100644 index 0000000000..c5113e483b --- /dev/null +++ b/crates/openshell-sandbox/src/accept_errors.rs @@ -0,0 +1,197 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Classification of listener accept errors that must not end a serving loop. + +use std::io; +use std::time::{Duration, Instant}; + +/// Backoff after a transient accept resource error, so the retry does not +/// spin while descriptors or memory are still exhausted. +pub const RESOURCE_ERROR_BACKOFF: Duration = Duration::from_millis(50); + +/// Longest silent stretch while retrying a transient accept resource error: +/// one warn line when the exhaustion episode starts, then at most one per +/// interval, so a sustained condition stays visible without flooding the log. +pub const RESOURCE_WARN_INTERVAL: Duration = Duration::from_secs(30); + +/// Transient resource exhaustion reported by `accept`: the listener itself +/// stays healthy, and the call succeeds again once descriptors or memory free +/// up. These errno values have no `io::ErrorKind` of their own. +pub fn is_transient_resource_error(error: &io::Error) -> bool { + matches!( + error.raw_os_error(), + Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM) + ) +} + +/// A peer that disappeared before the accepted connection could be served. +/// Skipping it must not affect later peers. +pub fn is_peer_reset(error: &io::Error) -> bool { + matches!( + error.kind(), + io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionReset + | io::ErrorKind::NotConnected + ) +} + +/// A network error already pending on a queued connection, which Linux +/// reports from `accept` itself. The `accept(2)` man page says to treat these +/// like `EAGAIN`, by retrying: they belong to one queued peer, not to the +/// listener. +pub fn is_pending_network_error(error: &io::Error) -> bool { + matches!( + error.raw_os_error(), + Some( + libc::ENETDOWN + | libc::EPROTO + | libc::ENOPROTOOPT + | libc::EHOSTDOWN + | libc::ENONET + | libc::EHOSTUNREACH + | libc::EOPNOTSUPP + | libc::ENETUNREACH + ) + ) +} + +/// Warn cadence for transient accept resource backoffs: one line when an +/// exhaustion episode starts, then at most one per +/// [`RESOURCE_WARN_INTERVAL`] while it lasts, and a fresh line when the +/// underlying errno changes. A successful accept ends the episode. +#[derive(Debug, Default)] +pub struct ResourceBackoffLogger { + logged: Option<(i32, Instant)>, +} + +impl ResourceBackoffLogger { + pub fn new() -> Self { + Self::default() + } + + /// Whether this retry deserves a warn line; records the decision. + pub fn should_log(&mut self, error: &io::Error) -> bool { + self.should_log_at(error, Instant::now()) + } + + fn should_log_at(&mut self, error: &io::Error, now: Instant) -> bool { + let code = error.raw_os_error().unwrap_or_default(); + let quiet = self.logged.is_some_and(|(logged_code, at)| { + logged_code == code && now.duration_since(at) < RESOURCE_WARN_INTERVAL + }); + if quiet { + false + } else { + self.logged = Some((code, now)); + true + } + } + + /// A successful accept ends the exhaustion episode. + pub fn reset(&mut self) { + self.logged = None; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resource_errors_are_transient() { + for code in [libc::EMFILE, libc::ENFILE, libc::ENOBUFS, libc::ENOMEM] { + assert!( + is_transient_resource_error(&io::Error::from_raw_os_error(code)), + "errno {code} should be a transient resource error" + ); + } + } + + #[test] + fn other_errors_are_not_transient_resource_errors() { + for error in [ + io::Error::from_raw_os_error(libc::EAGAIN), + io::Error::from_raw_os_error(libc::EBADF), + io::Error::from_raw_os_error(libc::ECONNABORTED), + io::Error::new(io::ErrorKind::OutOfMemory, "kind only"), + io::Error::new(io::ErrorKind::WouldBlock, "kind only"), + ] { + assert!( + !is_transient_resource_error(&error), + "{error} should not be a transient resource error" + ); + } + } + + #[test] + fn peer_resets_are_detected() { + for kind in [ + io::ErrorKind::ConnectionAborted, + io::ErrorKind::ConnectionReset, + io::ErrorKind::NotConnected, + ] { + assert!(is_peer_reset(&io::Error::new(kind, "reset"))); + } + assert!(!is_peer_reset(&io::Error::from_raw_os_error(libc::EMFILE))); + assert!(!is_peer_reset(&io::Error::new( + io::ErrorKind::Interrupted, + "interrupted" + ))); + } + + #[test] + fn pending_network_errors_are_detected() { + for code in [ + libc::ENETDOWN, + libc::EPROTO, + libc::ENOPROTOOPT, + libc::EHOSTDOWN, + libc::ENONET, + libc::EHOSTUNREACH, + libc::EOPNOTSUPP, + libc::ENETUNREACH, + ] { + assert!( + is_pending_network_error(&io::Error::from_raw_os_error(code)), + "errno {code} should be a pending network error" + ); + } + assert!(!is_pending_network_error(&io::Error::from_raw_os_error( + libc::EMFILE + ))); + assert!(!is_pending_network_error(&io::Error::new( + io::ErrorKind::ConnectionReset, + "reset" + ))); + } + + #[test] + fn resource_backoff_logging_is_once_per_episode_with_periodic_rewarn() { + let mut logger = ResourceBackoffLogger::new(); + let start = Instant::now(); + let error = |code| io::Error::from_raw_os_error(code); + + // One line when the episode starts, then silence while it lasts. + assert!(logger.should_log_at(&error(libc::EMFILE), start)); + assert!(!logger.should_log_at(&error(libc::EMFILE), start + Duration::from_secs(1))); + assert!(!logger.should_log_at(&error(libc::EMFILE), start + Duration::from_secs(29))); + // A sustained episode stays visible with a bounded cadence. + assert!(logger.should_log_at( + &error(libc::EMFILE), + start + RESOURCE_WARN_INTERVAL + Duration::from_secs(1) + )); + // A different errno during the episode gets its own line. + assert!(logger.should_log_at( + &error(libc::ENFILE), + start + RESOURCE_WARN_INTERVAL + Duration::from_secs(2) + )); + + // Recovery ends the episode: the next exhaustion warns again. + logger.reset(); + assert!(logger.should_log_at( + &error(libc::EMFILE), + start + RESOURCE_WARN_INTERVAL + Duration::from_secs(3) + )); + } +} diff --git a/crates/openshell-sandbox/src/boundary_server.rs b/crates/openshell-sandbox/src/boundary_server.rs index 4eb3f1ec02..2c7b957707 100644 --- a/crates/openshell-sandbox/src/boundary_server.rs +++ b/crates/openshell-sandbox/src/boundary_server.rs @@ -30,6 +30,10 @@ mod linux { use std::task::{Context, Poll}; use std::time::Duration; + use crate::accept_errors::{ + RESOURCE_ERROR_BACKOFF, ResourceBackoffLogger, is_peer_reset, is_pending_network_error, + is_transient_resource_error, + }; use crate::boundary_io::BoundaryRuntimeState; use crate::delegated::{AgentSignaler, spawn_workload}; use crate::main_session::{MainOutput, MainSession}; @@ -603,6 +607,36 @@ mod linux { Some(Ok(())) } + /// What the boundary control accept loop does with an accept error. + #[derive(Debug, PartialEq, Eq)] + enum ControlAcceptErrorAction { + /// Keep polling the listener immediately. A refused or already-dead + /// peer must not stop the listener. + Continue, + /// Retry after a short backoff: transient resource exhaustion. + Backoff, + /// Unrecoverable: end the boundary. + Fatal, + } + + fn control_accept_error_action(error: &io::Error) -> ControlAcceptErrorAction { + if is_peer_reset(error) + || is_pending_network_error(error) + || matches!( + error.kind(), + // PermissionDenied is this listener's own local-peer + // rejection (an injected EACCES), not an accept failure. + io::ErrorKind::Interrupted | io::ErrorKind::PermissionDenied + ) + { + ControlAcceptErrorAction::Continue + } else if is_transient_resource_error(error) { + ControlAcceptErrorAction::Backoff + } else { + ControlAcceptErrorAction::Fatal + } + } + fn serve(config: &BoundaryListenerConfig, runtime: Arc) -> Result<(), String> { let dial_tls = match config { BoundaryListenerConfig::RuntimeInitiatedTlsTcp { tls, .. } @@ -618,12 +652,16 @@ mod linux { let active_connections = Arc::new(AtomicUsize::new(0)); let pending_handshakes = Arc::new(tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES)); tracing::info!(?config, "Boundary control listener ready"); + // Warn once per exhaustion episode plus a bounded re-warn cadence, + // not on every backoff retry. + let mut resource_backoff_log = ResourceBackoffLogger::new(); loop { if let Some(result) = take_termination(&runtime) { return result; } match listener.accept() { Ok(stream) => { + resource_backoff_log.reset(); // Unauthenticated sockets use only a bounded async TLS // task, never an OS thread or an authenticated session slot. let Ok(pending) = pending_handshakes.clone().try_acquire_owned() else { @@ -647,20 +685,24 @@ mod linux { } }); } - Err(error) if error.kind() == io::ErrorKind::Interrupted => {} - // A refused or already-dead peer must not stop the listener. - Err(error) - if matches!( - error.kind(), - io::ErrorKind::PermissionDenied - | io::ErrorKind::NotConnected - | io::ErrorKind::ConnectionReset - | io::ErrorKind::ConnectionAborted - ) => {} Err(error) if error.kind() == io::ErrorKind::WouldBlock => { std::thread::sleep(Duration::from_millis(10)); } - Err(error) => return Err(format!("accept boundary control connection: {error}")), + Err(error) => match control_accept_error_action(&error) { + ControlAcceptErrorAction::Continue => {} + ControlAcceptErrorAction::Backoff => { + if resource_backoff_log.should_log(&error) { + tracing::warn!( + %error, + "Boundary control accept hit a transient resource limit; retrying" + ); + } + std::thread::sleep(RESOURCE_ERROR_BACKOFF); + } + ControlAcceptErrorAction::Fatal => { + return Err(format!("accept boundary control connection: {error}")); + } + }, } } } @@ -10049,6 +10091,244 @@ mod linux { Ok(openshell_isolation_interface::contract::BoundaryExitStatus::Exited(0)) )); } + + #[test] + fn transient_accept_resource_errors_back_off_instead_of_ending_the_boundary() { + for code in [libc::EMFILE, libc::ENFILE, libc::ENOBUFS, libc::ENOMEM] { + assert_eq!( + control_accept_error_action(&io::Error::from_raw_os_error(code)), + ControlAcceptErrorAction::Backoff, + "errno {code} should back off and retry" + ); + } + } + + #[test] + fn dead_peers_do_not_end_the_boundary_control_listener() { + for error in [ + io::Error::new(io::ErrorKind::Interrupted, "interrupted"), + io::Error::from_raw_os_error(libc::EACCES), + io::Error::new(io::ErrorKind::NotConnected, "not connected"), + io::Error::new(io::ErrorKind::ConnectionReset, "reset"), + io::Error::new(io::ErrorKind::ConnectionAborted, "aborted"), + // Linux reports already-pending network errors of a queued + // connection from accept; the man page says to retry them. + io::Error::from_raw_os_error(libc::ENETDOWN), + io::Error::from_raw_os_error(libc::EPROTO), + io::Error::from_raw_os_error(libc::EHOSTUNREACH), + io::Error::from_raw_os_error(libc::ENETUNREACH), + ] { + assert_eq!( + control_accept_error_action(&error), + ControlAcceptErrorAction::Continue + ); + } + } + + #[test] + fn unrecoverable_accept_errors_end_the_boundary_control_listener() { + for error in [ + io::Error::from_raw_os_error(libc::EBADF), + io::Error::new(io::ErrorKind::InvalidInput, "invalid"), + ] { + assert_eq!( + control_accept_error_action(&error), + ControlAcceptErrorAction::Fatal + ); + } + } + + /// Drive `serve` through a real EMFILE: exhaust the descriptor table + /// while a control connection is queued, require the accept loop to + /// keep retrying, then free a descriptor and require the loop to + /// still serve and shut down cleanly. + /// + /// The descriptor table is process-global, so the EMFILE window runs + /// in an isolated child of this test binary, like the reaper test in + /// `managed_children`. + #[test] + fn serve_keeps_serving_through_transient_resource_errors() { + const HELPER_ENV: &str = "OPENSHELL_BOUNDARY_SERVE_EMFILE_HELPER"; + + /// Read until the control listener closes the connection. It may + /// answer garbage with a TLS alert record before it closes. + fn drain_until_closed(socket: &socket2::Socket) { + let mut buffer = [std::mem::MaybeUninit::new(0_u8); 8]; + while socket.recv(&mut buffer).expect("control connection answer") > 0 {} + } + if std::env::var_os(HELPER_ENV).is_none() { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "boundary_server::linux::tests::serve_keeps_serving_through_transient_resource_errors", + "--nocapture", + ]) + .env(HELPER_ENV, "1") + .output() + .expect("start isolated boundary serve test process"); + assert!( + output.status.success(), + "boundary serve test failed:\n{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return; + } + + let directory = tempfile::tempdir().expect("temporary directory"); + let (server_tls, _client_tls) = stage_test_tls(directory.path(), "serve-emfile"); + let socket_path = directory.path().join("control.sock"); + let (broker, launcher) = test_network_broker(); + let (verification_key, _token) = test_auth_material("serve-emfile"); + let process_runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("test process runtime"); + let runtime = Arc::new( + BoundaryRuntime::new( + BoundaryConfig { + boundary_id: "serve-emfile".to_string(), + generation: "generation-1".to_string(), + session_id: test_session_id(), + session_rotation: openshell_core::jwt::SessionRotation::new(1) + .expect("session rotation"), + auth_epoch: CredentialEpoch::new(1).expect("auth epoch"), + gateway_id: "test-gateway".to_string(), + verification_keys: vec![verification_key], + listener: BoundaryListenerConfig::Unix { + socket_path: socket_path.clone(), + tls: server_tls.clone(), + }, + resource_claims: std::collections::BTreeMap::new(), + resource_claim_files: std::collections::BTreeMap::new(), + workload_identity: test_workload_identity(), + provisioning_evidence: test_provisioning_evidence(), + child_env: std::collections::HashMap::new(), + }, + process_runtime.handle().clone(), + broker, + launcher, + test_runtime_qualification(), + ) + .expect("test boundary runtime"), + ); + let config = BoundaryListenerConfig::Unix { + socket_path: socket_path.clone(), + tls: server_tls, + }; + let server = std::thread::spawn(move || serve(&config, runtime)); + + // The client socket is allocated before the descriptor table + // fills, so queueing the connection later needs no descriptor. + let client = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None) + .expect("client socket"); + client + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("client read timeout"); + let peer = socket2::SockAddr::unix(&socket_path).expect("client address"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !socket_path.exists() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + socket_path.exists(), + "boundary control listener was not bound" + ); + + // Readiness probe: a garbage connection that the server accepts + // and then closes proves serve finished loading its TLS identity + // (bind completed) and is polling accept, so filling the + // descriptor table cannot race the identity load. + let probe = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None) + .expect("probe socket"); + probe + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("probe read timeout"); + probe.connect(&peer).expect("connect probe"); + probe + .send(&[0_u8; 16]) + .expect("send garbage to the control listener"); + drain_until_closed(&probe); + drop(probe); + assert!( + !server.is_finished(), + "serve ended before the descriptor table was filled" + ); + + // Lower the soft limit to just above the current descriptor use + // and fill every slot below it, so the next accept that needs a + // descriptor fails with EMFILE. The hard limit caps the target so + // setrlimit cannot fail on a tight container limit. + let nofile = rustix::process::getrlimit(rustix::process::Resource::Nofile); + let open = u64::try_from( + std::fs::read_dir("/proc/self/fd") + .expect("count open descriptors") + .count(), + ) + .expect("descriptor count"); + let soft_target = nofile + .maximum + .map_or_else(|| open + 16, |max| (open + 16).min(max)); + rustix::process::setrlimit( + rustix::process::Resource::Nofile, + rustix::process::Rlimit { + current: Some(soft_target), + maximum: nofile.maximum, + }, + ) + .expect("lower descriptor soft limit"); + // An empty fill list means the table was already at the limit, + // which still gives accept its EMFILE. + let mut fillers = Vec::new(); + while let Ok(file) = File::open("/dev/null") { + fillers.push(file); + } + + // Queue the control connection: the accept loop now hits EMFILE + // with a pending peer and must back off and retry for as long as + // the table stays full. + client.connect(&peer).expect("queue control connection"); + std::thread::sleep(Duration::from_millis(300)); + assert!( + !server.is_finished(), + "serve ended on a transient EMFILE accept error" + ); + + // Freeing one descriptor lets the loop accept the queued + // connection: the garbage bytes are consumed, the listener + // answers with a TLS alert, and the connection closes, which + // only happens if accept recovered. + drop(fillers.pop().expect("a filler descriptor to free")); + client + .send(&[0_u8; 16]) + .expect("send garbage after recovery"); + drain_until_closed(&client); + assert!( + !server.is_finished(), + "serve ended after the descriptor table recovered" + ); + + // End the loop the way production does and require a termination + // result, not an accept failure. + request_boundary_termination(libc::SIGTERM); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !server.is_finished() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + server.is_finished(), + "serve ignored the termination request" + ); + let message = server + .join() + .expect("serve thread panicked") + .expect_err("serve returned success without a supervisor"); + assert!( + message.contains("SIGTERM") + && !message.contains("accept boundary control connection"), + "serve ended with an unexpected error: {message}" + ); + } } } diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 3a7fd8beb4..7b1f4400c1 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -3,6 +3,8 @@ //! Capability-free in-workload sandbox boundary. +#[cfg(target_os = "linux")] +mod accept_errors; pub mod binary_pins; #[cfg(target_os = "linux")] pub mod boundary_exec; diff --git a/crates/openshell-sandbox/src/network_broker.rs b/crates/openshell-sandbox/src/network_broker.rs index 70e9b444a1..09f283c82d 100644 --- a/crates/openshell-sandbox/src/network_broker.rs +++ b/crates/openshell-sandbox/src/network_broker.rs @@ -15,6 +15,10 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; +use crate::accept_errors::{ + RESOURCE_ERROR_BACKOFF, ResourceBackoffLogger, is_peer_reset, is_pending_network_error, + is_transient_resource_error, +}; use crate::linux::seccomp_notify::{Notification, NotificationListener}; use crate::linux::socket_registry::{ InetFamily, InetKind, SocketIdentity, SocketMetadata, SocketRegistry, SocketState, @@ -416,6 +420,107 @@ impl NetworkBroker { } } +/// What the DNS TCP relay accept loop does with an accept error. +#[derive(Debug, PartialEq, Eq)] +enum DnsTcpAcceptErrorAction { + /// Retry immediately: the call was interrupted, or the peer reset before + /// the accepted connection could be served. + Skip, + /// Retry after a short backoff: transient resource exhaustion. + Backoff, + /// Unrecoverable: stop serving TCP DNS. + Stop, +} + +fn dns_tcp_accept_error_action(error: &io::Error) -> DnsTcpAcceptErrorAction { + if error.kind() == io::ErrorKind::Interrupted + || is_peer_reset(error) + || is_pending_network_error(error) + { + DnsTcpAcceptErrorAction::Skip + } else if is_transient_resource_error(error) { + DnsTcpAcceptErrorAction::Backoff + } else { + DnsTcpAcceptErrorAction::Stop + } +} + +/// Serve accepted DNS TCP streams until an unrecoverable accept error. +/// +/// A transient resource error, or a peer that resets before its accept +/// completes, must not stop the relay: TCP DNS keeps serving for the rest of +/// the sandbox's life. +fn dns_tcp_accept_loop( + mut accept: impl FnMut() -> io::Result<(TcpStream, SocketAddr)>, + tcp_admissions: Arc>>, + active_workers: Arc, + pending: mpsc::Sender, +) { + // Warn once per exhaustion episode plus a bounded re-warn cadence, not on + // every backoff retry. + let mut resource_backoff_log = ResourceBackoffLogger::new(); + loop { + let (stream, peer) = match accept() { + Ok(accepted) => { + resource_backoff_log.reset(); + accepted + } + Err(error) => match dns_tcp_accept_error_action(&error) { + DnsTcpAcceptErrorAction::Skip => { + tracing::debug!(%error, "skipping DNS TCP accept of a reset peer"); + continue; + } + DnsTcpAcceptErrorAction::Backoff => { + if resource_backoff_log.should_log(&error) { + tracing::warn!( + %error, + "DNS TCP accept hit a transient resource limit; retrying" + ); + } + std::thread::sleep(RESOURCE_ERROR_BACKOFF); + continue; + } + DnsTcpAcceptErrorAction::Stop => { + tracing::warn!(%error, "stopping the DNS TCP relay accept loop"); + // Dual-emit: the relay stopping is a security finding + // (an unavailable control), not just an operational one. + openshell_ocsf::ocsf_emit!( + openshell_ocsf::DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(openshell_ocsf::ActivityId::Open) + .severity(openshell_ocsf::SeverityId::High) + .confidence(openshell_ocsf::ConfidenceId::High) + .is_alert(true) + .finding_info(openshell_ocsf::FindingInfo::new( + "sandbox-dns-tcp-relay-stopped", + "Sandbox DNS TCP Relay Stopped", + )) + .message("DNS TCP relay stopped serving; TCP DNS mediation unavailable") + .build() + ); + return; + } + }, + }; + // One admission authorizes exactly one accepted TCP stream; + // no peer mapping needs to outlive this accept. + if lock(&tcp_admissions).remove(&peer).is_none() { + tracing::warn!(%peer, "dropping DNS stream from unregistered socket"); + continue; + } + let Ok(worker_slot) = acquire_pending_dns_slot(&active_workers) else { + tracing::warn!(%peer, "dropping DNS stream because the worker quota is full"); + continue; + }; + let tcp_pending = pending.clone(); + let _ = std::thread::Builder::new() + .name("openshell-dns-tcp-query".to_string()) + .spawn(move || { + let _worker_slot = worker_slot; + serve_dns_tcp(stream, tcp_pending); + }); + } +} + fn start_dns_relay( address: SocketAddr, pending: mpsc::Sender, @@ -476,31 +581,17 @@ fn start_dns_relay( std::thread::Builder::new() .name("openshell-dns-tcp".to_string()) .spawn(move || { - for accepted in tcp.incoming() { - let Ok((stream, peer)) = accepted.and_then(|stream| { - let peer = stream.peer_addr()?; - Ok((stream, peer)) - }) else { - break; - }; - // One admission authorizes exactly one accepted TCP stream; - // no peer mapping needs to outlive this accept. - if lock(&tcp_admissions).remove(&peer).is_none() { - tracing::warn!(%peer, "dropping DNS stream from unregistered socket"); - continue; - } - let Ok(worker_slot) = acquire_pending_dns_slot(&tcp_active_workers) else { - tracing::warn!(%peer, "dropping DNS stream because the worker quota is full"); - continue; - }; - let tcp_pending = pending.clone(); - let _ = std::thread::Builder::new() - .name("openshell-dns-tcp-query".to_string()) - .spawn(move || { - let _worker_slot = worker_slot; - serve_dns_tcp(stream, tcp_pending); - }); - } + dns_tcp_accept_loop( + || { + tcp.accept().and_then(|(stream, _)| { + let peer = stream.peer_addr()?; + Ok((stream, peer)) + }) + }, + tcp_admissions, + tcp_active_workers, + pending, + ); }) .map_err(|error| io::Error::other(format!("start TCP DNS relay: {error}")))?; Ok(relay) @@ -3356,4 +3447,217 @@ mod tests { [0, 3, 4, 5, 6] ); } + + #[test] + fn dns_tcp_accept_errors_map_to_keep_serving_actions() { + for error in [ + io::Error::from_raw_os_error(libc::EMFILE), + io::Error::from_raw_os_error(libc::ENFILE), + ] { + assert_eq!( + dns_tcp_accept_error_action(&error), + DnsTcpAcceptErrorAction::Backoff + ); + } + for error in [ + io::Error::new(io::ErrorKind::Interrupted, "interrupted"), + io::Error::from_raw_os_error(libc::ECONNABORTED), + io::Error::from_raw_os_error(libc::ENOTCONN), + io::Error::from_raw_os_error(libc::EPROTO), + io::Error::from_raw_os_error(libc::ENETUNREACH), + ] { + assert_eq!( + dns_tcp_accept_error_action(&error), + DnsTcpAcceptErrorAction::Skip + ); + } + assert_eq!( + dns_tcp_accept_error_action(&io::Error::from_raw_os_error(libc::EBADF)), + DnsTcpAcceptErrorAction::Stop + ); + } + + #[test] + fn dns_tcp_relay_survives_transient_accept_errors() { + use std::io::{Read as _, Write as _}; + + // A real accepted TCP pair, so the loop serves an actual stream. + let listener = TcpListener::bind("127.0.0.1:0").expect("bind relay peer"); + let mut client = + TcpStream::connect(listener.local_addr().unwrap()).expect("connect client"); + client + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("client read timeout"); + let (stream, peer) = listener.accept().expect("accept client"); + + let tcp_admissions = Arc::new(Mutex::new(HashMap::new())); + lock(&tcp_admissions).insert( + peer, + SocketIdentity { + listener_generation: 0, + inode: 0, + cookie: 0, + }, + ); + let (pending_tx, mut pending_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); + let mut accepts = 0_u32; + let mut stream = Some(stream); + let relay = std::thread::spawn(move || { + dns_tcp_accept_loop( + || { + accepts += 1; + match accepts { + 1 => Err(io::Error::from_raw_os_error(libc::EMFILE)), + 2 => Err(io::Error::from_raw_os_error(libc::ECONNABORTED)), + 3 => Ok(( + stream + .take() + .expect("accept source served its stream twice"), + peer, + )), + 4 => Err(io::Error::from_raw_os_error(libc::EINVAL)), + _ => panic!("DNS TCP accept loop kept accepting past its stop error"), + } + }, + tcp_admissions, + Arc::new(AtomicUsize::new(0)), + pending_tx, + ); + }); + + // Both earlier accept errors were survived: the query still round-trips. + client + .write_all(&[0, 3, b'a', b'b', b'c']) + .expect("write DNS query"); + let query = pending_rx + .blocking_recv() + .expect("relay queued the DNS query after transient accept errors"); + assert_eq!(query.request, [0, 3, b'a', b'b', b'c']); + query + .response + .send(Ok(vec![0, 2, 0x01, 0x02])) + .expect("answer the DNS query"); + let mut answer = [0_u8; 4]; + client.read_exact(&mut answer).expect("read DNS answer"); + assert_eq!(answer, [0, 2, 0x01, 0x02]); + relay.join().expect("accept loop thread panicked"); + } + + #[test] + fn dns_tcp_relay_survives_a_peer_that_resets_before_peer_addr() { + use std::io::{Read as _, Write as _}; + + // The accept source reproduces the production wiring of + // start_dns_relay: accept the stream, then resolve the peer with + // peer_addr, which fails once the peer reset while its connection + // was still queued. + let listener = TcpListener::bind("127.0.0.1:0").expect("bind relay listener"); + let address = listener.local_addr().expect("relay address"); + let reset_client = socket2::Socket::new( + socket2::Domain::for_address(address), + socket2::Type::STREAM, + Some(socket2::Protocol::TCP), + ) + .expect("reset client socket"); + reset_client + .connect(&address.into()) + .expect("connect reset client"); + // Zero linger makes the close an RST while the connection is still + // queued, which is what makes `peer_addr` fail with ENOTCONN. + reset_client + .set_linger(Some(Duration::ZERO)) + .expect("zero linger"); + drop(reset_client); + // Let the RST land before the accept loop starts, so the first + // accepted stream is the reset one. + std::thread::sleep(Duration::from_millis(50)); + + let mut client = TcpStream::connect(address).expect("connect healthy client"); + client + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("client read timeout"); + let peer = client.local_addr().expect("healthy peer address"); + let tcp_admissions = Arc::new(Mutex::new(HashMap::new())); + lock(&tcp_admissions).insert( + peer, + SocketIdentity { + listener_generation: 0, + inode: 0, + cookie: 0, + }, + ); + let (pending_tx, mut pending_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); + let first_peer_addr_error = Arc::new(Mutex::new(None::)); + let observed_error = first_peer_addr_error.clone(); + // Test-only stop signal: once set, the accept source reports an + // unrecoverable error so the loop ends instead of leaking its thread. + let stop = Arc::new(AtomicBool::new(false)); + let stop_relay = stop.clone(); + let relay = std::thread::spawn(move || { + dns_tcp_accept_loop( + || { + if stop_relay.load(Ordering::Relaxed) { + return Err(io::Error::from_raw_os_error(libc::EINVAL)); + } + listener + .accept() + .and_then(|(stream, _)| match stream.peer_addr() { + Ok(peer) => Ok((stream, peer)), + Err(error) => { + *lock(&observed_error) = Some(error.kind()); + Err(error) + } + }) + }, + tcp_admissions, + Arc::new(AtomicUsize::new(0)), + pending_tx, + ); + }); + + // The reset peer reached the production wiring: accept succeeded and + // peer_addr failed, so the loop must have hit its skip path. + let deadline = Instant::now() + Duration::from_secs(5); + while lock(&first_peer_addr_error).is_none() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + let kind = lock(&first_peer_addr_error).expect("reset peer produced a peer_addr error"); + assert_eq!( + kind, + io::ErrorKind::NotConnected, + "the reset peer's peer_addr error must be a skip-not-stop kind" + ); + + // The relay outlived the reset peer: the healthy query round-trips. + client + .write_all(&[0, 3, b'x', b'y', b'z']) + .expect("write DNS query"); + let query = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime") + .block_on(async { + tokio::time::timeout(Duration::from_secs(5), pending_rx.recv()) + .await + .expect("relay queued the DNS query after the reset peer") + .expect("pending DNS queue stayed open") + }); + assert_eq!(query.request, [0, 3, b'x', b'y', b'z']); + query + .response + .send(Ok(vec![0, 2, 0x01, 0x02])) + .expect("answer the DNS query"); + let mut answer = [0_u8; 4]; + client.read_exact(&mut answer).expect("read DNS answer"); + assert_eq!(answer, [0, 2, 0x01, 0x02]); + assert!( + !relay.is_finished(), + "the DNS TCP accept loop stopped after the reset peer" + ); + + // Stop the relay without leaking its thread into later tests. + stop.store(true, Ordering::Relaxed); + let _wakeup = TcpStream::connect(address).expect("wake the accept loop"); + relay.join().expect("accept loop thread panicked"); + } } From 16d64f7b84aed0b8ecec1eb54953f6b7fceaa886 Mon Sep 17 00:00:00 2001 From: divesh Date: Wed, 4 Nov 2026 00:33:28 -0800 Subject: [PATCH 2/2] fix(sandbox): emit network event when TCP DNS relay stops Signed-off-by: divesh --- .../openshell-sandbox/src/network_broker.rs | 53 +++++++++++++++++-- 1 file changed, 50 insertions(+), 3 deletions(-) diff --git a/crates/openshell-sandbox/src/network_broker.rs b/crates/openshell-sandbox/src/network_broker.rs index 09f283c82d..4d84c8734c 100644 --- a/crates/openshell-sandbox/src/network_broker.rs +++ b/crates/openshell-sandbox/src/network_broker.rs @@ -451,6 +451,7 @@ fn dns_tcp_accept_error_action(error: &io::Error) -> DnsTcpAcceptErrorAction { /// completes, must not stop the relay: TCP DNS keeps serving for the rest of /// the sandbox's life. fn dns_tcp_accept_loop( + address: SocketAddr, mut accept: impl FnMut() -> io::Result<(TcpStream, SocketAddr)>, tcp_admissions: Arc>>, active_workers: Arc, @@ -481,9 +482,19 @@ fn dns_tcp_accept_loop( continue; } DnsTcpAcceptErrorAction::Stop => { - tracing::warn!(%error, "stopping the DNS TCP relay accept loop"); // Dual-emit: the relay stopping is a security finding // (an unavailable control), not just an operational one. + openshell_ocsf::ocsf_emit!( + openshell_ocsf::NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(openshell_ocsf::ActivityId::Fail) + .severity(openshell_ocsf::SeverityId::High) + .status(openshell_ocsf::StatusId::Failure) + .src_endpoint_addr(address.ip(), address.port()) + .connection_info(openshell_ocsf::ConnectionInfo::new("tcp")) + .status_detail(error.to_string()) + .message("DNS TCP relay stopped serving; TCP DNS mediation unavailable") + .build() + ); openshell_ocsf::ocsf_emit!( openshell_ocsf::DetectionFindingBuilder::new(openshell_ocsf::ctx::ctx()) .activity(openshell_ocsf::ActivityId::Open) @@ -582,6 +593,7 @@ fn start_dns_relay( .name("openshell-dns-tcp".to_string()) .spawn(move || { dns_tcp_accept_loop( + address, || { tcp.accept().and_then(|(stream, _)| { let peer = stream.peer_addr()?; @@ -3477,14 +3489,47 @@ mod tests { ); } + #[test] + fn dns_tcp_relay_stop_emits_network_failure_and_finding() { + let capture = openshell_ocsf::test_support::OcsfCapture::default(); + let _capture = capture.install(); + let address = SocketAddr::from(([127, 0, 0, 1], 1053)); + let error = io::Error::from_raw_os_error(libc::EBADF); + let (pending, _pending_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); + + dns_tcp_accept_loop( + address, + || Err(io::Error::from_raw_os_error(libc::EBADF)), + Arc::new(Mutex::new(HashMap::new())), + Arc::new(AtomicUsize::new(0)), + pending, + ); + + let events = capture.json_events(); + assert_eq!(events.len(), 2); + assert_eq!(events[0]["class_uid"], 4001); + assert_eq!(events[0]["activity_id"], 4); + assert_eq!(events[0]["status_id"], 2); + assert_eq!(events[0]["src_endpoint"]["ip"], "127.0.0.1"); + assert_eq!(events[0]["src_endpoint"]["port"], 1053); + assert_eq!(events[0]["connection_info"]["protocol_name"], "tcp"); + assert_eq!(events[0]["status_detail"], error.to_string()); + assert_eq!(events[1]["class_uid"], 2004); + assert_eq!(events[0]["message"], events[1]["message"]); + assert_eq!( + events[1]["finding_info"]["uid"], + "sandbox-dns-tcp-relay-stopped" + ); + } + #[test] fn dns_tcp_relay_survives_transient_accept_errors() { use std::io::{Read as _, Write as _}; // A real accepted TCP pair, so the loop serves an actual stream. let listener = TcpListener::bind("127.0.0.1:0").expect("bind relay peer"); - let mut client = - TcpStream::connect(listener.local_addr().unwrap()).expect("connect client"); + let address = listener.local_addr().expect("relay address"); + let mut client = TcpStream::connect(address).expect("connect client"); client .set_read_timeout(Some(Duration::from_secs(5))) .expect("client read timeout"); @@ -3504,6 +3549,7 @@ mod tests { let mut stream = Some(stream); let relay = std::thread::spawn(move || { dns_tcp_accept_loop( + address, || { accepts += 1; match accepts { @@ -3595,6 +3641,7 @@ mod tests { let stop_relay = stop.clone(); let relay = std::thread::spawn(move || { dns_tcp_accept_loop( + address, || { if stop_relay.load(Ordering::Relaxed) { return Err(io::Error::from_raw_os_error(libc::EINVAL));