From 0ba8d27261e656932a5382383eac3f0ce23d2a95 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Fri, 16 Oct 2026 22:46:34 +0000 Subject: [PATCH 1/6] Refuse channels the server opens back, and type the startup command into the shell russh's client handler accepts every channel a server opens towards the client: agent and X11 forwarding, forwarded and direct TCP/streamlocal, even a session. Clavyn requests none of those, and an accepted channel that nothing reads stays registered for the life of the connection, so a server could open them without limit. SshHandler now rejects them all. The host startup command was sent as an exec request after the shell request on the same channel. A channel takes one of shell, exec or subsystem, and OpenSSH ignores an exec on a channel that is already running a shell, so the command never ran. It is now written into the shell followed by a carriage return, as if typed. --- core/src/connection.rs | 92 +++++++++++++++- core/src/session.rs | 9 +- core/tests/server_requests.rs | 191 ++++++++++++++++++++++++++++++++++ 3 files changed, 288 insertions(+), 4 deletions(-) create mode 100644 core/tests/server_requests.rs diff --git a/core/src/connection.rs b/core/src/connection.rs index 7437f0e9..102b7d64 100644 --- a/core/src/connection.rs +++ b/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, Preferred}; use std::borrow::Cow; use std::collections::HashSet; use std::sync::{Arc, OnceLock, PoisonError}; @@ -169,6 +169,96 @@ 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. russh's defaults accept them, + // 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 server_channel_open_forwarded_tcpip( + &mut self, + _channel: Channel, + _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, + _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, + 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, + 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, + _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, + _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, + _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 diff --git a/core/src/session.rs b/core/src/session.rs index 9c776cda..a6793307 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -232,13 +232,16 @@ impl SessionManager { .await .map_err(|e| CoreError::Ssh(format!("request shell: {e}")))?; - // Run startup command if set + // The startup command is typed into the shell that was just started. + // A channel takes only one shell, exec or subsystem request, and + // OpenSSH ignores an exec that arrives after the shell, so it cannot + // be sent as a request of its own. if let Some(cmd) = &host.startup_command { if !cmd.is_empty() { channel - .exec(false, cmd.clone()) + .data(format!("{cmd}\r").as_bytes()) .await - .map_err(|e| CoreError::Ssh(format!("startup exec: {e}")))?; + .map_err(|e| CoreError::Ssh(format!("startup command: {e}")))?; } } diff --git a/core/tests/server_requests.rs b/core/tests/server_requests.rs new file mode 100644 index 00000000..e32a63f6 --- /dev/null +++ b/core/tests/server_requests.rs @@ -0,0 +1,191 @@ +//! What a connected server can ask of the client, checked against an +//! in-process russh server on loopback: channels it tries to open back are +//! refused, and a host's startup command reaches the shell it was meant for. + +use clavyn_core::host::Host; +use clavyn_core::known_hosts::KnownHosts; +use clavyn_core::session::SessionManager; +use russh::keys::{Algorithm, PrivateKey}; +use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; +use russh::{Channel, ChannelId}; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; +use tokio::net::TcpListener; +use tokio::sync::{mpsc, Mutex}; + +const USER: &str = "deploy"; +const PASSWORD: &str = "correct password"; + +/// What the server saw, reported back to the test. +#[derive(Debug)] +enum Seen { + /// Bytes the client sent on the shell channel. + Input(Vec), + /// Whether the client accepted each channel the server tried to open. + Opened { kind: &'static str, accepted: bool }, +} + +#[derive(Clone)] +struct TestServer { + seen: mpsc::UnboundedSender, + /// Try to open channels back to the client once its shell starts. + open_back: bool, +} + +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> { + reply.accept().await; + Ok(()) + } + + async fn shell_request( + &mut self, + _channel: ChannelId, + session: &mut Session, + ) -> Result<(), Self::Error> { + if self.open_back { + let handle = session.handle(); + let seen = self.seen.clone(); + tokio::spawn(async move { + let agent = handle.channel_open_agent().await.is_ok(); + let _ = seen.send(Seen::Opened { kind: "agent", accepted: agent }); + let x11 = handle.channel_open_x11("127.0.0.1", 6010).await.is_ok(); + let _ = seen.send(Seen::Opened { kind: "x11", accepted: x11 }); + let forwarded = handle + .channel_open_forwarded_tcpip("127.0.0.1", 8080, "127.0.0.1", 50000) + .await + .is_ok(); + let _ = seen.send(Seen::Opened { + kind: "forwarded-tcpip", + accepted: forwarded, + }); + let session = handle.channel_open_session().await.is_ok(); + let _ = seen.send(Seen::Opened { kind: "session", accepted: session }); + }); + } + Ok(()) + } + + async fn data( + &mut self, + _channel: ChannelId, + data: &[u8], + _session: &mut Session, + ) -> Result<(), Self::Error> { + let _ = self.seen.send(Seen::Input(data.to_vec())); + Ok(()) + } +} + +async fn start_server(open_back: bool) -> (u16, mpsc::UnboundedReceiver) { + let (seen, events) = mpsc::unbounded_channel(); + 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 handler = TestServer { seen, open_back }; + 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(); + let handler = handler.clone(); + tokio::spawn(async move { + if let Ok(running) = server::run_stream(config, stream, handler).await { + let _ = running.await; + } + }); + } + }); + (port, events) +} + +fn host(port: u16, startup_command: Option<&str>) -> 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"}}, + "startup_command": startup_command, + "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"), + )) +} + +fn sessions() -> SessionManager { + SessionManager::new(Arc::new(|_, _| {}), Arc::new(|_, _| {})) +} + +async fn open_shell(sessions: &SessionManager, host: &Host, dir: &TempDir) { + let claim = sessions.claim("shell").expect("claim"); + sessions + .create_ssh_session(&claim, host, None, known_hosts(dir), None, None, Some(PASSWORD), 80, 24) + .await + .expect("ssh session"); +} + +async fn next(events: &mut mpsc::UnboundedReceiver) -> Seen { + tokio::time::timeout(Duration::from_secs(10), events.recv()) + .await + .expect("the server saw nothing") + .expect("server gone") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn channels_the_server_opens_back_are_refused() { + let dir = tempfile::tempdir().expect("temp dir"); + let (port, mut events) = start_server(true).await; + let sessions = sessions(); + open_shell(&sessions, &host(port, None), &dir).await; + + for expected in ["agent", "x11", "forwarded-tcpip", "session"] { + match next(&mut events).await { + Seen::Opened { kind, accepted } => { + assert_eq!(kind, expected); + assert!(!accepted, "the client accepted a {kind} channel"); + } + other => panic!("unexpected {other:?}"), + } + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_startup_command_is_typed_into_the_shell() { + let dir = tempfile::tempdir().expect("temp dir"); + let (port, mut events) = start_server(false).await; + let sessions = sessions(); + open_shell(&sessions, &host(port, Some("tmux attach")), &dir).await; + + match next(&mut events).await { + Seen::Input(bytes) => assert_eq!(bytes, b"tmux attach\r"), + other => panic!("unexpected {other:?}"), + } +} From 8bc2f93b1cc37fc4492cff1d2763251fcda03110 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 06:44:50 +0000 Subject: [PATCH 2/6] Type the startup command once the output reader is running The command was written before the reader task started, so a long command whose echo filled the receive window could stall the connect waiting on its own output. It now goes through the session's ordinary write path after the reader is up, and a command that cannot be sent is logged instead of failing a session that is otherwise usable. --- core/src/session.rs | 25 ++++++++++++------------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index a6793307..91e91d61 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -232,19 +232,6 @@ impl SessionManager { .await .map_err(|e| CoreError::Ssh(format!("request shell: {e}")))?; - // The startup command is typed into the shell that was just started. - // A channel takes only one shell, exec or subsystem request, and - // OpenSSH ignores an exec that arrives after the shell, so it cannot - // be sent as a request of its own. - if let Some(cmd) = &host.startup_command { - if !cmd.is_empty() { - channel - .data(format!("{cmd}\r").as_bytes()) - .await - .map_err(|e| CoreError::Ssh(format!("startup command: {e}")))?; - } - } - let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32); let held = claim.commit(); @@ -326,6 +313,18 @@ impl SessionManager { } }); + // The startup command is typed into the shell once the reader is + // running, so whatever the remote echoes back is drained while it is + // sent. A channel takes only one shell, exec or subsystem request, and + // OpenSSH ignores an exec that arrives after the shell, so it cannot be + // sent as a request of its own. The session is usable either way, so a + // command that cannot be sent does not fail the connect. + if let Some(cmd) = host.startup_command.as_deref().filter(|cmd| !cmd.is_empty()) { + if let Err(error) = self.write(&session_id, format!("{cmd}\r").as_bytes()).await { + tracing::warn!("startup command for session {session_id} was not sent: {error}"); + } + } + Ok(()) } From a720254e2c9e2113d04162b0e78e5e6b39c29015 Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:04:07 +0000 Subject: [PATCH 3/6] Refuse channel types russh has no hook for in Clavyn's own handler Unknown server-initiated channel types were refused only because russh's should_accept_unknown_server_channel happens to default to false. The dependency is a caret requirement, so a patch release could change that default or add a typed hook whose default accepts, reopening the leak the other refusals close. The handler now answers both unknown-type hooks itself, and a unit test pins the refusal regardless of russh's default. --- core/src/connection.rs | 50 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 45 insertions(+), 5 deletions(-) diff --git a/core/src/connection.rs b/core/src/connection.rs index 102b7d64..650b7266 100644 --- a/core/src/connection.rs +++ b/core/src/connection.rs @@ -8,7 +8,7 @@ use russh::keys::{ decode_secret_key, Algorithm, EcdsaCurve, HashAlg, PrivateKeyWithHashAlg, PublicKeyOrCertificate, }; -use russh::{mac, Channel, Preferred}; +use russh::{mac, Channel, ChannelId, Preferred}; use std::borrow::Cow; use std::collections::HashSet; use std::sync::{Arc, OnceLock, PoisonError}; @@ -172,10 +172,29 @@ impl client::Handler for SshHandler { // 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. russh's defaults accept them, - // 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. + // 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, + reply: client::ChannelOpenHandle, + _session: &mut client::Session, + ) -> std::result::Result<(), Self::Error> { + refuse(reply).await + } async fn server_channel_open_forwarded_tcpip( &mut self, @@ -476,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> { let path = std::env::temp_dir().join(format!( From 50a74a0bd810e88f690433f15461e913963ca6ca Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:04:07 +0000 Subject: [PATCH 4/6] Send the startup command to the session just built, not to its id The startup write looked the session up by id again after inserting it. A shell that exits at once frees the id, so a session that took it in that gap would receive this one's command. The write now goes through the Arc that was just inserted, which also keeps the id reserved until the send is over. write() shares the same send path. --- core/src/session.rs | 37 ++++++++++++++++++++++++++----------- 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index 91e91d61..b16516f7 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -141,6 +141,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. @@ -244,7 +259,10 @@ impl SessionManager { claim: held.clone(), }); - self.sessions.lock().await.insert(session_id.clone(), session); + self.sessions + .lock() + .await + .insert(session_id.clone(), session.clone()); let sessions = self.sessions.clone(); // Spawn the reading task @@ -319,8 +337,13 @@ impl SessionManager { // OpenSSH ignores an exec that arrives after the shell, so it cannot be // sent as a request of its own. The session is usable either way, so a // command that cannot be sent does not fail the connect. + // + // It is sent to the session built above, not to whatever the id names + // by now: a shell that exits at once frees the id, and a session that + // takes it next must never receive this one's input. Holding `session` + // also keeps the id reserved until the send is over. if let Some(cmd) = host.startup_command.as_deref().filter(|cmd| !cmd.is_empty()) { - if let Err(error) = self.write(&session_id, format!("{cmd}\r").as_bytes()).await { + if let Err(error) = session.send(&session_id, format!("{cmd}\r").as_bytes()).await { tracing::warn!("startup command for session {session_id} was not sent: {error}"); } } @@ -348,15 +371,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. From 27c3cf6f5a2639f5704eda513a40c94800308e32 Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:04:07 +0000 Subject: [PATCH 5/6] Cover every channel type a server can open back, on a shared test server The refusal test opened only agent, x11, forwarded-tcpip and session channels, so dropping the forwarded-streamlocal, direct-tcpip or direct-streamlocal override would have passed unnoticed. It now opens all seven types russh can open back. The in-process loopback server, its config, the password host and the known-hosts setup were copied across server_requests, session_isolation and ssh_roundtrip. They now live in tests/common/loopback.rs. --- core/tests/common/loopback.rs | 80 +++++++++++++++++++++ core/tests/common/mod.rs | 2 + core/tests/server_requests.rs | 124 +++++++++++++++----------------- core/tests/session_isolation.rs | 73 ++++--------------- core/tests/ssh_roundtrip.rs | 55 +------------- 5 files changed, 154 insertions(+), 180 deletions(-) create mode 100644 core/tests/common/loopback.rs diff --git a/core/tests/common/loopback.rs b/core/tests/common/loopback.rs new file mode 100644 index 00000000..9976b543 --- /dev/null +++ b/core/tests/common/loopback.rs @@ -0,0 +1,80 @@ +//! An in-process russh server on loopback, and the client-side pieces a test +//! needs to connect to it the way the app does. + +use clavyn_core::host::Host; +use clavyn_core::known_hosts::KnownHosts; +use russh::keys::{Algorithm, PrivateKey}; +use russh::server; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; +use tokio::net::TcpListener; +use tokio::sync::Mutex; + +/// The account a loopback server's password check accepts. +pub const USER: &str = "deploy"; +pub const PASSWORD: &str = "correct password"; + +/// A fresh Ed25519 host key. +pub fn random_host_key() -> PrivateKey { + let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); + PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("generate host key") +} + +/// Server settings for `host_key`. A rejected authentication is answered at +/// once rather than after russh's usual delay, so tests that expect one do not +/// wait it out. +pub fn server_config(host_key: PrivateKey) -> server::Config { + server::Config { + keys: vec![host_key], + auth_rejection_time: Duration::from_millis(10), + auth_rejection_time_initial: Some(Duration::ZERO), + ..Default::default() + } +} + +/// Serves every connection on a fresh loopback port with a copy of `handler` +/// and returns that port. +pub async fn serve(config: server::Config, handler: H) -> u16 +where + H: server::Handler + Clone + Send + 'static, +{ + let config = Arc::new(config); + 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(); + let handler = handler.clone(); + tokio::spawn(async move { + if let Ok(running) = server::run_stream(config, stream, handler).await { + let _ = running.await; + } + }); + } + }); + port +} + +/// A saved host pointing at the loopback server on `port`, signing in as +/// [`USER`] with a password. +pub fn password_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") +} + +/// An empty known-hosts store in `dir`, so the first connection pins the +/// loopback server's key. +pub fn known_hosts(dir: &TempDir) -> Arc> { + Arc::new(Mutex::new( + KnownHosts::load(dir.path().join("known_hosts.json")).expect("load known hosts"), + )) +} diff --git a/core/tests/common/mod.rs b/core/tests/common/mod.rs index 32804452..41ecc788 100644 --- a/core/tests/common/mod.rs +++ b/core/tests/common/mod.rs @@ -8,6 +8,8 @@ // Every test crate compiles this module on its own and uses only part of it. #![allow(dead_code)] +pub mod loopback; + use clavyn_core::keys::{generate_ed25519, parse_openssh_private}; use russh::keys::ssh_key::{LineEnding, PrivateKey}; use std::path::PathBuf; diff --git a/core/tests/server_requests.rs b/core/tests/server_requests.rs index e32a63f6..7b262d9a 100644 --- a/core/tests/server_requests.rs +++ b/core/tests/server_requests.rs @@ -2,20 +2,18 @@ //! in-process russh server on loopback: channels it tries to open back are //! refused, and a host's startup command reaches the shell it was meant for. +mod common; + use clavyn_core::host::Host; -use clavyn_core::known_hosts::KnownHosts; use clavyn_core::session::SessionManager; -use russh::keys::{Algorithm, PrivateKey}; +use common::loopback::{known_hosts, password_host, random_host_key, serve, server_config}; +use common::loopback::{PASSWORD, USER}; use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; use russh::{Channel, ChannelId}; use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; -use tokio::net::TcpListener; -use tokio::sync::{mpsc, Mutex}; - -const USER: &str = "deploy"; -const PASSWORD: &str = "correct password"; +use tokio::sync::mpsc; /// What the server saw, reported back to the test. #[derive(Debug)] @@ -54,6 +52,8 @@ impl server::Handler for TestServer { Ok(()) } + /// Opens one channel of every type russh can open back, in the order + /// [`OPENED_BACK`] lists them. async fn shell_request( &mut self, _channel: ChannelId, @@ -63,20 +63,40 @@ impl server::Handler for TestServer { let handle = session.handle(); let seen = self.seen.clone(); tokio::spawn(async move { - let agent = handle.channel_open_agent().await.is_ok(); - let _ = seen.send(Seen::Opened { kind: "agent", accepted: agent }); - let x11 = handle.channel_open_x11("127.0.0.1", 6010).await.is_ok(); - let _ = seen.send(Seen::Opened { kind: "x11", accepted: x11 }); - let forwarded = handle - .channel_open_forwarded_tcpip("127.0.0.1", 8080, "127.0.0.1", 50000) - .await - .is_ok(); - let _ = seen.send(Seen::Opened { - kind: "forwarded-tcpip", - accepted: forwarded, - }); - let session = handle.channel_open_session().await.is_ok(); - let _ = seen.send(Seen::Opened { kind: "session", accepted: session }); + let report = |kind, accepted| { + let _ = seen.send(Seen::Opened { kind, accepted }); + }; + report("agent", handle.channel_open_agent().await.is_ok()); + report("x11", handle.channel_open_x11("127.0.0.1", 6010).await.is_ok()); + report( + "forwarded-tcpip", + handle + .channel_open_forwarded_tcpip("127.0.0.1", 8080, "127.0.0.1", 50000) + .await + .is_ok(), + ); + report( + "forwarded-streamlocal", + handle + .channel_open_forwarded_streamlocal("/tmp/forwarded.sock") + .await + .is_ok(), + ); + report( + "direct-tcpip", + handle + .channel_open_direct_tcpip("127.0.0.1", 22, "127.0.0.1", 50001) + .await + .is_ok(), + ); + report( + "direct-streamlocal", + handle + .channel_open_direct_streamlocal("/tmp/direct.sock") + .await + .is_ok(), + ); + report("session", handle.channel_open_session().await.is_ok()); }); } Ok(()) @@ -93,53 +113,23 @@ impl server::Handler for TestServer { } } +/// Every channel type the test server opens back, in the order it opens them. +const OPENED_BACK: [&str; 7] = [ + "agent", + "x11", + "forwarded-tcpip", + "forwarded-streamlocal", + "direct-tcpip", + "direct-streamlocal", + "session", +]; + async fn start_server(open_back: bool) -> (u16, mpsc::UnboundedReceiver) { let (seen, events) = mpsc::unbounded_channel(); - 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 handler = TestServer { seen, open_back }; - 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(); - let handler = handler.clone(); - tokio::spawn(async move { - if let Ok(running) = server::run_stream(config, stream, handler).await { - let _ = running.await; - } - }); - } - }); + let port = serve(server_config(random_host_key()), TestServer { seen, open_back }).await; (port, events) } -fn host(port: u16, startup_command: Option<&str>) -> 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"}}, - "startup_command": startup_command, - "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"), - )) -} - fn sessions() -> SessionManager { SessionManager::new(Arc::new(|_, _| {}), Arc::new(|_, _| {})) } @@ -164,9 +154,9 @@ async fn channels_the_server_opens_back_are_refused() { let dir = tempfile::tempdir().expect("temp dir"); let (port, mut events) = start_server(true).await; let sessions = sessions(); - open_shell(&sessions, &host(port, None), &dir).await; + open_shell(&sessions, &password_host(port), &dir).await; - for expected in ["agent", "x11", "forwarded-tcpip", "session"] { + for expected in OPENED_BACK { match next(&mut events).await { Seen::Opened { kind, accepted } => { assert_eq!(kind, expected); @@ -182,7 +172,9 @@ async fn the_startup_command_is_typed_into_the_shell() { let dir = tempfile::tempdir().expect("temp dir"); let (port, mut events) = start_server(false).await; let sessions = sessions(); - open_shell(&sessions, &host(port, Some("tmux attach")), &dir).await; + let mut host = password_host(port); + host.startup_command = Some("tmux attach".into()); + open_shell(&sessions, &host, &dir).await; match next(&mut events).await { Seen::Input(bytes) => assert_eq!(bytes, b"tmux attach\r"), diff --git a/core/tests/session_isolation.rs b/core/tests/session_isolation.rs index 7362bc3d..29383bd6 100644 --- a/core/tests/session_isolation.rs +++ b/core/tests/session_isolation.rs @@ -5,23 +5,18 @@ //! 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; +mod common; + use clavyn_core::session::SessionManager; use clavyn_core::sftp::SftpManager; -use russh::keys::{Algorithm, PrivateKey}; +use common::loopback::{known_hosts, password_host, random_host_key, serve, server_config}; +use common::loopback::{PASSWORD, USER}; 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. @@ -153,58 +148,14 @@ 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 { - 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(); - let handler = handler.clone(); - tokio::spawn(async move { - if let Ok(running) = server::run_stream(config, stream, handler.clone()).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"), - )) + serve(server_config(random_host_key()), TestServer::default()).await } #[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 host = password_host(port); let pins = known_hosts(&dir); let manager = Arc::new(SftpManager::new()); for id in ["stuck", "other"] { @@ -249,7 +200,7 @@ async fn closing_one_sftp_session_does_not_wait_on_its_listing() { 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)) + .connect("stuck".into(), &password_host(port), None, known_hosts(&dir), None, None, Some(PASSWORD)) .await .expect("sftp connect"); @@ -289,7 +240,7 @@ async fn a_shell_that_exits_leaves_no_session_behind() { let claim = sessions.claim("shell").expect("claim"); sessions - .create_ssh_session(&claim, &host(port), None, known_hosts(&dir), None, None, Some(PASSWORD), 80, 24) + .create_ssh_session(&claim, &password_host(port), None, known_hosts(&dir), None, None, Some(PASSWORD), 80, 24) .await .expect("ssh session"); drop(claim); @@ -316,15 +267,15 @@ async fn a_shell_that_exits_leaves_no_session_behind() { #[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 { + let stalling = TestServer { stall_on_input: true, ..TestServer::default() - }) - .await; + }; + let port = serve(server_config(random_host_key()), stalling).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) + .create_ssh_session(&claim, &password_host(port), None, known_hosts(&dir), None, None, Some(PASSWORD), 80, 24) .await .expect("ssh session"); drop(claim); diff --git a/core/tests/ssh_roundtrip.rs b/core/tests/ssh_roundtrip.rs index 2f5160a6..c85dda1f 100644 --- a/core/tests/ssh_roundtrip.rs +++ b/core/tests/ssh_roundtrip.rs @@ -12,22 +12,17 @@ use clavyn_core::host::Host; use clavyn_core::keys::{ generate_ed25519, import_openssh_private, parse_openssh_private, KeyMeta, KeyType, }; -use clavyn_core::known_hosts::KnownHosts; use clavyn_core::vault::Vault; use clavyn_core::CoreError; use common::{copy_fixture, passphrase_protected_ed25519, PASSPHRASE}; +use common::loopback::{known_hosts, password_host, random_host_key, serve, server_config}; +use common::loopback::{PASSWORD, USER}; use russh::keys::{decode_secret_key, parse_public_key_base64, Algorithm, PrivateKey, PublicKey}; use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; use russh::{Channel, ChannelId, ChannelMsg, Preferred}; use std::borrow::Cow; -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"; const VAULT_PASSPHRASE: &str = "vault passphrase"; #[derive(Clone)] @@ -79,20 +74,6 @@ impl server::Handler for TestServer { } } -fn random_host_key() -> PrivateKey { - let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); - PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("generate host key") -} - -fn server_config(host_key: PrivateKey) -> server::Config { - server::Config { - keys: vec![host_key], - auth_rejection_time: Duration::from_millis(10), - auth_rejection_time_initial: Some(Duration::ZERO), - ..Default::default() - } -} - /// Serves every connection on a fresh loopback port and returns that port. async fn start_server(host_key: PrivateKey, handler: TestServer) -> u16 { serve(server_config(host_key), handler).await @@ -114,24 +95,6 @@ async fn start_server_offering( serve(config, handler).await } -async fn serve(config: server::Config, handler: TestServer) -> u16 { - let config = Arc::new(config); - 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(); - let handler = handler.clone(); - tokio::spawn(async move { - if let Ok(running) = server::run_stream(config, stream, handler).await { - let _ = running.await; - } - }); - } - }); - port -} - fn host(port: u16, auth: serde_json::Value, key_id: Option) -> Host { serde_json::from_value(serde_json::json!({ "id": "00000000-0000-0000-0000-000000000001", @@ -146,20 +109,6 @@ fn host(port: u16, auth: serde_json::Value, key_id: Option) -> Host { .expect("host") } -fn password_host(port: u16) -> Host { - host( - port, - serde_json::json!({"password": {"credential_key": "unused"}}), - None, - ) -} - -fn known_hosts(dir: &TempDir) -> Arc> { - Arc::new(Mutex::new( - KnownHosts::load(dir.path().join("known_hosts.json")).expect("load known hosts"), - )) -} - #[tokio::test] async fn password_auth_connects_and_pins_the_host_key() { let dir = tempfile::tempdir().expect("temp dir"); From 22fce01dba9520bcacd80c462bb970d0ace2f35d Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:14:47 +0000 Subject: [PATCH 6/6] Type the startup command once the shell has printed its prompt The command was written into the PTY right after the shell request, before the login shell had run its startup files. Anything there that reads input could take it, and the terminal echoed it in the middle of the login banner and again after the prompt. It now waits for the shell's output to start and then pause for a second, which is how a shell sitting at its prompt looks from outside, with a five-second cap so a shell that prints nothing, or never stops, still gets it. The wait runs on its own task, so the connect returns as before. The command takes the session's first write turn before the session is reachable, so input sent while it waits reaches the shell after it, in order, instead of being spliced in front of it. --- core/src/session.rs | 133 ++++++++++++++++++++---- core/tests/server_requests.rs | 186 ++++++++++++++++++++++++---------- 2 files changed, 246 insertions(+), 73 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index b16516f7..67645b64 100644 --- a/core/src/session.rs +++ b/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>; /// 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>, resize_tx: mpsc::Sender<(u32, u32)>, /// Held so the id stays reserved while this entry exists. #[allow(dead_code)] @@ -254,11 +271,28 @@ impl SessionManager { 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(), }); + // 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 @@ -292,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); } @@ -331,21 +370,13 @@ impl SessionManager { } }); - // The startup command is typed into the shell once the reader is - // running, so whatever the remote echoes back is drained while it is - // sent. A channel takes only one shell, exec or subsystem request, and - // OpenSSH ignores an exec that arrives after the shell, so it cannot be - // sent as a request of its own. The session is usable either way, so a - // command that cannot be sent does not fail the connect. - // - // It is sent to the session built above, not to whatever the id names - // by now: a shell that exits at once frees the id, and a session that - // takes it next must never receive this one's input. Holding `session` - // also keeps the id reserved until the send is over. - if let Some(cmd) = host.startup_command.as_deref().filter(|cmd| !cmd.is_empty()) { - if let Err(error) = session.send(&session_id, format!("{cmd}\r").as_bytes()).await { - tracing::warn!("startup command for session {session_id} was not sent: {error}"); - } + // 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(()) @@ -453,6 +484,72 @@ async fn disconnect(handle: &Handle) { .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, 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; diff --git a/core/tests/server_requests.rs b/core/tests/server_requests.rs index 7b262d9a..9165a292 100644 --- a/core/tests/server_requests.rs +++ b/core/tests/server_requests.rs @@ -1,6 +1,7 @@ -//! What a connected server can ask of the client, checked against an -//! in-process russh server on loopback: channels it tries to open back are -//! refused, and a host's startup command reaches the shell it was meant for. +//! What a connected server can ask of the client, and what the client sends it +//! unasked, checked against an in-process russh server on loopback: channels +//! the server tries to open back are refused, and a host's startup command +//! reaches the shell it was meant for once that shell is ready for it. mod common; @@ -22,13 +23,30 @@ enum Seen { Input(Vec), /// Whether the client accepted each channel the server tried to open. Opened { kind: &'static str, accepted: bool }, + /// The shell finished printing its prompt. + Prompted, } +/// How the test server's shell behaves once it starts. +#[derive(Clone, Copy)] +enum Shell { + /// Prints nothing at all. + Silent, + /// Prints a login banner, runs its startup files without a word for + /// [`STARTUP_FILES`], then prints a prompt and waits. + Prompt, + /// Prints nothing and opens channels back to the client. + OpenBack, +} + +/// How long the [`Shell::Prompt`] shell is quiet between its banner and its +/// prompt, the way a shell busy with its startup files is. +const STARTUP_FILES: Duration = Duration::from_millis(400); + #[derive(Clone)] struct TestServer { seen: mpsc::UnboundedSender, - /// Try to open channels back to the client once its shell starts. - open_back: bool, + shell: Shell, } impl server::Handler for TestServer { @@ -52,52 +70,26 @@ impl server::Handler for TestServer { Ok(()) } - /// Opens one channel of every type russh can open back, in the order - /// [`OPENED_BACK`] lists them. async fn shell_request( &mut self, - _channel: ChannelId, + channel: ChannelId, session: &mut Session, ) -> Result<(), Self::Error> { - if self.open_back { - let handle = session.handle(); - let seen = self.seen.clone(); - tokio::spawn(async move { - let report = |kind, accepted| { - let _ = seen.send(Seen::Opened { kind, accepted }); - }; - report("agent", handle.channel_open_agent().await.is_ok()); - report("x11", handle.channel_open_x11("127.0.0.1", 6010).await.is_ok()); - report( - "forwarded-tcpip", - handle - .channel_open_forwarded_tcpip("127.0.0.1", 8080, "127.0.0.1", 50000) - .await - .is_ok(), - ); - report( - "forwarded-streamlocal", - handle - .channel_open_forwarded_streamlocal("/tmp/forwarded.sock") - .await - .is_ok(), - ); - report( - "direct-tcpip", - handle - .channel_open_direct_tcpip("127.0.0.1", 22, "127.0.0.1", 50001) - .await - .is_ok(), - ); - report( - "direct-streamlocal", - handle - .channel_open_direct_streamlocal("/tmp/direct.sock") - .await - .is_ok(), - ); - report("session", handle.channel_open_session().await.is_ok()); - }); + let handle = session.handle(); + let seen = self.seen.clone(); + match self.shell { + Shell::Silent => {} + Shell::Prompt => { + tokio::spawn(async move { + let _ = handle.data(channel, &b"Last login: today\r\n"[..]).await; + tokio::time::sleep(STARTUP_FILES).await; + let _ = handle.data(channel, &b"$ "[..]).await; + let _ = seen.send(Seen::Prompted); + }); + } + Shell::OpenBack => { + tokio::spawn(open_back(handle, seen)); + } } Ok(()) } @@ -124,9 +116,48 @@ const OPENED_BACK: [&str; 7] = [ "session", ]; -async fn start_server(open_back: bool) -> (u16, mpsc::UnboundedReceiver) { +/// Opens one channel of every type russh can open back, in the order +/// [`OPENED_BACK`] lists them, and reports whether the client accepted each. +async fn open_back(handle: server::Handle, seen: mpsc::UnboundedSender) { + let report = |kind, accepted| { + let _ = seen.send(Seen::Opened { kind, accepted }); + }; + report("agent", handle.channel_open_agent().await.is_ok()); + report("x11", handle.channel_open_x11("127.0.0.1", 6010).await.is_ok()); + report( + "forwarded-tcpip", + handle + .channel_open_forwarded_tcpip("127.0.0.1", 8080, "127.0.0.1", 50000) + .await + .is_ok(), + ); + report( + "forwarded-streamlocal", + handle + .channel_open_forwarded_streamlocal("/tmp/forwarded.sock") + .await + .is_ok(), + ); + report( + "direct-tcpip", + handle + .channel_open_direct_tcpip("127.0.0.1", 22, "127.0.0.1", 50001) + .await + .is_ok(), + ); + report( + "direct-streamlocal", + handle + .channel_open_direct_streamlocal("/tmp/direct.sock") + .await + .is_ok(), + ); + report("session", handle.channel_open_session().await.is_ok()); +} + +async fn start_server(shell: Shell) -> (u16, mpsc::UnboundedReceiver) { let (seen, events) = mpsc::unbounded_channel(); - let port = serve(server_config(random_host_key()), TestServer { seen, open_back }).await; + let port = serve(server_config(random_host_key()), TestServer { seen, shell }).await; (port, events) } @@ -134,6 +165,12 @@ fn sessions() -> SessionManager { SessionManager::new(Arc::new(|_, _| {}), Arc::new(|_, _| {})) } +fn host_with_startup_command(port: u16) -> Host { + let mut host = password_host(port); + host.startup_command = Some("tmux attach".into()); + host +} + async fn open_shell(sessions: &SessionManager, host: &Host, dir: &TempDir) { let claim = sessions.claim("shell").expect("claim"); sessions @@ -152,7 +189,7 @@ async fn next(events: &mut mpsc::UnboundedReceiver) -> Seen { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn channels_the_server_opens_back_are_refused() { let dir = tempfile::tempdir().expect("temp dir"); - let (port, mut events) = start_server(true).await; + let (port, mut events) = start_server(Shell::OpenBack).await; let sessions = sessions(); open_shell(&sessions, &password_host(port), &dir).await; @@ -168,13 +205,52 @@ async fn channels_the_server_opens_back_are_refused() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn the_startup_command_is_typed_into_the_shell() { +async fn the_startup_command_waits_for_the_shell_prompt() { let dir = tempfile::tempdir().expect("temp dir"); - let (port, mut events) = start_server(false).await; + let (port, mut events) = start_server(Shell::Prompt).await; let sessions = sessions(); - let mut host = password_host(port); - host.startup_command = Some("tmux attach".into()); - open_shell(&sessions, &host, &dir).await; + open_shell(&sessions, &host_with_startup_command(port), &dir).await; + + // Typed any earlier, the command would land while the startup files are + // still running and could be read by them instead of the shell. + match next(&mut events).await { + Seen::Prompted => {} + other => panic!("the client sent something before the prompt: {other:?}"), + } + match next(&mut events).await { + Seen::Input(bytes) => assert_eq!(bytes, b"tmux attach\r"), + other => panic!("unexpected {other:?}"), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn input_sent_while_the_startup_command_waits_follows_it() { + let dir = tempfile::tempdir().expect("temp dir"); + let (port, mut events) = start_server(Shell::Prompt).await; + let sessions = Arc::new(sessions()); + open_shell(&sessions, &host_with_startup_command(port), &dir).await; + + let typed = tokio::spawn({ + let sessions = sessions.clone(); + async move { sessions.write("shell", b"ls\r").await } + }); + + let mut inputs = Vec::new(); + while inputs.len() < 2 { + if let Seen::Input(bytes) = next(&mut events).await { + inputs.push(bytes); + } + } + assert_eq!(inputs, [b"tmux attach\r".to_vec(), b"ls\r".to_vec()]); + typed.await.expect("write task").expect("write"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_shell_that_prints_nothing_still_gets_the_startup_command() { + let dir = tempfile::tempdir().expect("temp dir"); + let (port, mut events) = start_server(Shell::Silent).await; + let sessions = sessions(); + open_shell(&sessions, &host_with_startup_command(port), &dir).await; match next(&mut events).await { Seen::Input(bytes) => assert_eq!(bytes, b"tmux attach\r"),