From 5ed5f7a7302d5fb3de400d0e46f71d80ae73a519 Mon Sep 17 00:00:00 2001 From: politerealism Date: Tue, 11 Aug 2026 14:10:27 -0400 Subject: [PATCH] fix(ssh): add EMFILE backoff and exit notification to SSH accept loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apply the same two-layer defense from the proxy accept loop (#2369/#2370) to the SSH accept loop: classify transient vs terminal accept errors with exponential backoff on EMFILE/resource-exhaustion, and notify the sandbox when the accept loop exits so the container terminates instead of running without SSH access. - Add SshAcceptAction enum and classify_ssh_accept_error in ssh.rs, mirroring the proxy pattern (EMFILE/ENFILE/ENOBUFS → Retry with backoff, unknown errors → Terminal after 10 consecutive failures) - Replace the bare accept().await in run_ssh_server with a classify-and- retry loop; resets consecutive-error counter on each successful accept - Thread ssh_exit_tx: Option> through run_process; hold it as a drop-guard inside the SSH spawn so the receiver fires when the task ends for any reason - Wire ssh_exited future in lib.rs (created only when ssh_socket_path is Some) and select! on it in both process_enabled paths, returning an error so the sandbox container restarts Closes #2372 Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 46 ++++ .../openshell-supervisor-process/src/run.rs | 2 + .../openshell-supervisor-process/src/ssh.rs | 216 +++++++++++++++--- 3 files changed, 228 insertions(+), 36 deletions(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index b4c2cfd9dd..b1c226cebd 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -852,6 +852,21 @@ pub async fn run_sandbox( } }); + let (ssh_exit_tx, ssh_exit_rx) = if ssh_socket_path.is_some() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + let ssh_exited: Pin + Send>> = if let Some(rx) = ssh_exit_rx { + Box::pin(async { + let _ = rx.await; + }) + } else { + Box::pin(std::future::pending()) + }; + tokio::pin!(ssh_exited); + let entrypoint_started_tx = if process_uses_sidecar_control && let Some(writer) = process_control_writer.clone() { let (tx, rx) = tokio::sync::oneshot::channel(); @@ -914,6 +929,7 @@ pub async fn run_sandbox( openshell_endpoint.as_deref(), ssh_socket_path, sidecar_network_enforcement, + ssh_exit_tx, &process_policy, resolved_process_identity, process_enforcement_mode, @@ -965,6 +981,21 @@ pub async fn run_sandbox( "proxy accept loop exited unexpectedly" )); } + () = &mut ssh_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "SSH accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "SSH accept loop exited unexpectedly" + )); + } } } else { tokio::select! { @@ -984,6 +1015,21 @@ pub async fn run_sandbox( "proxy accept loop exited unexpectedly" )); } + () = &mut ssh_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "SSH accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "SSH accept loop exited unexpectedly" + )); + } } } } else { diff --git a/crates/openshell-supervisor-process/src/run.rs b/crates/openshell-supervisor-process/src/run.rs index 8a2080f217..f0e792306d 100644 --- a/crates/openshell-supervisor-process/src/run.rs +++ b/crates/openshell-supervisor-process/src/run.rs @@ -67,6 +67,7 @@ pub async fn run_process( openshell_endpoint: Option<&str>, ssh_socket_path: Option, shared_ssh_socket: bool, + ssh_exit_tx: Option>, policy: &SandboxPolicy, resolved_process_identity: ResolvedProcessIdentity, enforcement_mode: ProcessEnforcementMode, @@ -292,6 +293,7 @@ pub async fn run_process( let (ssh_ready_tx, ssh_ready_rx) = tokio::sync::oneshot::channel(); tokio::spawn(async move { + let _ssh_exit_guard = ssh_exit_tx; if let Err(err) = crate::ssh::run_ssh_server( listen_path, ssh_ready_tx, diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index a56ab8f2d7..893967b2ac 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -12,6 +12,8 @@ use crate::process::{ drop_privileges_with_identity, is_supervisor_only_env_var, session_user_and_home, }; use crate::sandbox; +#[cfg(unix)] +use libc; use miette::{IntoDiagnostic, Result}; use nix::pty::{Winsize, openpty}; use nix::unistd::setsid; @@ -143,44 +145,186 @@ pub async fn run_ssh_server( } }; - loop { - let (stream, _peer) = listener.accept().await.into_diagnostic()?; - let config = config.clone(); - let policy = policy.clone(); - let workspace = workspace.clone(); - let proxy_url = proxy_url.clone(); - let ca_paths = ca_paths.clone(); - let provider_credentials = provider_credentials.clone(); - let user_environment = user_environment.clone(); - let main_session = Arc::clone(&main_session); + let mut consecutive_resource_errors: u32 = 0; + let mut consecutive_unknown_errors: u32 = 0; - tokio::spawn(async move { - if let Err(err) = handle_connection( - stream, - config, - policy, - workspace, - netns_fd, - proxy_url, - ca_paths, - provider_credentials, - user_environment, - resolved_identity, - enforcement_mode, - main_session, - ) - .await - { - ocsf_emit!( - SshActivityBuilder::new(openshell_ocsf::ctx::ctx()) - .activity(ActivityId::Fail) - .severity(SeverityId::Low) - .status(StatusId::Failure) - .message(format!("SSH connection failed: {err}")) - .build() - ); + loop { + match listener.accept().await { + Ok((stream, _peer)) => { + consecutive_resource_errors = 0; + consecutive_unknown_errors = 0; + let config = config.clone(); + let policy = policy.clone(); + let workspace = workspace.clone(); + let proxy_url = proxy_url.clone(); + let ca_paths = ca_paths.clone(); + let provider_credentials = provider_credentials.clone(); + let user_environment = user_environment.clone(); + let main_session = Arc::clone(&main_session); + + tokio::spawn(async move { + if let Err(err) = handle_connection( + stream, + config, + policy, + workspace, + netns_fd, + proxy_url, + ca_paths, + provider_credentials, + user_environment, + resolved_identity, + enforcement_mode, + main_session, + ) + .await + { + ocsf_emit!( + SshActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::Low) + .status(StatusId::Failure) + .message(format!("SSH connection failed: {err}")) + .build() + ); + } + }); } - }); + Err(err) => { + match classify_ssh_accept_error( + &err, + &mut consecutive_resource_errors, + &mut consecutive_unknown_errors, + ) { + SshAcceptAction::Terminal => { + ocsf_emit!( + SshActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message(format!( + "SSH accept loop exiting on terminal error: {err}" + )) + .build() + ); + break; + } + SshAcceptAction::Retry { backoff, severity } => { + ocsf_emit!( + SshActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(severity) + .status(StatusId::Failure) + .message(format!( + "SSH accept error (retrying in {}ms): {err}", + backoff.as_millis(), + )) + .build() + ); + tokio::time::sleep(backoff).await; + } + } + } + } + } + + Ok(()) +} + +const MAX_CONSECUTIVE_UNKNOWN_SSH_ACCEPT_ERRORS: u32 = 10; + +#[derive(Debug, PartialEq)] +enum SshAcceptAction { + Terminal, + Retry { + backoff: Duration, + severity: SeverityId, + }, +} + +fn classify_ssh_accept_error( + err: &std::io::Error, + consecutive_resource_errors: &mut u32, + consecutive_unknown_errors: &mut u32, +) -> SshAcceptAction { + #[cfg(unix)] + if matches!( + err.raw_os_error(), + Some(libc::EBADF | libc::EINVAL | libc::ENOTSOCK) + ) { + return SshAcceptAction::Terminal; + } + + #[cfg(unix)] + if matches!( + err.raw_os_error(), + Some( + libc::EMFILE + | libc::ENFILE + | libc::ENOBUFS + | libc::ENOMEM + | libc::ECONNABORTED + | libc::ECONNRESET + | libc::EINTR + | libc::ENETDOWN + | libc::EPROTO + | libc::ENOPROTOOPT + | libc::EHOSTDOWN + | libc::EHOSTUNREACH + | libc::EOPNOTSUPP + | libc::ENETUNREACH + | libc::ENOSR + | libc::ESOCKTNOSUPPORT + | libc::EPROTONOSUPPORT + | libc::ETIMEDOUT + ) + ) { + *consecutive_unknown_errors = 0; + + #[cfg(unix)] + let is_resource_pressure = matches!( + err.raw_os_error(), + Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM | libc::ENOSR) + ); + #[cfg(not(unix))] + let is_resource_pressure = false; + + if is_resource_pressure { + *consecutive_resource_errors = consecutive_resource_errors.saturating_add(1); + let backoff_ms = 100u64 + .saturating_mul(1u64 << (*consecutive_resource_errors).min(7).saturating_sub(1)) + .min(5_000); + return SshAcceptAction::Retry { + backoff: Duration::from_millis(backoff_ms), + severity: SeverityId::Medium, + }; + } + + *consecutive_resource_errors = 0; + return SshAcceptAction::Retry { + backoff: Duration::from_millis(100), + severity: SeverityId::Low, + }; + } + + #[cfg(unix)] + #[cfg(target_os = "linux")] + if matches!(err.raw_os_error(), Some(libc::ENONET)) { + *consecutive_unknown_errors = 0; + *consecutive_resource_errors = 0; + return SshAcceptAction::Retry { + backoff: Duration::from_millis(100), + severity: SeverityId::Low, + }; + } + + *consecutive_unknown_errors = consecutive_unknown_errors.saturating_add(1); + if *consecutive_unknown_errors >= MAX_CONSECUTIVE_UNKNOWN_SSH_ACCEPT_ERRORS { + return SshAcceptAction::Terminal; + } + SshAcceptAction::Retry { + backoff: Duration::from_millis(100), + severity: SeverityId::Low, } }