From cbd83489b274df32bfd084c8135911bce0b7bb1e Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Sun, 11 Oct 2026 16:33:17 +0100 Subject: [PATCH] rust: session manager (lifecycle, reconnect backoff, login scripts, recording, host-key policy, credential modes) and SSH tunnels --- rust/app/src/main.rs | 6 + rust/core/src/clients/ssh.rs | 34 +- rust/core/src/lib.rs | 1 + rust/core/src/secrets.rs | 50 ++ rust/core/src/sessions.rs | 1163 ++++++++++++++++++++++++++++++++++ rust/core/tests/sessions.rs | 131 ++++ 6 files changed, 1383 insertions(+), 2 deletions(-) create mode 100644 rust/core/src/sessions.rs create mode 100644 rust/core/tests/sessions.rs diff --git a/rust/app/src/main.rs b/rust/app/src/main.rs index f328e4d..660c693 100644 --- a/rust/app/src/main.rs +++ b/rust/app/src/main.rs @@ -1 +1,7 @@ +//! Nexus Terminal: SSH, Telnet, RAW TCP, Serial, VNC, IRC, GitHub and local terminals. +#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")] + +mod terminal; +mod terminal_themes; + fn main() {} diff --git a/rust/core/src/clients/ssh.rs b/rust/core/src/clients/ssh.rs index f5b5e4f..839ccb6 100644 --- a/rust/core/src/clients/ssh.rs +++ b/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, check: HostKeyCheck, password: Option, passphrase: Option) -> 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, check: HostKeyCheck, password: Option, passphrase: Option) -> 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 { + 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, check: HostKeyCheck, password: Option, passphrase: Option) -> Result<()> { + async fn connect_async(self: &Arc, check: HostKeyCheck, password: Option, passphrase: Option, 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) { diff --git a/rust/core/src/lib.rs b/rust/core/src/lib.rs index 4d5c119..9b1131f 100644 --- a/rust/core/src/lib.rs +++ b/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; diff --git a/rust/core/src/secrets.rs b/rust/core/src/secrets.rs index 0e2203b..9dc55f3 100644 --- a/rust/core/src/secrets.rs +++ b/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 { + 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::*; diff --git a/rust/core/src/sessions.rs b/rust/core/src/sessions.rs new file mode 100644 index 0000000..a4a237e --- /dev/null +++ b/rust/core/src/sessions.rs @@ -0,0 +1,1163 @@ +//! The session manager: one entry per open tab, each owning a protocol client. A port of +//! internal/sessions — same lifecycle, reconnect backoff, login scripts, recording hooks and +//! credential resolution — reporting to the UI through `Event`s. + +use crate::clients::{self, Callbacks, github, irc, raw, serial, telnet}; +use crate::clients::local::LocalShell; +use crate::clients::ssh::{HostKeyCheck, HostKeyInfo, SecretFn, SshClient}; +use crate::models::*; +use crate::recording::{self, Recorder}; +use crate::secrets::Secrets; +use crate::storage::Store; +use crate::Result; +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// Everything the UI hears about. Prompts carry their reply channel; dropping it cancels. +pub enum Event { + Data { session: String, data: Vec }, + State { session: String, state: SessionState, error: String, retry: Option }, + Closed { session: String }, + RawData { session: String, data: Vec, direction: String, timestamp: i64 }, + Debug { session: String, message: String }, + GithubStatus { session: String, status: String }, + GithubRepos { session: String, login: String, repos: Vec }, + GithubError { session: String, error: String }, + Irc { session: String, event: irc::Event }, + ForwardStarted { session: String, forward: PortForward }, + ForwardStopped { session: String, forward_id: String }, + ForwardError { session: String, forward_id: String, label: String, error: String }, + RecordingStarted { session: String }, + RecordingStopped { session: String, duration: f64, events: usize }, + SftpProgress { session: String, progress: clients::sftp::Progress }, + SecretPrompt { session: String, title: String, message: String, reply: mpsc::Sender }, + HostKeyVerify { session: String, host: String, key_type: String, fingerprint: String, reply: mpsc::Sender }, + VncReady { session: String }, + VncFrame { session: String }, +} + +#[derive(Clone, Copy, Debug)] +pub struct Retry { + pub attempt: i64, + pub max: i64, + pub delay_secs: u64, +} + +pub type EventSink = Arc; +/// Asks the user for a credential (blocks the calling worker thread). +pub type Prompt = Arc Result + Send + Sync>; + +#[derive(Clone)] +enum Client { + None, + Ssh(Arc), + Telnet(Arc), + Raw(Arc), + Serial(Arc), + Local(Arc), + Github(Arc, Arc), + Irc(Arc, Option>), + Vnc(Arc, Option>), +} + +impl Client { + fn close(&self) { + match self { + Client::None => {} + Client::Ssh(c) => c.close(), + Client::Telnet(c) => c.close(), + Client::Raw(c) => c.close(), + Client::Serial(c) => c.close(), + Client::Local(c) => c.close(), + Client::Github(_, stop) => stop.store(true, Ordering::SeqCst), + Client::Irc(c, tunnel) => { + c.close(); + if let Some(t) = tunnel { + t.close(); + } + } + Client::Vnc(c, tunnel) => { + c.close(); + if let Some(t) = tunnel { + t.close(); + } + } + } + } +} + +#[derive(Default)] +struct LoginBuf { + active: bool, + buf: Vec, +} + +struct Session { + info: SessionInfo, + /// Distinguishes this Session object from a later one with the same id (Go compared pointers). + epoch: u64, + closed: bool, + client: Client, + profile: Option, + local_shell: String, + local_cwd: Option, + recorder: Option>, + reconnect_count: i64, + cancel_reconnect: Option>, + pending_resize: Option<(u16, u16)>, + login: Arc>, +} + +pub struct Manager { + sessions: Mutex>, + store: Arc, + secrets: Arc, + recordings: recording::Store, + sink: Mutex>, + next_epoch: std::sync::atomic::AtomicU64, +} + +fn err(msg: String) -> Result { + Err(msg.into()) +} + +impl Manager { + pub fn new(store: Arc, secrets: Arc, recordings_dir: PathBuf) -> Arc { + Arc::new(Manager { + sessions: Mutex::new(HashMap::new()), + store, + secrets, + recordings: recording::Store::new(recordings_dir), + sink: Mutex::new(None), + next_epoch: 1.into(), + }) + } + + pub fn set_sink(&self, sink: EventSink) { + *self.sink.lock().unwrap() = Some(sink); + } + + fn emit(&self, e: Event) { + let sink = self.sink.lock().unwrap().clone(); + if let Some(s) = sink { + s(e); + } + } + + fn emit_state(&self, id: &str, state: SessionState, error: &str) { + self.emit(Event::State { session: id.into(), state, error: error.into(), retry: None }); + } + + /// True while `id` still names the Session object with `epoch` and it isn't closed. + fn is_current(&self, id: &str, epoch: u64) -> bool { + self.sessions.lock().unwrap().get(id).is_some_and(|s| s.epoch == epoch && !s.closed) + } + + fn with(&self, id: &str, f: impl FnOnce(&mut Session) -> R) -> Result { + let mut map = self.sessions.lock().unwrap(); + match map.get_mut(id) { + Some(s) => Ok(f(s)), + None => err(format!("sessions: session {id:?} not found")), + } + } + + fn client(&self, id: &str) -> Result { + self.with(id, |s| s.client.clone()) + } + + fn new_session(&self, info: SessionInfo, profile: Option, shell: String) -> u64 { + let epoch = self.next_epoch.fetch_add(1, Ordering::SeqCst); + let s = Session { + info, + epoch, + closed: false, + client: Client::None, + profile, + local_shell: shell, + local_cwd: None, + recorder: None, + reconnect_count: 0, + cancel_reconnect: None, + pending_resize: None, + login: Arc::default(), + }; + let id = s.info.id.clone(); + self.sessions.lock().unwrap().insert(id, s); + epoch + } + + /// Opens a session for `profile` and connects it on a worker thread. Returns the session id + /// immediately; progress arrives as `Event::State`. + pub fn connect(self: &Arc, profile: Profile, prompt: Prompt) -> String { + let id = new_id(); + let info = SessionInfo { + id: id.clone(), + profile_id: profile.id.clone(), + profile_name: profile.display_name.clone(), + protocol: profile.protocol, + host: profile.host.clone(), + port: profile.port, + state: SessionState::Connecting, + tab_name: profile.tab_name(), + pinned: false, + connected_at: now(), + error: String::new(), + }; + let epoch = self.new_session(info, Some(profile.clone()), String::new()); + self.emit_state(&id, SessionState::Connecting, ""); + let me = self.clone(); + let id2 = id.clone(); + std::thread::spawn(move || { + let r = me.connect_protocol(&id2, epoch, &profile, prompt); + me.finish_connect(&id2, epoch, r, |me| me.store.add_audit("connect", &profile.id, &id2, &format!("Connected via {} to {}", profile.protocol.id(), profile.host))); + }); + id + } + + fn finish_connect(&self, id: &str, epoch: u64, r: Result, on_ok: impl FnOnce(&Self)) { + let mut map = self.sessions.lock().unwrap(); + let current = map.get_mut(id).filter(|s| s.epoch == epoch && !s.closed); + match (r, current) { + (Ok(client), None) => { + drop(map); + client.close(); + } + (Err(_), None) => {} + (Err(e), Some(s)) => { + s.info.state = SessionState::Error; + s.info.error = e.to_string(); + drop(map); + self.emit_state(id, SessionState::Error, &e.to_string()); + } + (Ok(client), Some(s)) => { + s.client = client.clone(); + s.info.state = SessionState::Connected; + s.info.connected_at = now(); + s.reconnect_count = 0; + let pending = s.pending_resize.take(); + drop(map); + if let Some((c, r)) = pending { + let _ = self.resize_client(&client, c, r); + } + self.emit_state(id, SessionState::Connected, ""); + on_ok(self); + } + } + } + + fn connect_protocol(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { + match p.protocol { + Protocol::Ssh => self.connect_ssh(id, epoch, p, prompt), + Protocol::Telnet => { + let debug = p.telnet.as_ref().is_some_and(|t| t.negotiation_debug).then(|| { + let (me, id) = (self.clone(), id.to_string()); + Arc::new(move |m: String| me.emit(Event::Debug { session: id.clone(), message: m })) as clients::DebugFn + }); + let c = Arc::new(telnet::TelnetClient::new(p, self.terminal_callbacks(id, epoch), debug)); + c.connect()?; + self.start_login_scripts(id, epoch, p.telnet.as_ref().map(|t| t.login_scripts.clone()).unwrap_or_default()); + Ok(Client::Telnet(c)) + } + Protocol::Raw => { + let cb = self.terminal_callbacks(id, epoch); + let (me, sid) = (self.clone(), id.to_string()); + let on_data: raw::RawDataFn = Arc::new(move |d, dir, ts| { + me.emit(Event::RawData { session: sid.clone(), data: d.to_vec(), direction: dir.into(), timestamp: ts }) + }); + let c = Arc::new(raw::RawClient::new(p, on_data, cb.on_error, cb.on_close)); + c.connect()?; + Ok(Client::Raw(c)) + } + Protocol::Serial => { + let c = Arc::new(serial::SerialClient::new(p, self.terminal_callbacks(id, epoch))); + c.connect()?; + Ok(Client::Serial(c)) + } + Protocol::Github => self.connect_github(id, epoch, p, prompt), + Protocol::Irc => self.connect_irc(id, epoch, p, prompt), + Protocol::Vnc => self.connect_vnc(id, epoch, p, prompt), + Protocol::Local => err("unsupported protocol \"local\"".into()), + } + } + + /// Callbacks for terminal-style clients: output to the UI, recording and login scripts. + fn terminal_callbacks(self: &Arc, id: &str, epoch: u64) -> Callbacks { + let login = self.sessions.lock().unwrap().get(id).map(|s| s.login.clone()).unwrap_or_default(); + let (me, sid) = (self.clone(), id.to_string()); + let on_data = Arc::new(move |d: &[u8]| { + me.emit(Event::Data { session: sid.clone(), data: d.to_vec() }); + let rec = me.sessions.lock().unwrap().get(&sid).and_then(|s| s.recorder.clone()); + if let Some(r) = rec { + r.write_output(d); + } + let mut l = login.lock().unwrap(); + if l.active { + l.buf.extend_from_slice(d); + if l.buf.len() > 16384 { + let cut = l.buf.len() - 16384; + l.buf.drain(..cut); + } + } + }); + let (me, sid) = (self.clone(), id.to_string()); + let on_error = Arc::new(move |e: String| { + let ok = me + .with(&sid, |s| { + if s.epoch != epoch { + return false; + } + s.info.state = SessionState::Error; + s.info.error = e.clone(); + true + }) + .unwrap_or(false); + if ok { + me.emit_state(&sid, SessionState::Error, &e); + } + }); + let (me, sid) = (self.clone(), id.to_string()); + let on_close = Arc::new(move || { + let me = me.clone(); + let sid = sid.clone(); + // Close callbacks can fire inside a client's own lock; handle them off-thread. + std::thread::spawn(move || me.handle_closed(&sid, epoch)); + }); + Callbacks { on_data, on_error, on_close } + } + + // --- credentials -------------------------------------------------------------------- + + /// Resolves a credential per its SecretMode (Go's buildSecretCallback): saved values come from + /// the keychain then the vault; session-only values are asked once per connection. + pub fn secret_fn(&self, profile_id: &str, kind: &str, mode: SecretMode, prompt: Prompt) -> SecretFn { + let key = format!("profile:{profile_id}:{kind}"); + let (secrets, store) = (self.secrets.clone(), self.store.clone()); + let (pid, kind) = (profile_id.to_string(), kind.to_string()); + let cache: Arc>> = Arc::default(); + Arc::new(move || { + if mode == SecretMode::Saved + && let Some(v) = secrets.get(&key) + { + store.add_audit("secret.access", &pid, "", &format!("Accessed {kind} credential")); + return Ok(v); + } + if mode == SecretMode::SessionOnly + && let Some(v) = cache.lock().unwrap().clone() + { + return Ok(v); + } + let v = prompt(&format!("Enter {kind}:"))?; + match mode { + SecretMode::Saved => secrets.set(&key, &v), + SecretMode::SessionOnly => *cache.lock().unwrap() = Some(v.clone()), + SecretMode::AskEveryTime => {} + } + Ok(v) + }) + } + + /// Blocks until the user answers a credential prompt (5 minutes, or cancel). + pub fn prompt(&self, session: &str, title: &str, message: &str, timeout: Duration) -> Result { + let (tx, rx) = mpsc::channel(); + self.emit(Event::SecretPrompt { session: session.into(), title: title.into(), message: message.into(), reply: tx }); + match rx.recv_timeout(timeout) { + Ok(v) => Ok(v), + Err(mpsc::RecvTimeoutError::Timeout) => err("secret prompt timed out".into()), + Err(_) => err("user cancelled secret prompt".into()), + } + } + + /// The standard interactive prompt for a session's credentials. + pub fn prompt_for(self: &Arc, session: &str, title: &str, timeout: Duration) -> Prompt { + let (me, sid, title) = (self.clone(), session.to_string(), title.to_string()); + Arc::new(move |msg| me.prompt(&sid, &title, msg, timeout)) + } + + /// Host-key policy check against the known-hosts table (Go's HostKeyVerifier). + pub fn host_key_check(self: &Arc, session: &str, policy: HostKeyPolicy) -> HostKeyCheck { + let (me, sid) = (self.clone(), session.to_string()); + Arc::new(move |k: &HostKeyInfo| { + let stored = me.store.get_known_host(&k.host, k.port as i64).map_err(|e| format!("failed to query known hosts: {e}"))?; + let save = || { + let n = now(); + me.store + .save_known_host(&KnownHost { + id: String::new(), + host: k.host.clone(), + port: k.port as i64, + key_type: k.key_type.clone(), + fingerprint: k.fingerprint.clone(), + public_key: k.public_key_b64.clone(), + first_seen: n, + last_seen: n, + trusted: true, + }) + .map_err(|e| e.to_string()) + }; + let verify = |mut s: KnownHost| { + if s.fingerprint != k.fingerprint { + return Err(format!( + "host key for {} has changed! stored fingerprint: {}, received: {} — possible MITM attack", + k.host, s.fingerprint, k.fingerprint + )); + } + s.last_seen = now(); + let _ = me.store.save_known_host(&s); + Ok(()) + }; + match (policy, stored) { + (HostKeyPolicy::Insecure, None) | (HostKeyPolicy::AcceptNew, None) => save(), + (HostKeyPolicy::Insecure, Some(mut s)) => { + s.last_seen = now(); + let _ = me.store.save_known_host(&s); + Ok(()) + } + (HostKeyPolicy::Strict, None) => Err(format!("host key for {}:{} is not known (policy: strict)", k.host, k.port)), + (HostKeyPolicy::Ask, None) => { + let (tx, rx) = mpsc::channel(); + me.emit(Event::HostKeyVerify { + session: sid.clone(), + host: k.host.clone(), + key_type: k.key_type.clone(), + fingerprint: k.fingerprint.clone(), + reply: tx, + }); + match rx.recv_timeout(Duration::from_secs(60)) { + Ok(true) => save(), + Ok(false) | Err(mpsc::RecvTimeoutError::Disconnected) => Err(format!("host key for {}:{} rejected by user", k.host, k.port)), + Err(_) => Err("host key prompt failed: host key prompt timed out waiting for user response".into()), + } + } + (_, Some(s)) => verify(s), + } + }) + } + + // --- per-protocol connect ----------------------------------------------------------- + + fn connect_ssh(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { + let mut p = p.clone(); + let sc = p.ssh.get_or_insert_with(SshConfig::defaults); + if sc.totp_secret.is_empty() + && let Some(t) = self.secrets.get(&format!("profile:{}:totp", p.id)) + { + sc.totp_secret = t; + } + let sc = sc.clone(); + let password = self.secret_fn(&p.id, "password", sc.secret_mode, prompt.clone()); + let passphrase = self.secret_fn(&p.id, "passphrase", sc.key_secret_mode, prompt); + let c = SshClient::new(&p, self.terminal_callbacks(id, epoch)); + c.connect(self.host_key_check(id, sc.host_key_policy), Some(password), Some(passphrase))?; + for pf in sc.port_forwards.iter().filter(|f| f.enabled) { + match c.start_forward(pf.clone()) { + Ok(()) => self.emit(Event::ForwardStarted { session: id.into(), forward: pf.clone() }), + Err(e) => self.emit(Event::ForwardError { session: id.into(), forward_id: pf.id.clone(), label: pf.label.clone(), error: e.to_string() }), + } + } + self.start_login_scripts(id, epoch, sc.login_scripts.clone()); + Ok(Client::Ssh(c)) + } + + /// An SSH connection used only to tunnel another protocol (IRC, VNC). + #[allow(clippy::too_many_arguments)] + fn tunnel(self: &Arc, id: &str, profile_id: &str, host: &str, port: i64, user: &str, method: &str, key: &str, agent: bool, mode: SecretMode, kind: &str, prompt: Prompt) -> Result> { + let mut tp = Profile::new(Protocol::Ssh); + tp.host = host.into(); + tp.port = if port == 0 { 22 } else { port }; + tp.username = user.into(); + let c = tp.ssh.as_mut().unwrap(); + c.auth_methods = vec![if method.is_empty() { "publickey".into() } else { method.into() }]; + c.private_key_path = key.into(); + c.agent_auth = agent; + let password = (method == "password").then(|| self.secret_fn(profile_id, &format!("{kind}-ssh-password"), mode, prompt.clone())); + let passphrase = self.secret_fn(profile_id, &format!("{kind}-ssh-passphrase"), mode, prompt); + let t = SshClient::new(&tp, Callbacks::noop()); + t.connect_tunnel_only(self.host_key_check(id, HostKeyPolicy::Ask), password, Some(passphrase))?; + Ok(t) + } + + fn connect_github(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { + let gh = p.github.clone().ok_or_else(|| format!("GitHub configuration missing on profile {:?}", p.id))?; + let token = self.secret_fn(&p.id, "github-token", gh.secret_mode, prompt.clone())().map_err(|e| format!("obtaining GitHub token: {e}"))?; + if token.is_empty() { + return err("GitHub token cannot be empty".into()); + } + let mut client = github::Client::new(&gh.base_url, &token); + let login = match client.validate_token() { + Ok(l) => l, + Err(e) => { + let msg = e.to_string().to_lowercase(); + let auth = ["token validation failed", "authentication failed", "http 401", "http 403"].iter().any(|k| msg.contains(k)); + let retry = auth + .then(|| prompt("GitHub token did not work. Please enter a valid token:").ok()) + .flatten() + .filter(|t| !t.is_empty()); + let Some(t) = retry else { return err(format!("GitHub authentication failed: {e}")) }; + if gh.secret_mode == SecretMode::Saved { + self.secrets.set(&format!("profile:{}:github-token", p.id), &t); + } + client = github::Client::new(&gh.base_url, &t); + client.validate_token().map_err(|e| format!("GitHub authentication failed: {e}"))? + } + }; + let client = Arc::new(client); + if !self.is_current(id, epoch) { + return err(format!("session {id:?} was closed while connecting")); + } + let fetch = { + let (me, sid, client) = (self.clone(), id.to_string(), client.clone()); + move |status: &str| { + me.emit(Event::GithubStatus { session: sid.clone(), status: status.into() }); + let (me2, sid2) = (me.clone(), sid.clone()); + client.fetch_repos_with_progress(move |done, total| { + if total > 0 { + me2.emit(Event::GithubStatus { session: sid2.clone(), status: format!("Loading commit info ({done}/{total})") }); + } + }) + } + }; + let repos = fetch("Loading repositories...").map_err(|e| format!("fetching repositories: {e}"))?; + self.emit(Event::GithubRepos { session: id.into(), login: login.clone(), repos }); + let interval = Duration::from_secs(if gh.refresh_interval_secs < 60 { 300 } else { gh.refresh_interval_secs as u64 }); + let stop = Arc::new(AtomicBool::new(false)); + let (me, sid, stop2) = (self.clone(), id.to_string(), stop.clone()); + std::thread::spawn(move || { + loop { + let deadline = std::time::Instant::now() + interval; + while std::time::Instant::now() < deadline { + if stop2.load(Ordering::SeqCst) { + return; + } + std::thread::sleep(Duration::from_millis(500)); + } + match fetch("Refreshing repositories...") { + Ok(repos) => me.emit(Event::GithubRepos { session: sid.clone(), login: login.clone(), repos }), + Err(e) => me.emit(Event::GithubError { session: sid.clone(), error: e.to_string() }), + } + } + }); + Ok(Client::Github(client, stop)) + } + + fn connect_irc(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { + let ic = p.irc.clone().ok_or("IRC configuration missing")?; + let server_pass = if ic.server_password_required { + self.secret_fn(&p.id, "irc-server-password", ic.server_password_mode, prompt.clone())().map_err(|e| format!("IRC server password: {e}"))? + } else { + String::new() + }; + let ns_pass = if ic.nick_serv_identify { + self.secret_fn(&p.id, "irc-nickserv-password", ic.nick_serv_mode, prompt.clone())().map_err(|e| format!("NickServ password: {e}"))? + } else { + String::new() + }; + let mut rooms = ic.auto_join_rooms.clone(); + if !p.id.is_empty() { + let last = self.store.irc_last_rooms(&p.id).map_err(|e| format!("loading remembered IRC rooms: {e}"))?; + rooms = crate::storage::dedup_rooms(&[&rooms, &last]); + } + let tunnel = if ic.ssh_tunnel { + Some(self.tunnel(id, &p.id, &ic.ssh_host, ic.ssh_port, &ic.ssh_username, &ic.ssh_auth_method, &ic.ssh_private_key, ic.ssh_agent_auth, ic.ssh_secret_mode, "irc", prompt)?) + } else { + None + }; + let cb = self.terminal_callbacks(id, epoch); + let (me, sid) = (self.clone(), id.to_string()); + let on_event: irc::EventFn = Arc::new(move |e| me.emit(Event::Irc { session: sid.clone(), event: e })); + let (me, sid, pid) = (self.clone(), id.to_string(), p.id.clone()); + let on_rooms: irc::RoomsChangedFn = Arc::new(move |rooms: Vec| { + if pid.is_empty() { + return; + } + if let Err(e) = me.store.save_irc_last_rooms(&pid, &rooms) { + me.emit(Event::Irc { + session: sid.clone(), + event: irc::Event { + kind: "system".into(), + target: irc::SERVER_TARGET.into(), + text: format!("Could not remember open IRC rooms: {e}"), + timestamp: chrono::Utc::now().timestamp_millis(), + ..Default::default() + }, + }); + } + }); + let connector: Option = tunnel.clone().map(|t| { + let (host, port) = (p.host.clone(), p.port as u16); + Arc::new(move || { + let s = t.open_tunnel(&host, port)?; + s.set_read_timeout(Some(Duration::from_millis(50)))?; + Ok(Box::new(s) as Box) + }) as irc::Connector + }); + let cfg = irc::Config { + server: p.host.clone(), + port: p.port as u16, + nick: ic.nick.clone(), + user: if p.username.is_empty() { ic.nick.clone() } else { p.username.clone() }, + real_name: ic.real_name.clone(), + use_tls: ic.use_tls, + server_pass, + auto_join_rooms: rooms, + nickserv_identify: ic.nick_serv_identify, + nickserv_account: ic.nick_serv_account.clone(), + nickserv_password: ns_pass, + connector, + connect_timeout: Duration::from_secs(p.connect_timeout.max(1) as u64), + on_event, + on_error: cb.on_error, + on_close: cb.on_close, + on_rooms_changed: on_rooms, + }; + let c = irc::Client::new(cfg); + if let Err(e) = c.connect() { + if let Some(t) = tunnel { + t.close(); + } + return Err(e); + } + Ok(Client::Irc(c, tunnel)) + } + + fn connect_vnc(self: &Arc, id: &str, epoch: u64, p: &Profile, prompt: Prompt) -> Result { + let vc = p.vnc.clone().ok_or("VNC configuration missing")?; + let host = if vc.vnc_host.is_empty() { p.host.clone() } else { vc.vnc_host.clone() }; + let port = if vc.vnc_port == 0 { 5900 } else { vc.vnc_port } as u16; + let tunnel = if vc.ssh_tunnel { + Some(self.tunnel(id, &p.id, &vc.ssh_host, vc.ssh_port, &vc.ssh_username, &vc.ssh_auth_method, &vc.ssh_private_key, vc.ssh_agent_auth, vc.ssh_secret_mode, "vnc", prompt.clone())?) + } else { + None + }; + let password = self.secret_fn(&p.id, "vnc-password", vc.secret_mode, prompt); + let cb = self.terminal_callbacks(id, epoch); + let (me, sid) = (self.clone(), id.to_string()); + let c = vnc_stub::connect_profile(&host, port, &vc, p.connect_timeout, password, tunnel.clone(), cb, Arc::new(move || me.emit(Event::VncFrame { session: sid.clone() }))); + match c { + Ok(c) => { + self.emit(Event::VncReady { session: id.into() }); + Ok(Client::Vnc(c, tunnel)) + } + Err(e) => { + if let Some(t) = tunnel { + t.close(); + } + Err(e) + } + } + } + + // --- local shell -------------------------------------------------------------------- + + pub fn connect_local(self: &Arc, shell: &str) -> String { + let id = new_id(); + let info = SessionInfo { + id: id.clone(), + profile_id: String::new(), + profile_name: "Local Machine".into(), + protocol: Protocol::Local, + host: "localhost".into(), + port: 0, + state: SessionState::Connecting, + tab_name: "Local".into(), + pinned: false, + connected_at: now(), + error: String::new(), + }; + let epoch = self.new_session(info, None, shell.into()); + self.emit_state(&id, SessionState::Connecting, ""); + let me = self.clone(); + let id2 = id.clone(); + std::thread::spawn(move || me.start_local(&id2, epoch, 80, 24)); + id + } + + /// Starts a fresh shell in an exited local tab, keeping its scrollback and last directory. + pub fn restart_local(self: &Arc, id: &str, cols: u16, rows: u16) -> Result<()> { + let epoch = self.with(id, |s| { + if s.info.protocol != Protocol::Local { + return Err(format!("sessions: session {id:?} is not local")); + } + if matches!(s.client, Client::Local(_)) || matches!(s.info.state, SessionState::Connecting | SessionState::Connected) { + return Err(format!("sessions: local session {id:?} is still running")); + } + s.info.state = SessionState::Connecting; + s.info.error.clear(); + Ok(s.epoch) + })??; + self.emit_state(id, SessionState::Connecting, ""); + let (me, id2) = (self.clone(), id.to_string()); + std::thread::spawn(move || me.start_local(&id2, epoch, cols, rows)); + Ok(()) + } + + fn start_local(self: &Arc, id: &str, epoch: u64, cols: u16, rows: u16) { + let Ok((shell, cwd, pending)) = self.with(id, |s| (s.local_shell.clone(), s.local_cwd.clone(), s.pending_resize.take())) else { return }; + let (cols, rows) = pending.unwrap_or((cols, rows)); + let mut cb = self.terminal_callbacks(id, epoch); + // A local shell exiting keeps its tab ("press enter to restart"). + let (me, sid) = (self.clone(), id.to_string()); + cb.on_close = Arc::new(move || { + let (me, sid) = (me.clone(), sid.clone()); + std::thread::spawn(move || me.local_closed(&sid, epoch)); + }); + let c = LocalShell::new(&shell, cwd, cols, rows, cb); + let r = c.start().map(|_| Client::Local(c.clone())); + self.finish_connect(id, epoch, r, |me| me.store.add_audit("connect", "", id, "Connected via local to localhost")); + } + + fn local_closed(self: &Arc, id: &str, epoch: u64) { + let cwd = self.with(id, |s| { + if s.epoch != epoch || s.closed { + return None; + } + match std::mem::replace(&mut s.client, Client::None) { + Client::Local(c) => Some(c.working_directory()), + other => { + s.client = other; + None + } + } + }); + if let Ok(Some(cwd)) = cwd { + let _ = self.with(id, |s| s.local_cwd = Some(cwd)); + self.handle_closed(id, epoch); + } + } + + // --- lifecycle ---------------------------------------------------------------------- + + /// Explicit close: the tab is gone, so no reconnect. + pub fn disconnect(&self, id: &str) -> Result<()> { + let s = self.sessions.lock().unwrap().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; + self.close_session(s); + self.store.add_audit("disconnect", "", id, "Session disconnected"); + Ok(()) + } + + /// Closes and re-opens a session from its profile, keeping the tab name and pin; returns the + /// new session id. + pub fn reconnect(self: &Arc, id: &str, prompt: Prompt) -> Result { + let s = self.sessions.lock().unwrap().remove(id).ok_or_else(|| format!("sessions: session {id:?} not found"))?; + let (profile, tab, pinned, proto, shell) = (s.profile.clone(), s.info.tab_name.clone(), s.info.pinned, s.info.protocol, s.local_shell.clone()); + self.close_session(s); + let new = match (proto, profile) { + (Protocol::Local, _) => self.connect_local(&shell), + (_, Some(p)) => self.connect(p, prompt), + _ => return err("sessions: nothing to reconnect".into()), + }; + let _ = self.with(&new, |s| { + s.info.tab_name = tab; + s.info.pinned = pinned; + }); + Ok(new) + } + + fn close_session(&self, mut s: Session) { + s.closed = true; + if let Some(c) = s.cancel_reconnect.take() { + c.store(true, Ordering::SeqCst); + } + if let Some(r) = s.recorder.take() { + let _ = r.close(); + } + s.client.close(); + } + + fn handle_closed(self: &Arc, id: &str, epoch: u64) { + let profile = { + let mut map = self.sessions.lock().unwrap(); + let Some(s) = map.get_mut(id).filter(|s| s.epoch == epoch && !s.closed) else { return }; + s.info.state = SessionState::Disconnected; + if !matches!(s.client, Client::Local(_)) { + s.client = Client::None; + } + s.profile.clone() + }; + self.store.add_audit("disconnect", "", id, "Session disconnected"); + if let Some(p) = profile.filter(|p| p.auto_reconnect()) { + self.schedule_reconnect(id, epoch, p); + return; + } + self.emit(Event::Closed { session: id.into() }); + self.emit_state(id, SessionState::Disconnected, ""); + } + + /// Exponential backoff 1, 2, 4 … 30 s, up to the profile's attempt limit. + fn schedule_reconnect(self: &Arc, id: &str, epoch: u64, p: Profile) { + let max = p.max_reconnect_tries(); + let cancel = Arc::new(AtomicBool::new(false)); + let attempt = self.with(id, |s| { + s.reconnect_count += 1; + if s.reconnect_count > max { + s.info.state = SessionState::Disconnected; + return None; + } + s.info.state = SessionState::Reconnecting; + s.info.error.clear(); + if let Some(c) = s.cancel_reconnect.replace(cancel.clone()) { + c.store(true, Ordering::SeqCst); + } + Some(s.reconnect_count) + }); + let Ok(attempt) = attempt else { return }; + let Some(attempt) = attempt else { + self.emit(Event::Closed { session: id.into() }); + self.emit_state(id, SessionState::Disconnected, "auto-reconnect failed after maximum attempts"); + return; + }; + let delay = (1u64 << (attempt - 1).min(5)).min(30); + self.emit(Event::State { session: id.into(), state: SessionState::Reconnecting, error: String::new(), retry: Some(Retry { attempt, max, delay_secs: delay }) }); + let (me, id) = (self.clone(), id.to_string()); + std::thread::spawn(move || { + let deadline = std::time::Instant::now() + Duration::from_secs(delay); + while std::time::Instant::now() < deadline { + if cancel.load(Ordering::SeqCst) { + return; + } + std::thread::sleep(Duration::from_millis(100)); + } + let ok = me + .with(&id, |s| { + if s.epoch != epoch || s.closed || cancel.load(Ordering::SeqCst) { + return false; + } + s.client = Client::None; + s.info.state = SessionState::Connecting; + true + }) + .unwrap_or(false); + if !ok { + return; + } + me.emit_state(&id, SessionState::Connecting, ""); + let prompt = me.prompt_for(&id, "Reconnection Credentials", Duration::from_secs(60)); + match me.connect_protocol(&id, epoch, &p, prompt) { + Ok(c) => me.finish_connect(&id, epoch, Ok(c), |_| {}), + Err(e) => { + let current = me.with(&id, |s| { + if s.epoch != epoch || s.closed { + return false; + } + s.info.state = SessionState::Error; + s.info.error = e.to_string(); + true + }); + if current.unwrap_or(false) { + me.emit_state(&id, SessionState::Error, &e.to_string()); + me.schedule_reconnect(&id, epoch, p); + } + } + } + }); + } + + pub fn cancel_reconnect(&self, id: &str) { + let found = self + .with(id, |s| { + if let Some(c) = s.cancel_reconnect.take() { + c.store(true, Ordering::SeqCst); + s.reconnect_count = 0; + s.info.state = SessionState::Disconnected; + } + }) + .is_ok(); + if found { + self.emit_state(id, SessionState::Disconnected, ""); + } + } + + pub fn close_all(&self) { + let all: Vec = self.sessions.lock().unwrap().drain().map(|(_, s)| s).collect(); + for s in all { + self.close_session(s); + } + } + + // --- login scripts ------------------------------------------------------------------ + + fn start_login_scripts(self: &Arc, id: &str, epoch: u64, steps: Vec) { + if steps.is_empty() { + return; + } + let Ok(login) = self.with(id, |s| s.login.clone()) else { return }; + *login.lock().unwrap() = LoginBuf { active: true, buf: vec![] }; + let (me, id) = (self.clone(), id.to_string()); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(500)); + for (i, step) in steps.iter().enumerate() { + // Wait until the connect has been recorded as Connected. + let mut waited = 0; + while me.with(&id, |s| s.info.state).ok() == Some(SessionState::Connecting) && waited < 100 { + std::thread::sleep(Duration::from_millis(50)); + waited += 1; + } + if !me.is_current(&id, epoch) || me.with(&id, |s| s.info.state).ok() != Some(SessionState::Connected) { + break; + } + match step.kind.as_str() { + "wait" => { + let timeout = Duration::from_secs(if step.timeout <= 0 { 10 } else { step.timeout as u64 }); + if !wait_for_output(&login, &step.value, timeout) { + me.emit(Event::Debug { session: id.clone(), message: format!("login script step {} timed out waiting for {:?}", i + 1, step.value) }); + break; + } + } + "send" => { + if let Err(e) = me.write(&id, format!("{}\n", step.value).as_bytes()) { + me.emit(Event::Debug { session: id.clone(), message: format!("login script step {} failed to send: {e}", i + 1) }); + break; + } + std::thread::sleep(Duration::from_millis(200)); + } + _ => {} + } + } + *login.lock().unwrap() = LoginBuf::default(); + }); + } + + // --- I/O ---------------------------------------------------------------------------- + + pub fn write(&self, id: &str, data: &[u8]) -> Result<()> { + let (client, rec) = self.with(id, |s| (s.client.clone(), s.recorder.clone()))?; + if let Some(r) = rec { + r.write_input(data); + } + match client { + Client::Ssh(c) => c.write(data), + Client::Telnet(c) => c.write(data), + Client::Raw(c) => c.write(data), + Client::Serial(c) => c.write(data), + Client::Local(c) => c.write(data), + _ => err(format!("sessions: no active client for session {id:?}")), + } + } + + fn resize_client(&self, c: &Client, cols: u16, rows: u16) -> Result<()> { + if cols == 0 || rows == 0 { + return Ok(()); + } + match c { + Client::Ssh(c) => c.resize(cols as u32, rows as u32), + Client::Telnet(c) => c.resize(cols, rows), + Client::Local(c) => c.resize(cols, rows), + _ => Ok(()), + } + } + + /// Resizes the remote PTY; before a client exists the size is kept and applied on connect. + pub fn resize(&self, id: &str, cols: u16, rows: u16) -> Result<()> { + let c = self.with(id, |s| { + if matches!(s.client, Client::None) { + s.pending_resize = Some((cols, rows)); + } + s.client.clone() + })?; + self.resize_client(&c, cols, rows) + } + + pub fn info(&self, id: &str) -> Option { + self.sessions.lock().unwrap().get(id).map(|s| s.info.clone()) + } + + pub fn list(&self) -> Vec { + self.sessions.lock().unwrap().values().map(|s| s.info.clone()).collect() + } + + pub fn profile(&self, id: &str) -> Option { + self.sessions.lock().unwrap().get(id).and_then(|s| s.profile.clone()) + } + + pub fn rename(&self, id: &str, name: &str) -> Result<()> { + self.with(id, |s| s.info.tab_name = name.into()) + } + + pub fn pin(&self, id: &str, pinned: bool) -> Result<()> { + self.with(id, |s| s.info.pinned = pinned) + } + + pub fn send_break(&self, id: &str) -> Result<()> { + match self.client(id)? { + Client::Telnet(c) => c.send_break(), + Client::Serial(c) => c.send_break(), + _ => err(format!("sessions: session {id:?} does not support break")), + } + } + + pub fn set_serial_lines(&self, id: &str, dtr: bool, rts: bool) -> Result<()> { + match self.client(id)? { + Client::Serial(c) => c.set_lines(dtr, rts), + _ => err(format!("sessions: session {id:?} is not a serial connection")), + } + } + + // --- RAW ---------------------------------------------------------------------------- + + fn raw(&self, id: &str) -> Result> { + match self.client(id)? { + Client::Raw(c) => Ok(c), + _ => err(format!("sessions: session {id:?} is not a RAW connection")), + } + } + + pub fn raw_write_text(&self, id: &str, s: &str) -> Result<()> { + self.raw(id)?.write_text(s) + } + + pub fn raw_write_hex(&self, id: &str, s: &str) -> Result<()> { + self.raw(id)?.write_hex(s) + } + + pub fn raw_write_escaped(&self, id: &str, s: &str) -> Result<()> { + self.raw(id)?.write_escaped(s) + } + + pub fn raw_history(&self, id: &str) -> Result> { + Ok(self.raw(id)?.history()) + } + + pub fn raw_clear_history(&self, id: &str) -> Result<()> { + self.raw(id)?.clear_history(); + Ok(()) + } + + pub fn raw_save_capture(&self, id: &str, path: &std::path::Path) -> Result<()> { + self.raw(id)?.save_capture(path) + } + + // --- IRC ---------------------------------------------------------------------------- + + pub fn irc(&self, id: &str) -> Result> { + match self.client(id)? { + Client::Irc(c, _) => Ok(c), + _ => err(format!("sessions: session {id:?} is not an IRC connection")), + } + } + + // --- GitHub ------------------------------------------------------------------------- + + pub fn github(&self, id: &str) -> Result> { + match self.client(id)? { + Client::Github(c, _) => Ok(c), + _ => err(format!("sessions: session {id:?} is not a GitHub connection")), + } + } + + pub fn vnc(&self, id: &str) -> Result> { + match self.client(id)? { + Client::Vnc(c, _) => Ok(c), + _ => err(format!("sessions: session {id:?} is not a VNC connection")), + } + } + + // --- SSH extras --------------------------------------------------------------------- + + pub fn ssh(&self, id: &str) -> Result> { + match self.client(id)? { + Client::Ssh(c) => Ok(c), + _ => err(format!("sessions: session {id:?} is not an SSH connection")), + } + } + + pub fn ping(&self, id: &str) -> Result { + self.ssh(id)?.ping() + } + + pub fn cwd(&self, id: &str) -> Result { + self.ssh(id)?.run_command("pwd") + } + + pub fn start_forward(&self, id: &str, rule: PortForward) -> Result<()> { + self.ssh(id)?.start_forward(rule.clone())?; + self.emit(Event::ForwardStarted { session: id.into(), forward: rule }); + Ok(()) + } + + pub fn stop_forward(&self, id: &str, fid: &str) -> Result<()> { + self.ssh(id)?.stop_forward(fid)?; + self.emit(Event::ForwardStopped { session: id.into(), forward_id: fid.into() }); + Ok(()) + } + + pub fn forwards(&self, id: &str) -> Result> { + Ok(self.ssh(id)?.active_forwards()) + } + + pub fn sftp_download(&self, id: &str, remote: &str, local: &std::path::Path) -> Result<()> { + let s = self.ssh(id)?.sftp()?; + clients::sftp::download(&s, remote, local, &|p| self.emit(Event::SftpProgress { session: id.into(), progress: p })) + } + + pub fn sftp_upload(&self, id: &str, local: &std::path::Path, remote: &str) -> Result<()> { + let s = self.ssh(id)?.sftp()?; + clients::sftp::upload(&s, local, remote, &|p| self.emit(Event::SftpProgress { session: id.into(), progress: p })) + } + + // --- recording ---------------------------------------------------------------------- + + pub fn start_recording(&self, id: &str, cols: u16, rows: u16) -> Result<()> { + let (exists, title) = self.with(id, |s| (s.recorder.is_some(), format!("{} — {}", s.info.profile_name, s.info.host)))?; + if exists { + return err(format!("sessions: session {id:?} is already being recorded")); + } + let r = Arc::new(Recorder::new(self.recordings.dir(), id, &title, cols as i64, rows as i64).map_err(|e| format!("sessions: failed to start recording: {e}"))?); + self.with(id, |s| s.recorder = Some(r))?; + self.emit(Event::RecordingStarted { session: id.into() }); + Ok(()) + } + + pub fn stop_recording(&self, id: &str) -> Result<()> { + let r = self.with(id, |s| s.recorder.take())?.ok_or_else(|| format!("sessions: session {id:?} is not being recorded"))?; + r.close().map_err(|e| format!("sessions: failed to stop recording: {e}"))?; + self.emit(Event::RecordingStopped { session: id.into(), duration: r.duration(), events: r.event_count() }); + Ok(()) + } + + pub fn is_recording(&self, id: &str) -> bool { + self.with(id, |s| s.recorder.is_some()).unwrap_or(false) + } + + pub fn recordings(&self) -> &recording::Store { + &self.recordings + } +} + +// ponytail: placeholder until the VNC client lands; replaced by clients::vnc. +mod vnc_stub { + use crate::clients::Callbacks; + use crate::clients::ssh::{SecretFn, SshClient}; + use std::sync::Arc; + pub struct VncClient; + impl VncClient { + pub fn close(&self) {} + } + #[allow(clippy::too_many_arguments)] + pub fn connect_profile(_: &str, _: u16, _: &crate::models::VncConfig, _: i64, _: SecretFn, _: Option>, _: Callbacks, _: Arc) -> crate::Result> { + Err("VNC is not available yet".into()) + } +} + +fn wait_for_output(login: &Mutex, pattern: &str, timeout: Duration) -> bool { + if pattern.is_empty() { + return true; + } + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + { + let mut l = login.lock().unwrap(); + let text = String::from_utf8_lossy(&l.buf).into_owned(); + if let Some(i) = text.find(pattern) { + let consumed = text[..i + pattern.len()].len(); + let n = consumed.min(l.buf.len()); + l.buf.drain(..n); + return true; + } + } + std::thread::sleep(Duration::from_millis(100)); + } + false +} diff --git a/rust/core/tests/sessions.rs b/rust/core/tests/sessions.rs new file mode 100644 index 0000000..aba647e --- /dev/null +++ b/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, mpsc::Receiver, 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(rx: &mpsc::Receiver, mut f: impl FnMut(&Event) -> Option) -> 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 { + 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); +}