fix(ssh): add EMFILE backoff and exit notification to SSH accept loop

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<Box<dyn Future<Output = ()> + 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 {

crates/openshell-supervisor-process/src/run.rs

@@ -67,6 +67,7 @@ pub async fn run_process(
  openshell_endpoint: Option<&str>,
  ssh_socket_path: Option<String>,
  shared_ssh_socket: bool,
+ ssh_exit_tx: Option<tokio::sync::oneshot::Sender<()>>,
  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,

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,
  }
 }