diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 78e502795..f186e4db8 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -16,6 +16,7 @@ mod sidecar_control; use miette::{IntoDiagnostic, Result, WrapErr}; use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::AtomicU32; use std::time::Duration; @@ -361,7 +362,7 @@ pub async fn run_sandbox( #[cfg(not(target_os = "linux"))] drop(bypass_activity_tx); - let networking = if network_enabled { + let mut networking = if network_enabled { #[cfg(target_os = "linux")] let proxy_bind_ip = netns .as_ref() @@ -628,6 +629,19 @@ pub async fn run_sandbox( .zip(bootstrap.proxy_ca_bundle_path.clone()) }); + let proxy_exited: Pin + Send>> = if let Some(rx) = networking + .as_mut() + .and_then(|n| n.proxy.as_mut()) + .map(|p| p.take_exit_receiver()) + { + Box::pin(async { + let _ = rx.await; + }) + } else { + Box::pin(std::future::pending()) + }; + tokio::pin!(proxy_exited); + let exit_code = if process_enabled { let ca_file_paths = networking .as_ref() @@ -707,9 +721,41 @@ pub async fn run_sandbox( "authoritative network-sidecar control channel closed" )); } + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } } } else { - process.await? + tokio::select! { + result = process => result?, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } } else { // Network-only sidecar mode: keep the proxy and its background @@ -725,15 +771,62 @@ pub async fn run_sandbox( warn!(?result, "Authoritative sidecar control channel exited; restarting sidecar"); 1 } + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } } } else { - wait_for_shutdown_signal().await; - 0 + tokio::select! { + () = wait_for_shutdown_signal() => 0, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } #[cfg(not(target_os = "linux"))] { - wait_for_shutdown_signal().await; - 0 + tokio::select! { + () = wait_for_shutdown_signal() => 0, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } }; diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index ab2313b89..48c41bae0 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -174,11 +174,11 @@ impl InferenceContext { } } -#[derive(Debug)] pub struct ProxyHandle { #[allow(dead_code)] http_addr: Option, join: JoinHandle<()>, + exited_rx: tokio::sync::oneshot::Receiver<()>, } impl ProxyHandle { @@ -240,7 +240,13 @@ impl ProxyHandle { ); } + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); let join = tokio::spawn(async move { + // Hold the sender for the lifetime of this task — when the task + // exits (panic, abort, or loop break), the sender drops and the + // receiver fires, notifying the sandbox that the proxy is gone. + let _proxy_exit_guard = exited_tx; + // Wait for the OPA engine's symlink resolution reload to complete // before accepting connections. This prevents requests from // observing a generation transition mid-flight, which would cause @@ -265,9 +271,13 @@ impl ProxyHandle { } } + let mut consecutive_fd_errors: u32 = 0; + let mut consecutive_unknown_errors: u32 = 0; loop { match listener.accept().await { Ok((stream, _addr)) => { + consecutive_fd_errors = 0; + consecutive_unknown_errors = 0; let opa = opa_engine.clone(); let cache = identity_cache.clone(); let spid = entrypoint_pid.clone(); @@ -314,14 +324,37 @@ impl ProxyHandle { }); } Err(err) => { - let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) - .activity(ActivityId::Fail) - .severity(SeverityId::Low) - .status(StatusId::Failure) - .message(format!("Proxy accept error: {err}")) - .build(); - ocsf_emit!(event); - break; + match classify_accept_error( + &err, + &mut consecutive_fd_errors, + &mut consecutive_unknown_errors, + ) { + AcceptAction::Terminal => { + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message(format!( + "Proxy accept loop exiting on terminal error: {err}", + )) + .build(); + ocsf_emit!(event); + break; + } + AcceptAction::Retry { backoff, severity } => { + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(severity) + .status(StatusId::Failure) + .message(format!( + "Proxy accept error (retrying in {}ms): {err}", + backoff.as_millis(), + )) + .build(); + ocsf_emit!(event); + tokio::time::sleep(backoff).await; + } + } } } } @@ -330,6 +363,7 @@ impl ProxyHandle { Ok(Self { http_addr: Some(local_addr), join, + exited_rx, }) } @@ -337,6 +371,13 @@ impl ProxyHandle { pub const fn http_addr(&self) -> Option { self.http_addr } + + /// Take the exit notification receiver for health monitoring. + /// Resolves when the proxy accept loop task exits for any reason. + pub fn take_exit_receiver(&mut self) -> tokio::sync::oneshot::Receiver<()> { + let (_tx, dummy_rx) = tokio::sync::oneshot::channel(); + std::mem::replace(&mut self.exited_rx, dummy_rx) + } } impl Drop for ProxyHandle { @@ -345,6 +386,53 @@ impl Drop for ProxyHandle { } } +const MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS: u32 = 10; + +#[derive(Debug, PartialEq)] +enum AcceptAction { + Terminal, + Retry { + backoff: std::time::Duration, + severity: SeverityId, + }, +} + +fn classify_accept_error( + err: &std::io::Error, + consecutive_fd_errors: &mut u32, + consecutive_unknown_errors: &mut u32, +) -> AcceptAction { + // EBADF (9) / EINVAL (22) / ENOTSOCK (88) — the listener socket is + // permanently unusable; retrying will never succeed. EINVAL from + // accept() means the socket is no longer listening (shut down or + // corrupted), not a generic invalid-argument condition. + if matches!(err.raw_os_error(), Some(9 | 22 | 88)) { + return AcceptAction::Terminal; + } + + // EMFILE (24) / ENFILE (23) — process or system FD table is full. + if matches!(err.raw_os_error(), Some(24 | 23)) { + *consecutive_unknown_errors = 0; + *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); + let backoff_ms = 100u64 + .saturating_mul(1u64 << (*consecutive_fd_errors).min(7).saturating_sub(1)) + .min(5_000); + return AcceptAction::Retry { + backoff: std::time::Duration::from_millis(backoff_ms), + severity: SeverityId::Medium, + }; + } + + *consecutive_unknown_errors = consecutive_unknown_errors.saturating_add(1); + if *consecutive_unknown_errors >= MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + return AcceptAction::Terminal; + } + AcceptAction::Retry { + backoff: std::time::Duration::from_millis(100), + severity: SeverityId::Low, + } +} + fn emit_activity(tx: &Option, denied: bool, deny_group: &'static str) { if let Some(tx) = tx { let _ = try_record_activity(tx, denied, deny_group); @@ -9827,4 +9915,271 @@ network_policies: } } } + + #[tokio::test] + async fn test_exit_receiver_fires_when_task_exits() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + }); + handle.await.unwrap(); + // The sender was dropped when the task completed, so the receiver + // should resolve immediately with an Err (sender dropped). + assert!(exited_rx.await.is_err()); + } + + #[tokio::test] + async fn test_exit_receiver_fires_when_task_is_aborted() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + std::future::pending::<()>().await; + }); + handle.abort(); + // Abort drops the task's locals, including the sender guard. + assert!(exited_rx.await.is_err()); + } + + #[tokio::test] + async fn test_take_exit_receiver_returns_real_receiver() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let mut taken = handle.take_exit_receiver(); + // Sender still alive — receiver should not be ready yet. + assert!(taken.try_recv().is_err()); + // Drop the original sender — the taken receiver should resolve. + drop(tx); + assert!(taken.await.is_err()); + } + + #[tokio::test] + async fn test_take_exit_receiver_second_call_returns_instantly() { + let (_tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let _first = handle.take_exit_receiver(); + let second = handle.take_exit_receiver(); + // The dummy's sender was dropped inside take_exit_receiver, so + // the second receiver resolves immediately — this demonstrates + // the double-call hazard. + assert!(second.await.is_err()); + } + + // --- classify_accept_error tests --- + + #[test] + fn test_classify_terminal_error_ebadf() { + let err = std::io::Error::from_raw_os_error(9); // EBADF + let mut fd = 0; + let mut unk = 0; + assert_eq!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal + ); + } + + #[test] + fn test_classify_terminal_error_einval() { + let err = std::io::Error::from_raw_os_error(22); // EINVAL + let mut fd = 0; + let mut unk = 0; + assert_eq!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn test_classify_terminal_error_enotsock() { + let err = std::io::Error::from_raw_os_error(88); // ENOTSOCK (Linux) + let mut fd = 0; + let mut unk = 0; + assert_eq!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal + ); + } + + #[test] + fn test_classify_fd_exhaustion_returns_retry_medium() { + let err = std::io::Error::from_raw_os_error(24); // EMFILE + let mut fd = 0; + let mut unk = 0; + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!( + action, + AcceptAction::Retry { + severity: SeverityId::Medium, + .. + } + ), + "expected Retry/Medium for EMFILE, got {action:?}", + ); + } + + #[test] + fn test_classify_unknown_error_returns_retry_low() { + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let mut fd = 0; + let mut unk = 0; + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!( + action, + AcceptAction::Retry { + severity: SeverityId::Low, + .. + } + ), + "expected Retry/Low for unknown error, got {action:?}", + ); + } + + #[test] + fn test_classify_fd_exhaustion_backoff_increases_and_caps() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(24); // EMFILE + + let mut prev_backoff = std::time::Duration::ZERO; + for _ in 0..6 { + match classify_accept_error(&err, &mut fd, &mut unk) { + AcceptAction::Retry { backoff, .. } => { + assert!( + backoff > prev_backoff, + "backoff should increase: {backoff:?} <= {prev_backoff:?}", + ); + prev_backoff = backoff; + } + AcceptAction::Terminal => panic!("expected Retry, got Terminal"), + } + } + + // After enough consecutive errors the backoff should hit the 5s cap. + for _ in 6..12 { + classify_accept_error(&err, &mut fd, &mut unk); + } + match classify_accept_error(&err, &mut fd, &mut unk) { + AcceptAction::Retry { backoff, .. } => { + assert_eq!( + backoff, + std::time::Duration::from_secs(5), + "backoff should cap at 5000ms", + ); + } + AcceptAction::Terminal => panic!("expected Retry, got Terminal"), + } + } + + #[test] + fn test_classify_unknown_errors_exit_after_threshold() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "call {i} should be Retry, got {action:?}", + ); + } + let final_action = classify_accept_error(&err, &mut fd, &mut unk); + assert_eq!( + final_action, + AcceptAction::Terminal, + "call {MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS} should be Terminal", + ); + } + + #[test] + fn test_classify_success_resets_counters() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + + // Accumulate 9 unknown errors (one below threshold). + for _ in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + classify_accept_error(&err, &mut fd, &mut unk); + } + + // Simulate a successful accept — production loop resets both counters. + fd = 0; + unk = 0; + + // Another 9 unknowns should NOT trigger Terminal. + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "after reset, call {i} should be Retry, got {action:?}", + ); + } + } + + #[test] + fn test_classify_fd_error_resets_unknown_counter() { + let mut fd = 0; + let mut unk = 0; + let unknown_err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let fd_err = std::io::Error::from_raw_os_error(24); // EMFILE + + // Accumulate 5 unknown errors. + for _ in 0..5 { + classify_accept_error(&unknown_err, &mut fd, &mut unk); + } + + // One FD error resets the unknown counter. + classify_accept_error(&fd_err, &mut fd, &mut unk); + + // Another 9 unknowns should NOT trigger Terminal (counter was reset). + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&unknown_err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "after FD reset, call {i} should be Retry, got {action:?}", + ); + } + } + + #[test] + fn test_classify_fd_exhaustion_and_terminal_are_disjoint() { + let mut fd = 0; + let mut unk = 0; + + for errno in [24, 23] { + // EMFILE, ENFILE + let err = std::io::Error::from_raw_os_error(errno); + assert!( + matches!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Retry { .. } + ), + "errno {errno} should be Retry", + ); + fd = 0; + unk = 0; + } + + for errno in [9, 22] { + // EBADF, EINVAL (portable values) + let err = std::io::Error::from_raw_os_error(errno); + assert_eq!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal, + "errno {errno} should be Terminal", + ); + } + } }