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

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<oneshot::Sender<()>> 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 <burdcat17@gmail.com>

Author: politerealism

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