fix: negotiate SFTP read limits

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

crates/bssh-russh-sftp/src/client/fs/file.rs

@@ -71,6 +71,59 @@ impl File {
         }
     }
 
+    fn effective_read_len(&self) -> SftpResult<usize> {
+        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<usize> {
+        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<Metadata> {
         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<u64, Vec<u8>> = BTreeMap::new();
         let mut in_flight = FuturesUnordered::new();
         let mut eof = false;
+        let read_request = |session: Arc<RawSftpSession>, 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,32 +285,43 @@ 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);
             }
 
             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();

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,

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<u8>,
+    ) -> Result<Packet, Self::Error> {
+        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.