use std::{ collections::HashMap, io::{self, Read, Write}, net::{Shutdown, TcpStream}, sync::Mutex, thread, time::Duration, }; use ssh2::Session; use tauri::{AppHandle, Emitter}; use uuid::Uuid; use crate::domain::ssh::{ ConnectionClosedEvent, ExecResponse, ForwardStatus, ShellClosedEvent, ShellDataEvent, SshConnectRequest, }; use super::tunnel::TunnelManager; mod auth; mod known_hosts; mod path; mod proxy; mod sftp; use auth::authenticate; use known_hosts::verify_host_key; use proxy::connect_transport; #[derive(Clone)] struct Connection { session: Session, config: SshConnectRequest, } struct Shell { connection_id: String, session: Session, channel: ssh2::Channel, } #[derive(Default)] pub struct SshService { connections: Mutex>, shells: Mutex>, sftps: Mutex>, tunnels: TunnelManager, } impl SshService { pub fn connect(&self, config: SshConnectRequest) -> Result { validate_config(&config)?; let session = establish_session(&config)?; let id = Uuid::new_v4().to_string(); self.connections .lock() .map_err(lock_error)? .insert(id.clone(), Connection { session, config }); Ok(id) } pub fn disconnect(&self, app: &AppHandle, connection_id: &str) -> Result<(), String> { self.tunnels.stop_connection(connection_id)?; let shell_ids = { let shells = self.shells.lock().map_err(lock_error)?; shells .iter() .filter_map(|(id, shell)| { (shell.connection_id == connection_id).then_some(id.clone()) }) .collect::>() }; for shell_id in shell_ids { if let Some(mut shell) = self.shells.lock().map_err(lock_error)?.remove(&shell_id) { let _ = shell.channel.close(); let _ = shell.session.disconnect(None, "closed by user", None); } } self.sftps .lock() .map_err(lock_error)? .retain(|_, id| id != connection_id); if let Some(connection) = self .connections .lock() .map_err(lock_error)? .remove(connection_id) { let _ = connection.session.disconnect(None, "closed by user", None); } let _ = app.emit( "ssh:connection-closed", ConnectionClosedEvent { connection_id: connection_id.to_string(), reason: "closed by user".to_string(), }, ); Ok(()) } pub fn exec_with_status( &self, connection_id: &str, command: &str, ) -> Result<(String, i32), String> { let connection = self.connection(connection_id)?; let mut channel = connection .session .channel_session() .map_err(|error| format!("创建命令通道失败: {error}"))?; channel .exec(command) .map_err(|error| format!("执行命令失败: {error}"))?; let mut output = String::new(); channel .read_to_string(&mut output) .map_err(|error| format!("读取命令输出失败: {error}"))?; let mut stderr = String::new(); let _ = channel.stderr().read_to_string(&mut stderr); let _ = channel.wait_close(); let exit_code = channel.exit_status().unwrap_or(-1); if !stderr.is_empty() { output.push_str(&stderr); } Ok((output, exit_code)) } pub fn keepalive(&self, connection_id: &str) -> Result<(), String> { self.connection(connection_id)? .session .keepalive_send() .map(|_| ()) .map_err(|error| format!("SSH 保活失败: {error}"))?; // Interactive panes use dedicated SSH sessions so a busy SFTP/exec // channel cannot stall terminal IO. Keep those transports alive too. let shell_sessions = self .shells .lock() .map_err(lock_error)? .values() .filter(|shell| shell.connection_id == connection_id) .map(|shell| shell.session.clone()) .collect::>(); for session in shell_sessions { session .keepalive_send() .map_err(|error| format!("Shell SSH 保活失败: {error}"))?; } Ok(()) } pub fn create_shell( self: &std::sync::Arc, app: AppHandle, connection_id: &str, cols: u32, rows: u32, preferred_id: Option, ) -> Result { let connection = self.connection(connection_id)?; // A dedicated libssh2 session keeps the interactive non-blocking reader // from serializing SFTP and exec operations on the pooled connection. let session = establish_session(&connection.config)?; let mut channel = session .channel_session() .map_err(|error| format!("创建 Shell 通道失败: {error}"))?; channel .request_pty("xterm-256color", None, Some((cols, rows, 0, 0))) .map_err(|error| format!("申请远程 PTY 失败: {error}"))?; channel .shell() .map_err(|error| format!("启动远程 Shell 失败: {error}"))?; session.set_blocking(false); let shell_id = preferred_id.unwrap_or_else(|| format!("shell-{}", Uuid::new_v4())); let reader = channel.clone(); self.shells.lock().map_err(lock_error)?.insert( shell_id.clone(), Shell { connection_id: connection_id.to_string(), session, channel, }, ); spawn_shell_reader( std::sync::Arc::clone(self), app, connection_id.to_string(), shell_id.clone(), reader, ); Ok(shell_id) } pub fn shell_write(&self, shell_id: &str, data: &[u8]) -> Result<(), String> { let mut shells = self.shells.lock().map_err(lock_error)?; let shell = shells .get_mut(shell_id) .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?; write_nonblocking(&mut shell.channel, data) .map_err(|error| format!("写入远程 Shell 失败: {error}")) } pub fn shell_resize(&self, shell_id: &str, cols: u32, rows: u32) -> Result<(), String> { let mut shells = self.shells.lock().map_err(lock_error)?; let shell = shells .get_mut(shell_id) .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?; retry_would_block(|| shell.channel.request_pty_size(cols, rows, None, None)) .map_err(|error| format!("调整远程终端尺寸失败: {error}")) } pub fn forward_local( &self, connection_id: &str, forward_id: Option, local_port: u16, remote_host: String, remote_port: u16, ) -> Result { let config = self.connection(connection_id)?.config; self.tunnels.start_local( forward_id, connection_id, config, local_port, remote_host, remote_port, ) } pub fn forward_dynamic( &self, connection_id: &str, forward_id: Option, local_port: u16, ) -> Result { let config = self.connection(connection_id)?.config; self.tunnels .start_dynamic(forward_id, connection_id, config, local_port) } pub fn forward_remote( &self, connection_id: &str, forward_id: Option, remote_port: u16, local_host: String, local_port: u16, ) -> Result { let config = self.connection(connection_id)?.config; self.tunnels.start_remote( forward_id, connection_id, config, remote_port, local_host, local_port, ) } pub fn forward_stop(&self, forward_id: &str) -> Result<(), String> { self.tunnels.stop(forward_id) } pub fn forward_list(&self, connection_id: Option<&str>) -> Result, String> { self.tunnels.list(connection_id) } fn connection(&self, connection_id: &str) -> Result { self.connections .lock() .map_err(lock_error)? .get(connection_id) .cloned() .ok_or_else(|| format!("SSH 连接不存在: {connection_id}")) } } fn validate_config(config: &SshConnectRequest) -> Result<(), String> { if config.host.trim().is_empty() || config.username.trim().is_empty() { return Err("主机和用户名不能为空".to_string()); } if config.port == 0 { return Err("SSH 端口必须在 1-65535 范围内".to_string()); } Ok(()) } pub(super) fn establish_session(config: &SshConnectRequest) -> Result { let timeout_seconds = config.timeout_seconds.unwrap_or(15).clamp(3, 120); let address = format!("{}:{}", config.host.trim(), config.port); let tcp = connect_transport(config, timeout_seconds)?; tcp.set_read_timeout(Some(Duration::from_secs(timeout_seconds))) .ok(); tcp.set_write_timeout(Some(Duration::from_secs(timeout_seconds))) .ok(); let mut session = Session::new().map_err(|error| format!("初始化 SSH 失败: {error}"))?; session.set_tcp_stream(tcp); session.set_timeout(timeout_seconds.saturating_mul(1_000) as u32); session .handshake() .map_err(|error| format!("SSH 握手失败 ({address}): {error}"))?; session.set_keepalive(config.keep_alive.unwrap_or(true), 30); verify_host_key(&session, &config.host, config.port)?; authenticate(&session, config)?; Ok(session) } fn spawn_shell_reader( service: std::sync::Arc, app: AppHandle, connection_id: String, shell_id: String, mut channel: ssh2::Channel, ) { thread::spawn(move || { let mut buffer = [0_u8; 8192]; let mut output_tail = Vec::with_capacity(SHELL_OUTPUT_TAIL_LIMIT); let mut read_error = None; loop { match channel.read(&mut buffer) { Ok(0) if channel.eof() => break, Ok(0) => thread::sleep(Duration::from_millis(8)), Ok(length) => { append_output_tail(&mut output_tail, &buffer[..length]); let _ = app.emit( "ssh:shell-data", ShellDataEvent { connection_id: connection_id.clone(), shell_id: shell_id.clone(), data: buffer[..length].to_vec(), }, ); } Err(error) if error.kind() == io::ErrorKind::WouldBlock => { thread::sleep(Duration::from_millis(8)); } Err(error) => { read_error = Some(error.to_string()); break; } } } // disconnect() removes the shell before closing its channel. In that // case the connection-closed event is authoritative and a second // shell-closed event would briefly show a misleading recovery prompt. let was_active = service .shells .lock() .map(|mut shells| shells.remove(&shell_id).is_some()) .unwrap_or(false); if !was_active { return; } // Interactive shells use a dedicated SSH session. EOF, Ctrl+D, logout // and server-side TMOUT close only this shell; the pooled SSH // connection remains available to SFTP, tunnels and sibling panes. let (reason, idle_timeout) = classify_shell_close(&output_tail, read_error.as_deref()); let _ = app.emit( "ssh:shell-closed", ShellClosedEvent { connection_id, shell_id, reason, idle_timeout, }, ); }); } const SHELL_OUTPUT_TAIL_LIMIT: usize = 4096; fn append_output_tail(tail: &mut Vec, chunk: &[u8]) { if chunk.len() >= SHELL_OUTPUT_TAIL_LIMIT { tail.clear(); tail.extend_from_slice(&chunk[chunk.len() - SHELL_OUTPUT_TAIL_LIMIT..]); return; } let overflow = tail .len() .saturating_add(chunk.len()) .saturating_sub(SHELL_OUTPUT_TAIL_LIMIT); if overflow > 0 { tail.drain(..overflow); } tail.extend_from_slice(chunk); } fn classify_shell_close(output_tail: &[u8], read_error: Option<&str>) -> (String, bool) { let output = String::from_utf8_lossy(output_tail).to_ascii_lowercase(); let idle_timeout = output.contains("timed out waiting for input") || output.contains("auto-logout") || output.contains("auto logout"); if idle_timeout { return ("远端 Shell 因空闲策略(TMOUT)自动退出".to_string(), true); } if let Some(error) = read_error { return (format!("远端 Shell 读取失败: {error}"), false); } ( "远端 Shell 已退出(可能执行了 exit、Ctrl+D,或被服务器策略关闭)".to_string(), false, ) } fn write_nonblocking(channel: &mut ssh2::Channel, data: &[u8]) -> io::Result<()> { let mut written = 0; while written < data.len() { match channel.write(&data[written..]) { Ok(0) => thread::sleep(Duration::from_millis(5)), Ok(length) => written += length, Err(error) if error.kind() == io::ErrorKind::WouldBlock => { thread::sleep(Duration::from_millis(5)); } Err(error) => return Err(error), } } Ok(()) } fn retry_would_block( mut operation: impl FnMut() -> Result, ) -> Result { loop { match operation() { Err(error) if error.code() == ssh2::ErrorCode::Session(-37) => { thread::sleep(Duration::from_millis(5)); } result => return result, } } } fn lock_error(_: std::sync::PoisonError) -> String { "SSH 服务状态锁异常".to_string() } fn relay_channel(session: Session, mut channel: ssh2::Channel, mut socket: TcpStream) { session.set_blocking(false); let _ = socket.set_nonblocking(true); let mut network_buffer = [0_u8; 32 * 1024]; let mut channel_buffer = [0_u8; 32 * 1024]; let mut network_eof = false; loop { let mut progressed = false; if !network_eof { match socket.read(&mut network_buffer) { Ok(0) => { network_eof = true; let _ = channel.send_eof(); } Ok(length) => { if write_nonblocking(&mut channel, &network_buffer[..length]).is_err() { break; } progressed = true; } Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} Err(_) => break, } } match channel.read(&mut channel_buffer) { Ok(0) if channel.eof() => break, Ok(0) => {} Ok(length) => { if write_socket_nonblocking(&mut socket, &channel_buffer[..length]).is_err() { break; } progressed = true; } Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} Err(_) => break, } if !progressed { thread::sleep(Duration::from_millis(5)); } } let _ = socket.shutdown(Shutdown::Both); let _ = channel.close(); } fn write_socket_nonblocking(socket: &mut TcpStream, data: &[u8]) -> io::Result<()> { let mut written = 0; while written < data.len() { match socket.write(&data[written..]) { Ok(0) => thread::sleep(Duration::from_millis(5)), Ok(length) => written += length, Err(error) if error.kind() == io::ErrorKind::WouldBlock => { thread::sleep(Duration::from_millis(5)) } Err(error) => return Err(error), } } Ok(()) } #[allow(dead_code)] fn _exec_response(result: Result) -> ExecResponse { match result { Ok(output) => ExecResponse { success: true, result: Some(output), exit_code: None, error: None, }, Err(error) => ExecResponse { success: false, result: None, exit_code: None, error: Some(error), }, } } #[cfg(test)] mod tests { use std::{ fs, io::{Seek, SeekFrom}, path::Path, }; use super::sftp::{copy_with_progress, download_file, local_partial_path, upload_file}; use super::*; #[test] fn detects_remote_shell_idle_timeout_from_recent_output() { let (reason, idle_timeout) = classify_shell_close(b"timed out waiting for input: auto-logout\r\n", None); assert!(idle_timeout); assert!(reason.contains("TMOUT")); } #[test] fn shell_output_tail_is_bounded_and_keeps_latest_marker() { let mut tail = vec![b'x'; SHELL_OUTPUT_TAIL_LIMIT]; append_output_tail(&mut tail, b"auto-logout"); assert_eq!(tail.len(), SHELL_OUTPUT_TAIL_LIMIT); assert!(tail.ends_with(b"auto-logout")); } #[test] fn reports_normal_shell_eof_without_marking_connection_dead() { let (reason, idle_timeout) = classify_shell_close(b"logout\r\n", None); assert!(!idle_timeout); assert!(reason.contains("Shell 已退出")); } #[test] fn resumes_copy_from_existing_offset() { let mut source = io::Cursor::new(b"abcdef".to_vec()); source.seek(SeekFrom::Start(3)).unwrap(); let mut target = io::Cursor::new(b"abc".to_vec()); target.seek(SeekFrom::Start(3)).unwrap(); let mut progress = Vec::new(); copy_with_progress(&mut source, &mut target, 3, 6, |value| progress.push(value)).unwrap(); assert_eq!(target.into_inner(), b"abcdef"); assert_eq!(progress.first(), Some(&3)); assert_eq!(progress.last(), Some(&6)); } #[test] #[ignore = "requires an explicitly configured SSH endpoint"] fn real_ssh_exec_and_sftp_probe() { let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST"); let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string()); let port = std::env::var("OPSX_TEST_SSH_PORT") .ok() .and_then(|value| value.parse().ok()) .unwrap_or(22); let private_key = std::env::var("OPSX_TEST_SSH_KEY").ok(); let service = SshService::default(); let connection_id = service .connect(SshConnectRequest { host, port, username, password: None, private_key, passphrase: None, proxy: None, jump_host: None, jump_hosts: None, timeout_seconds: Some(15), keep_alive: Some(true), }) .expect("SSH connect"); let (output, exit_code) = service .exec_with_status(&connection_id, "printf OPSX_TAURI_OK") .unwrap(); assert_eq!(output, "OPSX_TAURI_OK"); assert_eq!(exit_code, 0); service .keepalive(&connection_id) .expect("SSH protocol keepalive"); let sftp_id = service.create_sftp(&connection_id).expect("SFTP connect"); assert!(!service .sftp_list(&sftp_id, "/") .expect("SFTP list") .is_empty()); let probe_id = Uuid::new_v4().to_string(); let local_source = std::env::temp_dir().join(format!("opsxshell-upload-{probe_id}.txt")); let local_download = std::env::temp_dir().join(format!("opsxshell-download-{probe_id}.txt")); let remote_path = format!("/tmp/opsxshell-sftp-{probe_id}.txt"); let payload = format!("OPSX_SFTP_ROUNDTRIP_{probe_id}\n").into_bytes(); fs::write(&local_source, &payload).expect("write local SFTP fixture"); let sftp = service.sftp(&sftp_id).expect("SFTP channel"); let transfer_result = (|| -> Result<(), String> { upload_file(&sftp, &local_source, &remote_path, |_, _| {})?; download_file(&sftp, &remote_path, &local_download, |_, _| {})?; let downloaded = fs::read(&local_download) .map_err(|error| format!("read downloaded probe: {error}"))?; if downloaded != payload { return Err("SFTP round-trip content mismatch".to_string()); } Ok(()) })(); let _ = sftp.unlink(Path::new(&remote_path)); let _ = sftp.unlink(Path::new(&format!("{remote_path}.opsx-part"))); let _ = fs::remove_file(&local_source); let _ = fs::remove_file(&local_download); let _ = fs::remove_file(local_partial_path(&local_download)); transfer_result.expect("SFTP upload/download round trip"); let connection = service.connection(&connection_id).unwrap(); let shell_session = establish_session(&connection.config).expect("shell SSH session"); let mut shell = shell_session.channel_session().expect("shell channel"); shell .request_pty("xterm-256color", None, Some((80, 24, 0, 0))) .expect("request PTY"); shell.shell().expect("start shell"); shell .write_all(b"printf OPSX_SHELL_OK; exit\n") .expect("write shell"); let mut output = String::new(); shell.read_to_string(&mut output).expect("read shell"); assert!( output.contains("OPSX_SHELL_OK"), "unexpected shell output: {output}" ); } #[test] #[ignore = "requires an explicitly configured SSH endpoint"] fn real_dynamic_forward_probe() { let _ = tracing_subscriber::fmt().with_test_writer().try_init(); let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST"); let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string()); let port = std::env::var("OPSX_TEST_SSH_PORT") .ok() .and_then(|value| value.parse().ok()) .unwrap_or(22); let private_key = std::env::var("OPSX_TEST_SSH_KEY").ok(); let service = SshService::default(); let connection_id = service .connect(SshConnectRequest { host: host.clone(), port, username, password: None, private_key, passphrase: None, proxy: None, jump_host: None, jump_hosts: None, timeout_seconds: Some(20), keep_alive: Some(true), }) .expect("SSH connect"); let forward_id = format!("real-socks-{}", Uuid::new_v4()); let status = service .forward_dynamic(&connection_id, Some(forward_id.clone()), 0) .expect("start dynamic forward"); let mut client = TcpStream::connect(("127.0.0.1", status.bound_port)) .expect("connect local SOCKS5 listener"); client .set_read_timeout(Some(Duration::from_secs(25))) .unwrap(); client.write_all(&[5, 1, 0]).unwrap(); let mut greeting = [0_u8; 2]; client.read_exact(&mut greeting).unwrap(); assert_eq!(greeting, [5, 0]); let target = host.as_bytes(); assert!(target.len() <= 255); let mut request = vec![5, 1, 0, 3, target.len() as u8]; request.extend_from_slice(target); request.extend_from_slice(&port.to_be_bytes()); client.write_all(&request).unwrap(); let mut reply = [0_u8; 10]; client.read_exact(&mut reply).unwrap(); assert_eq!(&reply[..2], &[5, 0]); let mut banner = [0_u8; 4]; client.read_exact(&mut banner).unwrap(); assert_eq!(&banner, b"SSH-"); service.forward_stop(&forward_id).expect("stop forward"); assert!(service .forward_list(Some(&connection_id)) .unwrap() .is_empty()); } #[test] #[ignore = "requires an explicitly configured SSH endpoint reachable through a jump host"] fn real_proxy_jump_probe() { let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST"); let jump_host = std::env::var("OPSX_TEST_SSH_JUMP_HOST").expect("OPSX_TEST_SSH_JUMP_HOST"); let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string()); let port = std::env::var("OPSX_TEST_SSH_PORT") .ok() .and_then(|value| value.parse().ok()) .unwrap_or(22); let jump_port = std::env::var("OPSX_TEST_SSH_JUMP_PORT") .ok() .and_then(|value| value.parse().ok()) .unwrap_or(22); let jump = serde_json::json!({ "host": jump_host, "port": jump_port, "username": username, }); let jump_count = std::env::var("OPSX_TEST_SSH_JUMP_COUNT") .ok() .and_then(|value| value.parse::().ok()) .unwrap_or(1) .clamp(1, 8); let service = SshService::default(); let connection_id = service .connect(SshConnectRequest { host, port, username, password: None, private_key: None, passphrase: None, proxy: None, jump_host: (jump_count == 1).then(|| jump.clone()), jump_hosts: (jump_count > 1).then(|| vec![jump; jump_count]), timeout_seconds: Some(20), keep_alive: Some(true), }) .expect("SSH connect through jump host"); let (output, exit_code) = service .exec_with_status(&connection_id, "printf OPSX_PROXY_JUMP_OK") .unwrap(); assert_eq!(output, "OPSX_PROXY_JUMP_OK"); assert_eq!(exit_code, 0); } }