diff --git a/docs/features/remote-workspaces.md b/docs/features/remote-workspaces.md index 7ae7312650..1fa87de482 100644 --- a/docs/features/remote-workspaces.md +++ b/docs/features/remote-workspaces.md @@ -99,6 +99,27 @@ The configured Docker CLI remains the security boundary. OpenBitFun does not exp the Docker daemon over the network or bypass the current user's Docker permissions. +### SFTP handle ownership + +Whole-file SFTP transfers wait for CLOSE acknowledgement before reporting +success, including reads. The file guard retains cleanup ownership after cancellation, +I/O errors, or a dropped streaming reader; it also receives and closes late OPEN +replies after the caller stops waiting. Writes are not replayed if their outcome +is uncertain. A failed close or timed-out OPEN retires the affected SFTP subsystem +so its unknown handles and client accounting cannot poison later operations. + +Full and bounded directory enumeration use the same serialized raw SFTP path, +which closes directory handles on errors as well as success. Cancellation retires +that directory subsystem, and subsequent enumeration replaces it without +invalidating the SSH transport or the separate file subsystem. No persisted +profile, workspace, or wire shape changes are required. + +The locked russh-sftp 2.3 dependency sends CLOSE on ordinary file drop without +reducing its client-side handle count. Relying on that drop alone can therefore +produce `Limit exceeded: handle limit reached` even after the server has closed +every file. See [Desktop troubleshooting](../../src/apps/desktop/README.md#remote-ssh-file-handle-errors) +for recovery guidance. + ## Search on hosts without ripgrep Agent Grep keeps one matching and result-processing implementation. For diff --git a/src/apps/desktop/README.md b/src/apps/desktop/README.md index 7ae0278027..d7056ea473 100644 --- a/src/apps/desktop/README.md +++ b/src/apps/desktop/README.md @@ -47,3 +47,21 @@ controls can continue driving a host session, but do not expose WSL connection setup. Detached Dispatch does not provision WSL connections. SSH port forwarding is unavailable for native WSL targets; use Windows WSL networking to reach a Linux service. + +## Remote SSH file handle errors + +If writing files and browsing directories both start failing with +`Limit exceeded: handle limit reached`, update OpenBitFun to a build containing +the SFTP handle-lifecycle fix. Earlier builds can exhaust a client-side counter +even when the server has already closed the files. Save ongoing work before +manually disconnecting and reconnecting the remote workspace as a temporary +recovery; reconnecting can interrupt its terminals and commands. + +This message alone does not establish a server configuration problem. Raising +server limits only delays a leaked-counter failure. Running `ulimit` in a new +SSH shell does not change the limits of the already-running SFTP subsystem. +OpenBitFun does not modify the remote user's shell startup files, SSH daemon +configuration, or OS limits automatically. If the problem persists after the +fix, capture the OpenBitFun version and logs plus the server's SFTP implementation +and advertised limits so genuine concurrent-handle or server resource exhaustion +can be distinguished from a client lifecycle problem. diff --git a/src/crates/services/services-integrations/AGENTS.md b/src/crates/services/services-integrations/AGENTS.md index 16865ed8f5..5752924f0a 100644 --- a/src/crates/services/services-integrations/AGENTS.md +++ b/src/crates/services/services-integrations/AGENTS.md @@ -125,3 +125,13 @@ On Windows with an initialized WSL distribution, set `OPENBITFUN_TEST_WSL_DISTRO and run `cargo test -p openbitfun-services-integrations --no-default-features --features remote-ssh-concrete --lib wsl_windows_workspace_transport -- --ignored` for binary filesystem/stdio, exit status, cancellation, and saved reconnect. + +For SFTP handle ownership and cancellation regressions, run +`cargo test --locked -p openbitfun-services-integrations --no-default-features +--features remote-ssh-concrete --lib +remote_ssh::manager::tests::workspace_sftp::`. These loopback SSH/SFTP tests +advertise a small handle limit and are included in the existing CI +`workspace_` filter. To exercise real OpenSSH file IO over loopback SSH, set +`OPENBITFUN_TEST_SFTP_SERVER` to an installed `sftp-server` executable and run +the same command with the filter ending in +`workspace_sftp::openssh_real_files_over_loopback_ssh -- --ignored`. diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager.rs b/src/crates/services/services-integrations/src/remote_ssh/manager.rs index db94c8cb56..362a0d5a50 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/manager.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/manager.rs @@ -2,6 +2,7 @@ //! //! This module manages SSH connections using the pure-Russ SSH implementation +use super::sftp_file::ManagedSftpSession; use crate::remote_ssh::password_vault::SSHPasswordVault; use crate::remote_ssh::types::{ ConnectionTestReport, ConnectionTestStage, ContainerAccess, ContainerWorkspaceConfig, @@ -16,7 +17,6 @@ use russh::Sig; use russh_keys::key::{KeyPair, PublicKey}; use russh_keys::PublicKeyBase64; use russh_sftp::client::error::Error as SftpError; -use russh_sftp::client::fs::ReadDir; use russh_sftp::client::{RawSftpSession, SftpSession}; use russh_sftp::protocol::{File as SftpFile, StatusCode as SftpStatusCode}; #[cfg(feature = "ssh_config")] @@ -566,7 +566,7 @@ struct ActiveConnection { } struct SftpCache { - session: tokio::sync::RwLock>>, + session: tokio::sync::RwLock>>, init_lock: tokio::sync::Mutex<()>, } @@ -579,13 +579,8 @@ impl SftpCache { } } -#[derive(Clone)] -struct SftpSessionLease { - session: Arc, - cache: Arc, -} - struct BoundedSftpChannel { + retired: AtomicBool, session: Arc, read_lock: tokio::sync::Mutex<()>, } @@ -611,14 +606,14 @@ struct BoundedSftpSession { } struct BoundedSftpReadGuard { - session: Arc, + channel: Arc, armed: bool, } impl BoundedSftpReadGuard { - fn new(session: Arc) -> Self { + fn new(channel: Arc) -> Self { Self { - session, + channel, armed: true, } } @@ -631,7 +626,8 @@ impl BoundedSftpReadGuard { impl Drop for BoundedSftpReadGuard { fn drop(&mut self) { if self.armed { - let _ = self.session.close_session(); + self.channel.retired.store(true, Ordering::Release); + let _ = self.channel.session.close_session(); } } } @@ -5677,11 +5673,7 @@ impl SSHConnectionManager { /// so a transient SSH disconnect (e.g. NAT timeout while the user is idly /// browsing the remote folder picker) is recovered transparently instead /// of cascading into a stale cached SFTP handle that fails forever. - pub async fn get_sftp(&self, connection_id: &str) -> anyhow::Result> { - Ok(self.get_sftp_lease(connection_id).await?.session) - } - - async fn get_sftp_lease(&self, connection_id: &str) -> anyhow::Result { + async fn get_sftp(&self, connection_id: &str) -> anyhow::Result> { self.ensure_alive_or_reconnect(connection_id).await?; // Capture the transport and cache from the same connection generation. @@ -5705,21 +5697,29 @@ impl SSHConnectionManager { (handle, connection.sftp_session.clone()) }; - if let Some(session) = cache.session.read().await.as_ref().cloned() { - return Ok(SftpSessionLease { - session, - cache: cache.clone(), - }); + if let Some(session) = cache + .session + .read() + .await + .as_ref() + .filter(|session| !session.is_retired()) + .cloned() + { + return Ok(session); } // Serialize initialization within one generation so concurrent callers // cannot exhaust the server's channel/session limit. let _init_guard = cache.init_lock.lock().await; - if let Some(session) = cache.session.read().await.as_ref().cloned() { - return Ok(SftpSessionLease { - session, - cache: cache.clone(), - }); + if let Some(session) = cache + .session + .read() + .await + .as_ref() + .filter(|session| !session.is_retired()) + .cloned() + { + return Ok(session); } // Open a channel and request SFTP subsystem @@ -5736,20 +5736,15 @@ impl SSHConnectionManager { .await .map_err(|e| anyhow!("Failed to create SFTP session: {}", e))?; - let sftp = Arc::new(sftp); + let sftp = Arc::new(ManagedSftpSession::new(sftp)); *cache.session.write().await = Some(sftp.clone()); - Ok(SftpSessionLease { - session: sftp, - cache: cache.clone(), - }) + Ok(sftp) } - /// Get or create the raw SFTP session used by bounded directory reads. - /// - /// The high-level russh-sftp `read_dir` API buffers until EOF. Keeping a - /// separate raw session lets callers stop issuing `readdir` requests as - /// soon as their entry budget is satisfied. + /// Reuse a dedicated SFTP channel for serialized directory enumeration. + /// The high-level API leaks directory handles on errors/cancellation. The + /// raw path closes on all exits and can also stop at an entry budget. async fn get_bounded_sftp(&self, connection_id: &str) -> anyhow::Result { self.ensure_alive_or_reconnect(connection_id).await?; @@ -5770,7 +5765,13 @@ impl SSHConnectionManager { (handle, connection.bounded_sftp_session.clone()) }; - let cached_channel = cache.channel.read().await.as_ref().cloned(); + let cached_channel = cache + .channel + .read() + .await + .as_ref() + .filter(|channel| !channel.retired.load(Ordering::Acquire)) + .cloned(); if let Some(channel) = cached_channel { return Ok(BoundedSftpSession { channel, @@ -5779,7 +5780,13 @@ impl SSHConnectionManager { } let _init_guard = cache.init_lock.lock().await; - let cached_channel = cache.channel.read().await.as_ref().cloned(); + let cached_channel = cache + .channel + .read() + .await + .as_ref() + .filter(|channel| !channel.retired.load(Ordering::Acquire)) + .cloned(); if let Some(channel) = cached_channel { return Ok(BoundedSftpSession { channel, @@ -5801,6 +5808,7 @@ impl SSHConnectionManager { .await .map_err(|error| anyhow!("Failed to create bounded SFTP session: {}", error))?; let channel = Arc::new(BoundedSftpChannel { + retired: AtomicBool::new(false), session: Arc::new(session), read_lock: tokio::sync::Mutex::new(()), }); @@ -5829,6 +5837,9 @@ impl SSHConnectionManager { .await .map_err(|e| anyhow!("Failed to read remote file '{}': {}", path, e))?; + file.close() + .await + .context("Failed to close remote file after reading")?; Ok(buffer) } @@ -5883,6 +5894,9 @@ impl SSHConnectionManager { // Ensure final 100% progress is reported even if metadata returned 0. on_progress(bytes_read, total); + file.close() + .await + .context("Failed to close remote file after reading")?; Ok(buffer) } @@ -5909,6 +5923,9 @@ impl SSHConnectionManager { .await .map_err(|e| anyhow!("Failed to flush remote file '{}': {}", path, e))?; + file.close() + .await + .context("Failed to close remote file after writing")?; Ok(()) } @@ -5986,6 +6003,10 @@ impl SSHConnectionManager { .flush() .await .map_err(|error| anyhow!("Failed to flush remote file '{}': {}", path, error))?; + remote + .close() + .await + .context("Failed to close remote upload file")?; Ok(written) } @@ -6027,49 +6048,20 @@ impl SSHConnectionManager { .await .map_err(|e| anyhow!("Failed to flush remote file '{}': {}", path, e))?; + file.close() + .await + .context("Failed to close remote file after writing")?; Ok(()) } - /// Read directory via SFTP. - /// - /// Retries once after dropping the cached SFTP session and forcing a - /// reconnect attempt, so a stale SFTP channel left over from a prior - /// network blip does not permanently break the remote folder picker. - pub async fn sftp_read_dir(&self, connection_id: &str, path: &str) -> anyhow::Result { - let resolved = self.resolve_sftp_path(connection_id, path).await?; - let lease = self.get_sftp_lease(connection_id).await?; - match lease.session.read_dir(&resolved).await { - Ok(entries) => Ok(entries), - Err(first_err) => { - let generation_is_current = - self.sftp_generation_is_current(connection_id, &lease).await; - if generation_is_current && !sftp_error_may_be_stale_transport(&first_err) { - return Err(anyhow!( - "Failed to read directory '{}': {}", - resolved, - first_err - )); - } - if generation_is_current { - log::warn!( - "SFTP read_dir '{}' failed (will retry once after refreshing session): {}", - resolved, - first_err - ); - self.invalidate_sftp_generation(connection_id, &lease).await; - // Force the alive flag to false so ensure_alive_or_reconnect rebuilds - // the underlying SSH transport too — the previous failure may indicate - // the channel was torn down even though the keepalive callback has not - // fired yet. - } - let lease = self.get_sftp_lease(connection_id).await?; - lease - .session - .read_dir(&resolved) - .await - .map_err(|e| anyhow!("Failed to read directory '{}': {}", resolved, e)) - } - } + /// Read a directory with the same handle cleanup as bounded enumeration. + pub async fn sftp_read_dir( + &self, + connection_id: &str, + path: &str, + ) -> anyhow::Result> { + self.sftp_read_dir_bounded(connection_id, path, usize::MAX) + .await } /// Read at most `max_entries` directory entries without asking the SFTP @@ -6098,7 +6090,10 @@ impl SSHConnectionManager { first_error )); } - if generation_is_current { + // A cancelled enumeration may have retired this channel while + // another caller waited on its read lock. Replace that SFTP + // subsystem without marking the shared SSH transport dead. + if generation_is_current && !session.channel.retired.load(Ordering::Acquire) { log::warn!( "Bounded SFTP read_dir '{}' failed (will retry once after refreshing session): {}", resolved, @@ -6121,12 +6116,17 @@ impl SSHConnectionManager { max_entries: usize, ) -> Result, SftpError> { let _read_lock = bounded.channel.read_lock.lock().await; + if bounded.channel.retired.load(Ordering::Acquire) { + return Err(SftpError::IO("SFTP directory channel was retired".into())); + } let session = bounded.channel.session.as_ref(); - let mut read_guard = BoundedSftpReadGuard::new(bounded.channel.session.clone()); + let mut read_guard = BoundedSftpReadGuard::new(bounded.channel.clone()); let handle = match session.opendir(path.to_string()).await { Ok(handle) => handle.handle, Err(error) => { - read_guard.disarm(); + if !sftp_error_may_be_stale_transport(&error) { + read_guard.disarm(); + } return Err(error); } }; @@ -6208,38 +6208,6 @@ impl SSHConnectionManager { connection.alive.store(false, Ordering::SeqCst); } - async fn sftp_generation_is_current( - &self, - connection_id: &str, - lease: &SftpSessionLease, - ) -> bool { - self.connections - .read() - .await - .get(connection_id) - .is_some_and(|connection| Arc::ptr_eq(&connection.sftp_session, &lease.cache)) - } - - async fn invalidate_sftp_generation(&self, connection_id: &str, failed: &SftpSessionLease) { - let guard = self.connections.read().await; - let Some(connection) = guard.get(connection_id) else { - return; - }; - if !Arc::ptr_eq(&connection.sftp_session, &failed.cache) { - return; - } - - let mut cached = failed.cache.session.write().await; - if !cached - .as_ref() - .is_some_and(|session| Arc::ptr_eq(session, &failed.session)) - { - return; - } - *cached = None; - connection.alive.store(false, Ordering::SeqCst); - } - /// Create directory via SFTP pub async fn sftp_mkdir(&self, connection_id: &str, path: &str) -> anyhow::Result<()> { let path = self.resolve_sftp_path(connection_id, path).await?; @@ -6567,6 +6535,8 @@ fn sftp_mkdir_all_prefixes(path: &str) -> Vec { mod tests { use super::*; + mod workspace_sftp; + struct UnpublishedSessionTestServer { opens: usize, delayed_open: Option, diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs new file mode 100644 index 0000000000..8993d1fdb3 --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_ssh/manager/tests/workspace_sftp.rs @@ -0,0 +1,746 @@ +//! Real SSH + SFTP protocol regressions with small advertised handle limits. +use super::*; +use russh_sftp::protocol::{ + Attrs, Data, ExtendedReply, FileAttributes, Handle as FileHandle, Name, OpenFlags, Packet, + Status, Version, +}; +use std::sync::atomic::AtomicUsize; +use tokio::io::AsyncReadExt; + +const HANDLE_LIMIT: usize = 4; + +#[derive(Default)] +struct State { + opened: AtomicUsize, + closed: AtomicUsize, + live: AtomicUsize, + channels: AtomicUsize, + pause_open: AtomicBool, + pause_read: AtomicBool, + pause_dir: AtomicBool, + pause_close: AtomicBool, + fail_read: AtomicBool, + fail_write: AtomicBool, + fail_close: AtomicBool, + requested: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +struct SftpServer { + state: Arc, + handles: HashMap, + data: Vec, +} +impl Drop for SftpServer { + fn drop(&mut self) { + self.state + .live + .fetch_sub(self.handles.len(), Ordering::SeqCst); + } +} +impl SftpServer { + async fn open_handle( + &mut self, + id: u32, + directory: bool, + ) -> Result { + if self.handles.len() >= HANDLE_LIMIT { + return Err(SftpStatusCode::Failure); + } + let number = self.state.opened.fetch_add(1, Ordering::SeqCst); + let handle = number.to_string(); + self.handles.insert(handle.clone(), directory); + self.state.live.fetch_add(1, Ordering::SeqCst); + if !directory && self.state.pause_open.swap(false, Ordering::SeqCst) { + self.state.requested.notify_one(); + self.state.release.notified().await; + } + Ok(FileHandle { id, handle }) + } +} +fn ok(id: u32) -> Status { + Status { + id, + status_code: SftpStatusCode::Ok, + error_message: String::new(), + language_tag: String::new(), + } +} +fn metadata() -> FileAttributes { + FileAttributes { + size: Some(4), + permissions: Some(0o100644), + ..Default::default() + } +} +impl russh_sftp::server::Handler for SftpServer { + type Error = SftpStatusCode; + fn unimplemented(&self) -> Self::Error { + SftpStatusCode::OpUnsupported + } + async fn init( + &mut self, + _version: u32, + _extensions: HashMap, + ) -> Result { + let mut version = Version::new(); + version + .extensions + .insert("limits@openssh.com".into(), "1".into()); + Ok(version) + } + async fn extended( + &mut self, + id: u32, + request: String, + _data: Vec, + ) -> Result { + assert_eq!(request, "limits@openssh.com"); + let data = [262144_u64, 65536, 65536, HANDLE_LIMIT as u64] + .into_iter() + .flat_map(u64::to_be_bytes) + .collect(); + Ok(Packet::ExtendedReply(ExtendedReply { id, data })) + } + async fn open( + &mut self, + id: u32, + _filename: String, + flags: OpenFlags, + _attrs: FileAttributes, + ) -> Result { + if flags.contains(OpenFlags::TRUNCATE) { + self.data.clear(); + } + self.open_handle(id, false).await + } + async fn close(&mut self, id: u32, handle: String) -> Result { + if self.state.pause_close.swap(false, Ordering::SeqCst) { + self.state.requested.notify_one(); + self.state.release.notified().await; + } + if self.state.fail_close.swap(false, Ordering::SeqCst) { + return Err(SftpStatusCode::Failure); + } + self.handles + .remove(&handle) + .ok_or(SftpStatusCode::Failure)?; + self.state.closed.fetch_add(1, Ordering::SeqCst); + self.state.live.fetch_sub(1, Ordering::SeqCst); + Ok(ok(id)) + } + async fn read( + &mut self, + id: u32, + handle: String, + offset: u64, + len: u32, + ) -> Result { + assert!(self.handles.contains_key(&handle)); + if self.state.pause_read.swap(false, Ordering::SeqCst) { + self.state.requested.notify_one(); + self.state.release.notified().await; + } + if self.state.fail_read.swap(false, Ordering::SeqCst) { + return Err(SftpStatusCode::PermissionDenied); + } + let start = offset as usize; + if start >= self.data.len() { + return Err(SftpStatusCode::Eof); + } + Ok(Data { + id, + data: self.data[start..(start + len as usize).min(self.data.len())].to_vec(), + }) + } + async fn write( + &mut self, + id: u32, + handle: String, + offset: u64, + data: Vec, + ) -> Result { + assert!(self.handles.contains_key(&handle)); + if self.state.fail_write.swap(false, Ordering::SeqCst) { + return Err(SftpStatusCode::PermissionDenied); + } + let start = offset as usize; + self.data.resize(self.data.len().max(start + data.len()), 0); + self.data[start..start + data.len()].copy_from_slice(&data); + Ok(ok(id)) + } + async fn stat(&mut self, id: u32, _path: String) -> Result { + Ok(Attrs { + id, + attrs: metadata(), + }) + } + async fn fstat(&mut self, id: u32, _handle: String) -> Result { + Ok(Attrs { + id, + attrs: metadata(), + }) + } + async fn opendir(&mut self, id: u32, _path: String) -> Result { + self.open_handle(id, true).await + } + async fn readdir(&mut self, id: u32, handle: String) -> Result { + if self.state.pause_dir.swap(false, Ordering::SeqCst) { + self.state.requested.notify_one(); + self.state.release.notified().await; + } + if self.state.fail_read.swap(false, Ordering::SeqCst) { + return Err(SftpStatusCode::PermissionDenied); + } + let first = self.handles.get_mut(&handle).unwrap(); + if !*first { + return Err(SftpStatusCode::Eof); + } + *first = false; + Ok(Name { + id, + files: vec![SftpFile { + filename: "file".into(), + longname: String::new(), + attrs: metadata(), + }], + }) + } +} + +struct SshServer { + openssh: Option, + state: Arc, + channels: HashMap>, +} +#[async_trait] +impl russh::server::Handler for SshServer { + type Error = russh::Error; + async fn auth_none(&mut self, _user: &str) -> Result { + Ok(russh::server::Auth::Accept) + } + async fn channel_open_session( + &mut self, + channel: russh::Channel, + _session: &mut russh::server::Session, + ) -> Result { + self.channels.insert(channel.id(), channel); + Ok(true) + } + async fn subsystem_request( + &mut self, + id: russh::ChannelId, + name: &str, + session: &mut russh::server::Session, + ) -> Result<(), Self::Error> { + assert_eq!(name, "sftp"); + session.channel_success(id); + self.state.channels.fetch_add(1, Ordering::SeqCst); + let channel = self.channels.remove(&id).unwrap(); + if let Some(executable) = &self.openssh { + let mut command = process_manager::create_tokio_command(executable); + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + .unwrap(); + let mut stdin = child.stdin.take().unwrap(); + let mut stdout = child.stdout.take().unwrap(); + let (mut reader, mut writer) = tokio::io::split(channel.into_stream()); + tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let input = async { + tokio::io::copy(&mut reader, &mut stdin).await?; + stdin.shutdown().await + }; + let output = async { + tokio::io::copy(&mut stdout, &mut writer).await?; + writer.shutdown().await + }; + tokio::select! { + _ = async { tokio::try_join!(input, output) } => {}, + _ = child.wait() => {}, + } + let _ = child.kill().await; + }); + return Ok(()); + } + let handler = SftpServer { + state: self.state.clone(), + handles: HashMap::new(), + data: vec![0, 1, 254, 255], + }; + tokio::spawn(russh_sftp::server::run(channel.into_stream(), handler)); + Ok(()) + } +} + +struct Fixture { + manager: Arc, + state: Arc, + task: tokio::task::JoinHandle<()>, + _dir: tempfile::TempDir, +} +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} +impl Fixture { + async fn new() -> Self { + Self::with_openssh(None).await + } + + async fn with_openssh(openssh: Option) -> Self { + let state = Arc::new(State::default()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = SshServer { + openssh, + state: state.clone(), + channels: HashMap::new(), + }; + let config = Arc::new(russh::server::Config { + keys: vec![KeyPair::generate_ed25519().unwrap()], + ..Default::default() + }); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let _ = russh::server::run_stream(config, socket, server) + .await + .unwrap() + .await; + }); + let mut handle = russh::client::connect( + Arc::new(russh::client::Config::default()), + address, + SSHHandler::with_verify_callback(|_, _, _| true), + ) + .await + .unwrap(); + assert!(handle.authenticate_none("sftp-test").await.unwrap()); + let dir = tempfile::tempdir().unwrap(); + let manager = Arc::new(SSHConnectionManager::new(dir.path().into())); + let config = SSHConnectionConfig { + id: "sftp-test".into(), + name: "SFTP loopback fixture".into(), + host: "127.0.0.1".into(), + port: address.port(), + username: "sftp-test".into(), + auth: SSHAuthMethod::Agent { + key_fingerprint: None, + fallback_key_path: None, + }, + default_workspace: None, + proxy_jump: None, + container: None, + wsl: None, + options: Default::default(), + }; + manager.connections.write().await.insert( + config.id.clone(), + ActiveConnection { + handle: Some(Arc::new(handle)), + jump_handles: Vec::new(), + effective_config: config.clone(), + config, + server_info: None, + sftp_session: Arc::new(SftpCache::new()), + bounded_sftp_session: Arc::new(BoundedSftpCache::new()), + server_key: None, + alive: Arc::new(AtomicBool::new(true)), + }, + ); + Self { + manager, + state, + task, + _dir: dir, + } + } + async fn drained(&self) { + tokio::time::timeout(Duration::from_secs(5), async { + while self.state.live.load(Ordering::SeqCst) != 0 { + tokio::task::yield_now().await; + } + // CLOSE packet receipt precedes the client's counter update. A + // metadata round trip ensures that response was consumed as well. + self.manager.sftp_stat("sftp-test", "/file").await.unwrap(); + }) + .await + .expect("SFTP handles were not reclaimed"); + } + async fn requested(&self) { + tokio::time::timeout(Duration::from_secs(5), self.state.requested.notified()) + .await + .unwrap(); + } +} + +#[tokio::test] +#[ignore = "Diagnostic for the locked russh-sftp drop bug; not a contract for future dependency versions"] +async fn dependency_drop_reproduces_client_limit_with_zero_server_handles() { + let f = Fixture::new().await; + let session = f.manager.get_sftp("sftp-test").await.unwrap(); + let raw_high_level: &SftpSession = &session; + for _ in 0..HANDLE_LIMIT { + drop(raw_high_level.open("/file").await.unwrap()); + raw_high_level.metadata("/file").await.unwrap(); + } + assert_eq!(f.state.live.load(Ordering::SeqCst), 0); + assert!(matches!( + raw_high_level.create("/file").await, + Err(SftpError::Limited(_)) + )); + assert!(matches!( + raw_high_level.read_dir("/").await, + Err(SftpError::Limited(_)) + )); +} + +#[tokio::test] +async fn repeated_manager_transfers_release_handles_and_preserve_bytes() { + let f = Fixture::new().await; + let local = f._dir.path().join("upload"); + let bytes = vec![0, 128, 255, 10]; + tokio::fs::write(&local, &bytes).await.unwrap(); + for _ in 0..HANDLE_LIMIT * 3 { + f.manager + .sftp_write("sftp-test", "/file", &bytes) + .await + .unwrap(); + assert_eq!( + f.manager.sftp_read("sftp-test", "/file").await.unwrap(), + bytes + ); + f.manager + .sftp_write_with_progress("sftp-test", "/file", &bytes, 2, &mut |_, _| true) + .await + .unwrap(); + assert_eq!( + f.manager + .sftp_read_with_progress("sftp-test", "/file", 2, &mut |_, _| true) + .await + .unwrap(), + bytes + ); + assert_eq!( + f.manager + .sftp_write_from_file("sftp-test", "/file", &local, 4) + .await + .unwrap(), + 4 + ); + assert_eq!( + f.manager + .sftp_read_dir("sftp-test", "/") + .await + .unwrap() + .len(), + 1 + ); + assert_eq!(f.state.live.load(Ordering::SeqCst), 0); + } + assert_eq!( + f.state.channels.load(Ordering::SeqCst), + 2, + "reuse file and directory subsystems" + ); +} + +#[tokio::test] +async fn cancelled_progress_and_dropped_streams_reclaim_handles() { + let f = Fixture::new().await; + for _ in 0..HANDLE_LIMIT * 2 { + assert!(f + .manager + .sftp_read_with_progress("sftp-test", "/file", 1, &mut |_, _| false) + .await + .is_err()); + f.drained().await; + assert!(f + .manager + .sftp_write_with_progress("sftp-test", "/file", &[1, 2, 3, 4], 1, &mut |_, _| false) + .await + .is_err()); + f.drained().await; + let mut reader = f + .manager + .open_workspace_file_read("sftp-test", "/file") + .await + .unwrap(); + let mut byte = [0]; + reader.read_exact(&mut byte).await.unwrap(); + drop(reader); + f.drained().await; + } + assert_eq!(f.state.channels.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn cancellation_during_open_and_read_closes_late_handles() { + let f = Fixture::new().await; + for opening in [true, false] { + if opening { + f.state.pause_open.store(true, Ordering::SeqCst); + } else { + f.state.pause_read.store(true, Ordering::SeqCst); + } + let manager = f.manager.clone(); + let caller = tokio::spawn(async move { manager.sftp_read("sftp-test", "/file").await }); + f.requested().await; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + f.state.release.notify_one(); + f.drained().await; + f.manager.sftp_read("sftp-test", "/file").await.unwrap(); + } + assert_eq!(f.state.channels.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn io_errors_and_close_failure_do_not_poison_following_transfers() { + let f = Fixture::new().await; + for _ in 0..HANDLE_LIMIT * 2 { + f.state.fail_read.store(true, Ordering::SeqCst); + assert!(f.manager.sftp_read("sftp-test", "/file").await.is_err()); + f.drained().await; + } + f.state.fail_write.store(true, Ordering::SeqCst); + assert!(f + .manager + .sftp_write("sftp-test", "/file", &[1]) + .await + .is_err()); + f.drained().await; + f.state.fail_close.store(true, Ordering::SeqCst); + assert!(f + .manager + .sftp_write("sftp-test", "/file", &[2]) + .await + .unwrap_err() + .to_string() + .contains("close")); + f.manager + .sftp_write("sftp-test", "/file", &[3]) + .await + .unwrap(); + assert_eq!( + f.manager.sftp_read("sftp-test", "/file").await.unwrap(), + vec![3] + ); + assert_eq!(f.state.channels.load(Ordering::SeqCst), 2); +} + +#[tokio::test] +async fn directory_errors_and_cancellation_preserve_ssh_and_sibling_file() { + let f = Fixture::new().await; + let mut sibling = f + .manager + .open_workspace_file_read("sftp-test", "/file") + .await + .unwrap(); + for _ in 0..HANDLE_LIMIT * 2 { + f.state.fail_read.store(true, Ordering::SeqCst); + assert!(f.manager.sftp_read_dir("sftp-test", "/").await.is_err()); + assert_eq!(f.state.live.load(Ordering::SeqCst), 1); + } + f.state.pause_dir.store(true, Ordering::SeqCst); + let manager = f.manager.clone(); + let caller = tokio::spawn(async move { manager.sftp_read_dir("sftp-test", "/").await }); + f.requested().await; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + f.state.release.notify_one(); + assert_eq!( + f.manager + .sftp_read_dir_bounded("sftp-test", "/", 1) + .await + .unwrap() + .len(), + 1 + ); + let mut bytes = Vec::new(); + sibling.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, vec![0, 1, 254, 255]); + drop(sibling); + f.drained().await; + assert_eq!(f.state.channels.load(Ordering::SeqCst), 3); + assert!(f + .manager + .connections + .read() + .await + .get("sftp-test") + .unwrap() + .alive + .load(Ordering::SeqCst)); +} + +#[tokio::test] +async fn concurrent_transfers_reuse_the_negotiated_handle_capacity() { + let f = Fixture::new().await; + for _ in 0..8 { + let mut tasks = tokio::task::JoinSet::new(); + for _ in 0..HANDLE_LIMIT { + let manager = f.manager.clone(); + tasks.spawn(async move { manager.sftp_read("sftp-test", "/file").await.unwrap() }); + } + while let Some(result) = tasks.join_next().await { + assert_eq!(result.unwrap(), vec![0, 1, 254, 255]); + } + assert_eq!(f.state.live.load(Ordering::SeqCst), 0); + } + assert_eq!(f.state.channels.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn timed_out_create_retires_only_sftp_without_replaying_mutation() { + let f = Fixture::new().await; + f.manager + .get_sftp("sftp-test") + .await + .unwrap() + .set_timeout(1); + f.state.pause_open.store(true, Ordering::SeqCst); + assert!(f + .manager + .sftp_write("sftp-test", "/file", &[7]) + .await + .is_err()); + assert_eq!( + f.state.opened.load(Ordering::SeqCst), + 1, + "CREATE must not be replayed" + ); + f.state.release.notify_one(); + f.manager + .sftp_write("sftp-test", "/file", &[8]) + .await + .unwrap(); + assert_eq!( + f.manager.sftp_read("sftp-test", "/file").await.unwrap(), + vec![8] + ); + assert_eq!(f.state.channels.load(Ordering::SeqCst), 2); +} + +#[tokio::test] +#[ignore = "Set OPENBITFUN_TEST_SFTP_SERVER to the OpenSSH sftp-server executable"] +async fn openssh_real_files_over_loopback_ssh() { + use tokio::io::{AsyncSeekExt, AsyncWriteExt}; + let executable = std::env::var("OPENBITFUN_TEST_SFTP_SERVER") + .expect("OpenSSH sftp-server executable required"); + let f = Fixture::with_openssh(Some(executable)).await; + let path = f + ._dir + .path() + .join("remote.bin") + .to_str() + .unwrap() + .to_owned(); + let directory = f._dir.path().to_str().unwrap().to_owned(); + let bytes: Vec = (0..65536).map(|n| n as u8).collect(); + for _ in 0..128 { + f.manager + .sftp_write("sftp-test", &path, &bytes) + .await + .unwrap(); + assert_eq!( + f.manager.sftp_read("sftp-test", &path).await.unwrap(), + bytes + ); + assert_eq!( + f.manager + .sftp_read_dir("sftp-test", &directory) + .await + .unwrap() + .len(), + 1 + ); + } + let mut stream = f + .manager + .open_workspace_file_read("sftp-test", &path) + .await + .unwrap(); + stream.seek(std::io::SeekFrom::Start(100)).await.unwrap(); + let mut tail = Vec::new(); + stream.read_to_end(&mut tail).await.unwrap(); + assert_eq!(tail, bytes[100..]); + drop(stream); + assert!(f + .manager + .sftp_read_with_progress("sftp-test", &path, 1024, &mut |_, _| false) + .await + .is_err()); + // A rejected directory-as-file open must close its FSTAT-checked handle. + assert!(f + .manager + .open_workspace_file_read("sftp-test", &directory) + .await + .is_err()); + let session = f.manager.get_sftp("sftp-test").await.unwrap(); + let mut file = session.create(&path).await.unwrap(); + file.write_all(&bytes).await.unwrap(); + file.shutdown().await.unwrap(); + file.shutdown().await.unwrap(); // idempotent; must not double-CLOSE a reused id + drop(file); + assert_eq!(tokio::fs::read(&path).await.unwrap(), bytes); + assert_eq!( + f.manager.sftp_read("sftp-test", &path).await.unwrap(), + bytes + ); + assert_eq!(f.state.channels.load(Ordering::SeqCst), 2); +} + +#[tokio::test] +async fn cancellation_during_close_finishes_the_existing_close_once() { + let f = Fixture::new().await; + for _ in 0..HANDLE_LIMIT * 2 { + f.state.pause_close.store(true, Ordering::SeqCst); + let manager = f.manager.clone(); + let caller = + tokio::spawn(async move { manager.sftp_write("sftp-test", "/file", &[42]).await }); + f.requested().await; + caller.abort(); + assert!(caller.await.unwrap_err().is_cancelled()); + f.state.release.notify_one(); + f.drained().await; + } + assert_eq!( + f.state.opened.load(Ordering::SeqCst), + f.state.closed.load(Ordering::SeqCst) + ); + assert_eq!(f.state.channels.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn a_directory_waiter_detects_retirement_after_acquiring_the_lock() { + let f = Fixture::new().await; + let session = f.manager.get_bounded_sftp("sftp-test").await.unwrap(); + let lock = session.channel.read_lock.lock().await; + let stale = session.clone(); + let waiter = tokio::spawn(async move { + SSHConnectionManager::read_bounded_sftp_entries(&stale, "/", 1).await + }); + drop(BoundedSftpReadGuard::new(session.channel.clone())); + drop(lock); + assert!(tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .unwrap() + .unwrap() + .is_err()); + assert_eq!( + f.manager + .sftp_read_dir("sftp-test", "/") + .await + .unwrap() + .len(), + 1 + ); + assert_eq!(f.state.channels.load(Ordering::SeqCst), 2); + assert_eq!(f.state.opened.load(Ordering::SeqCst), 1); +} diff --git a/src/crates/services/services-integrations/src/remote_ssh/mod.rs b/src/crates/services/services-integrations/src/remote_ssh/mod.rs index a86907384e..bbe832b786 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/mod.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/mod.rs @@ -7,6 +7,8 @@ mod file_name_search; pub mod paths; mod product_paths; pub mod remote_git; +#[cfg(feature = "remote-ssh-concrete")] +mod sftp_file; mod shell; #[cfg(feature = "remote-ssh-concrete")] mod transport; diff --git a/src/crates/services/services-integrations/src/remote_ssh/remote_fs.rs b/src/crates/services/services-integrations/src/remote_ssh/remote_fs.rs index 2f40864d3b..f760342433 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/remote_fs.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/remote_fs.rs @@ -297,12 +297,9 @@ impl RemoteFileService { None => Ok(manager .sftp_read_dir(connection_id, path) .await? + .into_iter() .map(|entry| { - remote_dir_entry_from_metadata( - &path_resolved, - entry.file_name(), - entry.metadata(), - ) + remote_dir_entry_from_metadata(&path_resolved, entry.filename, entry.attrs) }) .collect()), } diff --git a/src/crates/services/services-integrations/src/remote_ssh/sftp_file.rs b/src/crates/services/services-integrations/src/remote_ssh/sftp_file.rs new file mode 100644 index 0000000000..436a014cd6 --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_ssh/sftp_file.rs @@ -0,0 +1,181 @@ +//! Own SFTP files through CLOSE acknowledgement, including cancelled callers. +//! +//! russh-sftp 2.3 File::drop sends CLOSE without decrementing the client's +//! handle counter. Explicit shutdown is required even for read-only files. +use russh_sftp::client::{ + error::Error, + fs::{File, Metadata}, + SftpSession, +}; +use std::{ + io, + pin::Pin, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + task::{Context, Poll}, +}; +use tokio::io::{AsyncRead, AsyncSeek, AsyncWrite, AsyncWriteExt, ReadBuf}; + +pub(super) struct ManagedSftpSession { + session: Arc, + retired: AtomicBool, +} + +impl std::ops::Deref for ManagedSftpSession { + type Target = SftpSession; + fn deref(&self) -> &Self::Target { + &self.session + } +} + +impl ManagedSftpSession { + pub(super) fn new(session: SftpSession) -> Self { + Self { + session: Arc::new(session), + retired: AtomicBool::new(false), + } + } + + pub(super) fn is_retired(&self) -> bool { + self.retired.load(Ordering::Acquire) + } + + async fn retire(&self) { + self.retired.store(true, Ordering::Release); + let _ = self.session.close().await; + } + + pub(super) async fn open(self: &Arc, path: &str) -> Result { + self.open_owned(path, false).await + } + + pub(super) async fn create(self: &Arc, path: &str) -> Result { + self.open_owned(path, true).await + } + + async fn open_owned( + self: &Arc, + path: &str, + create: bool, + ) -> Result { + let session = self.clone(); + let path = path.to_owned(); + let (send, receive) = tokio::sync::oneshot::channel(); + // Keep OPEN alive until its reply. If the caller disappears, the + // undelivered guard still closes a late handle. Never replay CREATE. + tokio::spawn(async move { + if send.is_closed() { + return; + } + let result = if create { + session.session.create(path).await + } else { + session.session.open(path).await + }; + if matches!( + &result, + Err(Error::Timeout + | Error::IO(_) + | Error::UnexpectedBehavior(_) + | Error::UnexpectedPacket) + ) { + // A timed-out OPEN may have succeeded remotely without a known + // handle; retire only this SFTP channel to reclaim that handle. + session.retire().await; + } + let result = result.map(|file| ManagedSftpFile { + file: Some(file), + session, + runtime: tokio::runtime::Handle::current(), + }); + let _ = send.send(result); + }); + receive + .await + .map_err(|_| Error::UnexpectedBehavior("SFTP open owner stopped".into()))? + } +} + +pub(super) struct ManagedSftpFile { + file: Option, + session: Arc, + runtime: tokio::runtime::Handle, +} + +impl ManagedSftpFile { + pub(super) async fn metadata(&self) -> Result { + self.file.as_ref().expect("open SFTP file").metadata().await + } + + pub(super) async fn close(&mut self) -> io::Result<()> { + self.shutdown().await + } + + fn file_mut(&mut self) -> io::Result<&mut File> { + self.file + .as_mut() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "SFTP file is closed")) + } +} + +impl Drop for ManagedSftpFile { + fn drop(&mut self) { + if let Some(mut file) = self.file.take() { + let session = self.session.clone(); + self.runtime.spawn(async move { + if let Err(error) = file.shutdown().await { + log::warn!("SFTP file cleanup failed; retiring subsystem: {}", error); + session.retire().await; + } + }); + } + } +} + +impl AsyncRead for ManagedSftpFile { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(self.file_mut()?).poll_read(cx, buf) + } +} +impl AsyncWrite for ManagedSftpFile { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(self.file_mut()?).poll_write(cx, buf) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(self.file_mut()?).poll_flush(cx) + } + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let Some(file) = self.file.as_mut() else { + return Poll::Ready(Ok(())); + }; + let result = std::task::ready!(Pin::new(file).poll_shutdown(cx)); + if result.is_err() { + // Do not publish a poisoned counter/channel to another operation. + self.session.retired.store(true, Ordering::Release); + let session = self.session.clone(); + self.runtime.spawn(async move { + session.retire().await; + }); + } + self.file.take(); + Poll::Ready(result) + } +} +impl AsyncSeek for ManagedSftpFile { + fn start_seek(mut self: Pin<&mut Self>, pos: io::SeekFrom) -> io::Result<()> { + Pin::new(self.file_mut()?).start_seek(pos) + } + fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(self.file_mut()?).poll_complete(cx) + } +}