From 90c8b9436eb3a3139a2b498503f117951bae8d90 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Fri, 16 Oct 2026 22:34:23 +0000 Subject: [PATCH 1/8] Keep the session maps unlocked across network I/O SessionManager and SftpManager each kept one map lock held across every network await: a keystroke send, a disconnect, a directory listing, and in SFTP's case an entire file transfer. One peer that stopped reading, or a large download, therefore stalled every other session, and close_all could not drain the map. The vault reset gives close_all five seconds and then drops it, so an SFTP session busy in a transfer or an endless listing stayed reachable after the reset that was meant to cut it off. Entries are now Arc'd and looked up under a short lock; the I/O runs after the guard is released. Writes to one SSH session still go out in order through a per-session lock. close and close_all remove entries first and bound the goodbye. SFTP listings get a 30 second overall deadline, since russh-sftp reads until the server says end-of-file with no total limit. The SSH reader also removes its own entry and disconnects when the remote shell exits, instead of leaving the connection and its keepalives up until the pane is closed, and it no longer spins on a closed resize channel while a close is in flight. --- core/src/session.rs | 142 +++++++++++---- core/src/sftp.rs | 138 +++++++------- core/tests/session_isolation.rs | 291 ++++++++++++++++++++++++++++++ desktop/src-tauri/src/commands.rs | 6 +- 4 files changed, 475 insertions(+), 102 deletions(-) create mode 100644 core/tests/session_isolation.rs diff --git a/core/src/session.rs b/core/src/session.rs index 3bf22469..e7cfda27 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -8,9 +8,15 @@ use russh::client::Handle; use russh::ChannelMsg; use std::collections::{HashMap, HashSet}; use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use tokio::sync::{mpsc, Mutex}; +/// How long a polite disconnect may take before the connection is simply +/// dropped. A peer that has stopped reading never acknowledges one. +const DISCONNECT_GRACE: Duration = Duration::from_secs(5); + +type SessionMap = HashMap>; + /// Callback invoked when a session receives data from the remote end. /// The Tauri layer uses this to forward a batch of output to the frontend. pub type DataCallback = Arc; @@ -28,7 +34,11 @@ async fn sleep_until(deadline: Option) { /// Manages all active terminal sessions (SSH and local). pub struct SessionManager { - sessions: Mutex>, + /// Live sessions by id. Held only to look an entry up, insert or remove + /// it, never across network I/O: one peer that stops reading would + /// otherwise stall writes, resizes and closes on every other session, and + /// keep a vault reset from draining the map. + sessions: Arc>, /// Every id held by an SSH session, from the moment it is claimed — before /// the connection exists — until the session is closed. `sessions` only /// learns an id once the connection is up, seconds later, so it cannot tell @@ -106,10 +116,12 @@ fn lock_claims( } struct SshSession { - #[allow(dead_code)] - id: String, - handle: Arc>>, + handle: Handle, channel_id: russh::ChannelId, + /// Serializes writes to this session so keystrokes reach the channel in + /// the order they were sent. Per session, so a stalled peer only holds up + /// its own input. + writes: Mutex<()>, resize_tx: mpsc::Sender<(u32, u32)>, /// Held so the id stays reserved while this entry exists. #[allow(dead_code)] @@ -122,7 +134,7 @@ pub use connection::SshHandler; impl SessionManager { pub fn new(data_callback: DataCallback, close_callback: CloseCallback) -> Self { Self { - sessions: Mutex::new(HashMap::new()), + sessions: Arc::new(Mutex::new(HashMap::new())), claimed: Arc::new(std::sync::Mutex::new(HashSet::new())), data_callback, close_callback, @@ -207,19 +219,19 @@ impl SessionManager { } } - let handle = Arc::new(Mutex::new(handle)); let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32); let held = claim.commit(); - let session = SshSession { - id: session_id.clone(), - handle: handle.clone(), + let session = Arc::new(SshSession { + handle, channel_id, + writes: Mutex::new(()), resize_tx, claim: held.clone(), - }; + }); self.sessions.lock().await.insert(session_id.clone(), session); + let sessions = self.sessions.clone(); // Spawn the reading task let sid = session_id.clone(); @@ -230,7 +242,7 @@ impl SessionManager { tokio::spawn(async move { // Dropped when this task ends, after the close has been announced, // which is the moment the id is free for another session. - let _held = held; + let held = held; let mut pending_resize: Option<(u32, u32)> = None; let mut batcher = OutputBatcher::new(Instant::now()); let flush = |batcher: &mut OutputBatcher| { @@ -259,13 +271,23 @@ impl SessionManager { // order; see `AppState::init`. flush(&mut batcher); close_cb(&sid, "session closed"); + // A shell that exits leaves the connection up; + // end it, unless a close already took the entry. + // The goodbye runs on its own so the id is free + // for a reconnect as soon as this task ends. + if let Some(session) = take_if_held(&sessions, &sid, &held).await { + tokio::spawn(async move { disconnect(&session).await }); + } break; } _ => {} } } - resize = resize_rx.recv() => { - pending_resize = resize; + // Once the entry is gone the sender is dropped and `recv` + // answers `None` at once; the pattern disables the branch + // then, so the loop waits on the channel instead of spinning. + Some(size) = resize_rx.recv() => { + pending_resize = Some(size); } () = sleep_until(flush_at) => { flush(&mut batcher); @@ -280,14 +302,30 @@ impl SessionManager { Ok(()) } + /// The live session behind `session_id`, taken out from under the map lock. + async fn get(&self, session_id: &str) -> Result> { + self.sessions + .lock() + .await + .get(session_id) + .cloned() + .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string())) + } + + /// Whether an SSH session is live under `session_id`. + pub async fn contains(&self, session_id: &str) -> bool { + self.sessions.lock().await.contains_key(session_id) + } + /// Write data to a session's terminal. + /// + /// The send waits for as long as the peer leaves the connection unread, + /// and only this session's writes wait behind it. pub async fn write(&self, session_id: &str, data: &[u8]) -> Result<()> { - let sessions = self.sessions.lock().await; - let session = sessions - .get(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; - let handle = session.handle.lock().await; - handle + let session = self.get(session_id).await?; + let _order = session.writes.lock().await; + session + .handle .data(session.channel_id, data.to_vec()) .await .map_err(|_| CoreError::Ssh("write failed".into()))?; @@ -296,10 +334,7 @@ impl SessionManager { /// Resize a session's terminal. pub async fn resize(&self, session_id: &str, cols: u32, rows: u32) -> Result<()> { - let sessions = self.sessions.lock().await; - let session = sessions - .get(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let session = self.get(session_id).await?; session .resize_tx .send((cols, rows)) @@ -309,13 +344,15 @@ impl SessionManager { } /// Close a session. + /// + /// The entry leaves the map before the disconnect is sent, so the session + /// stops being addressable at once even if the peer never answers. The + /// entry's share of the id goes with it; the reader task still holds one + /// until it has flushed and announced the close. pub async fn close(&self, session_id: &str) -> Result<()> { - let mut sessions = self.sessions.lock().await; - if let Some(session) = sessions.remove(session_id) { - // The entry's share of the id goes with it; the reader task still - // holds one until it has flushed and announced the close. - let handle = session.handle.lock().await; - let _ = handle.disconnect(russh::Disconnect::ByApplication, "", "en").await; + let removed = self.sessions.lock().await.remove(session_id); + if let Some(session) = removed { + disconnect(&session).await; } Ok(()) } @@ -329,24 +366,53 @@ impl SessionManager { /// /// The map is drained under the lock and the guard is released before any /// network work, so a session stops being addressable the moment this is - /// called. A remote that never acknowledges the disconnect can therefore - /// only delay the polite goodbye, never keep a session reachable. + /// called. The disconnects then run side by side, each bounded by + /// [`DISCONNECT_GRACE`], so one peer that never acknowledges cannot hold + /// up the goodbye to the others. pub async fn close_all(&self) -> usize { - let drained: Vec = { + let drained: Vec> = { let mut sessions = self.sessions.lock().await; sessions.drain().map(|(_, session)| session).collect() }; let count = drained.len(); - for session in &drained { - let handle = session.handle.lock().await; - let _ = handle - .disconnect(russh::Disconnect::ByApplication, "", "en") - .await; + let goodbyes: Vec<_> = drained + .into_iter() + .map(|session| tokio::spawn(async move { disconnect(&session).await })) + .collect(); + for goodbye in goodbyes { + let _ = goodbye.await; } count } } +/// Remove `session_id` from the map if the entry there is still the session +/// `held` belongs to. An id is reused once its session has ended, and a +/// reader finishing late must not take a newer session's entry. +async fn take_if_held( + sessions: &Mutex, + session_id: &str, + held: &Arc, +) -> Option> { + let mut sessions = sessions.lock().await; + match sessions.get(session_id) { + Some(session) if Arc::ptr_eq(&session.claim, held) => sessions.remove(session_id), + _ => None, + } +} + +/// Say goodbye to the peer, giving up after [`DISCONNECT_GRACE`]. The +/// connection ends either way once the last reference to the handle is gone. +async fn disconnect(session: &SshSession) { + let _ = tokio::time::timeout( + DISCONNECT_GRACE, + session + .handle + .disconnect(russh::Disconnect::ByApplication, "", "en"), + ) + .await; +} + #[cfg(test)] mod claim_tests { use super::SessionManager; diff --git a/core/src/sftp.rs b/core/src/sftp.rs index 6c71aed4..274eaf3e 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -9,9 +9,19 @@ use russh_sftp::client::SftpSession; use std::collections::HashMap; use std::path::Path; use std::sync::Arc; +use std::time::Duration; use tokio::io::AsyncWriteExt; use tokio::sync::Mutex; +/// The longest a directory listing may take in total. russh-sftp reads a +/// directory until the server reports its end, with a timeout on each request +/// but none overall, so a server that keeps answering with more entries +/// would otherwise keep the listing, and its memory, growing for good. +const LIST_DIR_DEADLINE: Duration = Duration::from_secs(30); + +/// How long a polite goodbye may take before the connection is dropped. +const CLOSE_GRACE: Duration = Duration::from_secs(5); + /// A directory entry returned by SFTP list operations. #[derive(Debug, Clone, serde::Serialize)] pub struct SftpEntry { @@ -26,16 +36,36 @@ pub struct SftpEntry { } /// Manages SFTP sessions for multiple hosts. +/// +/// The session map is locked only to look an entry up, insert or remove it, +/// never across network I/O. A transfer or a listing therefore holds up +/// nothing but itself: other sessions keep working, and `close` and +/// `close_all` can always take an entry out of reach at once. pub struct SftpManager { - sessions: Mutex>, + sessions: Mutex>>, } struct SftpConnection { - #[allow(dead_code)] - handle: Arc>>, + handle: Handle, sftp: SftpSession, } +impl SftpConnection { + /// End the SFTP channel, which fails every request still in flight on it, + /// then the connection. Bounded, so an unresponsive peer cannot hold the + /// caller. + async fn close(&self) { + let _ = tokio::time::timeout(CLOSE_GRACE, async { + let _ = self.sftp.close().await; + let _ = self + .handle + .disconnect(russh::Disconnect::ByApplication, "", "en") + .await; + }) + .await; + } +} + impl SftpManager { pub fn new() -> Self { Self { @@ -88,9 +118,7 @@ impl SftpManager { .await .map_err(|e| CoreError::Ssh(format!("sftp init: {e}")))?; - let handle = Arc::new(Mutex::new(handle)); - - let conn = SftpConnection { handle, sftp }; + let conn = Arc::new(SftpConnection { handle, sftp }); // The check at the top ran before a connect that takes seconds, so a // second connect on the same id may have finished in the meantime. The // session already published keeps the id; this one is dropped rather @@ -106,18 +134,30 @@ impl SftpManager { } } + /// The live connection behind `session_id`, taken out from under the map + /// lock. + async fn get(&self, session_id: &str) -> Result> { + self.sessions + .lock() + .await + .get(session_id) + .cloned() + .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string())) + } + /// List directory entries at the given path. pub async fn list_dir(&self, session_id: &str, path: &str) -> Result> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; let mut entries = Vec::new(); - let mut dir = conn - .sftp - .read_dir(path) + let mut dir = tokio::time::timeout(LIST_DIR_DEADLINE, conn.sftp.read_dir(path)) .await + .map_err(|_| { + CoreError::Ssh(format!( + "read_dir: the listing took longer than {}s", + LIST_DIR_DEADLINE.as_secs() + )) + })? .map_err(|e| CoreError::Ssh(format!("read_dir: {e}")))?; // ReadDir is a plain Iterator (not async) @@ -155,10 +195,7 @@ impl SftpManager { /// Get the canonical absolute path (e.g., resolve "~"). pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .canonicalize(path) @@ -171,10 +208,7 @@ impl SftpManager { /// Kept for API compatibility. Interactive file transfers should use /// `download_to_local` so large files never cross Tauri IPC as JSON arrays. pub async fn read_file(&self, session_id: &str, path: &str) -> Result> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .read(path) @@ -187,10 +221,7 @@ impl SftpManager { /// Kept for API compatibility. Interactive file transfers should use /// `upload_from_local` so large files never cross Tauri IPC as JSON arrays. pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .write(path, data) @@ -208,10 +239,7 @@ impl SftpManager { local_path: &Path, overwrite: bool, ) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; let mut remote = conn .sftp @@ -234,10 +262,7 @@ impl SftpManager { remote_path: &str, overwrite: bool, ) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; if !overwrite && conn @@ -266,10 +291,7 @@ impl SftpManager { /// Create a directory. pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .create_dir(path) @@ -279,10 +301,7 @@ impl SftpManager { /// Remove a file. pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .remove_file(path) @@ -292,10 +311,7 @@ impl SftpManager { /// Remove a directory. pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .remove_dir(path) @@ -310,10 +326,7 @@ impl SftpManager { old_path: &str, new_path: &str, ) -> Result<()> { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; conn.sftp .rename(old_path, new_path) @@ -323,10 +336,7 @@ impl SftpManager { /// Get file metadata (size, permissions, modified time). pub async fn stat(&self, session_id: &str, path: &str) -> Result { - let mut sessions = self.sessions.lock().await; - let conn = sessions - .get_mut(session_id) - .ok_or_else(|| CoreError::SessionNotFound(session_id.to_string()))?; + let conn = self.get(session_id).await?; let attrs = conn .sftp @@ -352,11 +362,12 @@ impl SftpManager { }) } - /// Close an SFTP session. + /// Close an SFTP session. The entry leaves the map before anything is + /// sent, and closing the channel ends any transfer still running on it. pub async fn close(&self, session_id: &str) -> Result<()> { - let mut sessions = self.sessions.lock().await; - if let Some(conn) = sessions.remove(session_id) { - let _ = conn.sftp.close().await; + let removed = self.sessions.lock().await.remove(session_id); + if let Some(conn) = removed { + conn.close().await; } Ok(()) } @@ -370,15 +381,20 @@ impl SftpManager { /// /// The map is drained under the lock and the guard is released before any /// network work, so a channel stops being addressable the moment this is - /// called even if the remote never answers the close. + /// called even if the remote never answers the close. Transfers and + /// listings never hold the lock, so the drain does not wait on them. pub async fn close_all(&self) -> usize { - let drained: Vec = { + let drained: Vec> = { let mut sessions = self.sessions.lock().await; sessions.drain().map(|(_, conn)| conn).collect() }; let count = drained.len(); - for conn in &drained { - let _ = conn.sftp.close().await; + let goodbyes: Vec<_> = drained + .into_iter() + .map(|conn| tokio::spawn(async move { conn.close().await })) + .collect(); + for goodbye in goodbyes { + let _ = goodbye.await; } count } diff --git a/core/tests/session_isolation.rs b/core/tests/session_isolation.rs new file mode 100644 index 00000000..a0bf2e5b --- /dev/null +++ b/core/tests/session_isolation.rs @@ -0,0 +1,291 @@ +//! One misbehaving peer must hold up nothing but its own session. +//! +//! These run the SSH and SFTP managers against an in-process russh server on +//! loopback whose SFTP side never finishes a directory listing and whose shell +//! exits as soon as it starts, and check that other sessions, a close and the +//! drain a vault reset relies on all carry on regardless. + +use clavyn_core::host::Host; +use clavyn_core::known_hosts::KnownHosts; +use clavyn_core::session::SessionManager; +use clavyn_core::sftp::SftpManager; +use russh::keys::{Algorithm, PrivateKey}; +use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; +use russh::{Channel, ChannelId}; +use russh_sftp::protocol::{File, Handle, Name, Status, StatusCode, Version}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; +use tokio::net::TcpListener; +use tokio::sync::Mutex; + +const USER: &str = "deploy"; +const PASSWORD: &str = "correct password"; + +/// An SFTP server whose directories never end: every READDIR answers with one +/// more entry and never with end-of-file. +struct EndlessDirectory; + +impl russh_sftp::server::Handler for EndlessDirectory { + type Error = StatusCode; + + fn unimplemented(&self) -> Self::Error { + StatusCode::OpUnsupported + } + + async fn init( + &mut self, + _version: u32, + _extensions: HashMap, + ) -> Result { + Ok(Version::new()) + } + + async fn opendir(&mut self, id: u32, _path: String) -> Result { + Ok(Handle { + id, + handle: "dir".into(), + }) + } + + async fn readdir(&mut self, id: u32, _handle: String) -> Result { + tokio::time::sleep(Duration::from_millis(5)).await; + Ok(Name { + id, + files: vec![File::dummy("entry")], + }) + } + + async fn realpath(&mut self, id: u32, _path: String) -> Result { + Ok(Name { + id, + files: vec![File::dummy("/home/deploy")], + }) + } + + async fn close(&mut self, id: u32, _handle: String) -> Result { + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: "Ok".into(), + language_tag: "en-US".into(), + }) + } +} + +#[derive(Clone, Default)] +struct TestServer { + channels: Arc>>>, +} + +impl server::Handler for TestServer { + type Error = russh::Error; + + async fn auth_password(&mut self, user: &str, password: &str) -> Result { + Ok(if user == USER && password == PASSWORD { + Auth::Accept + } else { + Auth::reject() + }) + } + + async fn channel_open_session( + &mut self, + channel: Channel, + reply: ChannelOpenHandle, + _session: &mut Session, + ) -> Result<(), Self::Error> { + self.channels + .lock() + .expect("channels") + .insert(channel.id(), channel); + reply.accept().await; + Ok(()) + } + + /// The shell says one line and exits. + async fn shell_request( + &mut self, + channel: ChannelId, + session: &mut Session, + ) -> Result<(), Self::Error> { + session.data(channel, b"bye\r\n".to_vec())?; + session.eof(channel)?; + session.close(channel)?; + Ok(()) + } + + async fn subsystem_request( + &mut self, + channel_id: ChannelId, + name: &str, + session: &mut Session, + ) -> Result<(), Self::Error> { + let channel = self.channels.lock().expect("channels").remove(&channel_id); + match channel { + Some(channel) if name == "sftp" => { + session.channel_success(channel_id)?; + russh_sftp::server::run(channel.into_stream(), EndlessDirectory).await; + } + _ => session.channel_failure(channel_id)?, + } + Ok(()) + } +} + +async fn start_server() -> u16 { + let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); + let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key"); + let config = Arc::new(server::Config { + keys: vec![host_key], + auth_rejection_time: Duration::from_millis(10), + auth_rejection_time_initial: Some(Duration::ZERO), + ..Default::default() + }); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let port = listener.local_addr().expect("addr").port(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let config = config.clone(); + tokio::spawn(async move { + if let Ok(running) = + server::run_stream(config, stream, TestServer::default()).await + { + let _ = running.await; + } + }); + } + }); + port +} + +fn host(port: u16) -> Host { + serde_json::from_value(serde_json::json!({ + "id": "00000000-0000-0000-0000-000000000001", + "label": "Loopback", + "hostname": "127.0.0.1", + "port": port, + "username": USER, + "auth": {"password": {"credential_key": "unused"}}, + "tags": [] + })) + .expect("host") +} + +fn known_hosts(dir: &TempDir) -> Arc> { + Arc::new(Mutex::new( + KnownHosts::load(dir.path().join("known_hosts.json")).expect("load known hosts"), + )) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_listing_that_never_ends_holds_up_no_other_sftp_session() { + let dir = tempfile::tempdir().expect("temp dir"); + let port = start_server().await; + let host = host(port); + let pins = known_hosts(&dir); + let manager = Arc::new(SftpManager::new()); + for id in ["stuck", "other"] { + manager + .connect(id.into(), &host, None, pins.clone(), None, None, Some(PASSWORD)) + .await + .expect("sftp connect"); + } + + let listing = tokio::spawn({ + let manager = manager.clone(); + async move { manager.list_dir("stuck", "/").await } + }); + tokio::time::sleep(Duration::from_millis(300)).await; + assert!(!listing.is_finished(), "the endless listing ended on its own"); + + let home = tokio::time::timeout(Duration::from_secs(5), manager.canonicalize("other", ".")) + .await + .expect("another session waited on the listing") + .expect("canonicalize"); + assert_eq!(home, "/home/deploy"); + + // The drain a vault reset runs must take both sessions out of reach + // without waiting for the listing. + let closed = tokio::time::timeout(Duration::from_secs(5), manager.close_all()) + .await + .expect("close_all waited on the listing"); + assert_eq!(closed, 2); + assert!(manager.list().await.is_empty()); + + // Closing the channel ends the listing rather than leaving it running. + let result = tokio::time::timeout(Duration::from_secs(15), listing) + .await + .expect("the listing outlived its session") + .expect("listing task"); + assert!(result.is_err()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn closing_one_sftp_session_does_not_wait_on_its_listing() { + let dir = tempfile::tempdir().expect("temp dir"); + let port = start_server().await; + let manager = Arc::new(SftpManager::new()); + manager + .connect("stuck".into(), &host(port), None, known_hosts(&dir), None, None, Some(PASSWORD)) + .await + .expect("sftp connect"); + + let listing = tokio::spawn({ + let manager = manager.clone(); + async move { manager.list_dir("stuck", "/").await } + }); + tokio::time::sleep(Duration::from_millis(300)).await; + + tokio::time::timeout(Duration::from_secs(5), manager.close("stuck")) + .await + .expect("close waited on the listing") + .expect("close"); + assert!(manager.list().await.is_empty()); + let _ = tokio::time::timeout(Duration::from_secs(15), listing) + .await + .expect("the listing outlived its session"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_shell_that_exits_leaves_no_session_behind() { + let dir = tempfile::tempdir().expect("temp dir"); + let port = start_server().await; + let (closed_tx, mut closed_rx) = tokio::sync::mpsc::unbounded_channel::(); + let output = Arc::new(std::sync::Mutex::new(Vec::::new())); + let sessions = SessionManager::new( + { + let output = output.clone(); + Arc::new(move |_, data: &[u8]| output.lock().expect("output").extend_from_slice(data)) + }, + Arc::new(move |id: &str, _| { + let _ = closed_tx.send(id.to_string()); + }), + ); + + let claim = sessions.claim("shell").expect("claim"); + sessions + .create_ssh_session(&claim, &host(port), None, known_hosts(&dir), None, None, Some(PASSWORD), 80, 24) + .await + .expect("ssh session"); + drop(claim); + + let closed = tokio::time::timeout(Duration::from_secs(10), closed_rx.recv()) + .await + .expect("the close was never announced"); + assert_eq!(closed.as_deref(), Some("shell")); + assert_eq!(output.lock().expect("output").as_slice(), b"bye\r\n"); + + // The reader removes its entry after announcing the close; give it the + // moment that takes, then the id must be gone and free again. + let gone = async { + while sessions.contains("shell").await || sessions.is_claimed("shell") { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }; + tokio::time::timeout(Duration::from_secs(10), gone) + .await + .expect("a session whose shell exited stayed registered"); + assert!(sessions.write("shell", b"ls\n").await.is_err()); +} diff --git a/desktop/src-tauri/src/commands.rs b/desktop/src-tauri/src/commands.rs index 38c153bd..1d80e9b6 100644 --- a/desktop/src-tauri/src/commands.rs +++ b/desktop/src-tauri/src/commands.rs @@ -869,7 +869,7 @@ pub async fn session_write( data: Vec, ) -> ApiResult<()> { // Try SSH session first - if state.sessions.list().await.contains(&session_id) { + if state.sessions.contains(&session_id).await { return state.sessions.write(&session_id, &data).await.map_err(err); } // Try local terminal. The writer is taken out from under the map lock and @@ -928,7 +928,7 @@ pub async fn session_resize( let cols = pty_dimension("cols", cols)?; let rows = pty_dimension("rows", rows)?; // Try SSH session first - if state.sessions.list().await.contains(&session_id) { + if state.sessions.contains(&session_id).await { return state .sessions .resize(&session_id, cols.into(), rows.into()) @@ -959,7 +959,7 @@ pub async fn close_session(state: State<'_, Arc>, session_id: String) // stream. Releasing it here would drop the sink first and discard the tail, // which also happens on the reconnect path. // Try SSH session - if state.sessions.list().await.contains(&session_id) { + if state.sessions.contains(&session_id).await { return state.sessions.close(&session_id).await.map_err(err); } // Try local terminal From 216ac838a8cf60c1a3f33e563da571c9d8ea8764 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Fri, 16 Oct 2026 23:05:03 +0000 Subject: [PATCH 2/8] End a write, listing or transfer the moment its session is closed Taking an entry out of the map stopped new calls from reaching a session, but a call already holding the entry kept it alive: a keystroke send stuck behind a peer that stopped reading, or an SFTP listing or transfer still running, held the connection and the session id until it finished on its own. Each SSH session and SFTP connection now carries a closed signal that close, close_all and the reader's own cleanup flip. Writes, listings and transfers race against it and return an error once it fires. The goodbye sent after a close holds only the connection handle, so it no longer keeps the session id claimed while a stalled peer fails to answer. --- core/src/session.rs | 55 ++++++++++----- core/src/sftp.rs | 117 ++++++++++++++++++++------------ core/tests/session_isolation.rs | 96 +++++++++++++++++++++++--- 3 files changed, 202 insertions(+), 66 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index e7cfda27..9c776cda 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -116,7 +116,13 @@ fn lock_claims( } struct SshSession { - handle: Handle, + /// Shared with the goodbye sent after the session is closed, which needs + /// nothing else from it, so that goodbye does not keep the id claimed. + handle: Arc>, + /// Flipped once the session is closed. A write stuck behind a peer that + /// stopped reading watches it and gives up, so it cannot keep the session, + /// its connection and its id alive after the close. + closed: tokio::sync::watch::Sender, channel_id: russh::ChannelId, /// Serializes writes to this session so keystrokes reach the channel in /// the order they were sent. Per session, so a stalled peer only holds up @@ -128,6 +134,23 @@ struct SshSession { claim: Arc, } +impl SshSession { + /// Resolves once the session has been closed. + async fn closed(&self) { + let mut closed = self.closed.subscribe(); + let _ = closed.wait_for(|closed| *closed).await; + } + + /// Mark the session closed and say goodbye to the peer. The goodbye holds + /// only the connection handle, so the id is released as soon as the map + /// entry and the reader are gone, however long the peer takes. + fn retire(&self) -> impl std::future::Future + Send + 'static { + self.closed.send_replace(true); + let handle = self.handle.clone(); + async move { disconnect(&handle).await } + } +} + // Re-export the handler type so the session manager can use it. pub use connection::SshHandler; @@ -223,7 +246,8 @@ impl SessionManager { let held = claim.commit(); let session = Arc::new(SshSession { - handle, + handle: Arc::new(handle), + closed: tokio::sync::watch::Sender::new(false), channel_id, writes: Mutex::new(()), resize_tx, @@ -276,7 +300,7 @@ impl SessionManager { // The goodbye runs on its own so the id is free // for a reconnect as soon as this task ends. if let Some(session) = take_if_held(&sessions, &sid, &held).await { - tokio::spawn(async move { disconnect(&session).await }); + tokio::spawn(session.retire()); } break; } @@ -323,13 +347,14 @@ impl SessionManager { /// and only this session's writes wait behind it. pub async fn write(&self, session_id: &str, data: &[u8]) -> Result<()> { let session = self.get(session_id).await?; - let _order = session.writes.lock().await; - session - .handle - .data(session.channel_id, data.to_vec()) - .await - .map_err(|_| CoreError::Ssh("write failed".into()))?; - Ok(()) + let send = async { + let _order = session.writes.lock().await; + session.handle.data(session.channel_id, data.to_vec()).await + }; + tokio::select! { + sent = send => sent.map_err(|_| CoreError::Ssh("write failed".into())), + () = session.closed() => Err(CoreError::SessionNotFound(session_id.to_string())), + } } /// Resize a session's terminal. @@ -352,7 +377,7 @@ impl SessionManager { pub async fn close(&self, session_id: &str) -> Result<()> { let removed = self.sessions.lock().await.remove(session_id); if let Some(session) = removed { - disconnect(&session).await; + session.retire().await; } Ok(()) } @@ -377,7 +402,7 @@ impl SessionManager { let count = drained.len(); let goodbyes: Vec<_> = drained .into_iter() - .map(|session| tokio::spawn(async move { disconnect(&session).await })) + .map(|session| tokio::spawn(session.retire())) .collect(); for goodbye in goodbyes { let _ = goodbye.await; @@ -403,12 +428,10 @@ async fn take_if_held( /// Say goodbye to the peer, giving up after [`DISCONNECT_GRACE`]. The /// connection ends either way once the last reference to the handle is gone. -async fn disconnect(session: &SshSession) { +async fn disconnect(handle: &Handle) { let _ = tokio::time::timeout( DISCONNECT_GRACE, - session - .handle - .disconnect(russh::Disconnect::ByApplication, "", "en"), + handle.disconnect(russh::Disconnect::ByApplication, "", "en"), ) .await; } diff --git a/core/src/sftp.rs b/core/src/sftp.rs index 274eaf3e..228ce79c 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -48,13 +48,32 @@ pub struct SftpManager { struct SftpConnection { handle: Handle, sftp: SftpSession, + /// Flipped once the connection is closed. Listings and transfers watch it + /// and stop, so one that is stuck behind a peer that stopped reading, or + /// simply still running, cannot outlive the close. + closed: tokio::sync::watch::Sender, } impl SftpConnection { - /// End the SFTP channel, which fails every request still in flight on it, - /// then the connection. Bounded, so an unresponsive peer cannot hold the - /// caller. + /// Run `operation` until it finishes or the connection is closed. + async fn until_closed( + &self, + operation: impl std::future::Future>, + ) -> Result { + let mut closed = self.closed.subscribe(); + tokio::select! { + result = operation => result, + _ = closed.wait_for(|closed| *closed) => { + Err(CoreError::Ssh("the SFTP session was closed".into())) + } + } + } + + /// Stop every listing and transfer on this connection, end the SFTP + /// channel, then the connection. Bounded, so an unresponsive peer cannot + /// hold the caller. async fn close(&self) { + self.closed.send_replace(true); let _ = tokio::time::timeout(CLOSE_GRACE, async { let _ = self.sftp.close().await; let _ = self @@ -118,7 +137,11 @@ impl SftpManager { .await .map_err(|e| CoreError::Ssh(format!("sftp init: {e}")))?; - let conn = Arc::new(SftpConnection { handle, sftp }); + let conn = Arc::new(SftpConnection { + handle, + sftp, + closed: tokio::sync::watch::Sender::new(false), + }); // The check at the top ran before a connect that takes seconds, so a // second connect on the same id may have finished in the meantime. The // session already published keeps the id; this one is dropped rather @@ -150,15 +173,19 @@ impl SftpManager { let conn = self.get(session_id).await?; let mut entries = Vec::new(); - let mut dir = tokio::time::timeout(LIST_DIR_DEADLINE, conn.sftp.read_dir(path)) - .await - .map_err(|_| { - CoreError::Ssh(format!( - "read_dir: the listing took longer than {}s", - LIST_DIR_DEADLINE.as_secs() - )) - })? - .map_err(|e| CoreError::Ssh(format!("read_dir: {e}")))?; + let mut dir = conn + .until_closed(async { + tokio::time::timeout(LIST_DIR_DEADLINE, conn.sftp.read_dir(path)) + .await + .map_err(|_| { + CoreError::Ssh(format!( + "read_dir: the listing took longer than {}s", + LIST_DIR_DEADLINE.as_secs() + )) + })? + .map_err(|e| CoreError::Ssh(format!("read_dir: {e}"))) + }) + .await?; // ReadDir is a plain Iterator (not async) for entry in dir.by_ref() { @@ -241,16 +268,19 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - let mut remote = conn - .sftp - .open(remote_path) - .await - .map_err(|e| CoreError::Ssh(format!("open remote file: {e}")))?; - let mut local = open_download_target(local_path, overwrite).await?; + conn.until_closed(async { + let mut remote = conn + .sftp + .open(remote_path) + .await + .map_err(|e| CoreError::Ssh(format!("open remote file: {e}")))?; + let mut local = open_download_target(local_path, overwrite).await?; - tokio::io::copy(&mut remote, &mut local).await?; - local.flush().await?; - Ok(()) + tokio::io::copy(&mut remote, &mut local).await?; + local.flush().await?; + Ok(()) + }) + .await } /// Stream a local file directly to the remote host. Existing remote files @@ -264,29 +294,32 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - if !overwrite - && conn + conn.until_closed(async { + if !overwrite + && conn + .sftp + .try_exists(remote_path) + .await + .map_err(|e| CoreError::Ssh(format!("check remote file: {e}")))? + { + return Err(CoreError::InvalidInput(format!( + "remote file already exists: {remote_path}" + ))); + } + + let mut local = tokio::fs::File::open(local_path).await?; + let mut remote = conn .sftp - .try_exists(remote_path) + .create(remote_path) .await - .map_err(|e| CoreError::Ssh(format!("check remote file: {e}")))? - { - return Err(CoreError::InvalidInput(format!( - "remote file already exists: {remote_path}" - ))); - } + .map_err(|e| CoreError::Ssh(format!("create remote file: {e}")))?; - let mut local = tokio::fs::File::open(local_path).await?; - let mut remote = conn - .sftp - .create(remote_path) - .await - .map_err(|e| CoreError::Ssh(format!("create remote file: {e}")))?; - - tokio::io::copy(&mut local, &mut remote).await?; - remote.flush().await?; - remote.shutdown().await?; - Ok(()) + tokio::io::copy(&mut local, &mut remote).await?; + remote.flush().await?; + remote.shutdown().await?; + Ok(()) + }) + .await } /// Create a directory. diff --git a/core/tests/session_isolation.rs b/core/tests/session_isolation.rs index a0bf2e5b..7362bc3d 100644 --- a/core/tests/session_isolation.rs +++ b/core/tests/session_isolation.rs @@ -77,6 +77,8 @@ impl russh_sftp::server::Handler for EndlessDirectory { #[derive(Clone, Default)] struct TestServer { channels: Arc>>>, + /// Stop reading once the first input arrives, the way a hung peer does. + stall_on_input: bool, } impl server::Handler for TestServer { @@ -104,18 +106,34 @@ impl server::Handler for TestServer { Ok(()) } - /// The shell says one line and exits. + /// The shell says one line and exits, unless this server stalls on input, + /// in which case it stays open. async fn shell_request( &mut self, channel: ChannelId, session: &mut Session, ) -> Result<(), Self::Error> { + if self.stall_on_input { + return Ok(()); + } session.data(channel, b"bye\r\n".to_vec())?; session.eof(channel)?; session.close(channel)?; Ok(()) } + async fn data( + &mut self, + _channel: ChannelId, + _data: &[u8], + _session: &mut Session, + ) -> Result<(), Self::Error> { + if self.stall_on_input { + std::future::pending::<()>().await; + } + Ok(()) + } + async fn subsystem_request( &mut self, channel_id: ChannelId, @@ -135,6 +153,10 @@ impl server::Handler for TestServer { } async fn start_server() -> u16 { + serve(TestServer::default()).await +} + +async fn serve(handler: TestServer) -> u16 { let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key"); let config = Arc::new(server::Config { @@ -148,10 +170,9 @@ async fn start_server() -> u16 { tokio::spawn(async move { while let Ok((stream, _)) = listener.accept().await { let config = config.clone(); + let handler = handler.clone(); tokio::spawn(async move { - if let Ok(running) = - server::run_stream(config, stream, TestServer::default()).await - { + if let Ok(running) = server::run_stream(config, stream, handler.clone()).await { let _ = running.await; } }); @@ -214,8 +235,8 @@ async fn a_listing_that_never_ends_holds_up_no_other_sftp_session() { assert_eq!(closed, 2); assert!(manager.list().await.is_empty()); - // Closing the channel ends the listing rather than leaving it running. - let result = tokio::time::timeout(Duration::from_secs(15), listing) + // Closing the session ends the listing rather than leaving it running. + let result = tokio::time::timeout(Duration::from_secs(2), listing) .await .expect("the listing outlived its session") .expect("listing task"); @@ -243,9 +264,11 @@ async fn closing_one_sftp_session_does_not_wait_on_its_listing() { .expect("close waited on the listing") .expect("close"); assert!(manager.list().await.is_empty()); - let _ = tokio::time::timeout(Duration::from_secs(15), listing) + let result = tokio::time::timeout(Duration::from_secs(2), listing) .await - .expect("the listing outlived its session"); + .expect("the listing outlived its session") + .expect("listing task"); + assert!(result.is_err()); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -289,3 +312,60 @@ async fn a_shell_that_exits_leaves_no_session_behind() { .expect("a session whose shell exited stayed registered"); assert!(sessions.write("shell", b"ls\n").await.is_err()); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_write_stuck_behind_a_peer_that_stopped_reading_ends_with_the_close() { + let dir = tempfile::tempdir().expect("temp dir"); + let port = serve(TestServer { + stall_on_input: true, + ..TestServer::default() + }) + .await; + let sessions = Arc::new(SessionManager::new(Arc::new(|_, _| {}), Arc::new(|_, _| {}))); + let claim = sessions.claim("stuck").expect("claim"); + sessions + .create_ssh_session(&claim, &host(port), None, known_hosts(&dir), None, None, Some(PASSWORD), 80, 24) + .await + .expect("ssh session"); + drop(claim); + + // Keep writing until the peer's window and the socket buffers are full + // and a write stops returning. + let written = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let writer = tokio::spawn({ + let sessions = sessions.clone(); + let written = written.clone(); + async move { + loop { + sessions.write("stuck", &[b'x'; 64 * 1024]).await?; + written.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + #[allow(unreachable_code)] + Ok::<(), clavyn_core::CoreError>(()) + } + }); + let stuck = async { + loop { + let before = written.load(std::sync::atomic::Ordering::Relaxed); + tokio::time::sleep(Duration::from_millis(500)).await; + if before > 0 && written.load(std::sync::atomic::Ordering::Relaxed) == before { + return; + } + } + }; + tokio::time::timeout(Duration::from_secs(20), stuck) + .await + .expect("writes never stalled"); + assert!(!writer.is_finished()); + + tokio::time::timeout(Duration::from_secs(10), sessions.close("stuck")) + .await + .expect("close waited on the stuck write") + .expect("close"); + let result = tokio::time::timeout(Duration::from_secs(2), writer) + .await + .expect("the stuck write outlived the close") + .expect("writer task"); + assert!(result.is_err()); + assert!(!sessions.contains("stuck").await); +} From f72acb2de6410230d90b3abaeecb9b2bf9e9d224 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 06:38:07 +0000 Subject: [PATCH 3/8] Stop single SFTP requests at the close too Listings and transfers already stopped when their connection closed, but mkdir, remove, rename, canonicalize and stat did not: a request issued just before a close or a vault reset could still complete and report success after teardown began. Every SftpManager operation now runs under the connection's closed signal. --- core/src/sftp.rs | 99 ++++++++++++++++++++++++++++++------------------ 1 file changed, 62 insertions(+), 37 deletions(-) diff --git a/core/src/sftp.rs b/core/src/sftp.rs index 228ce79c..bda7c62b 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -48,9 +48,10 @@ pub struct SftpManager { struct SftpConnection { handle: Handle, sftp: SftpSession, - /// Flipped once the connection is closed. Listings and transfers watch it - /// and stop, so one that is stuck behind a peer that stopped reading, or - /// simply still running, cannot outlive the close. + /// Flipped once the connection is closed. Every operation watches it and + /// stops, so one that is stuck behind a peer that stopped reading, or + /// simply still running, cannot outlive the close or report success + /// after it. closed: tokio::sync::watch::Sender, } @@ -69,8 +70,8 @@ impl SftpConnection { } } - /// Stop every listing and transfer on this connection, end the SFTP - /// channel, then the connection. Bounded, so an unresponsive peer cannot + /// Stop every request, listing and transfer on this connection, end the + /// SFTP channel, then the connection. Bounded, so an unresponsive peer cannot /// hold the caller. async fn close(&self) { self.closed.send_replace(true); @@ -224,10 +225,13 @@ impl SftpManager { pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result { let conn = self.get(session_id).await?; - conn.sftp - .canonicalize(path) - .await - .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) + conn.until_closed(async { + conn.sftp + .canonicalize(path) + .await + .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) + }) + .await } /// Read a file's contents into a byte vector. @@ -237,10 +241,13 @@ impl SftpManager { pub async fn read_file(&self, session_id: &str, path: &str) -> Result> { let conn = self.get(session_id).await?; - conn.sftp - .read(path) - .await - .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) + conn.until_closed(async { + conn.sftp + .read(path) + .await + .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) + }) + .await } /// Write data to a remote file (creates or overwrites). @@ -250,10 +257,13 @@ impl SftpManager { pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .write(path, data) - .await - .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) + conn.until_closed(async { + conn.sftp + .write(path, data) + .await + .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) + }) + .await } /// Stream a remote file directly to a local path. An existing local file is @@ -326,30 +336,39 @@ impl SftpManager { pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .create_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) + conn.until_closed(async { + conn.sftp + .create_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) + }) + .await } /// Remove a file. pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .remove_file(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) + conn.until_closed(async { + conn.sftp + .remove_file(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) + }) + .await } /// Remove a directory. pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .remove_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) + conn.until_closed(async { + conn.sftp + .remove_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) + }) + .await } /// Rename a file or directory. @@ -361,10 +380,13 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .rename(old_path, new_path) - .await - .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) + conn.until_closed(async { + conn.sftp + .rename(old_path, new_path) + .await + .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) + }) + .await } /// Get file metadata (size, permissions, modified time). @@ -372,10 +394,13 @@ impl SftpManager { let conn = self.get(session_id).await?; let attrs = conn - .sftp - .metadata(path) - .await - .map_err(|e| CoreError::Ssh(format!("stat: {e}")))?; + .until_closed(async { + conn.sftp + .metadata(path) + .await + .map_err(|e| CoreError::Ssh(format!("stat: {e}"))) + }) + .await?; let name = path.rsplit('/').next().unwrap_or(path).to_string(); let modified = attrs From 4cfacfe1c6bf63acad350ce996ea0cfc950071dc Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Fri, 16 Oct 2026 22:39:40 +0000 Subject: [PATCH 4/8] Leave the original file alone when a download fails, and widen upload writes An overwrite download truncated the user's file before the first byte arrived, so a dropped connection or a server timeout part way through left a cut-off file where their original had been. Overwrites now go to a temporary file beside the target that is renamed over it once it is complete and synced; a new file is removed again if the transfer fails. Uploads read through a 256 KiB buffer. russh-sftp pipelines up to 16 writes, each the size of the slice it is given, and tokio::io::copy's 8 KiB reads capped that at 128 KiB in flight per round trip. --- core/src/sftp.rs | 212 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 201 insertions(+), 11 deletions(-) diff --git a/core/src/sftp.rs b/core/src/sftp.rs index bda7c62b..bb8e1e49 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -7,10 +7,10 @@ use crate::{CoreError, Result}; use russh::client::Handle; use russh_sftp::client::SftpSession; use std::collections::HashMap; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncRead, AsyncWriteExt}; use tokio::sync::Mutex; /// The longest a directory listing may take in total. russh-sftp reads a @@ -22,6 +22,12 @@ const LIST_DIR_DEADLINE: Duration = Duration::from_secs(30); /// How long a polite goodbye may take before the connection is dropped. const CLOSE_GRACE: Duration = Duration::from_secs(5); +/// Read buffer for uploads. russh-sftp keeps up to 16 writes in flight, each +/// as large as the slice it is handed, so the read size sets how much data is +/// in the air per round trip: `tokio::io::copy`'s 8 KiB would cap it at +/// 128 KiB, far below what a long link can carry. +const UPLOAD_BUFFER: usize = 256 * 1024; + /// A directory entry returned by SFTP list operations. #[derive(Debug, Clone, serde::Serialize)] pub struct SftpEntry { @@ -284,11 +290,7 @@ impl SftpManager { .open(remote_path) .await .map_err(|e| CoreError::Ssh(format!("open remote file: {e}")))?; - let mut local = open_download_target(local_path, overwrite).await?; - - tokio::io::copy(&mut remote, &mut local).await?; - local.flush().await?; - Ok(()) + write_download(&mut remote, local_path, overwrite).await }) .await } @@ -317,14 +319,17 @@ impl SftpManager { ))); } - let mut local = tokio::fs::File::open(local_path).await?; + let mut local = tokio::io::BufReader::with_capacity( + UPLOAD_BUFFER, + tokio::fs::File::open(local_path).await?, + ); let mut remote = conn .sftp .create(remote_path) .await .map_err(|e| CoreError::Ssh(format!("create remote file: {e}")))?; - tokio::io::copy(&mut local, &mut remote).await?; + tokio::io::copy_buf(&mut local, &mut remote).await?; remote.flush().await?; remote.shutdown().await?; Ok(()) @@ -464,6 +469,67 @@ impl Default for SftpManager { } } +/// Write a download to `target` without ever leaving a partial file behind. +/// +/// Overwriting goes through a temporary file beside the target that is +/// renamed over it only once every byte is on disk, so a dropped connection +/// or a server timeout leaves the user's existing file as it was. A new file +/// is written in place and removed again if the transfer fails. Either way a +/// failed download leaves nothing that looks like a finished one, and that +/// holds when the transfer is cancelled too, as closing its session does. +async fn write_download( + source: &mut R, + target: &Path, + overwrite: bool, +) -> Result<()> { + let staging = if overwrite { + let name = target + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + Some(target.with_file_name(format!(".{name}.{}.part", uuid::Uuid::new_v4()))) + } else { + None + }; + let written_to = staging.as_deref().unwrap_or(target); + // The staging name is fresh, so `create_new` holds for it as well. + let file = open_download_target(written_to, false).await?; + let mut partial = PartialDownload { + path: written_to.to_path_buf(), + file: Some(file), + }; + let file = partial.file.as_mut().expect("just opened"); + tokio::io::copy(source, file).await?; + file.flush().await?; + file.sync_all().await?; + partial.file = None; + if let Some(staging) = &staging { + tokio::fs::rename(staging, target).await?; + } + partial.path = PathBuf::new(); + Ok(()) +} + +/// A download that has not finished. Dropped before it is complete, whether +/// by an error or because the transfer future itself was dropped, it removes +/// the file written so far. +struct PartialDownload { + /// Emptied once the download is in place. + path: PathBuf, + file: Option, +} + +impl Drop for PartialDownload { + fn drop(&mut self) { + // The handle goes first: Windows refuses to remove a file that is + // still open without delete sharing. + self.file = None; + if !self.path.as_os_str().is_empty() { + let _ = std::fs::remove_file(&self.path); + } + } +} + /// Open the local file a download writes into. Without `overwrite` an existing /// file is refused; `create_new` checks and creates in one step, so a file that /// appears between a check and the open is refused too. @@ -486,8 +552,132 @@ async fn open_download_target(path: &Path, overwrite: bool) -> Result, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.0.is_empty() { + return Poll::Ready(Err(std::io::ErrorKind::ConnectionReset.into())); + } + let n = self.0.len().min(buf.remaining()); + buf.put_slice(&self.0[..n]); + self.0 = &self.0[n..]; + Poll::Ready(Ok(())) + } + } + + fn entries(dir: &std::path::Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .expect("read dir") + .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names + } + + #[tokio::test] + async fn an_overwrite_replaces_the_file_only_once_complete() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + + write_download(&mut &b"replacement"[..], &path, true) + .await + .expect("download"); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "replacement"); + assert_eq!(entries(dir.path()), ["report.txt"]); + } + + #[tokio::test] + async fn a_failed_overwrite_keeps_the_original_and_no_partial_file() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + + assert!(write_download(&mut BrokenAfter(b"half of the"), &path, true) + .await + .is_err()); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); + assert_eq!(entries(dir.path()), ["report.txt"]); + } + + #[tokio::test] + async fn a_failed_new_download_leaves_nothing_behind() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + + assert!(write_download(&mut BrokenAfter(b"half of the"), &path, false) + .await + .is_err()); + assert!(entries(dir.path()).is_empty()); + } + + /// Yields `data`, then never answers again, like a peer that went quiet. + struct StallsAfter(&'static [u8]); + + impl AsyncRead for StallsAfter { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.0.is_empty() { + return Poll::Pending; + } + let n = self.0.len().min(buf.remaining()); + buf.put_slice(&self.0[..n]); + self.0 = &self.0[n..]; + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn a_cancelled_download_leaves_the_original_and_nothing_else() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + + let mut source = StallsAfter(b"half of the"); + let cancelled = tokio::time::timeout( + std::time::Duration::from_millis(200), + write_download(&mut source, &path, true), + ) + .await; + assert!(cancelled.is_err(), "the stalled download finished"); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); + assert_eq!(entries(dir.path()), ["report.txt"]); + + let fresh = dir.path().join("new.txt"); + let mut source = StallsAfter(b"half of the"); + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + write_download(&mut source, &fresh, false), + ) + .await; + assert_eq!(entries(dir.path()), ["report.txt"]); + } + + #[tokio::test] + async fn a_download_without_overwrite_keeps_an_existing_file() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + + assert!(write_download(&mut &b"replacement"[..], &path, false) + .await + .is_err()); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); + } #[tokio::test] async fn an_existing_file_is_kept_unless_overwrite_is_chosen() { From c96ad56e12d6947c9daefd0299e12977b302f9df Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 06:43:18 +0000 Subject: [PATCH 5/8] Make a replacing download keep permissions, follow links and stop cleanly Closing a session dropped the download future, which could interrupt the local open or the final rename halfway and leave a file behind. The close now stops only the copy, and the download cleans up after itself with ordinary awaits, logging a file it could not remove. The staging file is created owner-only and takes the replaced file's permissions before the rename, so a 0600 file is not left world-readable under a 022 umask. A symbolic link is resolved first, so the file it names is replaced and the link survives. Staging names keep only the start of the target's name, so a name at the length limit still fits, and a download to a name that does not exist yet writes it directly. --- core/src/sftp.rs | 241 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 174 insertions(+), 67 deletions(-) diff --git a/core/src/sftp.rs b/core/src/sftp.rs index bb8e1e49..d296bbc6 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -284,15 +284,21 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - let mut remote = conn - .sftp - .open(remote_path) - .await - .map_err(|e| CoreError::Ssh(format!("open remote file: {e}")))?; - write_download(&mut remote, local_path, overwrite).await - }) - .await + let mut remote = conn + .until_closed(async { + conn.sftp + .open(remote_path) + .await + .map_err(|e| CoreError::Ssh(format!("open remote file: {e}"))) + }) + .await?; + // The close stops the copy rather than dropping the whole download, + // so the local file is always cleaned up by the download itself. + let mut closed = conn.closed.subscribe(); + let closed = async move { + let _ = closed.wait_for(|closed| *closed).await; + }; + write_download(&mut remote, local_path, overwrite, closed).await } /// Stream a local file directly to the remote host. Existing remote files @@ -471,61 +477,113 @@ impl Default for SftpManager { /// Write a download to `target` without ever leaving a partial file behind. /// -/// Overwriting goes through a temporary file beside the target that is -/// renamed over it only once every byte is on disk, so a dropped connection -/// or a server timeout leaves the user's existing file as it was. A new file -/// is written in place and removed again if the transfer fails. Either way a -/// failed download leaves nothing that looks like a finished one, and that -/// holds when the transfer is cancelled too, as closing its session does. +/// Replacing an existing file goes through a staging file beside it that is +/// renamed over it only once every byte is on disk, so a dropped connection, +/// a server timeout or a closed session leaves the user's file as it was. A +/// new file is written in place and removed again if the transfer fails. +/// Either way a failed download leaves nothing that looks like a finished +/// one. `closed` resolving stops the copy; the cleanup still runs. +/// +/// A symbolic link is followed, so the file it names is replaced and the +/// link stays a link, as writing through it would have done. async fn write_download( source: &mut R, target: &Path, overwrite: bool, + closed: impl std::future::Future, ) -> Result<()> { - let staging = if overwrite { - let name = target - .file_name() - .map(|n| n.to_string_lossy().into_owned()) - .unwrap_or_default(); - Some(target.with_file_name(format!(".{name}.{}.part", uuid::Uuid::new_v4()))) + let target = follow_link(target).await; + let existing = if overwrite { + tokio::fs::metadata(&target).await.ok() } else { None }; - let written_to = staging.as_deref().unwrap_or(target); - // The staging name is fresh, so `create_new` holds for it as well. - let file = open_download_target(written_to, false).await?; - let mut partial = PartialDownload { - path: written_to.to_path_buf(), - file: Some(file), + let Some(existing) = existing else { + let mut file = open_download_target(&target, false).await?; + let written = copy_until(source, &mut file, closed).await; + drop(file); + if written.is_err() { + discard(&target).await; + } + return written.map_err(Into::into); + }; + + let staging = staging_path(&target); + let mut file = open_staging(&staging).await?; + let written = async { + copy_until(source, &mut file, closed).await?; + // The staging file is created owner-only so a private file being + // replaced is never readable by anyone else mid-download; the + // replacement then takes the original's permissions. + #[cfg(unix)] + tokio::fs::set_permissions(&staging, existing.permissions()).await?; + #[cfg(not(unix))] + let _ = &existing; + tokio::fs::rename(&staging, &target).await + } + .await; + drop(file); + if written.is_err() { + discard(&staging).await; + } + written.map_err(Into::into) +} + +/// Copy `source` into `file` and make it durable, unless `closed` resolves +/// first. +async fn copy_until( + source: &mut R, + file: &mut tokio::fs::File, + closed: impl std::future::Future, +) -> std::io::Result<()> { + tokio::select! { + copied = tokio::io::copy(source, file) => copied?, + () = closed => { + return Err(std::io::Error::new( + std::io::ErrorKind::Interrupted, + "the SFTP session was closed", + )); + } }; - let file = partial.file.as_mut().expect("just opened"); - tokio::io::copy(source, file).await?; file.flush().await?; - file.sync_all().await?; - partial.file = None; - if let Some(staging) = &staging { - tokio::fs::rename(staging, target).await?; + file.sync_all().await +} + +/// The file a symbolic link at `path` names, or `path` itself. +async fn follow_link(path: &Path) -> PathBuf { + match tokio::fs::symlink_metadata(path).await { + Ok(meta) if meta.file_type().is_symlink() => tokio::fs::canonicalize(path) + .await + .unwrap_or_else(|_| path.to_path_buf()), + _ => path.to_path_buf(), } - partial.path = PathBuf::new(); - Ok(()) } -/// A download that has not finished. Dropped before it is complete, whether -/// by an error or because the transfer future itself was dropped, it removes -/// the file written so far. -struct PartialDownload { - /// Emptied once the download is in place. - path: PathBuf, - file: Option, +/// A fresh hidden name beside `target`. Only the start of the target's name +/// is kept, so a name close to the file system's length limit still leaves +/// room for the suffix. +fn staging_path(target: &Path) -> PathBuf { + let name: String = target + .file_name() + .map(|n| n.to_string_lossy().chars().take(32).collect()) + .unwrap_or_default(); + target.with_file_name(format!(".{name}.{}.part", uuid::Uuid::new_v4())) +} + +async fn open_staging(path: &Path) -> Result { + let mut options = tokio::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600); + Ok(options.open(path).await?) } -impl Drop for PartialDownload { - fn drop(&mut self) { - // The handle goes first: Windows refuses to remove a file that is - // still open without delete sharing. - self.file = None; - if !self.path.as_os_str().is_empty() { - let _ = std::fs::remove_file(&self.path); +/// Remove what a failed download wrote. A file that cannot be removed is +/// logged, since the error the caller sees is the download's own. +async fn discard(path: &Path) { + if let Err(error) = tokio::fs::remove_file(path).await { + if error.kind() != std::io::ErrorKind::NotFound { + tracing::warn!("could not remove an incomplete download at {}: {error}", path.display()); } } } @@ -591,7 +649,7 @@ mod download_target_tests { let path = dir.path().join("report.txt"); std::fs::write(&path, "original").expect("write"); - write_download(&mut &b"replacement"[..], &path, true) + write_download(&mut &b"replacement"[..], &path, true, never()) .await .expect("download"); assert_eq!(std::fs::read_to_string(&path).expect("read"), "replacement"); @@ -604,7 +662,7 @@ mod download_target_tests { let path = dir.path().join("report.txt"); std::fs::write(&path, "original").expect("write"); - assert!(write_download(&mut BrokenAfter(b"half of the"), &path, true) + assert!(write_download(&mut BrokenAfter(b"half of the"), &path, true, never()) .await .is_err()); assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); @@ -616,7 +674,7 @@ mod download_target_tests { let dir = tempfile::tempdir().expect("temp dir"); let path = dir.path().join("report.txt"); - assert!(write_download(&mut BrokenAfter(b"half of the"), &path, false) + assert!(write_download(&mut BrokenAfter(b"half of the"), &path, false, never()) .await .is_err()); assert!(entries(dir.path()).is_empty()); @@ -641,39 +699,88 @@ mod download_target_tests { } } + fn never() -> std::future::Pending<()> { + std::future::pending() + } + + fn soon() -> tokio::time::Sleep { + tokio::time::sleep(std::time::Duration::from_millis(200)) + } + #[tokio::test] - async fn a_cancelled_download_leaves_the_original_and_nothing_else() { + async fn a_download_stopped_by_a_close_leaves_the_original_and_nothing_else() { let dir = tempfile::tempdir().expect("temp dir"); let path = dir.path().join("report.txt"); std::fs::write(&path, "original").expect("write"); - let mut source = StallsAfter(b"half of the"); - let cancelled = tokio::time::timeout( - std::time::Duration::from_millis(200), - write_download(&mut source, &path, true), - ) - .await; - assert!(cancelled.is_err(), "the stalled download finished"); + let stopped = write_download(&mut StallsAfter(b"half of the"), &path, true, soon()).await; + assert!(stopped.is_err(), "the stalled download finished"); assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); assert_eq!(entries(dir.path()), ["report.txt"]); let fresh = dir.path().join("new.txt"); - let mut source = StallsAfter(b"half of the"); - let _ = tokio::time::timeout( - std::time::Duration::from_millis(200), - write_download(&mut source, &fresh, false), - ) - .await; + assert!(write_download(&mut StallsAfter(b"half of the"), &fresh, true, soon()) + .await + .is_err()); assert_eq!(entries(dir.path()), ["report.txt"]); } + /// The replacement takes the original's permissions, so a private file + /// stays private. + #[cfg(unix)] + #[tokio::test] + async fn an_overwrite_keeps_the_original_permissions() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("id_backup"); + std::fs::write(&path, "original").expect("write"); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).expect("chmod"); + + write_download(&mut &b"replacement"[..], &path, true, never()) + .await + .expect("download"); + let mode = std::fs::metadata(&path).expect("stat").permissions().mode() & 0o777; + assert_eq!(mode, 0o600); + } + + /// Overwriting through a symbolic link replaces the file it names and + /// leaves the link in place. + #[cfg(unix)] + #[tokio::test] + async fn an_overwrite_through_a_link_replaces_the_file_it_names() { + let dir = tempfile::tempdir().expect("temp dir"); + let real = dir.path().join("real.txt"); + let link = dir.path().join("link.txt"); + std::fs::write(&real, "original").expect("write"); + std::os::unix::fs::symlink(&real, &link).expect("symlink"); + + write_download(&mut &b"replacement"[..], &link, true, never()) + .await + .expect("download"); + assert!(std::fs::symlink_metadata(&link).expect("lstat").file_type().is_symlink()); + assert_eq!(std::fs::read_to_string(&real).expect("read"), "replacement"); + assert_eq!(entries(dir.path()), ["link.txt", "real.txt"]); + } + + #[tokio::test] + async fn an_overwrite_works_for_a_name_at_the_length_limit() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join(format!("{}.txt", "n".repeat(251))); + std::fs::write(&path, "original").expect("write"); + + write_download(&mut &b"replacement"[..], &path, true, never()) + .await + .expect("download"); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "replacement"); + } + #[tokio::test] async fn a_download_without_overwrite_keeps_an_existing_file() { let dir = tempfile::tempdir().expect("temp dir"); let path = dir.path().join("report.txt"); std::fs::write(&path, "original").expect("write"); - assert!(write_download(&mut &b"replacement"[..], &path, false) + assert!(write_download(&mut &b"replacement"[..], &path, false, never()) .await .is_err()); assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); From 68bd6725073ff173981f50aa2cc021055d736bfc Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 08:31:58 +0000 Subject: [PATCH 6/8] Remove a failed new download only while its path still names that file A second download that chose to overwrite could rename its finished file over the path before the first one failed; the first then deleted the second's result by name. --- core/src/sftp.rs | 64 ++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 62 insertions(+), 2 deletions(-) diff --git a/core/src/sftp.rs b/core/src/sftp.rs index d296bbc6..e94812d8 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -501,10 +501,10 @@ async fn write_download( let Some(existing) = existing else { let mut file = open_download_target(&target, false).await?; let written = copy_until(source, &mut file, closed).await; - drop(file); if written.is_err() { - discard(&target).await; + discard_if_still(&target, &file).await; } + drop(file); return written.map_err(Into::into); }; @@ -588,6 +588,32 @@ async fn discard(path: &Path) { } } +/// Remove an incomplete download at `path`, but only while `path` still names +/// the file this download created. Another download that chose to overwrite +/// can have renamed its finished file there in the meantime, and that one +/// must survive this failure. +/// +/// Only Unix exposes a file's identity on stable Rust; elsewhere the path is +/// removed as before. +async fn discard_if_still(path: &Path, file: &tokio::fs::File) { + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + let (Ok(ours), Ok(there)) = ( + file.metadata().await, + tokio::fs::symlink_metadata(path).await, + ) else { + return; + }; + if (ours.dev(), ours.ino()) != (there.dev(), there.ino()) { + return; + } + } + #[cfg(not(unix))] + let _ = file; + discard(path).await; +} + /// Open the local file a download writes into. Without `overwrite` an existing /// file is refused; `create_new` checks and creates in one step, so a file that /// appears between a check and the open is refused too. @@ -669,6 +695,40 @@ mod download_target_tests { assert_eq!(entries(dir.path()), ["report.txt"]); } + /// Fails after another download has renamed its finished file over the + /// path this one was writing. + struct ReplacedThenBroken(std::path::PathBuf); + + impl AsyncRead for ReplacedThenBroken { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &mut ReadBuf<'_>, + ) -> Poll> { + let other = self.0.with_extension("other"); + std::fs::write(&other, "finished by another download").expect("write"); + std::fs::rename(&other, &self.0).expect("rename"); + Poll::Ready(Err(std::io::ErrorKind::ConnectionReset.into())) + } + } + + #[cfg(unix)] + #[tokio::test] + async fn a_failed_new_download_spares_a_file_that_replaced_it() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + + assert!( + write_download(&mut ReplacedThenBroken(path.clone()), &path, false, never()) + .await + .is_err() + ); + assert_eq!( + std::fs::read_to_string(&path).expect("the other download's file"), + "finished by another download" + ); + } + #[tokio::test] async fn a_failed_new_download_leaves_nothing_behind() { let dir = tempfile::tempdir().expect("temp dir"); From 9fb163a61519c5e1ee72b16d2d74649d50e2b80c Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 08:51:32 +0000 Subject: [PATCH 7/8] Close a failed download's handle before removing its file Removal no longer depends on the handle having been opened with delete sharing on Windows. --- core/src/sftp.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/core/src/sftp.rs b/core/src/sftp.rs index e94812d8..92c6b379 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -502,9 +502,10 @@ async fn write_download( let mut file = open_download_target(&target, false).await?; let written = copy_until(source, &mut file, closed).await; if written.is_err() { - discard_if_still(&target, &file).await; + discard_if_still(&target, file).await; + } else { + drop(file); } - drop(file); return written.map_err(Into::into); }; @@ -594,8 +595,9 @@ async fn discard(path: &Path) { /// must survive this failure. /// /// Only Unix exposes a file's identity on stable Rust; elsewhere the path is -/// removed as before. -async fn discard_if_still(path: &Path, file: &tokio::fs::File) { +/// removed as before. The handle is closed before the removal either way, so +/// it never depends on the file having been opened with delete sharing. +async fn discard_if_still(path: &Path, file: tokio::fs::File) { #[cfg(unix)] { use std::os::unix::fs::MetadataExt; @@ -609,8 +611,7 @@ async fn discard_if_still(path: &Path, file: &tokio::fs::File) { return; } } - #[cfg(not(unix))] - let _ = file; + drop(file); discard(path).await; } From dbcf4a4d877c60fcee733ecb50f30e3a3e2b9934 Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:11:17 +0000 Subject: [PATCH 8/8] Stage every download and refuse what an in-place write would refuse A download to a new name was written in place, so quitting the app or a crash part way left a truncated file under the final name, and on Windows a failed download could still delete a finished file another download had put at the same path. Every download now goes through a staging file. A download that may not overwrite takes its name through a hard link, which refuses an existing name in the same step, so a failure only ever removes its own staging file. Replacing by rename needs write access only to the directory, so a file marked read-only was silently replaced. The target is now opened for writing first and refused if that fails, and only a regular file is replaced. Set-ID bits are not carried over, and a failed chmod is logged instead of discarding a complete download. An overwrite through a dangling symlink failed with "local file already exists"; links are now followed with read_link, so the file they name is created. A close that lands after the last byte, while the file is being synced, still stops the download before it is put in place, and it reports the same SSH error as every other operation. Local writes go through a 256 KiB buffer, the handle is closed only after tokio's in-flight write finishes, and the directory is synced after the rename. The upload buffer comment claimed russh-sftp sends writes as large as the slice it is given; it caps them at 32 KiB by default, so the buffer keeps packets full (about 512 KiB in flight) rather than 4 MiB. --- core/src/fs_util.rs | 4 +- core/src/sftp.rs | 595 ++++++++++++++++++++++++-------- core/tests/session_isolation.rs | 94 ++++- 3 files changed, 545 insertions(+), 148 deletions(-) diff --git a/core/src/fs_util.rs b/core/src/fs_util.rs index 6f93755d..ee3408ef 100644 --- a/core/src/fs_util.rs +++ b/core/src/fs_util.rs @@ -164,7 +164,7 @@ fn remove_if_present(path: &Path) -> std::io::Result { } #[cfg(unix)] -fn sync_parent_directory(path: &Path) -> std::io::Result<()> { +pub(crate) fn sync_parent_directory(path: &Path) -> std::io::Result<()> { let parent = path .parent() .filter(|parent| !parent.as_os_str().is_empty()) @@ -176,7 +176,7 @@ fn sync_parent_directory(path: &Path) -> std::io::Result<()> { // on Windows. Keep the unlink behavior unchanged there rather than pretending a // file handle sync makes the parent directory entry durable. #[cfg(not(unix))] -fn sync_parent_directory(_path: &Path) -> std::io::Result<()> { +pub(crate) fn sync_parent_directory(_path: &Path) -> std::io::Result<()> { Ok(()) } diff --git a/core/src/sftp.rs b/core/src/sftp.rs index 92c6b379..d6052f4f 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -7,8 +7,11 @@ use crate::{CoreError, Result}; use russh::client::Handle; use russh_sftp::client::SftpSession; use std::collections::HashMap; +use std::future::Future; use std::path::{Path, PathBuf}; +use std::pin::Pin; use std::sync::Arc; +use std::task::Poll; use std::time::Duration; use tokio::io::{AsyncRead, AsyncWriteExt}; use tokio::sync::Mutex; @@ -22,11 +25,20 @@ const LIST_DIR_DEADLINE: Duration = Duration::from_secs(30); /// How long a polite goodbye may take before the connection is dropped. const CLOSE_GRACE: Duration = Duration::from_secs(5); -/// Read buffer for uploads. russh-sftp keeps up to 16 writes in flight, each -/// as large as the slice it is handed, so the read size sets how much data is -/// in the air per round trip: `tokio::io::copy`'s 8 KiB would cap it at -/// 128 KiB, far below what a long link can carry. -const UPLOAD_BUFFER: usize = 256 * 1024; +/// Buffer between a transfer and its local file. +/// +/// tokio runs every read or write of a `tokio::fs::File` as a separate job on +/// its blocking pool, so this sets how much each of those jobs moves. On an +/// upload it also keeps every SFTP write full: russh-sftp splits writes into +/// packets of at most `Config::max_write_packet_len` (32 KiB by default) and +/// keeps up to 16 in flight, but sends a smaller packet whenever it is handed +/// less, as it is by `tokio::io::copy`'s 8 KiB reads. With full packets an +/// upload keeps about 512 KiB in flight instead of 128 KiB. +const LOCAL_FILE_BUFFER: usize = 256 * 1024; + +/// Most symbolic links a download follows to find the file it writes, the +/// same bound Linux puts on a path lookup. +const MAX_LINKS: usize = 40; /// A directory entry returned by SFTP list operations. #[derive(Debug, Clone, serde::Serialize)] @@ -70,9 +82,7 @@ impl SftpConnection { let mut closed = self.closed.subscribe(); tokio::select! { result = operation => result, - _ = closed.wait_for(|closed| *closed) => { - Err(CoreError::Ssh("the SFTP session was closed".into())) - } + _ = closed.wait_for(|closed| *closed) => Err(session_closed()), } } @@ -292,8 +302,9 @@ impl SftpManager { .map_err(|e| CoreError::Ssh(format!("open remote file: {e}"))) }) .await?; - // The close stops the copy rather than dropping the whole download, - // so the local file is always cleaned up by the download itself. + // A close stops the download at any point before it is put in place + // but never cuts its cleanup short, so the staging file is always + // removed by the download itself. let mut closed = conn.closed.subscribe(); let closed = async move { let _ = closed.wait_for(|closed| *closed).await; @@ -326,7 +337,7 @@ impl SftpManager { } let mut local = tokio::io::BufReader::with_capacity( - UPLOAD_BUFFER, + LOCAL_FILE_BUFFER, tokio::fs::File::open(local_path).await?, ); let mut remote = conn @@ -475,89 +486,174 @@ impl Default for SftpManager { } } -/// Write a download to `target` without ever leaving a partial file behind. +/// The error every operation stopped by a close reports. +fn session_closed() -> CoreError { + CoreError::Ssh("the SFTP session was closed".into()) +} + +/// Write a download to `target` without ever leaving a partial file under its +/// name. /// -/// Replacing an existing file goes through a staging file beside it that is -/// renamed over it only once every byte is on disk, so a dropped connection, -/// a server timeout or a closed session leaves the user's file as it was. A -/// new file is written in place and removed again if the transfer fails. -/// Either way a failed download leaves nothing that looks like a finished -/// one. `closed` resolving stops the copy; the cleanup still runs. +/// Every download is written to a staging file beside the target and put in +/// place only once every byte is on disk, so a dropped connection, a server +/// timeout, a closed session or the app quitting mid-transfer never leaves a +/// file under the target's name that looks finished. A failed download removes +/// only its own staging file, so it never takes a file that another download +/// put at the same name. /// -/// A symbolic link is followed, so the file it names is replaced and the -/// link stays a link, as writing through it would have done. +/// With `overwrite`, whatever is at `target` is replaced, provided it is a +/// regular file the user could also have written in place: one the user has +/// no write access to, such as a file marked read-only, is refused rather than +/// renamed over. +/// The replacement keeps the replaced file's permission bits. It is still a +/// new file, though, so other hard links to the old one keep the old content, +/// its owner, extended attributes and Windows access control are its own, and +/// replacing needs room for both copies until the rename. Without `overwrite`, +/// an existing file is refused, and the download takes the name only if +/// nothing else has taken it in the meantime. +/// +/// A symbolic link is followed, even one whose file does not exist yet, so the +/// file it names is written and the link stays a link, as writing through it +/// would have done. `closed` resolving stops the download at any point before +/// it is put in place; the cleanup still runs. async fn write_download( source: &mut R, target: &Path, overwrite: bool, - closed: impl std::future::Future, + closed: impl Future, ) -> Result<()> { + let mut closed = std::pin::pin!(closed); + if !overwrite && tokio::fs::symlink_metadata(target).await.is_ok() { + return Err(already_exists(target)); + } let target = follow_link(target).await; - let existing = if overwrite { - tokio::fs::metadata(&target).await.ok() + let replaced = if overwrite { + replaceable(&target).await? } else { None }; - let Some(existing) = existing else { - let mut file = open_download_target(&target, false).await?; - let written = copy_until(source, &mut file, closed).await; - if written.is_err() { - discard_if_still(&target, file).await; - } else { - drop(file); - } - return written.map_err(Into::into); - }; let staging = staging_path(&target); - let mut file = open_staging(&staging).await?; - let written = async { - copy_until(source, &mut file, closed).await?; - // The staging file is created owner-only so a private file being - // replaced is never readable by anyone else mid-download; the - // replacement then takes the original's permissions. - #[cfg(unix)] - tokio::fs::set_permissions(&staging, existing.permissions()).await?; - #[cfg(not(unix))] - let _ = &existing; - tokio::fs::rename(&staging, &target).await + let mut file = open_staging(&staging, replaced.is_some()).await?; + let copied = copy_until(source, &mut file, closed.as_mut()).await; + if let (Ok(()), Some(replaced)) = (&copied, &replaced) { + keep_permissions(&file, replaced).await; } - .await; - drop(file); - if written.is_err() { + close(file).await; + + let published = match copied { + Err(error) => Err(error), + // A close that lands while the file is being made durable still stops + // the download, so nothing is put in place once the session is gone. + Ok(()) if has_resolved(closed.as_mut()).await => Err(session_closed()), + Ok(()) => publish(&staging, &target, overwrite).await, + }; + if published.is_err() { discard(&staging).await; + return published; } - written.map_err(Into::into) + sync_directory_of(&target).await; + Ok(()) } /// Copy `source` into `file` and make it durable, unless `closed` resolves /// first. -async fn copy_until( +async fn copy_until( source: &mut R, file: &mut tokio::fs::File, - closed: impl std::future::Future, -) -> std::io::Result<()> { + closed: Pin<&mut F>, +) -> Result<()> +where + R: AsyncRead + Unpin + ?Sized, + F: Future, +{ + let mut writer = tokio::io::BufWriter::with_capacity(LOCAL_FILE_BUFFER, &mut *file); tokio::select! { - copied = tokio::io::copy(source, file) => copied?, - () = closed => { - return Err(std::io::Error::new( - std::io::ErrorKind::Interrupted, - "the SFTP session was closed", - )); + copied = tokio::io::copy(source, &mut writer) => { + copied?; } - }; - file.flush().await?; - file.sync_all().await + () = closed => return Err(session_closed()), + } + writer.flush().await?; + drop(writer); + file.sync_all().await?; + Ok(()) } -/// The file a symbolic link at `path` names, or `path` itself. +/// Whether `future` has resolved, without waiting for it. Never called on a +/// future that has already resolved. +async fn has_resolved>(mut future: Pin<&mut F>) -> bool { + std::future::poll_fn(|cx| Poll::Ready(future.as_mut().poll(cx).is_ready())).await +} + +/// Close `file` once tokio has finished any write still in flight on it. +/// Dropping a `tokio::fs::File` leaves the handle open until then, and the +/// file is renamed or removed right after this. +async fn close(mut file: tokio::fs::File) { + let _ = file.flush().await; + drop(file.try_into_std()); +} + +/// The path a write through `path` ends up at. A symbolic link is followed +/// even when the file it names does not exist yet, since writing through it +/// would create that file. A chain longer than `MAX_LINKS` is left as it is, +/// so opening it fails the way the system reports a loop. async fn follow_link(path: &Path) -> PathBuf { - match tokio::fs::symlink_metadata(path).await { - Ok(meta) if meta.file_type().is_symlink() => tokio::fs::canonicalize(path) + let mut path = path.to_path_buf(); + for _ in 0..MAX_LINKS { + let is_link = tokio::fs::symlink_metadata(&path) .await - .unwrap_or_else(|_| path.to_path_buf()), - _ => path.to_path_buf(), + .is_ok_and(|meta| meta.file_type().is_symlink()); + if !is_link { + break; + } + let Ok(next) = tokio::fs::read_link(&path).await else { + break; + }; + // A relative link names a path from the directory holding the link; + // `join` keeps an absolute one as it is. + path = match path.parent() { + Some(dir) => dir.join(next), + None => next, + }; } + path +} + +/// The file an overwrite of `target` replaces, if there is one. +/// +/// Writing a file in place needs permission to write that file, while renaming +/// over it needs permission only on its directory. The download is put in +/// place by a rename, so the file is first opened for writing, and one the user +/// could not have written is refused rather than replaced. Only a regular file +/// is replaced. +async fn replaceable(target: &Path) -> Result> { + let replaced = match tokio::fs::metadata(target).await { + Ok(meta) => meta, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + if !replaced.is_file() { + return Err(CoreError::InvalidInput(format!( + "not a regular file: {}", + target.display() + ))); + } + tokio::fs::OpenOptions::new() + .write(true) + .open(target) + .await + .map_err(|error| { + std::io::Error::new( + error.kind(), + format!("cannot write {}: {error}", target.display()), + ) + })?; + Ok(Some(replaced)) +} + +fn already_exists(path: &Path) -> CoreError { + CoreError::InvalidInput(format!("local file already exists: {}", path.display())) } /// A fresh hidden name beside `target`. Only the start of the target's name @@ -571,76 +667,107 @@ fn staging_path(target: &Path) -> PathBuf { target.with_file_name(format!(".{name}.{}.part", uuid::Uuid::new_v4())) } -async fn open_staging(path: &Path) -> Result { +/// Create the staging file. One that replaces an existing file starts out +/// readable by its owner only on Unix, so a private file being replaced is +/// never readable by anyone else mid-download; a new download starts with the +/// default permissions it keeps. On Windows both take the access control the +/// directory passes on. +async fn open_staging(path: &Path, private: bool) -> Result { let mut options = tokio::fs::OpenOptions::new(); options.write(true).create_new(true); #[cfg(unix)] - options.mode(0o600); + if private { + options.mode(0o600); + } + #[cfg(not(unix))] + let _ = private; Ok(options.open(path).await?) } -/// Remove what a failed download wrote. A file that cannot be removed is -/// logged, since the error the caller sees is the download's own. -async fn discard(path: &Path) { - if let Err(error) = tokio::fs::remove_file(path).await { - if error.kind() != std::io::ErrorKind::NotFound { - tracing::warn!("could not remove an incomplete download at {}: {error}", path.display()); - } +/// Give a replacement the permission bits of the file it replaces. Set-user-ID +/// and set-group-ID are left off, as a write in place by anyone but root clears +/// them. A failure is logged and the staging file stays owner-only, which is +/// never more open than the original: some network and FUSE file systems +/// refuse every mode change, and a complete download is not thrown away for it. +#[cfg(unix)] +async fn keep_permissions(file: &tokio::fs::File, replaced: &std::fs::Metadata) { + use std::os::unix::fs::PermissionsExt; + let mode = replaced.permissions().mode() & 0o777; + if let Err(error) = file + .set_permissions(std::fs::Permissions::from_mode(mode)) + .await + { + tracing::warn!("could not give a download the replaced file's permissions: {error}"); } } -/// Remove an incomplete download at `path`, but only while `path` still names -/// the file this download created. Another download that chose to overwrite -/// can have renamed its finished file there in the meantime, and that one -/// must survive this failure. +/// Windows has no permission bits to carry over, and a file that is read-only +/// is refused before the download starts. +#[cfg(not(unix))] +async fn keep_permissions(_file: &tokio::fs::File, _replaced: &std::fs::Metadata) {} + +/// Put a finished staging file in place at `target`. /// -/// Only Unix exposes a file's identity on stable Rust; elsewhere the path is -/// removed as before. The handle is closed before the removal either way, so -/// it never depends on the file having been opened with delete sharing. -async fn discard_if_still(path: &Path, file: tokio::fs::File) { - #[cfg(unix)] - { - use std::os::unix::fs::MetadataExt; - let (Ok(ours), Ok(there)) = ( - file.metadata().await, - tokio::fs::symlink_metadata(path).await, - ) else { - return; - }; - if (ours.dev(), ours.ino()) != (there.dev(), there.ino()) { - return; +/// With `replace`, whatever is at `target` is replaced. Without it, the file +/// takes the name only if nothing else has: a hard link refuses an existing +/// name in the same step that creates it. A file system without hard links +/// (FAT, exFAT, some network shares) gets a check followed by a rename +/// instead, which leaves a short window in which a file created by someone +/// else could be replaced. +async fn publish(staging: &Path, target: &Path, replace: bool) -> Result<()> { + if replace { + return Ok(tokio::fs::rename(staging, target).await?); + } + match tokio::fs::hard_link(staging, target).await { + Ok(()) => { + discard(staging).await; + Ok(()) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + Err(already_exists(target)) + } + Err(_) => { + if tokio::fs::symlink_metadata(target).await.is_ok() { + return Err(already_exists(target)); + } + Ok(tokio::fs::rename(staging, target).await?) } } - drop(file); - discard(path).await; } -/// Open the local file a download writes into. Without `overwrite` an existing -/// file is refused; `create_new` checks and creates in one step, so a file that -/// appears between a check and the open is refused too. -async fn open_download_target(path: &Path, overwrite: bool) -> Result { - let mut options = tokio::fs::OpenOptions::new(); - options.write(true); - if overwrite { - options.create(true).truncate(true); - } else { - options.create_new(true); +/// Make the download's name durable as well as its bytes. A failure is +/// logged: the download is already in place, and reporting it as failed would +/// say the opposite. +async fn sync_directory_of(target: &Path) { + let target = target.to_path_buf(); + let synced = tokio::task::spawn_blocking(move || { + crate::fs_util::sync_parent_directory(&target).map_err(|error| (target, error)) + }) + .await; + if let Ok(Err((target, error))) = synced { + tracing::warn!("saved {} but could not sync its directory: {error}", target.display()); } - options.open(path).await.map_err(|e| { - if e.kind() == std::io::ErrorKind::AlreadyExists { - CoreError::InvalidInput(format!("local file already exists: {}", path.display())) - } else { - e.into() +} + +/// Remove what a failed download wrote. A file that cannot be removed is +/// logged, since the error the caller sees is the download's own. +async fn discard(path: &Path) { + if let Err(error) = tokio::fs::remove_file(path).await { + if error.kind() != std::io::ErrorKind::NotFound { + tracing::warn!("could not remove an incomplete download at {}: {error}", path.display()); } - }) + } } #[cfg(test)] mod download_target_tests { - use super::{open_download_target, write_download}; + use super::write_download; + use std::path::PathBuf; use std::pin::Pin; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; use std::task::{Context, Poll}; - use tokio::io::{AsyncRead, AsyncWriteExt, ReadBuf}; + use tokio::io::{AsyncRead, ReadBuf}; /// Yields `data`, then fails the way a dropped connection does. struct BrokenAfter(&'static [u8]); @@ -698,7 +825,7 @@ mod download_target_tests { /// Fails after another download has renamed its finished file over the /// path this one was writing. - struct ReplacedThenBroken(std::path::PathBuf); + struct ReplacedThenBroken(PathBuf); impl AsyncRead for ReplacedThenBroken { fn poll_read( @@ -713,7 +840,6 @@ mod download_target_tests { } } - #[cfg(unix)] #[tokio::test] async fn a_failed_new_download_spares_a_file_that_replaced_it() { let dir = tempfile::tempdir().expect("temp dir"); @@ -728,6 +854,7 @@ mod download_target_tests { std::fs::read_to_string(&path).expect("the other download's file"), "finished by another download" ); + assert_eq!(entries(dir.path()), ["report.txt"]); } #[tokio::test] @@ -741,6 +868,91 @@ mod download_target_tests { assert!(entries(dir.path()).is_empty()); } + /// Yields `data` and notes whether anything was at `path` while it did. + struct WatchesName { + path: PathBuf, + data: &'static [u8], + saw_the_name: bool, + } + + impl AsyncRead for WatchesName { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = &mut *self; + this.saw_the_name |= std::fs::symlink_metadata(&this.path).is_ok(); + let n = this.data.len().min(buf.remaining()); + buf.put_slice(&this.data[..n]); + this.data = &this.data[n..]; + Poll::Ready(Ok(())) + } + } + + /// A new download is staged too, so its name never holds a partial file: + /// not while it runs, and not if the app quits part way through. + #[tokio::test] + async fn a_new_download_takes_its_name_only_once_complete() { + let dir = tempfile::tempdir().expect("temp dir"); + for (name, overwrite) in [("a.txt", false), ("b.txt", true)] { + let path = dir.path().join(name); + let mut source = WatchesName { + path: path.clone(), + data: b"complete", + saw_the_name: false, + }; + write_download(&mut source, &path, overwrite, never()) + .await + .expect("download"); + assert!(!source.saw_the_name, "{name} existed mid-download"); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "complete"); + } + assert_eq!(entries(dir.path()), ["a.txt", "b.txt"]); + } + + /// Writes another file at `path` as the download starts, then yields + /// `data`. + struct NameTakenDuring { + path: Option, + data: &'static [u8], + } + + impl AsyncRead for NameTakenDuring { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if let Some(path) = self.path.take() { + std::fs::write(path, "finished by another download").expect("write"); + } + let n = self.data.len().min(buf.remaining()); + buf.put_slice(&self.data[..n]); + self.data = &self.data[n..]; + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn a_download_without_overwrite_never_replaces_a_file_that_took_its_name() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + let mut source = NameTakenDuring { + path: Some(path.clone()), + data: b"this download", + }; + + assert!(write_download(&mut source, &path, false, never()) + .await + .is_err()); + assert_eq!( + std::fs::read_to_string(&path).expect("read"), + "finished by another download" + ); + assert_eq!(entries(dir.path()), ["report.txt"]); + } + /// Yields `data`, then never answers again, like a peer that went quiet. struct StallsAfter(&'static [u8]); @@ -786,6 +998,54 @@ mod download_target_tests { assert_eq!(entries(dir.path()), ["report.txt"]); } + /// Yields `data`, then reports the session closed as the data runs out. + struct ClosesAtTheEnd { + data: &'static [u8], + closed: Arc, + } + + impl AsyncRead for ClosesAtTheEnd { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.data.is_empty() { + self.closed.store(true, Ordering::SeqCst); + return Poll::Ready(Ok(())); + } + let n = self.data.len().min(buf.remaining()); + buf.put_slice(&self.data[..n]); + self.data = &self.data[n..]; + Poll::Ready(Ok(())) + } + } + + /// A close that arrives once every byte has been read still stops the + /// download, wherever it is in making the file durable. + #[tokio::test] + async fn a_close_after_the_last_byte_still_leaves_the_original() { + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + let closed = Arc::new(AtomicBool::new(false)); + let mut source = ClosesAtTheEnd { + data: b"replacement", + closed: closed.clone(), + }; + let signal = std::future::poll_fn(move |_| { + if closed.load(Ordering::SeqCst) { + Poll::Ready(()) + } else { + Poll::Pending + } + }); + + assert!(write_download(&mut source, &path, true, signal).await.is_err()); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); + assert_eq!(entries(dir.path()), ["report.txt"]); + } + /// The replacement takes the original's permissions, so a private file /// stays private. #[cfg(unix)] @@ -804,6 +1064,68 @@ mod download_target_tests { assert_eq!(mode, 0o600); } + /// A write in place clears set-user-ID and set-group-ID, so a replacement + /// does not carry them over. + #[cfg(unix)] + #[tokio::test] + async fn an_overwrite_does_not_carry_set_id_bits_over() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("tool"); + std::fs::write(&path, "original").expect("write"); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o6755)).expect("chmod"); + + write_download(&mut &b"replacement"[..], &path, true, never()) + .await + .expect("download"); + let mode = std::fs::metadata(&path).expect("stat").permissions().mode() & 0o7777; + assert_eq!(mode, 0o755); + } + + /// A new download ends up with the permissions any new file in that + /// directory gets. + #[cfg(unix)] + #[tokio::test] + async fn a_new_download_gets_the_default_permissions() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().expect("temp dir"); + let reference = dir.path().join("reference"); + std::fs::write(&reference, "").expect("write"); + let path = dir.path().join("report.txt"); + + write_download(&mut &b"data"[..], &path, true, never()) + .await + .expect("download"); + let mode = |path: &std::path::Path| { + std::fs::metadata(path).expect("stat").permissions().mode() & 0o777 + }; + assert_eq!(mode(&path), mode(&reference)); + } + + /// A file the user could not have written in place is refused. A rename + /// needs write access only to the directory, so it must not stand in for + /// a write the user has no permission for. + #[cfg(unix)] + #[tokio::test] + async fn an_overwrite_refuses_a_file_the_user_cannot_write() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().expect("temp dir"); + let path = dir.path().join("report.txt"); + std::fs::write(&path, "original").expect("write"); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o444)).expect("chmod"); + // An account that may write any file, root for one, has nothing to be + // refused here. + if std::fs::OpenOptions::new().write(true).open(&path).is_ok() { + return; + } + + assert!(write_download(&mut &b"replacement"[..], &path, true, never()) + .await + .is_err()); + assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); + assert_eq!(entries(dir.path()), ["report.txt"]); + } + /// Overwriting through a symbolic link replaces the file it names and /// leaves the link in place. #[cfg(unix)] @@ -823,6 +1145,26 @@ mod download_target_tests { assert_eq!(entries(dir.path()), ["link.txt", "real.txt"]); } + /// Writing through a link whose file does not exist yet creates that file, + /// as writing through it in place would. + #[cfg(unix)] + #[tokio::test] + async fn an_overwrite_through_a_dangling_link_creates_the_file_it_names() { + let dir = tempfile::tempdir().expect("temp dir"); + let link = dir.path().join("link.txt"); + std::os::unix::fs::symlink("real.txt", &link).expect("symlink"); + + write_download(&mut &b"replacement"[..], &link, true, never()) + .await + .expect("download"); + assert!(std::fs::symlink_metadata(&link).expect("lstat").file_type().is_symlink()); + assert_eq!( + std::fs::read_to_string(dir.path().join("real.txt")).expect("read"), + "replacement" + ); + assert_eq!(entries(dir.path()), ["link.txt", "real.txt"]); + } + #[tokio::test] async fn an_overwrite_works_for_a_name_at_the_length_limit() { let dir = tempfile::tempdir().expect("temp dir"); @@ -845,27 +1187,6 @@ mod download_target_tests { .await .is_err()); assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); - } - - #[tokio::test] - async fn an_existing_file_is_kept_unless_overwrite_is_chosen() { - let dir = tempfile::tempdir().expect("temp dir"); - let path = dir.path().join("report.txt"); - std::fs::write(&path, "original").expect("write"); - - assert!(open_download_target(&path, false).await.is_err()); - assert_eq!(std::fs::read_to_string(&path).expect("read"), "original"); - - let mut file = open_download_target(&path, true).await.expect("overwrite"); - file.write_all(b"new").await.expect("write"); - file.flush().await.expect("flush"); - assert_eq!(std::fs::read_to_string(&path).expect("read"), "new"); - } - - #[tokio::test] - async fn a_new_file_is_created_either_way() { - let dir = tempfile::tempdir().expect("temp dir"); - assert!(open_download_target(&dir.path().join("a"), false).await.is_ok()); - assert!(open_download_target(&dir.path().join("b"), true).await.is_ok()); + assert_eq!(entries(dir.path()), ["report.txt"]); } } diff --git a/core/tests/session_isolation.rs b/core/tests/session_isolation.rs index 7362bc3d..14df57aa 100644 --- a/core/tests/session_isolation.rs +++ b/core/tests/session_isolation.rs @@ -1,9 +1,9 @@ //! One misbehaving peer must hold up nothing but its own session. //! //! These run the SSH and SFTP managers against an in-process russh server on -//! loopback whose SFTP side never finishes a directory listing and whose shell -//! exits as soon as it starts, and check that other sessions, a close and the -//! drain a vault reset relies on all carry on regardless. +//! loopback whose SFTP side never finishes a directory listing or a file read +//! and whose shell exits as soon as it starts, and check that other sessions, +//! a close and the drain a vault reset relies on all carry on regardless. use clavyn_core::host::Host; use clavyn_core::known_hosts::KnownHosts; @@ -12,7 +12,9 @@ use clavyn_core::sftp::SftpManager; use russh::keys::{Algorithm, PrivateKey}; use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; use russh::{Channel, ChannelId}; -use russh_sftp::protocol::{File, Handle, Name, Status, StatusCode, Version}; +use russh_sftp::protocol::{ + Data, File, FileAttributes, Handle, Name, OpenFlags, Status, StatusCode, Version, +}; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; @@ -23,11 +25,12 @@ use tokio::sync::Mutex; const USER: &str = "deploy"; const PASSWORD: &str = "correct password"; -/// An SFTP server whose directories never end: every READDIR answers with one -/// more entry and never with end-of-file. -struct EndlessDirectory; +/// An SFTP server that never finishes anything: every READDIR answers with one +/// more entry and never with end-of-file, and a file answers its first read +/// and then goes quiet. +struct NeverFinishes; -impl russh_sftp::server::Handler for EndlessDirectory { +impl russh_sftp::server::Handler for NeverFinishes { type Error = StatusCode; fn unimplemented(&self) -> Self::Error { @@ -57,6 +60,35 @@ impl russh_sftp::server::Handler for EndlessDirectory { }) } + async fn open( + &mut self, + id: u32, + _filename: String, + _pflags: OpenFlags, + _attrs: FileAttributes, + ) -> Result { + Ok(Handle { + id, + handle: "file".into(), + }) + } + + async fn read( + &mut self, + id: u32, + _handle: String, + offset: u64, + _len: u32, + ) -> Result { + if offset > 0 { + std::future::pending::<()>().await; + } + Ok(Data { + id, + data: vec![b'x'; 1024], + }) + } + async fn realpath(&mut self, id: u32, _path: String) -> Result { Ok(Name { id, @@ -144,7 +176,7 @@ impl server::Handler for TestServer { match channel { Some(channel) if name == "sftp" => { session.channel_success(channel_id)?; - russh_sftp::server::run(channel.into_stream(), EndlessDirectory).await; + russh_sftp::server::run(channel.into_stream(), NeverFinishes).await; } _ => session.channel_failure(channel_id)?, } @@ -271,6 +303,50 @@ async fn closing_one_sftp_session_does_not_wait_on_its_listing() { assert!(result.is_err()); } +/// Closing a session ends a download stuck behind a peer that went quiet, and +/// leaves the file the download was replacing as it was. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_download_from_a_quiet_peer_ends_with_the_close_and_keeps_the_original() { + let dir = tempfile::tempdir().expect("temp dir"); + let port = start_server().await; + let manager = Arc::new(SftpManager::new()); + manager + .connect("stuck".into(), &host(port), None, known_hosts(&dir), None, None, Some(PASSWORD)) + .await + .expect("sftp connect"); + + let downloads = tempfile::tempdir().expect("download dir"); + let target = downloads.path().join("report.txt"); + std::fs::write(&target, "original").expect("write"); + let download = tokio::spawn({ + let manager = manager.clone(); + let target = target.clone(); + async move { + manager + .download_to_local("stuck", "/srv/report.txt", &target, true) + .await + } + }); + tokio::time::sleep(Duration::from_millis(300)).await; + assert!(!download.is_finished(), "the stalled download ended on its own"); + + tokio::time::timeout(Duration::from_secs(5), manager.close("stuck")) + .await + .expect("close waited on the download") + .expect("close"); + let result = tokio::time::timeout(Duration::from_secs(2), download) + .await + .expect("the download outlived its session") + .expect("download task"); + assert!(result.is_err()); + assert_eq!(std::fs::read_to_string(&target).expect("read"), "original"); + let names: Vec<_> = std::fs::read_dir(downloads.path()) + .expect("read dir") + .map(|entry| entry.expect("entry").file_name()) + .collect(); + assert_eq!(names, ["report.txt"]); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_shell_that_exits_leaves_no_session_behind() { let dir = tempfile::tempdir().expect("temp dir");