diff --git a/core/src/connection.rs b/core/src/connection.rs index 7437f0e9..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, 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, + 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, + _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 @@ -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> { let path = std::env::temp_dir().join(format!( diff --git a/core/src/session.rs b/core/src/session.rs index 9c776cda..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)] @@ -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,16 +264,6 @@ 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(); @@ -249,12 +271,32 @@ 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(), }); - 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) { .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/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 new file mode 100644 index 00000000..9165a292 --- /dev/null +++ b/core/tests/server_requests.rs @@ -0,0 +1,259 @@ +//! 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; + +use clavyn_core::host::Host; +use clavyn_core::session::SessionManager; +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::sync::mpsc; + +/// 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 }, + /// 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, + shell: Shell, +} + +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> { + 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(()) + } + + 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(()) + } +} + +/// 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", +]; + +/// 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, shell }).await; + (port, events) +} + +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 + .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(Shell::OpenBack).await; + let sessions = sessions(); + open_shell(&sessions, &password_host(port), &dir).await; + + for expected in OPENED_BACK { + 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_waits_for_the_shell_prompt() { + let dir = tempfile::tempdir().expect("temp dir"); + let (port, mut events) = start_server(Shell::Prompt).await; + let sessions = sessions(); + 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"), + other => panic!("unexpected {other:?}"), + } +} 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");