diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 770e2762..79a222de 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -595,6 +595,7 @@ SSH server implementation using the russh library for accepting incoming connect - **SftpHandler**: SFTP subsystem handler (`src/server/sftp.rs`) - Implements `russh_sftp::server::Handler` trait for file transfer operations + - Advertises and serves `limits@openssh.com` so clients negotiate the server's packet, read, write, and handle ceilings before bulk transfers - Path traversal prevention with chroot-like isolation - File operations: open, read, write, close - Directory operations: opendir, readdir, mkdir, rmdir @@ -603,6 +604,7 @@ SSH server implementation using the russh library for accepting incoming connect - Symlink validation ensures targets remain within root directory - Handle limit enforcement to prevent resource exhaustion - Read size capping to prevent memory exhaustion + - Pipelined downloads tolerate legal short `READ` replies by re-requesting the missing byte range before ordered reassembly, and clamp advertised transfer ceilings to the negotiated packet payload budget - **ScpHandler**: SCP protocol handler (`src/server/scp.rs`) - Implements SCP server protocol for file transfers via the `scp` command diff --git a/crates/bssh-russh-sftp/src/client/fs/file.rs b/crates/bssh-russh-sftp/src/client/fs/file.rs index 7a42a555..9e5fa4c0 100644 --- a/crates/bssh-russh-sftp/src/client/fs/file.rs +++ b/crates/bssh-russh-sftp/src/client/fs/file.rs @@ -71,6 +71,59 @@ impl File { } } + fn effective_read_len(&self) -> SftpResult { + let packet_payload = self + .features + .max_packet_len + .saturating_sub(READ_OVERHEAD_LENGTH); + let advertised = self + .features + .limits + .and_then(|limits| limits.read_len) + .unwrap_or(u64::from(packet_payload)); + let effective = advertised + .min(u64::from(packet_payload)) + .min(u64::from(u32::MAX)); + + if effective == 0 { + return Err(Error::UnexpectedBehavior( + "effective SFTP read payload length is zero".to_owned(), + )); + } + + Ok(effective as usize) + } + + fn effective_write_len(&self) -> SftpResult { + let handle_len = u32::try_from(self.handle.len()).map_err(|_| { + Error::UnexpectedBehavior("SFTP handle length exceeds protocol limit".to_owned()) + })?; + let overhead = WRITE_OVERHEAD_LENGTH + .checked_add(handle_len) + .ok_or_else(|| { + Error::UnexpectedBehavior( + "SFTP write packet overhead exceeds protocol limit".to_owned(), + ) + })?; + let packet_payload = self.features.max_packet_len.saturating_sub(overhead); + let advertised = self + .features + .limits + .and_then(|limits| limits.write_len) + .unwrap_or(u64::from(packet_payload)); + let effective = advertised + .min(u64::from(packet_payload)) + .min(u64::from(u32::MAX)); + + if effective == 0 { + return Err(Error::UnexpectedBehavior( + "effective SFTP write payload length is zero".to_owned(), + )); + } + + Ok(effective as usize) + } + /// Queries metadata about the remote file. pub async fn metadata(&self) -> SftpResult { Ok(self.session.fstat(self.handle.as_str()).await?.attrs) @@ -126,14 +179,7 @@ impl File { )); } - let chunk_size = self - .features - .limits - .and_then(|l| l.write_len) - .unwrap_or_else(|| { - let overhead = WRITE_OVERHEAD_LENGTH + self.handle.len() as u32; - self.features.max_packet_len.saturating_sub(overhead) as u64 - }) as usize; + let chunk_size = self.effective_write_len()?; let mut total: u64 = 0; let mut offset = self.pos; @@ -203,15 +249,7 @@ impl File { )); } - let chunk_size = self - .features - .limits - .and_then(|l| l.read_len) - .unwrap_or_else(|| { - self.features - .max_packet_len - .saturating_sub(READ_OVERHEAD_LENGTH) as u64 - }) as usize; + let chunk_size = self.effective_read_len()?; let file_end = self .metadata() .await @@ -225,6 +263,15 @@ impl File { let mut pending: BTreeMap> = BTreeMap::new(); let mut in_flight = FuturesUnordered::new(); let mut eof = false; + let read_request = |session: Arc, handle: String, off: u64, len: u32| async move { + match session.read(handle, off, len).await { + Ok(data) => SftpResult::Ok((off, len, Some(data.data))), + Err(Error::Status(s)) if s.status_code == StatusCode::Eof => { + SftpResult::Ok((off, len, None)) + } + Err(e) => Err(e), + } + }; loop { while !eof @@ -238,15 +285,7 @@ impl File { (end - next_offset).min(chunk_size as u64) }) as u32; - in_flight.push(async move { - match session.read(handle, off, len).await { - Ok(data) => SftpResult::Ok((off, len, Some(data.data))), - Err(Error::Status(s)) if s.status_code == StatusCode::Eof => { - SftpResult::Ok((off, len, None)) - } - Err(e) => Err(e), - } - }); + in_flight.push(read_request(session, handle, off, len)); next_offset += u64::from(len); } @@ -254,16 +293,35 @@ impl File { match in_flight.next().await { Some(Ok((off, len, Some(data)))) => { if data.is_empty() { + if file_end.is_some_and(|end| off < end) { + return Err(Error::UnexpectedBehavior(format!( + "unexpected empty read before file size at offset {off}" + ))); + } eof = true; } else { + if data.len() > len as usize { + return Err(Error::UnexpectedBehavior(format!( + "read returned more data than requested at offset {off}: requested {len} bytes, received {} bytes", + data.len() + ))); + } + if let Some(end) = file_end { let got_end = off.saturating_add(data.len() as u64); - if data.len() != len as usize || got_end > end { + if got_end > end { return Err(Error::UnexpectedBehavior(format!( - "short read before EOF at offset {off}: requested {len} bytes, received {} bytes", + "read returned data past known EOF at offset {off}: requested {len} bytes, received {} bytes", data.len() ))); } + if data.len() < len as usize && got_end < end { + let remaining = u64::from(len) - data.len() as u64; + let retry_len = remaining.min(u64::from(u32::MAX)) as u32; + let session = self.session.clone(); + let handle = self.handle.clone(); + in_flight.push(read_request(session, handle, got_end, retry_len)); + } } else if data.len() < len as usize { eof = true; } @@ -362,15 +420,13 @@ impl AsyncRead for File { Some(f) => f, None => { let session = self.session.clone(); - let max_read_len = self - .features - .limits - .and_then(|l| l.read_len) - .unwrap_or_else(|| { - self.features - .max_packet_len - .saturating_sub(READ_OVERHEAD_LENGTH) as u64 - }) as usize; + let max_read_len = match self.effective_read_len() { + Ok(len) => len, + Err(e) => { + let message = e.to_string(); + return Poll::Ready(Err(io::Error::other(message))); + } + }; let file_handle = self.handle.clone(); @@ -479,14 +535,10 @@ impl AsyncWrite for File { } } - let max_write_len = self - .features - .limits - .and_then(|l| l.write_len) - .unwrap_or_else(|| { - let overhead = WRITE_OVERHEAD_LENGTH + self.handle.len() as u32; - self.features.max_packet_len.saturating_sub(overhead) as u64 - }) as usize; + let max_write_len = match self.effective_write_len() { + Ok(len) => len, + Err(e) => return Poll::Ready(Err(io::Error::other(e.to_string()))), + }; let len = usize::min(buf.len(), max_write_len); let data = buf[..len].to_vec(); diff --git a/crates/bssh-russh-sftp/src/extensions.rs b/crates/bssh-russh-sftp/src/extensions.rs index 857f51ac..974f6f8e 100644 --- a/crates/bssh-russh-sftp/src/extensions.rs +++ b/crates/bssh-russh-sftp/src/extensions.rs @@ -26,6 +26,8 @@ pub struct LimitsExtension { pub max_open_handles: u64, } +impl_try_into_bytes!(LimitsExtension); + #[derive(Debug, Serialize, Deserialize)] pub struct HardlinkExtension { pub oldpath: String, diff --git a/src/server/sftp.rs b/src/server/sftp.rs index 40f28086..3743c674 100644 --- a/src/server/sftp.rs +++ b/src/server/sftp.rs @@ -47,8 +47,12 @@ use std::os::unix::fs::{MetadataExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use russh_sftp::protocol::{ - Attrs, Data, FileAttributes, Handle, Name, OpenFlags, Status, StatusCode, Version, +use russh_sftp::{ + extensions::{self, LimitsExtension}, + protocol::{ + Attrs, Data, ExtendedReply, FileAttributes, Handle, Name, OpenFlags, Packet, Status, + StatusCode, Version, + }, }; use tokio::fs::{self, File, OpenOptions}; use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; @@ -189,6 +193,17 @@ const MAX_HANDLES: usize = 1000; /// per-request buffer of this size. const MAX_READ_SIZE: u32 = 261120; +/// Version string required by OpenSSH's `limits@openssh.com` extension. +const LIMITS_EXTENSION_VERSION: &str = "1"; + +/// Maximum write payload size advertised to clients via `limits@openssh.com`. +/// +/// The server's packet reader still enforces the true frame ceiling from +/// `russh_sftp::server::Config::max_client_packet_len`; advertising the same +/// 255 KiB payload ceiling used for reads leaves room for SFTP write packet +/// overhead and keeps pipelined uploads below that frame ceiling. +const MAX_ADVERTISED_WRITE_SIZE: u32 = MAX_READ_SIZE; + /// Normalize a path's `..` and `.` components without touching the filesystem. /// /// This is a logical normalization that does not follow symlinks. Used as @@ -402,6 +417,30 @@ impl SftpHandler { } } + /// Build the SFTP version reply with extensions supported by this server. + fn version_with_extensions() -> Version { + let mut version = Version::new(); + version.extensions.insert( + extensions::LIMITS.to_owned(), + LIMITS_EXTENSION_VERSION.to_owned(), + ); + version + } + + /// Build the limits advertised through OpenSSH's `limits@openssh.com`. + fn limits_extension() -> LimitsExtension { + let server_config = russh_sftp::server::Config::default(); + let max_write_len = + (MAX_ADVERTISED_WRITE_SIZE as usize).min(server_config.max_write_coalesce_len); + + LimitsExtension { + max_packet_len: u64::from(server_config.max_client_packet_len), + max_read_len: u64::from(MAX_READ_SIZE), + max_write_len: max_write_len as u64, + max_open_handles: MAX_HANDLES as u64, + } + } + /// Generate a new unique handle ID. fn new_handle(&mut self) -> String { self.handle_counter += 1; @@ -584,7 +623,25 @@ impl russh_sftp::server::Handler for SftpHandler { "SFTP session initialized" ); - async move { Ok(Version::new()) } + async move { Ok(SftpHandler::version_with_extensions()) } + } + + /// Handle SFTP extension requests. + async fn extended( + &mut self, + id: u32, + request: String, + _data: Vec, + ) -> Result { + if request != extensions::LIMITS { + return Err(SftpError::not_supported()); + } + + let data = russh_sftp::ser::to_bytes(&SftpHandler::limits_extension()) + .map(|bytes| bytes.to_vec()) + .map_err(|err| SftpError::failure(err.to_string()))?; + + Ok(Packet::ExtendedReply(ExtendedReply { id, data })) } /// Open a file. diff --git a/tests/sftp_limits_download_test.rs b/tests/sftp_limits_download_test.rs new file mode 100644 index 00000000..9ff5b3b8 --- /dev/null +++ b/tests/sftp_limits_download_test.rs @@ -0,0 +1,644 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +use bssh::server::sftp::{SftpError, SftpHandler}; +use bssh::shared::auth_types::UserInfo; +use russh_sftp::client::{Config as ClientConfig, RawSftpSession, SftpSession}; +use russh_sftp::extensions::{self, LimitsExtension}; +use russh_sftp::protocol::{ + Attrs, Data, ExtendedReply, FileAttributes, Handle, OpenFlags, Packet, Status, StatusCode, + Version, +}; +use tokio::io::AsyncWriteExt; + +const BSSH_MAX_READ_LEN: u64 = 261_120; +const BSSH_MAX_HANDLES: u64 = 1_000; +const MAX_INFLIGHT: usize = 64; +const READ_OVERHEAD_LEN: u32 = 9; +const WRITE_OVERHEAD_LEN: u32 = 21; +const SMALL_CLIENT_PACKET_LEN: u32 = 1_024; + +async fn sftp_session_with_handler(handler: H) -> SftpSession +where + H: russh_sftp::server::Handler + Send + 'static, +{ + sftp_session_with_handler_and_configs( + handler, + ClientConfig::default(), + russh_sftp::server::Config::default(), + ) + .await +} + +async fn sftp_session_with_handler_and_config(handler: H, cfg: ClientConfig) -> SftpSession +where + H: russh_sftp::server::Handler + Send + 'static, +{ + sftp_session_with_handler_and_configs(handler, cfg, russh_sftp::server::Config::default()).await +} + +async fn sftp_session_with_handler_and_configs( + handler: H, + cfg: ClientConfig, + server_cfg: russh_sftp::server::Config, +) -> SftpSession +where + H: russh_sftp::server::Handler + Send + 'static, +{ + let (client, server) = tokio::io::duplex(1 << 20); + russh_sftp::server::run_with_config(server, handler, server_cfg).await; + SftpSession::new_with_config(client, cfg) + .await + .expect("SFTP session should initialize") +} + +async fn raw_session_with_handler(handler: H) -> RawSftpSession +where + H: russh_sftp::server::Handler + Send + 'static, +{ + let (client, server) = tokio::io::duplex(1 << 20); + russh_sftp::server::run(server, handler).await; + RawSftpSession::new(client) +} + +fn bssh_handler(root: PathBuf) -> SftpHandler { + SftpHandler::new(UserInfo::new("testuser"), None, root) +} + +fn payload(size: usize) -> Vec { + (0..size) + .map(|i| ((i.wrapping_mul(31) ^ (i >> 7)) & 0xff) as u8) + .collect() +} + +async fn upload_payload(sftp: &SftpSession, path: &str, payload: &[u8]) { + let mut remote = sftp + .open_with_flags( + path, + OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE | OpenFlags::READ, + ) + .await + .expect("remote file should open for upload"); + let (mut reader, mut writer) = tokio::io::duplex(64 * 1024); + let payload_for_writer = payload.to_vec(); + let writer_task = tokio::spawn(async move { + writer + .write_all(&payload_for_writer) + .await + .expect("payload should feed upload reader"); + writer + .shutdown() + .await + .expect("payload reader should close"); + }); + + let written = remote + .write_all_pipelined(&mut reader, MAX_INFLIGHT) + .await + .expect("pipelined upload should complete"); + writer_task + .await + .expect("payload writer task should finish"); + assert_eq!(written as usize, payload.len()); + remote + .close() + .await + .expect("remote upload handle should close"); +} + +async fn download_payload(sftp: &SftpSession, path: &str) -> Vec { + let mut remote = sftp + .open(path) + .await + .expect("remote file should open for download"); + let mut downloaded = Vec::new(); + let read = remote + .read_to_writer_pipelined(&mut downloaded, MAX_INFLIGHT) + .await + .expect("pipelined download should complete"); + assert_eq!(read as usize, downloaded.len()); + remote + .close() + .await + .expect("remote download handle should close"); + downloaded +} + +#[derive(Clone, Default)] +struct ObservedChunks { + reads: Arc>>, + writes: Arc>>, +} + +struct HugeAdvertisedLimitsHandler { + data: Vec, + observed: ObservedChunks, +} + +impl HugeAdvertisedLimitsHandler { + fn new(data: Vec, observed: ObservedChunks) -> Self { + Self { data, observed } + } +} + +impl russh_sftp::server::Handler for HugeAdvertisedLimitsHandler { + type Error = SftpError; + + fn unimplemented(&self) -> Self::Error { + SftpError::not_supported() + } + + async fn init( + &mut self, + _version: u32, + _extensions: HashMap, + ) -> Result { + let mut version = Version::new(); + version + .extensions + .insert(extensions::LIMITS.to_owned(), "1".to_owned()); + Ok(version) + } + + async fn extended( + &mut self, + id: u32, + request: String, + _data: Vec, + ) -> Result { + if request != extensions::LIMITS { + return Err(SftpError::not_supported()); + } + + let data = russh_sftp::ser::to_bytes(&LimitsExtension { + max_packet_len: u64::MAX, + max_read_len: u64::MAX, + max_write_len: u64::MAX, + max_open_handles: u64::MAX, + }) + .map(|bytes| bytes.to_vec()) + .map_err(|err| SftpError::failure(err.to_string()))?; + + Ok(Packet::ExtendedReply(ExtendedReply { id, data })) + } + + async fn open( + &mut self, + id: u32, + _filename: String, + _pflags: OpenFlags, + _attrs: FileAttributes, + ) -> Result { + Ok(Handle { + id, + handle: "huge".to_owned(), + }) + } + + async fn fstat(&mut self, id: u32, handle: String) -> Result { + if handle != "huge" { + return Err(SftpError::invalid_handle()); + } + + Ok(Attrs { + id, + attrs: FileAttributes { + size: Some(self.data.len() as u64), + ..FileAttributes::default() + }, + }) + } + + async fn read( + &mut self, + id: u32, + handle: String, + offset: u64, + len: u32, + ) -> Result { + if handle != "huge" { + return Err(SftpError::invalid_handle()); + } + self.observed + .reads + .lock() + .expect("reads mutex poisoned") + .push(len); + if offset >= self.data.len() as u64 { + return Err(SftpError::eof()); + } + + let start = offset as usize; + let end = start.saturating_add(len as usize).min(self.data.len()); + Ok(Data { + id, + data: self.data[start..end].to_vec(), + }) + } + + async fn write( + &mut self, + id: u32, + handle: String, + _offset: u64, + data: Vec, + ) -> Result { + if handle != "huge" { + return Err(SftpError::invalid_handle()); + } + self.observed + .writes + .lock() + .expect("writes mutex poisoned") + .push(data.len()); + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en-US".to_owned(), + }) + } + + async fn close(&mut self, id: u32, _handle: String) -> Result { + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en-US".to_owned(), + }) + } +} + +#[tokio::test] +async fn bssh_server_advertises_limits_extension() { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let raw = raw_session_with_handler(bssh_handler(dir.path().to_path_buf())).await; + + let version = raw.init().await.expect("init should return version"); + assert_eq!( + version + .extensions + .get(russh_sftp::extensions::LIMITS) + .map(String::as_str), + Some("1") + ); + + let limits = raw.limits().await.expect("limits extension should reply"); + assert_eq!( + limits.max_packet_len, + u64::from(russh_sftp::server::Config::default().max_client_packet_len) + ); + assert_eq!(limits.max_read_len, BSSH_MAX_READ_LEN); + assert_eq!(limits.max_write_len, BSSH_MAX_READ_LEN); + assert_eq!(limits.max_open_handles, BSSH_MAX_HANDLES); + + match raw + .extended("unsupported@example.com", Vec::new()) + .await + .expect("unsupported extension should receive a status reply") + { + Packet::Status(status) => assert_eq!(status.status_code, StatusCode::OpUnsupported), + other => panic!("expected unsupported extension status, got {other:?}"), + } + + raw.close_session() + .expect("raw SFTP session should shut down cleanly"); +} + +#[tokio::test] +async fn bssh_to_bssh_pipelined_round_trip_above_read_cap() { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let sftp = sftp_session_with_handler(bssh_handler(dir.path().to_path_buf())).await; + + for (name, size) in [("payload-300k.bin", 300_000), ("payload-2m.bin", 2_100_123)] { + let expected = payload(size); + upload_payload(&sftp, name, &expected).await; + let downloaded = download_payload(&sftp, name).await; + assert_eq!( + downloaded, expected, + "downloaded payload must match for {name}" + ); + } + + sftp.close() + .await + .expect("high-level SFTP session should shut down cleanly"); +} + +#[tokio::test] +async fn client_clamps_advertised_read_limit_to_packet_payload() { + let observed = ObservedChunks::default(); + let expected = payload((SMALL_CLIENT_PACKET_LEN as usize * 3) + 17); + let sftp = sftp_session_with_handler_and_config( + HugeAdvertisedLimitsHandler::new(expected.clone(), observed.clone()), + ClientConfig { + max_packet_len: SMALL_CLIENT_PACKET_LEN, + ..ClientConfig::default() + }, + ) + .await; + + let mut remote = sftp + .open("ignored.bin") + .await + .expect("read-limit file should open"); + let mut downloaded = Vec::new(); + remote + .read_to_writer_pipelined(&mut downloaded, 3) + .await + .expect("oversized advertised read limit should be clamped"); + + let expected_max = SMALL_CLIENT_PACKET_LEN - READ_OVERHEAD_LEN; + let read_lengths = observed.reads.lock().expect("reads mutex poisoned").clone(); + assert!( + !read_lengths.is_empty(), + "handler should observe at least one read" + ); + assert_eq!(read_lengths[0], expected_max); + assert!( + read_lengths.iter().all(|&len| len <= expected_max), + "read requests exceeded packet payload ceiling: {read_lengths:?}" + ); + assert_eq!(downloaded, expected); +} + +#[tokio::test] +async fn client_clamps_advertised_write_limit_to_packet_payload() { + let observed = ObservedChunks::default(); + let data = payload((SMALL_CLIENT_PACKET_LEN as usize * 3) + 29); + let sftp = sftp_session_with_handler_and_configs( + HugeAdvertisedLimitsHandler::new(Vec::new(), observed.clone()), + ClientConfig { + max_packet_len: SMALL_CLIENT_PACKET_LEN, + ..ClientConfig::default() + }, + russh_sftp::server::Config { + max_write_coalesce_len: 0, + ..russh_sftp::server::Config::default() + }, + ) + .await; + + let mut remote = sftp + .create("ignored.bin") + .await + .expect("write-limit file should open"); + let (mut reader, mut writer) = tokio::io::duplex(2 * SMALL_CLIENT_PACKET_LEN as usize); + let payload_for_writer = data.clone(); + let writer_task = tokio::spawn(async move { + writer + .write_all(&payload_for_writer) + .await + .expect("payload should feed upload reader"); + writer + .shutdown() + .await + .expect("payload reader should close"); + }); + + let written = remote + .write_all_pipelined(&mut reader, 3) + .await + .expect("oversized advertised write limit should be clamped"); + writer_task + .await + .expect("payload writer task should finish"); + + let expected_max = SMALL_CLIENT_PACKET_LEN - (WRITE_OVERHEAD_LEN + 4); + let write_lengths = observed + .writes + .lock() + .expect("writes mutex poisoned") + .clone(); + assert_eq!(written as usize, data.len()); + assert!( + !write_lengths.is_empty(), + "handler should observe at least one write" + ); + assert_eq!(write_lengths[0], expected_max as usize); + assert!( + write_lengths + .iter() + .all(|&len| len <= expected_max as usize), + "write requests exceeded packet payload ceiling: {write_lengths:?}" + ); +} + +struct ShortReadHandler { + data: Vec, + max_chunk: usize, +} + +impl ShortReadHandler { + fn new(data: Vec, max_chunk: usize) -> Self { + Self { data, max_chunk } + } +} + +impl russh_sftp::server::Handler for ShortReadHandler { + type Error = SftpError; + + fn unimplemented(&self) -> Self::Error { + SftpError::not_supported() + } + + async fn init( + &mut self, + _version: u32, + _extensions: HashMap, + ) -> Result { + Ok(Version::new()) + } + + async fn open( + &mut self, + id: u32, + _filename: String, + _pflags: OpenFlags, + _attrs: FileAttributes, + ) -> Result { + Ok(Handle { + id, + handle: "short".to_owned(), + }) + } + + async fn fstat(&mut self, id: u32, handle: String) -> Result { + let len = self.data.len() as u64; + if handle != "short" { + return Err(SftpError::invalid_handle()); + } + Ok(Attrs { + id, + attrs: FileAttributes { + size: Some(len), + ..FileAttributes::default() + }, + }) + } + + async fn read( + &mut self, + id: u32, + handle: String, + offset: u64, + len: u32, + ) -> Result { + if handle != "short" { + Err(SftpError::invalid_handle()) + } else if offset >= self.data.len() as u64 { + Err(SftpError::eof()) + } else { + let start = offset as usize; + let requested_end = start.saturating_add(len as usize).min(self.data.len()); + let short_end = start + .saturating_add(self.max_chunk) + .min(requested_end) + .min(self.data.len()); + Ok(Data { + id, + data: self.data[start..short_end].to_vec(), + }) + } + } + + async fn close(&mut self, id: u32, _handle: String) -> Result { + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en-US".to_owned(), + }) + } +} + +#[tokio::test] +async fn read_to_writer_retries_short_reads_before_known_eof() { + let expected = payload(300_000); + let sftp = sftp_session_with_handler(ShortReadHandler::new(expected.clone(), 57_000)).await; + let mut remote = sftp + .open("ignored.bin") + .await + .expect("short-read file should open"); + let mut downloaded = Vec::new(); + + let read = remote + .read_to_writer_pipelined(&mut downloaded, 4) + .await + .expect("short reads before EOF should be retried"); + + assert_eq!(read as usize, expected.len()); + assert_eq!(downloaded, expected); + remote + .close() + .await + .expect("short-read handle should close"); + sftp.close() + .await + .expect("short-read session should shut down cleanly"); +} + +struct EmptyDataBeforeEofHandler { + data: Vec, +} + +impl russh_sftp::server::Handler for EmptyDataBeforeEofHandler { + type Error = SftpError; + + fn unimplemented(&self) -> Self::Error { + SftpError::not_supported() + } + + async fn init( + &mut self, + _version: u32, + _extensions: HashMap, + ) -> Result { + Ok(Version::new()) + } + + async fn open( + &mut self, + id: u32, + _filename: String, + _pflags: OpenFlags, + _attrs: FileAttributes, + ) -> Result { + Ok(Handle { + id, + handle: "empty".to_owned(), + }) + } + + async fn fstat(&mut self, id: u32, handle: String) -> Result { + if handle != "empty" { + return Err(SftpError::invalid_handle()); + } + Ok(Attrs { + id, + attrs: FileAttributes { + size: Some(self.data.len() as u64), + ..FileAttributes::default() + }, + }) + } + + async fn read( + &mut self, + id: u32, + handle: String, + offset: u64, + _len: u32, + ) -> Result { + if handle != "empty" { + return Err(SftpError::invalid_handle()); + } + if offset == 0 { + return Ok(Data { + id, + data: Vec::new(), + }); + } + Err(SftpError::eof()) + } + + async fn close(&mut self, id: u32, _handle: String) -> Result { + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en-US".to_owned(), + }) + } +} + +#[tokio::test] +async fn read_to_writer_rejects_empty_data_before_known_eof() { + let expected = payload(1024); + let sftp = sftp_session_with_handler(EmptyDataBeforeEofHandler { data: expected }).await; + let mut remote = sftp + .open("ignored.bin") + .await + .expect("empty-read file should open"); + let mut downloaded = Vec::new(); + + let err = remote + .read_to_writer_pipelined(&mut downloaded, 4) + .await + .expect_err("empty DATA before known EOF must not truncate successfully"); + + assert!( + err.to_string() + .contains("unexpected empty read before file size"), + "unexpected error: {err}" + ); + assert!(downloaded.is_empty()); + remote + .close() + .await + .expect("empty-read handle should close"); + sftp.close() + .await + .expect("empty-read session should shut down cleanly"); +}