Refuse channels the server opens back, and type the startup command into the shell

core/src/connection.rs

@@ -8,7 +8,7 @@ use russh::keys::{
     decode_secret_key, Algorithm, EcdsaCurve, HashAlg, PrivateKeyWithHashAlg,
     PublicKeyOrCertificate,
 };
-use russh::{mac, Preferred};
+use russh::{mac, Channel, ChannelId, Preferred};
 use std::borrow::Cow;
 use std::collections::HashSet;
 use std::sync::{Arc, OnceLock, PoisonError};
@@ -169,6 +169,115 @@ impl client::Handler for SshHandler {
         }
         kh.verify(&self.host, self.port, server_public_key, self.host_key_hash)
     }
+
+    // Clavyn never requests agent or X11 forwarding, remote port forwards or
+    // anything else a server could legitimately open a channel back for, so
+    // every server-initiated channel is refused, types russh has no hook for
+    // included. Each refusal is spelled out here rather than left to russh's
+    // defaults, so it does not move with a dependency upgrade: those defaults
+    // accept every typed channel, and an accepted channel nobody reads stays in
+    // the connection's channel table for as long as the session lives, so a
+    // server could keep opening them for free.
+
+    async fn should_accept_unknown_server_channel(
+        &mut self,
+        _id: ChannelId,
+        _channel_type: &str,
+    ) -> bool {
+        false
+    }
+
+    async fn server_channel_open_unknown(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_forwarded_tcpip(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        _connected_address: &str,
+        _connected_port: u32,
+        _originator_address: &str,
+        _originator_port: u32,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_forwarded_streamlocal(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        _socket_path: &str,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_agent_forward(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_session(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_direct_tcpip(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        _host_to_connect: &str,
+        _port_to_connect: u32,
+        _originator_address: &str,
+        _originator_port: u32,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_direct_streamlocal(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        _socket_path: &str,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+
+    async fn server_channel_open_x11(
+        &mut self,
+        _channel: Channel<client::Msg>,
+        _originator_address: &str,
+        _originator_port: u32,
+        reply: client::ChannelOpenHandle,
+        _session: &mut client::Session,
+    ) -> std::result::Result<(), Self::Error> {
+        refuse(reply).await
+    }
+}
+
+/// Decline a channel the server tried to open.
+async fn refuse(reply: client::ChannelOpenHandle) -> Result<()> {
+    reply
+        .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
+        .await;
+    Ok(())
 }
 
 /// Extra context for a key-exchange failure, which russh surfaces as the bare
@@ -386,6 +495,27 @@ pub async fn connect(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use russh::client::Handler as _;
+    use russh::keys::ssh_key::encoding::Decode;
+
+    /// A channel type russh has no hook for is refused whatever russh's own
+    /// default for it is. An end-to-end check cannot cover this, because a
+    /// russh server can only open the channel types russh knows.
+    #[tokio::test]
+    async fn channel_types_russh_has_no_hook_for_are_refused() {
+        let mut handler = SshHandler {
+            host: "must-not-connect.example.test".into(),
+            port: 22,
+            known_hosts: known_hosts(),
+            host_key_hash: None,
+        };
+        let id = ChannelId::decode(&mut &[0, 0, 0, 7][..]).unwrap();
+        assert!(
+            !handler
+                .should_accept_unknown_server_channel(id, "tun@openssh.com")
+                .await
+        );
+    }
 
     fn known_hosts() -> Arc<Mutex<KnownHosts>> {
         let path = std::env::temp_dir().join(format!(

core/src/session.rs

@@ -15,6 +15,22 @@ use tokio::sync::{mpsc, Mutex};
 /// dropped. A peer that has stopped reading never acknowledges one.
 const DISCONNECT_GRACE: Duration = Duration::from_secs(5);
 
+/// How long the shell's output has to pause before a startup command is typed.
+///
+/// The login banner arrives before the shell has run its startup files, and a
+/// shell still running them can read typed input meant for its prompt or have
+/// the terminal echo it ahead of the prompt. A shell prints its prompt last and
+/// then waits, so output that has started and then stopped for this long is
+/// taken as the prompt. It is longer than a typical interactive shell takes
+/// between the banner and its first prompt.
+const STARTUP_SETTLE: Duration = Duration::from_secs(1);
+
+/// Upper bound on waiting for the shell before the startup command is typed
+/// anyway, so a shell that prints nothing, or never stops printing, still gets
+/// it. Input sent in the meantime waits behind the command for at most this
+/// long.
+const STARTUP_WAIT_LIMIT: Duration = Duration::from_secs(5);
+
 type SessionMap = HashMap<String, Arc<SshSession>>;
 
 /// Callback invoked when a session receives data from the remote end.
@@ -126,8 +142,9 @@ struct SshSession {
     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<()>,
+    /// its own input. Shared so a startup command can hold its turn while it
+    /// waits for the shell.
+    writes: Arc<Mutex<()>>,
     resize_tx: mpsc::Sender<(u32, u32)>,
     /// Held so the id stays reserved while this entry exists.
     #[allow(dead_code)]
@@ -141,6 +158,21 @@ impl SshSession {
         let _ = closed.wait_for(|closed| *closed).await;
     }
 
+    /// Send `data` to this 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. A close ends the wait.
+    async fn send(&self, session_id: &str, data: &[u8]) -> Result<()> {
+        let send = async {
+            let _order = self.writes.lock().await;
+            self.handle.data(self.channel_id, data.to_vec()).await
+        };
+        tokio::select! {
+            sent = send => sent.map_err(|_| CoreError::Ssh("write failed".into())),
+            () = self.closed() => Err(CoreError::SessionNotFound(session_id.to_string())),
+        }
+    }
+
     /// 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.
@@ -232,29 +264,39 @@ impl SessionManager {
             .await
             .map_err(|e| CoreError::Ssh(format!("request shell: {e}")))?;
 
-        // Run startup command if set
-        if let Some(cmd) = &host.startup_command {
-            if !cmd.is_empty() {
-                channel
-                    .exec(false, cmd.clone())
-                    .await
-                    .map_err(|e| CoreError::Ssh(format!("startup exec: {e}")))?;
-            }
-        }
-
         let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32);
 
         let held = claim.commit();
         let session = Arc::new(SshSession {
             handle: Arc::new(handle),
             closed: tokio::sync::watch::Sender::new(false),
             channel_id,
-            writes: Mutex::new(()),
+            writes: Arc::new(Mutex::new(())),
             resize_tx,
             claim: held.clone(),
         });
 
-        self.sessions.lock().await.insert(session_id.clone(), session);
+        // A startup command takes the first turn to write before the session
+        // is reachable, so nothing written to it can overtake the command. The
+        // reader reports each batch of output for as long as it waits.
+        let (mut output_seen, startup) =
+            match host.startup_command.as_deref().filter(|cmd| !cmd.is_empty()) {
+                Some(command) => {
+                    let (seen, output) = tokio::sync::watch::channel(());
+                    let startup = StartupCommand {
+                        command: command.to_string(),
+                        turn: session.writes.clone().lock_owned().await,
+                        output,
+                    };
+                    (Some(seen), Some(startup))
+                }
+                None => (None, None),
+            };
+
+        self.sessions
+            .lock()
+            .await
+            .insert(session_id.clone(), session.clone());
         let sessions = self.sessions.clone();
 
         // Spawn the reading task
@@ -284,6 +326,11 @@ impl SessionManager {
                         match msg {
                             Some(ChannelMsg::Data { data })
                             | Some(ChannelMsg::ExtendedData { data, .. }) => {
+                                // Nobody listens once the startup command has
+                                // stopped waiting, and the sender goes with it.
+                                if output_seen.as_ref().is_some_and(|seen| seen.send(()).is_err()) {
+                                    output_seen = None;
+                                }
                                 if batcher.push(&data, Instant::now()) {
                                     flush(&mut batcher);
                                 }
@@ -323,6 +370,15 @@ impl SessionManager {
             }
         });
 
+        // A channel takes only one shell, exec or subsystem request, and
+        // OpenSSH ignores an exec that arrives after the shell, so the startup
+        // command is typed into the shell instead. It waits for the shell on a
+        // task of its own, so the connect does not, and the session is usable
+        // whether or not the command goes out.
+        if let Some(startup) = startup {
+            tokio::spawn(type_startup_command(session, session_id, startup));
+        }
+
         Ok(())
     }
 
@@ -346,15 +402,7 @@ impl SessionManager {
     /// 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 session = self.get(session_id).await?;
-        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())),
-        }
+        self.get(session_id).await?.send(session_id, data).await
     }
 
     /// Resize a session's terminal.
@@ -436,6 +484,72 @@ async fn disconnect(handle: &Handle<connection::SshHandler>) {
     .await;
 }
 
+/// A host's startup command, waiting for its shell to be ready for input.
+struct StartupCommand {
+    command: String,
+    /// The session's first turn to write. Input sent while the command waits
+    /// queues behind it and reaches the shell after the command, in order.
+    turn: tokio::sync::OwnedMutexGuard<()>,
+    /// Marked by the reader each time the shell prints something; closed when
+    /// the reader ends.
+    output: tokio::sync::watch::Receiver<()>,
+}
+
+/// Whether the shell is ready for typed input: `true` once its output has
+/// started and then paused for [`STARTUP_SETTLE`], or once
+/// [`STARTUP_WAIT_LIMIT`] has passed, and `false` if the output ends first,
+/// which means the session is ending.
+async fn shell_settled(mut output: tokio::sync::watch::Receiver<()>) -> bool {
+    let settled = async {
+        if output.changed().await.is_err() {
+            return false;
+        }
+        loop {
+            match tokio::time::timeout(STARTUP_SETTLE, output.changed()).await {
+                Err(_) => return true,
+                Ok(Ok(())) => {}
+                Ok(Err(_)) => return false,
+            }
+        }
+    };
+    tokio::time::timeout(STARTUP_WAIT_LIMIT, settled)
+        .await
+        .unwrap_or(true)
+}
+
+/// Type `startup.command` into the session's shell once [`shell_settled`]
+/// says the shell is ready, followed by `\r`, the byte Enter sends.
+///
+/// It goes to `session` itself, not to whatever the id names by then: a shell
+/// that exits at once frees the id, and a session that takes it next must
+/// never receive this one's input. A close ends the wait and releases the
+/// write turn.
+async fn type_startup_command(session: Arc<SshSession>, session_id: String, startup: StartupCommand) {
+    let StartupCommand {
+        command,
+        turn,
+        output,
+    } = startup;
+    let typed = async {
+        let _turn = turn;
+        if !shell_settled(output).await {
+            return Ok(());
+        }
+        session
+            .handle
+            .data(session.channel_id, format!("{command}\r").into_bytes())
+            .await
+    };
+    tokio::select! {
+        sent = typed => {
+            if sent.is_err() {
+                tracing::warn!("startup command for session {session_id} was not sent: the connection is gone");
+            }
+        }
+        () = session.closed() => {}
+    }
+}
+
 #[cfg(test)]
 mod claim_tests {
     use super::SessionManager;