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..4d84c8734c 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,118 @@ 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( + address: SocketAddr, + 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 => { + // 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) + .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 +592,18 @@ 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( + address, + || { + 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 +3459,252 @@ 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_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 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"); + 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( + address, + || { + 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( + address, + || { + 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"); + } }