fix(sftp): say why SFTP would not start instead of "Timeout"

Every way an SFTP channel can fail before the protocol starts ended the
same way: the library's 10s request timeout, shown in the Files panel as
"sftp init failed: Timeout". Each has its own cause and fix on the server:

- sshd without a `Subsystem sftp` line answers the request with
  CHANNEL_FAILURE. russh's request_subsystem only sends the request, so
  the refusal went unread and the INIT was sent into a channel nobody
  was listening on.
- A `Subsystem` line naming an sftp-server that is not installed: the
  channel is accepted and then closes without a byte.
- A shell startup file that prints on non-interactive logins puts its
  text where the version reply belongs; read as a packet length it asks
  for gigabytes that never arrive.

The subsystem reply is now read (bounded at 20s), and the first bytes
back are checked to be the header of an SSH_FXP_VERSION packet before
the stream is handed to the library. Each case is reported at once with
what to change on the server; text ahead of SFTP is quoted.

Fixes #1126

crates/tty7-core/src/daemon/ssh/sftp.rs

@@ -433,8 +433,23 @@ fn is_transport_failure(msg: &str) -> bool {
     MARKERS.iter().any(|m| msg.contains(m))
 }
 
+#[cfg(not(test))]
+const SUBSYSTEM_REPLY_TIMEOUT: Duration = Duration::from_secs(20);
+#[cfg(test)]
+const SUBSYSTEM_REPLY_TIMEOUT: Duration = Duration::from_secs(2);
+
+const SFTP_REFUSED: &str = "the server refused to start SFTP — its sshd_config needs a \
+     `Subsystem sftp` line";
+const SFTP_EXITED: &str = "the server's SFTP program exited before it answered — is \
+     sftp-server installed where its sshd_config `Subsystem sftp` line points?";
+
+/// SFTP failures that happen before the protocol starts all used to surface
+/// as the library's request timeout, ten seconds later, as "Timeout". Each
+/// has its own cause on the server and its own fix, so each is told apart
+/// here: the subsystem request's answer is read instead of ignored, and the
+/// first bytes back are checked to be an SFTP version reply.
 async fn open_sftp(conn: &Arc<SshConnection>) -> Result<Arc<SftpSession>, String> {
-    let channel = conn
+    let mut channel = conn
         .open_session_channel()
         .await
         .map_err(|e| format!("open sftp channel failed: {e}"))?;
@@ -442,10 +457,156 @@ async fn open_sftp(conn: &Arc<SshConnection>) -> Result<Arc<SftpSession>, String
         .request_subsystem(true, "sftp")
         .await
         .map_err(|e| format!("sftp subsystem request failed: {e}"))?;
-    let sftp = SftpSession::new(channel.into_stream())
-        .await
-        .map_err(|e| format!("sftp init failed: {e}"))?;
-    Ok(Arc::new(sftp))
+    match tokio::time::timeout(SUBSYSTEM_REPLY_TIMEOUT, subsystem_reply(&mut channel)).await {
+        Ok(reply) => reply?,
+        Err(_) => return Err("sftp subsystem request got no answer: Timeout".into()),
+    }
+    let (verdict_tx, verdict_rx) = tokio::sync::oneshot::channel();
+    let stream = FirstPacketProbe::new(channel.into_stream(), verdict_tx);
+    tokio::select! {
+        sftp = SftpSession::new(stream) => {
+            sftp.map(Arc::new).map_err(|e| format!("sftp init failed: {e}"))
+        }
+        Ok(reason) = verdict_rx => Err(reason),
+    }
+}
+
+async fn subsystem_reply(channel: &mut russh::Channel<russh::client::Msg>) -> Result<(), String> {
+    use russh::ChannelMsg;
+    loop {
+        match channel.wait().await {
+            Some(ChannelMsg::Success) => return Ok(()),
+            Some(ChannelMsg::Failure) => return Err(SFTP_REFUSED.into()),
+            Some(ChannelMsg::Data { data }) => return Err(not_sftp(&data)),
+            Some(ChannelMsg::Eof | ChannelMsg::Close | ChannelMsg::ExitStatus { .. }) | None => {
+                return Err(SFTP_EXITED.into());
+            }
+            Some(_) => {}
+        }
+    }
+}
+
+fn not_sftp(bytes: &[u8]) -> String {
+    let text = String::from_utf8_lossy(bytes);
+    let line: String = text
+        .lines()
+        .map(str::trim)
+        .find(|l| !l.is_empty())
+        .unwrap_or("")
+        .chars()
+        .filter(|c| !c.is_control())
+        .take(80)
+        .collect();
+    format!(
+        "the server printed text where SFTP should start (\"{line}\") — a shell startup \
+         file that prints on non-interactive logins breaks SFTP; guard it with an \
+         interactive-shell check"
+    )
+}
+
+/// The largest version reply worth believing: the library's own default cap
+/// on a packet. Text read as a length prefix lands far above it.
+const MAX_VERSION_PACKET: u32 = 256 * 1024;
+const SSH_FXP_VERSION: u8 = 2;
+
+/// Passes a channel through to the SFTP library once its first five bytes
+/// read as the header of a version reply. Anything else — text, or the end
+/// of the stream — is reported on `verdict` and read as end of stream, since
+/// the library itself would only wait out its request timeout on it.
+struct FirstPacketProbe<S> {
+    inner: S,
+    head: Vec<u8>,
+    passed: bool,
+    verdict: Option<tokio::sync::oneshot::Sender<String>>,
+}
+
+impl<S> FirstPacketProbe<S> {
+    fn new(inner: S, verdict: tokio::sync::oneshot::Sender<String>) -> Self {
+        Self {
+            inner,
+            head: Vec::new(),
+            passed: false,
+            verdict: Some(verdict),
+        }
+    }
+
+    fn reject(&mut self, reason: String) {
+        if let Some(tx) = self.verdict.take() {
+            let _ = tx.send(reason);
+        }
+    }
+}
+
+fn is_version_header(head: &[u8]) -> bool {
+    let len = u32::from_be_bytes([head[0], head[1], head[2], head[3]]);
+    (1..=MAX_VERSION_PACKET).contains(&len) && head[4] == SSH_FXP_VERSION
+}
+
+impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for FirstPacketProbe<S> {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        use std::task::Poll;
+        let this = self.get_mut();
+        if !this.passed {
+            if this.verdict.is_none() {
+                return Poll::Ready(Ok(()));
+            }
+            while this.head.len() < 5 {
+                let mut chunk = [0u8; 512];
+                let mut rb = tokio::io::ReadBuf::new(&mut chunk);
+                match std::pin::Pin::new(&mut this.inner).poll_read(cx, &mut rb) {
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
+                    Poll::Ready(Ok(())) if rb.filled().is_empty() => {
+                        this.reject(SFTP_EXITED.into());
+                        return Poll::Ready(Ok(()));
+                    }
+                    Poll::Ready(Ok(())) => this.head.extend_from_slice(rb.filled()),
+                }
+            }
+            if !is_version_header(&this.head) {
+                let reason = not_sftp(&this.head);
+                this.reject(reason);
+                return Poll::Ready(Ok(()));
+            }
+            this.passed = true;
+            this.verdict = None;
+        }
+        if !this.head.is_empty() {
+            let n = this.head.len().min(buf.remaining());
+            buf.put_slice(&this.head[..n]);
+            this.head.drain(..n);
+            return Poll::Ready(Ok(()));
+        }
+        std::pin::Pin::new(&mut this.inner).poll_read(cx, buf)
+    }
+}
+
+impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for FirstPacketProbe<S> {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        data: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        std::pin::Pin::new(&mut self.get_mut().inner).poll_write(cx, data)
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::pin::Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
+    }
 }
 
 async fn list_dir(sftp: &SftpSession, path: &str) -> Result<Vec<SftpEntry>, String> {
@@ -951,8 +1112,53 @@ fn preserve_mode(_path: &Path, _mode: Option<u32>) {}
 
 #[cfg(test)]
 mod tests {
+    use super::super::test_support::{FakeSshd, Subsystem};
     use super::*;
 
+    /// Opens SFTP on a server that answers as `subsystem` says, giving up
+    /// well inside the library's own 10s request timeout so a test that
+    /// waits that long fails instead of passing on the old message.
+    async fn start_sftp(subsystem: Subsystem) -> Result<Arc<SftpSession>, String> {
+        let sshd = FakeSshd::with_subsystem(subsystem).await;
+        tokio::time::timeout(Duration::from_secs(5), open_sftp(&sshd.conn))
+            .await
+            .unwrap_or_else(|_| Err("still waiting after 5s".into()))
+    }
+
+    #[tokio::test]
+    async fn a_server_that_speaks_sftp_opens_a_session() {
+        if let Err(e) = start_sftp(Subsystem::Sftp).await {
+            panic!("a working SFTP server must open: {e}");
+        }
+    }
+
+    /// #1126: sshd with no `Subsystem sftp` line answers the request with
+    /// CHANNEL_FAILURE. That answer used to go unread, so the Files panel sat
+    /// out the init timeout and then said only "Timeout".
+    #[tokio::test]
+    async fn a_refused_sftp_subsystem_is_reported_as_refused() {
+        let err = start_sftp(Subsystem::Refused).await.err().expect("refused");
+        assert!(err.contains("Subsystem sftp"), "{err}");
+    }
+
+    #[tokio::test]
+    async fn an_sftp_server_that_exits_unanswered_is_reported_as_such() {
+        let err = start_sftp(Subsystem::Exits).await.err().expect("exited");
+        assert!(err.contains("sftp-server"), "{err}");
+    }
+
+    /// A startup file that prints on a non-interactive login puts its text
+    /// where the SFTP version reply belongs; read as a packet length it asks
+    /// for gigabytes that never come.
+    #[tokio::test]
+    async fn text_ahead_of_sftp_is_reported_with_the_text() {
+        let err = start_sftp(Subsystem::Prints(b"Welcome to the box\r\n"))
+            .await
+            .err()
+            .expect("garbled");
+        assert!(err.contains("Welcome to the box"), "{err}");
+    }
+
     #[test]
     fn remote_join_handles_root_and_nested_and_slashes() {
         assert_eq!(remote_join("/", "file"), "/file");

crates/tty7-core/src/daemon/ssh/test_support.rs

@@ -6,6 +6,7 @@
 //! connection, accepts session channels up to a limit, answers `exec` the way
 //! the test asks, and counts what the client opened and closed.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
@@ -33,6 +34,21 @@ pub(crate) enum Exec {
     Hangs,
 }
 
+/// How the server answers a request for the `sftp` subsystem.
+#[derive(Clone, Copy)]
+pub(crate) enum Subsystem {
+    /// A working SFTP server behind the channel.
+    Sftp,
+    /// CHANNEL_FAILURE: sshd with no `Subsystem sftp` line.
+    Refused,
+    /// Success, then the program behind it exits without a byte: a
+    /// `Subsystem` line naming an sftp-server that is not installed.
+    Exits,
+    /// Success, then this text ahead of anything SFTP: a shell startup file
+    /// that prints on a non-interactive login.
+    Prints(&'static [u8]),
+}
+
 #[derive(Default)]
 struct Counts {
     opened: AtomicUsize,
@@ -42,6 +58,8 @@ struct Counts {
 
 struct Sshd {
     exec: Exec,
+    subsystem: Subsystem,
+    channels: HashMap<ChannelId, Channel<server::Msg>>,
     max_sessions: Option<usize>,
     counts: Arc<Counts>,
     /// `Some` for a server that has stopped answering: each open is held
@@ -59,7 +77,7 @@ impl server::Handler for Sshd {
 
     async fn channel_open_session(
         &mut self,
-        _channel: Channel<server::Msg>,
+        channel: Channel<server::Msg>,
         reply: ChannelOpenHandle,
         _session: &mut Session,
     ) -> Result<(), Self::Error> {
@@ -79,10 +97,39 @@ impl server::Handler for Sshd {
             return Ok(());
         }
         self.counts.opened.fetch_add(1, Ordering::SeqCst);
+        self.channels.insert(channel.id(), channel);
         reply.accept().await;
         Ok(())
     }
 
+    async fn subsystem_request(
+        &mut self,
+        channel: ChannelId,
+        _name: &str,
+        session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        match self.subsystem {
+            Subsystem::Refused => session.channel_failure(channel)?,
+            Subsystem::Exits => {
+                session.channel_success(channel)?;
+                session.exit_status_request(channel, 127)?;
+                session.eof(channel)?;
+                session.close(channel)?;
+            }
+            Subsystem::Prints(text) => {
+                session.channel_success(channel)?;
+                session.data(channel, text)?;
+            }
+            Subsystem::Sftp => {
+                session.channel_success(channel)?;
+                if let Some(ch) = self.channels.remove(&channel) {
+                    russh_sftp::server::run(ch.into_stream(), SftpServer).await;
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Held the same way when the server has stopped answering — which, from
     /// the client's side, is indistinguishable from a server still waiting
     /// on its own connect to a target that drops SYNs.
@@ -138,6 +185,16 @@ impl server::Handler for Sshd {
     }
 }
 
+struct SftpServer;
+
+impl russh_sftp::server::Handler for SftpServer {
+    type Error = russh_sftp::protocol::StatusCode;
+
+    fn unimplemented(&self) -> Self::Error {
+        russh_sftp::protocol::StatusCode::OpUnsupported
+    }
+}
+
 /// One connection to a server in this process, and what the server counted.
 pub(crate) struct FakeSshd {
     pub(crate) conn: Arc<SshConnection>,
@@ -146,16 +203,26 @@ pub(crate) struct FakeSshd {
 
 impl FakeSshd {
     pub(crate) async fn connect(exec: Exec, max_sessions: Option<usize>) -> FakeSshd {
-        Self::start(exec, max_sessions, false).await
+        Self::start(exec, Subsystem::Sftp, max_sessions, false).await
+    }
+
+    /// A server that answers the `sftp` subsystem request as `subsystem` says.
+    pub(crate) async fn with_subsystem(subsystem: Subsystem) -> FakeSshd {
+        Self::start(Exec::Hangs, subsystem, None, false).await
     }
 
     /// A server that completes the handshake and then never answers a channel
     /// open: the connection stays up and nothing on it errors.
     pub(crate) async fn connect_unanswering() -> FakeSshd {
-        Self::start(Exec::Hangs, None, true).await
+        Self::start(Exec::Hangs, Subsystem::Sftp, None, true).await
     }
 
-    async fn start(exec: Exec, max_sessions: Option<usize>, unanswering: bool) -> FakeSshd {
+    async fn start(
+        exec: Exec,
+        subsystem: Subsystem,
+        max_sessions: Option<usize>,
+        unanswering: bool,
+    ) -> FakeSshd {
         let counts = Arc::new(Counts::default());
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
@@ -169,6 +236,8 @@ impl FakeSshd {
             .push(PrivateKey::from(Ed25519Keypair::from_seed(&[7; 32])));
         let handler = Sshd {
             exec,
+            subsystem,
+            channels: HashMap::new(),
             max_sessions,
             counts: Arc::clone(&counts),
             unanswered: unanswering.then(Vec::new),