rust: session manager (lifecycle, reconnect backoff, login scripts, recording, host-key policy, credential modes) and SSH tunnels

rust/core/src/clients/ssh.rs

@@ -432,7 +432,33 @@ impl SshClient {
     /// Connects, authenticates and starts the shell. Blocks the calling thread.
     pub fn connect(self: &Arc<Self>, check: HostKeyCheck, password: Option<SecretFn>, passphrase: Option<SecretFn>) -> Result<()> {
         let me = self.clone();
-        rt().block_on(async move { me.connect_async(check, password, passphrase).await })
+        rt().block_on(async move { me.connect_async(check, password, passphrase, true).await })
+    }
+
+    /// Connects and authenticates without opening a shell, for tunnelling another protocol.
+    pub fn connect_tunnel_only(self: &Arc<Self>, check: HostKeyCheck, password: Option<SecretFn>, passphrase: Option<SecretFn>) -> Result<()> {
+        let me = self.clone();
+        rt().block_on(async move { me.connect_async(check, password, passphrase, false).await })
+    }
+
+    /// A `direct-tcpip` channel to `host:port` presented as an ordinary loopback TCP stream,
+    /// so blocking clients (IRC, VNC) can use it with read timeouts.
+    pub fn open_tunnel(&self, host: &str, port: u16) -> Result<std::net::TcpStream> {
+        let h = self.handle()?;
+        let host = host.to_string();
+        let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
+        let addr = listener.local_addr()?;
+        let ch = rt().block_on(h.channel_open_direct_tcpip(host.clone(), port as u32, "127.0.0.1", 0)).map_err(|e| format!("SSH tunnel to {host}:{port} failed: {e}"))?;
+        let client = std::net::TcpStream::connect(addr)?;
+        let (server, _) = listener.accept()?;
+        server.set_nonblocking(true)?;
+        rt().spawn(async move {
+            if let Ok(mut local) = tokio::net::TcpStream::from_std(server) {
+                let mut remote = ch.into_stream();
+                let _ = tokio::io::copy_bidirectional(&mut local, &mut remote).await;
+            }
+        });
+        Ok(client)
     }
 
     async fn handshake(
@@ -454,7 +480,7 @@ impl SshClient {
         }
     }
 
-    async fn connect_async(self: &Arc<Self>, check: HostKeyCheck, password: Option<SecretFn>, passphrase: Option<SecretFn>) -> Result<()> {
+    async fn connect_async(self: &Arc<Self>, check: HostKeyCheck, password: Option<SecretFn>, passphrase: Option<SecretFn>, shell: bool) -> Result<()> {
         let sc = self.sc();
         let p = &self.profile;
         let port = if p.port == 0 { 22 } else { p.port as u16 };
@@ -484,6 +510,10 @@ impl SshClient {
         )
         .await?;
         let handle = Arc::new(handle);
+        if !shell {
+            *self.handle.lock().unwrap() = Some(handle);
+            return Ok(());
+        }
 
         let ch = handle.channel_open_session().await.map_err(|e| format!("failed to create SSH session: {e}"))?;
         for (k, v) in crate::terminalenv::ssh_env_vars(&sc.env_vars) {

rust/core/src/lib.rs

@@ -9,6 +9,7 @@ pub mod keyfiles;
 pub mod models;
 pub mod recording;
 pub mod secrets;
+pub mod sessions;
 pub mod sshconfig;
 pub mod storage;
 pub mod terminalenv;

rust/core/src/secrets.rs

@@ -226,6 +226,56 @@ impl Keychain {
     }
 }
 
+/// Keychain first, vault second — how every credential lookup in the app works.
+pub struct Secrets {
+    pub vault: Vault,
+    pub keychain: Keychain,
+    /// Probed once at startup (the probe writes an item, so don't repeat it per lookup).
+    pub keychain_ok: bool,
+}
+
+impl Secrets {
+    pub fn new(vault: Vault, keychain: Keychain) -> Secrets {
+        let keychain_ok = keychain.available();
+        Secrets { vault, keychain, keychain_ok }
+    }
+
+    pub fn get(&self, key: &str) -> Option<String> {
+        if self.keychain_ok
+            && let Some(v) = self.keychain.get(key).filter(|v| !v.is_empty())
+        {
+            return Some(v);
+        }
+        self.vault.get(key).ok().filter(|v| !v.is_empty())
+    }
+
+    /// Stores where possible: the keychain, else the vault if it's unlocked.
+    pub fn set(&self, key: &str, value: &str) {
+        let _ = self.set_checked(key, value);
+    }
+
+    pub fn set_checked(&self, key: &str, value: &str) -> Result<()> {
+        if self.keychain_ok && self.keychain.set(key, value).is_ok() {
+            return Ok(());
+        }
+        if !self.vault.is_unlocked() {
+            return Err("secret store is locked; unlock it first".into());
+        }
+        self.vault.set(key, value)
+    }
+
+    pub fn delete(&self, key: &str) {
+        if self.keychain_ok {
+            self.keychain.delete(key);
+        }
+        let _ = self.vault.delete(key);
+    }
+
+    pub fn delete_profile(&self, profile_id: &str) {
+        let _ = self.vault.delete_prefix(&format!("profile:{profile_id}:"));
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

rust/core/tests/sessions.rs

@@ -0,0 +1,131 @@
+//! Session manager end to end: events, writes, host-key prompts, auto-reconnect.
+
+use nexus_core::models::{Profile, Protocol, SessionState};
+use nexus_core::secrets::{Keychain, Secrets, Vault};
+use nexus_core::sessions::{Event, Manager, Prompt};
+use nexus_core::storage::Store;
+use std::io::{Read, Write};
+use std::net::TcpListener;
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+fn manager(name: &str) -> (Arc<Manager>, mpsc::Receiver<Event>, std::path::PathBuf) {
+    let dir = std::env::temp_dir().join(format!("nexus-mgr-{name}-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    let store = Arc::new(Store::open(&dir.join("nexus.db")).unwrap());
+    let secrets = Arc::new(Secrets { vault: Vault::new(dir.join("vault")), keychain: Keychain::new("NexusTerminalTest", None), keychain_ok: false });
+    let m = Manager::new(store, secrets, dir.join("recordings"));
+    let (tx, rx) = mpsc::channel();
+    let tx = Mutex::new(tx);
+    m.set_sink(Arc::new(move |e| {
+        let _ = tx.lock().unwrap().send(e);
+    }));
+    (m, rx, dir)
+}
+
+fn no_prompt() -> Prompt {
+    Arc::new(|_| Err("no prompt in tests".into()))
+}
+
+/// Waits for an event matching `f`, answering host-key prompts with `accept`.
+fn wait<T>(rx: &mpsc::Receiver<Event>, mut f: impl FnMut(&Event) -> Option<T>) -> T {
+    let deadline = std::time::Instant::now() + Duration::from_secs(15);
+    loop {
+        let e = rx.recv_timeout(deadline.saturating_duration_since(std::time::Instant::now())).expect("timed out waiting for event");
+        if let Some(t) = f(&e) {
+            return t;
+        }
+        if let Event::HostKeyVerify { reply, .. } = e {
+            let _ = reply.send(true);
+        }
+    }
+}
+
+fn state(e: &Event, want: SessionState) -> Option<String> {
+    match e {
+        Event::State { state, error, session, .. } if *state == want => Some(format!("{session}:{error}")),
+        _ => None,
+    }
+}
+
+#[cfg(unix)]
+#[test]
+fn local_shell_round_trip_and_restart() {
+    let (m, rx, dir) = manager("local");
+    let id = m.connect_local("/bin/sh");
+    m.resize(&id, 100, 30).unwrap();
+    wait(&rx, |e| state(e, SessionState::Connected));
+    m.start_recording(&id, 100, 30).unwrap();
+    m.write(&id, b"echo mgr-$((6*7))\n").unwrap();
+    let mut seen = Vec::new();
+    wait(&rx, |e| match e {
+        Event::Data { data, .. } => {
+            seen.extend_from_slice(data);
+            String::from_utf8_lossy(&seen).contains("mgr-42").then_some(())
+        }
+        _ => None,
+    });
+    m.stop_recording(&id).unwrap();
+    assert_eq!(m.recordings().list().unwrap().len(), 1);
+    m.write(&id, b"exit\n").unwrap();
+    wait(&rx, |e| matches!(e, Event::Closed { .. }).then_some(()));
+    assert_eq!(m.info(&id).unwrap().state, SessionState::Disconnected);
+    m.restart_local(&id, 80, 24).unwrap();
+    wait(&rx, |e| state(e, SessionState::Connected));
+    m.disconnect(&id).unwrap();
+    assert!(m.info(&id).is_none());
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+#[test]
+fn raw_session_and_auto_reconnect() {
+    let (m, rx, dir) = manager("raw");
+    let l = TcpListener::bind("127.0.0.1:0").unwrap();
+    let port = l.local_addr().unwrap().port();
+    std::thread::spawn(move || {
+        // First connection: echo one message then hang up; the second stays open.
+        for (n, s) in l.incoming().flatten().enumerate() {
+            let mut s = s;
+            let mut buf = [0u8; 64];
+            if let Ok(k) = s.read(&mut buf) {
+                let _ = s.write_all(&buf[..k]);
+            }
+            if n > 0 {
+                std::thread::sleep(Duration::from_secs(30));
+            }
+        }
+    });
+    let mut p = Profile::new(Protocol::Raw);
+    p.host = "127.0.0.1".into();
+    p.port = port as i64;
+    p.raw.as_mut().unwrap().auto_reconnect = true;
+    let id = m.connect(p, no_prompt());
+    wait(&rx, |e| state(e, SessionState::Connected));
+    m.raw_write_text(&id, "ping").unwrap();
+    wait(&rx, |e| matches!(e, Event::RawData { direction, data, .. } if direction == "recv" && data.starts_with(b"ping")).then_some(()));
+    // Server hung up: reconnecting with a backoff, then connected again under the same id.
+    let retry = wait(&rx, |e| match e {
+        Event::State { state: SessionState::Reconnecting, retry, .. } => *retry,
+        _ => None,
+    });
+    assert_eq!((retry.attempt, retry.delay_secs), (1, 1));
+    let back = wait(&rx, |e| state(e, SessionState::Connected));
+    assert!(back.starts_with(&id));
+    m.disconnect(&id).unwrap();
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+#[test]
+fn connect_failure_reports_error_state() {
+    let (m, rx, dir) = manager("fail");
+    let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
+    let mut p = Profile::new(Protocol::Telnet);
+    p.host = "127.0.0.1".into();
+    p.port = port as i64;
+    let id = m.connect(p, no_prompt());
+    let err = wait(&rx, |e| state(e, SessionState::Error));
+    assert!(err.starts_with(&id) && err.len() > id.len() + 1, "{err}");
+    let _ = std::fs::remove_dir_all(dir);
+}