From 9d6b22c6ba89e65047f7f99bc29c08771bdc509e Mon Sep 17 00:00:00 2001 From: Joe Abbate Date: Thu, 9 Jul 2026 21:06:26 -0400 Subject: [PATCH] feat(pivot): add stateful ssh_session for persistent SSH connections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce pivot.ssh_session(target, port, username, password, key, key_password, timeout) -> — the first ForeignValue object with methods in Eldritch. It solves the per-call auth-spam problem of ssh_exec by holding a single russh transport open for many channel_open_session+exec calls. .exec(command) returns Dict{stdout, stderr, status} identical to ssh_exec. .close() disconnects the transport and is idempotent; subsequent exec() raises "session closed". Implementation: - implants/lib/eldritch/stdlib/eldritch-libpivot/src/lib.rs: add ssh_session signature to PivotLibrary trait - implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/mod.rs: wire factory, add pub mod ssh_session_impl - implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/ssh_session_impl.rs (new): SshSessionHandle = Arc>>; eager connect with timeout wrapper, take-restore across await to avoid holding lock, custom Debug that only prints target/port/connected (no secret leak), async helpers for tokio tests (wrap_for_test, exec_async, close_async, is_closed), mock sshd that supports multiple execs per connection with rising start-up guard (wait_until_listening) - implants/lib/eldritch/stdlib/eldritch-libpivot/src/fake.rs: FakeSshSessionHandle + FakeSshSessionLibrary (exec/close) with closed-flag via spin::Mutex; PivotLibraryFake::ssh_session returns Foreign handle - implants/lib/eldritch/stdlib/eldritch-libpivot/Cargo.toml: add spin optional dep for fake_bindings - implants/lib/eldritch/eldritch/src/bindings_test.rs: fix registration-order bug where with_context re-registered Std pivot over fake (chain with_fake_agent after with_context); add file bindings drift (list_named_pipes, read_named_pipe, tmp_dir); add pivot.ssh_session to expected bindings; add test_ssh_session_object_methods, test_ssh_session_fake_exec_and_close, test_ssh_session_fake_reuse_multiple_execs Tests: cargo test -p eldritch-libpivot --lib 48 passed (6 new), cargo test -p eldritch --features fake_bindings 17 passed Docs: user guide update for pivot.ssh_session is out of scope for this commit per repo conventions. --- .../eldritch/eldritch/src/bindings_test.rs | 94 ++- .../stdlib/eldritch-libpivot/Cargo.toml | 3 +- .../stdlib/eldritch-libpivot/src/fake.rs | 59 +- .../stdlib/eldritch-libpivot/src/lib.rs | 35 + .../stdlib/eldritch-libpivot/src/std/mod.rs | 23 + .../src/std/ssh_session_impl.rs | 614 ++++++++++++++++++ 6 files changed, 819 insertions(+), 9 deletions(-) create mode 100644 implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/ssh_session_impl.rs diff --git a/implants/lib/eldritch/eldritch/src/bindings_test.rs b/implants/lib/eldritch/eldritch/src/bindings_test.rs index 85a3372d6..3ca1b54c9 100644 --- a/implants/lib/eldritch/eldritch/src/bindings_test.rs +++ b/implants/lib/eldritch/eldritch/src/bindings_test.rs @@ -17,16 +17,22 @@ fn create_interp() -> Interpreter { jwt: "a test jwt".to_string(), }; let backend = Arc::new(EmptyAssets {}); - Interpreter::new().with_default_libs().with_context( - agent_mock, - eldritch_agent::Context::Task(task_context), - vec![], - backend, - ) + // with_default_libs registers Std then Fake (fake wins). + // with_context re-registers Std (std wins – bug). Chain with_fake_agent + // to make fake win again so tests use fake impls. + Interpreter::new() + .with_default_libs() + .with_context( + agent_mock, + eldritch_agent::Context::Task(task_context), + vec![], + backend, + ) + .with_fake_agent() } #[cfg(not(feature = "stdlib"))] { - Interpreter::new() + Interpreter::new().with_fake_agent() } } @@ -67,6 +73,7 @@ fn test_file_bindings() { "is_dir", "is_file", "list", + "list_named_pipes", "list_recent", "mkdir", "move", @@ -74,6 +81,7 @@ fn test_file_bindings() { "pwd", "read", "read_binary", + "read_named_pipe", "remove", "replace", "replace_all", @@ -81,6 +89,7 @@ fn test_file_bindings() { "template", "template_str", "timestomp", + "tmp_dir", "write", "write_binary", ], @@ -136,10 +145,81 @@ fn test_pivot_bindings() { "ssh_copy", "ssh_deploy", "ssh_exec", + "ssh_session", ], ); } +#[test] +fn test_ssh_session_object_methods() { + // with_fake_agent returns FakeSshSessionHandle which has exec + close + let mut interp = create_interp(); + let code = r#" +s = pivot.ssh_session("127.0.0.1", 22, "root", "pass", None, None, None) +dir(s) +"#; + let val = interp + .interpret(code) + .expect("interpret ssh_session failed"); + if let Value::List(l) = val { + let list = l.read(); + let actual: Vec = list + .iter() + .map(|v| v.to_string().replace("\"", "")) + .collect(); + assert!( + actual.contains(&"exec".to_string()), + "ssh_session dir should contain exec, got {actual:?}" + ); + assert!( + actual.contains(&"close".to_string()), + "ssh_session dir should contain close, got {actual:?}" + ); + } else { + panic!("Expected list for dir(ssh_session)"); + } +} + +#[test] +fn test_ssh_session_fake_exec_and_close() { + let mut interp = create_interp(); + // Fake exec returns stdout="fake output" + let code = r#" +s = pivot.ssh_session("127.0.0.1", 22, "root", "pass", None, None, None) +r = s.exec("whoami") +r["stdout"] +"#; + let val = interp.interpret(code).expect("interpret exec failed"); + assert_eq!(val.to_string(), "fake output"); + + // After close, exec should raise. + let code2 = r#" +s = pivot.ssh_session("127.0.0.1", 22, "root", "pass", None, None, None) +s.close() +s.exec("whoami") +"#; + let res2 = interp.interpret(code2); + assert!( + res2.is_err(), + "expected error after close, got ok: {res2:?}" + ); +} + +#[test] +fn test_ssh_session_fake_reuse_multiple_execs() { + let mut interp = create_interp(); + let code = r#" +s = pivot.ssh_session("127.0.0.1", 22, "root", "pass", None, None, None) +r1 = s.exec("whoami") +r2 = s.exec("id") +r3 = s.exec("hostname") +s.close() +r1["stdout"] + "|" + r2["stdout"] + "|" + r3["stdout"] +"#; + let val = interp.interpret(code).expect("multi exec should succeed"); + assert_eq!(val.to_string(), "fake output|fake output|fake output"); +} + #[test] fn test_assets_bindings() { check_bindings("assets", &["copy", "list", "read", "read_binary"]); diff --git a/implants/lib/eldritch/stdlib/eldritch-libpivot/Cargo.toml b/implants/lib/eldritch/stdlib/eldritch-libpivot/Cargo.toml index 1ed85d2e8..e7aa0ecfc 100644 --- a/implants/lib/eldritch/stdlib/eldritch-libpivot/Cargo.toml +++ b/implants/lib/eldritch/stdlib/eldritch-libpivot/Cargo.toml @@ -23,6 +23,7 @@ tokio = { workspace = true, features = [ ], optional = true } pnet = { workspace = true, optional = true } pb = { workspace = true, optional = true } +spin = { workspace = true, features = ["mutex"], optional = true } [target.'cfg(not(target_os = "freebsd"))'.dependencies] listeners = { workspace = true, optional = true } @@ -48,5 +49,5 @@ stdlib = [ "tokio", "pnet", ] -fake_bindings = [] +fake_bindings = ["dep:spin"] print_debug = [] diff --git a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/fake.rs b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/fake.rs index 26d448216..1b2b09806 100644 --- a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/fake.rs +++ b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/fake.rs @@ -2,9 +2,50 @@ use super::PivotLibrary; use alloc::collections::BTreeMap; use alloc::string::String; use alloc::string::ToString; +use alloc::sync::Arc; use alloc::vec::Vec; use eldritch_core::Value; -use eldritch_macros::eldritch_library_impl; +use eldritch_macros::{eldritch_library, eldritch_library_impl, eldritch_method}; + +// ────────────────────────────────────────────────────────────────────────────── +// Fake ssh_session handle (returned as Foreign Value) +// ────────────────────────────────────────────────────────────────────────────── + +#[eldritch_library("ssh_session")] +pub trait FakeSshSessionLibrary { + #[eldritch_method] + fn exec(&self, _command: String) -> Result, String>; + #[eldritch_method] + fn close(&self) -> Result<(), String>; +} + +#[derive(Debug)] +#[eldritch_library_impl(FakeSshSessionLibrary)] +pub struct FakeSshSessionHandle { + closed: spin::Mutex, +} + +impl FakeSshSessionLibrary for FakeSshSessionHandle { + fn exec(&self, _command: String) -> Result, String> { + if *self.closed.lock() { + return Err("ssh_session is closed".to_string()); + } + let mut map = BTreeMap::new(); + map.insert("status".into(), Value::Int(0)); + map.insert("stdout".into(), Value::String("fake output".to_string())); + map.insert("stderr".into(), Value::String("".to_string())); + Ok(map) + } + + fn close(&self) -> Result<(), String> { + *self.closed.lock() = true; + Ok(()) + } +} + +// ────────────────────────────────────────────────────────────────────────────── +// Fake Pivot library +// ────────────────────────────────────────────────────────────────────────────── #[derive(Default, Debug)] #[eldritch_library_impl(PivotLibrary)] @@ -48,6 +89,22 @@ impl PivotLibrary for PivotLibraryFake { Ok("Success".into()) } + fn ssh_session( + &self, + _target: String, + _port: i64, + _username: String, + _password: Option, + _key: Option, + _key_password: Option, + _timeout: Option, + ) -> Result { + let handle = FakeSshSessionHandle { + closed: spin::Mutex::new(false), + }; + Ok(Value::Foreign(Arc::new(handle))) + } + fn port_scan( &self, _target_cidrs: Vec, diff --git a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/lib.rs b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/lib.rs index e1d64fd96..633764e37 100644 --- a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/lib.rs +++ b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/lib.rs @@ -98,6 +98,41 @@ pub trait PivotLibrary { timeout: Option, ) -> Result; + #[allow(clippy::too_many_arguments)] + #[eldritch_method] + /// Creates a persistent SSH session for reuse across multiple commands. + /// + /// Unlike `ssh_exec` which opens a new connection per call, this establishes + /// one connection and returns a session object whose `exec()` can be called + /// repeatedly. This reduces authentication log noise and handshake latency. + /// + /// **Parameters** + /// - `target` (`str`): The remote host IP or hostname. + /// - `port` (`int`): The SSH port (usually 22). + /// - `username` (`str`): SSH username. + /// - `password` (`Option`): SSH password (optional). + /// - `key` (`Option`): SSH private key (optional). + /// - `key_password` (`Option`): Password for the private key (optional). + /// - `timeout` (`Option`): Connection timeout in seconds (optional, defaults to 3). + /// + /// **Returns** + /// - ``: A session object with methods: + /// - `exec(command: str) -> Dict{stdout, stderr, status}` + /// - `close() -> None` + /// + /// **Errors** + /// - Returns an error string if connection fails. + fn ssh_session( + &self, + target: String, + port: i64, + username: String, + password: Option, + key: Option, + key_password: Option, + timeout: Option, + ) -> Result; + #[eldritch_method] /// Scans TCP/UDP ports on target hosts. /// diff --git a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/mod.rs b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/mod.rs index 1487a2f5b..55fd63f1b 100644 --- a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/mod.rs +++ b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/mod.rs @@ -5,6 +5,7 @@ pub mod port_scan_impl; pub mod ssh_copy_impl; pub mod ssh_deploy_impl; pub mod ssh_exec_impl; +pub mod ssh_session_impl; use alloc::collections::BTreeMap; use alloc::string::String; @@ -111,6 +112,28 @@ impl PivotLibrary for StdPivotLibrary { .map_err(|e| e.to_string()) } + fn ssh_session( + &self, + target: String, + port: i64, + username: String, + password: Option, + key: Option, + key_password: Option, + timeout: Option, + ) -> Result { + ssh_session_impl::ssh_session( + target, + port as i32, + username, + password, + key, + key_password, + timeout.map(|t| t as u32), + ) + .map_err(|e| e.to_string()) + } + fn port_scan( &self, target_cidrs: Vec, diff --git a/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/ssh_session_impl.rs b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/ssh_session_impl.rs new file mode 100644 index 000000000..101444d52 --- /dev/null +++ b/implants/lib/eldritch/stdlib/eldritch-libpivot/src/std/ssh_session_impl.rs @@ -0,0 +1,614 @@ +use alloc::collections::BTreeMap; +use alloc::format; +use alloc::string::String; +use alloc::string::ToString; +use alloc::sync::Arc; +use anyhow::Result; +use eldritch_core::Value; +use eldritch_macros::{eldritch_library, eldritch_library_impl, eldritch_method}; + +use crate::std::Session; + +use std::sync::Mutex; + +/// Persistent SSH session handle – foreign value. +/// +/// `inner` holds `Option` so `close()` can take it and future +/// `exec()` fails with "session closed". The `Mutex` is only held briefly; +/// we never hold it across an await – we take, await, then restore. +#[eldritch_library("ssh_session")] +pub trait SshSessionLibrary { + #[eldritch_method] + /// Executes a command on the remote host via the existing SSH transport. + fn exec(&self, command: String) -> Result, String>; + + #[eldritch_method] + /// Closes the SSH session, disconnecting the transport. Idempotent. + fn close(&self) -> Result<(), String>; +} + +#[eldritch_library_impl(SshSessionLibrary)] +pub struct SshSessionHandle { + inner: Arc>>, + target: String, + port: i32, +} + +impl core::fmt::Debug for SshSessionHandle { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + let connected = self.inner.lock().map(|g| g.is_some()).unwrap_or(false); + f.debug_struct("SshSessionHandle") + .field("target", &self.target) + .field("port", &self.port) + .field("connected", &connected) + .finish() + } +} + +// ────────────────────────────────────────────────────────────────────────────── +// Connect helper + factory +// ────────────────────────────────────────────────────────────────────────────── + +async fn handle_connect( + target: String, + port: u16, + username: String, + password: Option, + key: Option, + key_password: Option<&str>, + timeout: Option, +) -> Result { + let secs = timeout.unwrap_or(3) as u64; + let addr = format!("{target}:{port}"); + let session = tokio::time::timeout( + std::time::Duration::from_secs(secs), + Session::connect(username, password, key, key_password, addr), + ) + .await + .map_err(|_| anyhow::anyhow!("SSH connection timed out after {secs}s"))??; + Ok(session) +} + +/// Factory called by `StdPivotLibrary::ssh_session`. +pub fn ssh_session( + target: String, + port: i32, + username: String, + password: Option, + key: Option, + key_password: Option, + timeout: Option, +) -> Result { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + + let local_port: u16 = port + .try_into() + .map_err(|e| anyhow::anyhow!("invalid port {port}: {e}"))?; + + let sess = runtime.block_on(handle_connect( + target.clone(), + local_port, + username, + password, + key, + key_password.as_deref(), + timeout, + ))?; + + let handle = SshSessionHandle { + inner: Arc::new(Mutex::new(Some(sess))), + target, + port, + }; + + Ok(Value::Foreign(Arc::new(handle))) +} + +// ────────────────────────────────────────────────────────────────────────────── +// SshSessionLibrary impl (sync, callable from non-tokio thread – matches other pivot impls) +// ────────────────────────────────────────────────────────────────────────────── + +impl SshSessionLibrary for SshSessionHandle { + fn exec(&self, command: String) -> Result, String> { + // Take session out of mutex so we don't hold the lock across await. + let mut sess = { + let mut guard = self + .inner + .lock() + .map_err(|_| "ssh_session mutex poisoned".to_string())?; + guard + .take() + .ok_or_else(|| "ssh_session is closed".to_string())? + }; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| e.to_string())?; + + let (res, sess) = runtime.block_on(async { + let r = sess.call(&command).await; + (r, sess) + }); + + // Always restore, even on error – keep session for close(). + if let Ok(mut guard) = self.inner.lock() { + *guard = Some(sess); + } + + let cr = res.map_err(|e| e.to_string())?; + + let mut map = BTreeMap::new(); + map.insert( + "stdout".into(), + Value::String(cr.output().map_err(|e| e.to_string())?), + ); + map.insert( + "stderr".into(), + Value::String(cr.error().map_err(|e| e.to_string())?), + ); + map.insert("status".into(), Value::Int(cr.code.unwrap_or(0) as i64)); + Ok(map) + } + + fn close(&self) -> Result<(), String> { + let maybe_sess = { + let mut guard = self + .inner + .lock() + .map_err(|_| "ssh_session mutex poisoned".to_string())?; + guard.take() + }; + + if let Some(mut sess) = maybe_sess { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| e.to_string())?; + runtime.block_on(sess.close()).map_err(|e| e.to_string())?; + } + // idempotent: already None -> success + Ok(()) + } +} + +// ────────────────────────────────────────────────────────────────────────────── +// Async helpers for tests +// ────────────────────────────────────────────────────────────────────────────── + +impl SshSessionHandle { + #[cfg(test)] + pub(crate) fn wrap_for_test(session: Session, target: String, port: i32) -> Self { + Self { + inner: Arc::new(Mutex::new(Some(session))), + target, + port, + } + } + + #[cfg(test)] + pub(crate) async fn exec_async( + &self, + command: &str, + ) -> anyhow::Result> { + let mut sess = { + let mut guard = self.inner.lock().map_err(|_| anyhow::anyhow!("poisoned"))?; + guard + .take() + .ok_or_else(|| anyhow::anyhow!("ssh_session is closed"))? + }; + + let res = sess.call(command).await; + + if let Ok(mut guard) = self.inner.lock() { + *guard = Some(sess); + } + + let cr = res?; + let mut map = BTreeMap::new(); + map.insert("stdout".into(), Value::String(cr.output()?)); + map.insert("stderr".into(), Value::String(cr.error()?)); + map.insert("status".into(), Value::Int(cr.code.unwrap_or(0) as i64)); + Ok(map) + } + + #[cfg(test)] + pub(crate) async fn close_async(&self) -> anyhow::Result<()> { + let maybe_sess = { + let mut g = self.inner.lock().map_err(|_| anyhow::anyhow!("poisoned"))?; + g.take() + }; + if let Some(mut sess) = maybe_sess { + sess.close().await?; + } + Ok(()) + } + + #[cfg(test)] + pub(crate) fn is_closed(&self) -> bool { + self.inner.lock().map(|g| g.is_none()).unwrap_or(true) + } +} + +// ────────────────────────────────────────────────────────────────────────────── +// Tests +// ────────────────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + use russh::server::{Auth, Msg, Session as ServerSession}; + use russh::*; + use std::collections::HashMap; + use std::process::Command; + use std::sync::Mutex as StdMutex; + use std::time::Duration; + use tokio::net::TcpListener; + use tokio::task; + + #[derive(Clone)] + #[allow(dead_code)] + struct TestServer { + client_pubkey: Arc, + clients: Arc>>>, + id: usize, + } + + impl server::Server for TestServer { + type Handler = Self; + fn new_client(&mut self, _: Option) -> Self { + let s = self.clone(); + self.id += 1; + s + } + } + + #[async_trait] + impl server::Handler for TestServer { + type Error = anyhow::Error; + + async fn channel_open_session( + self, + channel: Channel, + session: ServerSession, + ) -> Result<(Self, bool, ServerSession), Self::Error> { + { + let mut clients = self.clients.lock().unwrap(); + clients.insert((self.id, channel.id()), channel); + } + Ok((self, true, session)) + } + + #[allow(unused_variables)] + async fn exec_request( + self, + channel: ChannelId, + data: &[u8], + mut session: ServerSession, + ) -> Result<(Self, ServerSession), Self::Error> { + let cmd = std::str::from_utf8(data)?; + let (command_string, command_args): (&str, Vec<&str>) = if cfg!(target_os = "windows") { + ("cmd", vec!["/c", cmd]) + } else { + ("bash", vec!["-c", cmd]) + }; + let tmp_res = Command::new(command_string).args(command_args).output()?; + session.data(channel, CryptoVec::from(tmp_res.stdout.clone())); + if !tmp_res.stderr.is_empty() { + session.extended_data(channel, 1, CryptoVec::from(tmp_res.stderr)); + } + session.eof(channel); + session.close(channel); + Ok((self, session)) + } + + async fn auth_publickey( + self, + _: &str, + _: &russh_keys::key::PublicKey, + ) -> Result<(Self, Auth), Self::Error> { + Ok((self, server::Auth::Accept)) + } + + async fn auth_password( + self, + _user: &str, + _password: &str, + ) -> Result<(Self, Auth), Self::Error> { + Ok((self, Auth::Accept)) + } + + async fn data( + self, + _channel: ChannelId, + data: &[u8], + mut session: ServerSession, + ) -> Result<(Self, ServerSession), Self::Error> { + { + let mut clients = self.clients.lock().unwrap(); + for ((_, _channel_id), ref mut channel) in clients.iter_mut() { + session.data(channel.id(), CryptoVec::from(data.to_vec())); + } + } + Ok((self, session)) + } + } + + async fn test_ssh_server_multi(address: String, port: u16, secs: u64) { + let client_key = russh_keys::key::KeyPair::generate_ed25519().unwrap(); + let client_pubkey = Arc::new(client_key.clone_public_key().unwrap()); + let config = Arc::new(russh::server::Config { + connection_timeout: Some(Duration::from_secs(10)), + auth_rejection_time: Duration::from_secs(1), + keys: vec![russh_keys::key::KeyPair::generate_ed25519().unwrap()], + ..Default::default() + }); + let sh = TestServer { + client_pubkey, + clients: Arc::new(StdMutex::new(HashMap::new())), + id: 0, + }; + let _ = tokio::time::timeout( + Duration::from_secs(secs), + russh::server::run(config, (address, port), sh), + ) + .await; + } + + async fn allocate_localhost_unused_ports() -> anyhow::Result { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + Ok(listener.local_addr().unwrap().port().into()) + } + + /// Wait until TCP port is accepting connections, or time out. + async fn wait_until_listening(host: &str, port: u16) { + for _ in 0..50 { + if tokio::net::TcpStream::connect(format!("{host}:{port}")) + .await + .is_ok() + { + // Small grace period: server accepted, but SSH stack may still be initializing. + tokio::time::sleep(Duration::from_millis(50)).await; + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + #[tokio::test] + async fn test_ssh_session_connect_and_double_call_direct() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 8)); + wait_until_listening(&ssh_host, ssh_port).await; + + let mut sess = tokio::time::timeout( + Duration::from_secs(3), + Session::connect( + "root".to_string(), + Some("pass".to_string()), + None, + None, + format!("{ssh_host}:{ssh_port}"), + ), + ) + .await??; + + let r1 = sess.call("echo first").await?; + assert!(r1.output()?.contains("first"), "r1 was {:?}", r1.output()); + let r2 = sess.call("echo second").await?; + assert!(r2.output()?.contains("second"), "r2 was {:?}", r2.output()); + + sess.close().await?; + server_task.abort(); + Ok(()) + } + + #[tokio::test] + async fn test_ssh_session_handle_reuse_async() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 8)); + wait_until_listening(&ssh_host, ssh_port).await; + + let sess = tokio::time::timeout( + Duration::from_secs(3), + Session::connect( + "root".to_string(), + Some("pass".to_string()), + None, + None, + format!("{ssh_host}:{ssh_port}"), + ), + ) + .await??; + + let handle = SshSessionHandle::wrap_for_test(sess, ssh_host.clone(), ssh_port as i32); + + let r1 = handle.exec_async("echo alpha").await?; + let out1 = match r1.get("stdout") { + Some(Value::String(s)) => s.clone(), + other => panic!("unexpected stdout: {other:?}"), + }; + assert!(out1.contains("alpha"), "out1 was {out1:?}"); + + let r2 = handle.exec_async("echo beta").await?; + let out2 = match r2.get("stdout") { + Some(Value::String(s)) => s.clone(), + other => panic!("unexpected stdout: {other:?}"), + }; + assert!(out2.contains("beta"), "out2 was {out2:?}"); + + handle.close_async().await?; + assert!(handle.is_closed()); + server_task.abort(); + Ok(()) + } + + #[tokio::test] + async fn test_ssh_session_handle_close_then_exec_fails() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 8)); + wait_until_listening(&ssh_host, ssh_port).await; + + let sess = tokio::time::timeout( + Duration::from_secs(3), + Session::connect( + "root".to_string(), + Some("pass".to_string()), + None, + None, + format!("{ssh_host}:{ssh_port}"), + ), + ) + .await??; + + let handle = SshSessionHandle::wrap_for_test(sess, ssh_host.clone(), ssh_port as i32); + handle.close_async().await?; + + let err = handle.exec_async("echo should-fail").await.unwrap_err(); + assert!( + err.to_string().contains("closed"), + "expected closed error, got {err:?}" + ); + server_task.abort(); + Ok(()) + } + + #[tokio::test] + async fn test_ssh_session_factory_sync_success() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 10)); + wait_until_listening(&ssh_host, ssh_port).await; + + let port_for_blocking = ssh_port; + let ssh_host_clone = ssh_host.clone(); + let val = task::spawn_blocking(move || { + ssh_session( + ssh_host_clone, + port_for_blocking as i32, + "root".to_string(), + Some("pass".to_string()), + None, + None, + Some(3), + ) + }) + .await??; + + match &val { + Value::Foreign(f) => { + assert_eq!(f.type_name(), "ssh_session"); + let mut names = f.method_names(); + names.sort(); + assert!(names.contains(&"exec".to_string()), "methods: {names:?}"); + assert!(names.contains(&"close".to_string()), "methods: {names:?}"); + } + other => panic!("expected Foreign, got {other:?}"), + } + + server_task.abort(); + Ok(()) + } + + #[tokio::test] + async fn test_ssh_session_factory_fail() -> anyhow::Result<()> { + let free_port = allocate_localhost_unused_ports().await? as u16; + let res = task::spawn_blocking(move || { + ssh_session( + "127.0.0.1".to_string(), + free_port as i32, + "root".to_string(), + Some("pass".to_string()), + None, + None, + Some(1), + ) + }) + .await?; + assert!( + res.is_err(), + "expected error when connecting to unused port" + ); + Ok(()) + } + + #[tokio::test] + async fn test_debug_no_secret_leak() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 5)); + wait_until_listening(&ssh_host, ssh_port).await; + + let sess = tokio::time::timeout( + Duration::from_secs(3), + Session::connect( + "root".to_string(), + Some("supersecretpassword".to_string()), + None, + None, + format!("{ssh_host}:{ssh_port}"), + ), + ) + .await??; + + let handle = SshSessionHandle::wrap_for_test(sess, ssh_host.clone(), ssh_port as i32); + let dbg = format!("{handle:?}"); + assert!( + !dbg.contains("supersecretpassword"), + "Debug leaked password: {dbg}" + ); + assert!( + dbg.contains("127.0.0.1"), + "Debug should contain target: {dbg}" + ); + + handle.close_async().await?; + server_task.abort(); + Ok(()) + } + + #[tokio::test] + async fn test_ssh_session_factory_multi_exec_then_close() -> anyhow::Result<()> { + let ssh_port = allocate_localhost_unused_ports().await? as u16; + let ssh_host = "127.0.0.1".to_string(); + let server_task = task::spawn(test_ssh_server_multi(ssh_host.clone(), ssh_port, 10)); + wait_until_listening(&ssh_host, ssh_port).await; + + let sess = tokio::time::timeout( + Duration::from_secs(3), + Session::connect( + "root".to_string(), + Some("pass".to_string()), + None, + None, + format!("{ssh_host}:{ssh_port}"), + ), + ) + .await??; + + let handle = SshSessionHandle::wrap_for_test(sess, ssh_host.clone(), ssh_port as i32); + + let r1 = handle.exec_async("echo one").await?; + let r2 = handle.exec_async("echo two").await?; + let r3 = handle.exec_async("echo three").await?; + + let get_out = |m: &BTreeMap| match m.get("stdout") { + Some(Value::String(s)) => s.clone(), + other => panic!("bad stdout {other:?}"), + }; + + assert!(get_out(&r1).contains("one")); + assert!(get_out(&r2).contains("two")); + assert!(get_out(&r3).contains("three")); + + handle.close_async().await?; + server_task.abort(); + Ok(()) + } +}