rust: SSH client on russh (auth chain, TOTP, jump hosts, proxies, forwards, agent forwarding) and SFTP, with sshd integration tests
This commit is contained in:
Andy Dixon
@@ -14,3 +14,9 @@ aes-gcm = "0.10"
|
||||
rand = "0.8"
|
||||
base64 = "0.22"
|
||||
keyring = { version = "3", features = ["apple-native", "windows-native", "async-secret-service", "tokio", "crypto-rust"] }
|
||||
russh = { version = "0.64.1", default-features = false, features = ["flate2", "ring", "rsa"] }
|
||||
russh-sftp = "3.0.1"
|
||||
tokio = { version = "1.53.2", features = ["full"] }
|
||||
sha2 = "0.11.0"
|
||||
hmac = "0.13.0"
|
||||
sha1 = "0.11.0"
|
||||
|
||||
@@ -6,7 +6,18 @@
|
||||
//! - `write`/`resize`/`close` are callable from any thread; `close` is idempotent and must not
|
||||
//! fire `on_close` re-entrantly while holding locks.
|
||||
|
||||
use std::sync::Arc;
|
||||
pub mod sftp;
|
||||
pub mod ssh;
|
||||
|
||||
use std::sync::{Arc, OnceLock};
|
||||
|
||||
/// The shared tokio runtime for async protocol stacks (russh). Blocking clients use threads.
|
||||
pub fn rt() -> &'static tokio::runtime::Runtime {
|
||||
static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
|
||||
RT.get_or_init(|| {
|
||||
tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().thread_name("nexus-net").build().expect("tokio runtime")
|
||||
})
|
||||
}
|
||||
|
||||
pub type DataFn = Arc<dyn Fn(&[u8]) + Send + Sync>;
|
||||
pub type ErrorFn = Arc<dyn Fn(String) + Send + Sync>;
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
//! SFTP operations over an SSH session's subsystem (internal/sftpclient). Every call blocks the
|
||||
//! caller, so the UI runs them on worker threads.
|
||||
|
||||
use super::rt;
|
||||
use crate::Result;
|
||||
use russh_sftp::client::SftpSession;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct FileEntry {
|
||||
pub name: String,
|
||||
/// Full remote path.
|
||||
pub path: String,
|
||||
pub size: u64,
|
||||
pub is_dir: bool,
|
||||
/// Unix permission string, e.g. "-rwxr-xr-x".
|
||||
pub mode: String,
|
||||
/// Unix epoch milliseconds.
|
||||
pub modified_at: i64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct Progress {
|
||||
pub transferred: u64,
|
||||
pub total: u64,
|
||||
pub path: String,
|
||||
pub done: bool,
|
||||
pub error: String,
|
||||
}
|
||||
|
||||
fn mode_string(m: &russh_sftp::protocol::FileAttributes) -> String {
|
||||
let kind = if m.is_dir() { 'd' } else if m.is_symlink() { 'L' } else { '-' };
|
||||
format!("{kind}{}", m.permissions())
|
||||
}
|
||||
|
||||
fn entry(name: String, path: String, m: &russh_sftp::protocol::FileAttributes) -> FileEntry {
|
||||
FileEntry {
|
||||
name,
|
||||
path,
|
||||
size: m.size.unwrap_or(0),
|
||||
is_dir: m.is_dir(),
|
||||
mode: mode_string(m),
|
||||
modified_at: m.mtime.map_or(0, |t| t as i64 * 1000),
|
||||
}
|
||||
}
|
||||
|
||||
/// `path.Join` for remote (always '/') paths.
|
||||
pub fn join(dir: &str, name: &str) -> String {
|
||||
if dir.is_empty() {
|
||||
name.to_string()
|
||||
} else if dir.ends_with('/') {
|
||||
format!("{dir}{name}")
|
||||
} else {
|
||||
format!("{dir}/{name}")
|
||||
}
|
||||
}
|
||||
|
||||
/// Directories first, then case-insensitive name order.
|
||||
pub fn list(s: &Arc<SftpSession>, path: &str) -> Result<Vec<FileEntry>> {
|
||||
let rd = rt().block_on(s.read_dir(path)).map_err(|e| format!("sftpclient: failed to list {path}: {e}"))?;
|
||||
let mut out: Vec<FileEntry> = rd
|
||||
.filter(|e| e.file_name() != "." && e.file_name() != "..")
|
||||
.map(|e| entry(e.file_name(), join(path, &e.file_name()), &e.metadata()))
|
||||
.collect();
|
||||
out.sort_by(|a, b| b.is_dir.cmp(&a.is_dir).then_with(|| a.name.to_lowercase().cmp(&b.name.to_lowercase())));
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
pub fn stat(s: &Arc<SftpSession>, path: &str) -> Result<FileEntry> {
|
||||
let m = rt().block_on(s.metadata(path)).map_err(|e| format!("sftpclient: stat failed for {path}: {e}"))?;
|
||||
let name = path.rsplit('/').next().unwrap_or(path).to_string();
|
||||
Ok(entry(name, path.to_string(), &m))
|
||||
}
|
||||
|
||||
pub fn mkdir(s: &Arc<SftpSession>, path: &str) -> Result<()> {
|
||||
Ok(rt().block_on(s.create_dir(path))?)
|
||||
}
|
||||
|
||||
/// Removes a file, or an empty directory.
|
||||
pub fn remove(s: &Arc<SftpSession>, path: &str) -> Result<()> {
|
||||
rt().block_on(async {
|
||||
match s.remove_file(path).await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(e) => s.remove_dir(path).await.map_err(|_| e),
|
||||
}
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn rename(s: &Arc<SftpSession>, from: &str, to: &str) -> Result<()> {
|
||||
Ok(rt().block_on(s.rename(from, to))?)
|
||||
}
|
||||
|
||||
pub fn pwd(s: &Arc<SftpSession>) -> Result<String> {
|
||||
Ok(rt().block_on(s.canonicalize("."))?)
|
||||
}
|
||||
|
||||
async fn pump(
|
||||
mut src: impl tokio::io::AsyncRead + Unpin,
|
||||
mut dst: impl tokio::io::AsyncWrite + Unpin,
|
||||
total: u64,
|
||||
path: &str,
|
||||
on: &dyn Fn(Progress),
|
||||
) -> Result<()> {
|
||||
let mut buf = vec![0u8; 32 * 1024];
|
||||
let (mut done, mut last) = (0u64, Instant::now());
|
||||
loop {
|
||||
let n = src.read(&mut buf).await.map_err(|e| format!("sftpclient: read failed: {e}"))?;
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
dst.write_all(&buf[..n]).await.map_err(|e| format!("sftpclient: write failed: {e}"))?;
|
||||
done += n as u64;
|
||||
if last.elapsed() > Duration::from_millis(100) {
|
||||
on(Progress { transferred: done, total, path: path.into(), ..Default::default() });
|
||||
last = Instant::now();
|
||||
}
|
||||
}
|
||||
dst.flush().await?;
|
||||
dst.shutdown().await?;
|
||||
on(Progress { transferred: done, total, path: path.into(), done: true, ..Default::default() });
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn download(s: &Arc<SftpSession>, remote: &str, local: &std::path::Path, on: &dyn Fn(Progress)) -> Result<()> {
|
||||
rt().block_on(async {
|
||||
let total = s.metadata(remote).await.map_err(|e| format!("sftpclient: stat failed: {e}"))?.size.unwrap_or(0);
|
||||
let src = s.open(remote).await.map_err(|e| format!("sftpclient: failed to open remote file: {e}"))?;
|
||||
let dst = tokio::fs::File::create(local).await.map_err(|e| format!("sftpclient: failed to create local file: {e}"))?;
|
||||
pump(src, dst, total, remote, on).await
|
||||
})
|
||||
}
|
||||
|
||||
pub fn upload(s: &Arc<SftpSession>, local: &std::path::Path, remote: &str, on: &dyn Fn(Progress)) -> Result<()> {
|
||||
rt().block_on(async {
|
||||
let src = tokio::fs::File::open(local).await.map_err(|e| format!("sftpclient: failed to open local file: {e}"))?;
|
||||
let total = src.metadata().await?.len();
|
||||
let dst = s.create(remote).await.map_err(|e| format!("sftpclient: failed to create remote file: {e}"))?;
|
||||
pump(src, dst, total, remote, on).await
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#[test]
|
||||
fn join_paths() {
|
||||
assert_eq!(super::join("/", "a"), "/a");
|
||||
assert_eq!(super::join("/home/x", "a"), "/home/x/a");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,766 @@
|
||||
//! SSH client on russh: interactive PTY shell, the profile's auth chain (agent, public key,
|
||||
//! password, keyboard-interactive with TOTP), jump hosts, SOCKS5/HTTP proxies, local and remote
|
||||
//! port forwards, agent forwarding and SFTP. A port of internal/sshclient + sftpclient.
|
||||
|
||||
use super::{Callbacks, rt};
|
||||
use crate::models::{ForwardDirection, JumpHost, PortForward, Profile, SshConfig};
|
||||
use crate::Result;
|
||||
use base64::Engine;
|
||||
use russh::client::{self, Handle, Msg};
|
||||
use russh::keys::{HashAlg, PrivateKeyWithHashAlg, PublicKey};
|
||||
use russh::{Channel, ChannelMsg, Pty};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
const MAX_PRIVATE_KEY_BYTES: u64 = 1 << 20;
|
||||
|
||||
/// What a host-key decision needs. The manager's checker consults the known-hosts table and,
|
||||
/// per policy, prompts the user; it returns Err with the reason to reject.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct HostKeyInfo {
|
||||
pub host: String,
|
||||
pub port: u16,
|
||||
pub key_type: String,
|
||||
pub fingerprint: String,
|
||||
pub public_key_b64: String,
|
||||
}
|
||||
|
||||
pub type HostKeyCheck = Arc<dyn Fn(&HostKeyInfo) -> std::result::Result<(), String> + Send + Sync>;
|
||||
/// Blocking credential source (may prompt the user).
|
||||
pub type SecretFn = Arc<dyn Fn() -> Result<String> + Send + Sync>;
|
||||
|
||||
pub fn host_key_info(host: &str, port: u16, key: &PublicKey) -> HostKeyInfo {
|
||||
HostKeyInfo {
|
||||
host: host.into(),
|
||||
port,
|
||||
key_type: key.algorithm().as_str().into(),
|
||||
fingerprint: key.fingerprint(HashAlg::Sha256).to_string(),
|
||||
public_key_b64: base64::engine::general_purpose::STANDARD.encode(key.to_bytes().unwrap_or_default()),
|
||||
}
|
||||
}
|
||||
|
||||
/// State the russh handler shares with the client: remote-forward targets and the last
|
||||
/// host-key rejection reason (russh only reports "unknown key").
|
||||
#[derive(Default)]
|
||||
struct Shared {
|
||||
remote_targets: Mutex<HashMap<(String, u32), (String, u16)>>,
|
||||
rejection: Mutex<Option<String>>,
|
||||
agent_forwarding: std::sync::atomic::AtomicBool,
|
||||
}
|
||||
|
||||
struct Handler {
|
||||
host: String,
|
||||
port: u16,
|
||||
check: HostKeyCheck,
|
||||
shared: Arc<Shared>,
|
||||
}
|
||||
|
||||
impl client::Handler for Handler {
|
||||
type Error = russh::Error;
|
||||
|
||||
async fn check_server_key(&mut self, key: &russh::keys::PublicKeyOrCertificate) -> std::result::Result<bool, Self::Error> {
|
||||
let key = match key {
|
||||
russh::keys::PublicKeyOrCertificate::PublicKey { key, .. } => key.clone(),
|
||||
russh::keys::PublicKeyOrCertificate::Certificate(c) => c.public_key().clone().into(),
|
||||
};
|
||||
let info = host_key_info(&self.host, self.port, &key);
|
||||
let check = self.check.clone();
|
||||
// The check may block on a UI prompt.
|
||||
let res = tokio::task::spawn_blocking(move || check(&info)).await.unwrap_or(Err("host key check failed".into()));
|
||||
match res {
|
||||
Ok(()) => Ok(true),
|
||||
Err(why) => {
|
||||
*self.shared.rejection.lock().unwrap() = Some(why);
|
||||
Ok(false)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn server_channel_open_forwarded_tcpip(
|
||||
&mut self,
|
||||
channel: Channel<Msg>,
|
||||
connected_address: &str,
|
||||
connected_port: u32,
|
||||
_originator_address: &str,
|
||||
_originator_port: u32,
|
||||
reply: client::ChannelOpenHandle,
|
||||
_session: &mut client::Session,
|
||||
) -> std::result::Result<(), Self::Error> {
|
||||
let dest = self.shared.remote_targets.lock().unwrap().get(&(connected_address.to_string(), connected_port)).cloned();
|
||||
let Some((host, port)) = dest else {
|
||||
reply.reject(russh::ChannelOpenFailure::AdministrativelyProhibited).await;
|
||||
return Ok(());
|
||||
};
|
||||
reply.accept().await;
|
||||
tokio::spawn(async move {
|
||||
if let Ok(mut local) = tokio::net::TcpStream::connect((host.as_str(), port)).await {
|
||||
let mut remote = channel.into_stream();
|
||||
let _ = tokio::io::copy_bidirectional(&mut remote, &mut local).await;
|
||||
}
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn server_channel_open_agent_forward(
|
||||
&mut self,
|
||||
channel: Channel<Msg>,
|
||||
reply: client::ChannelOpenHandle,
|
||||
_session: &mut client::Session,
|
||||
) -> std::result::Result<(), Self::Error> {
|
||||
if !self.shared.agent_forwarding.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
reply.reject(russh::ChannelOpenFailure::AdministrativelyProhibited).await;
|
||||
return Ok(());
|
||||
}
|
||||
reply.accept().await;
|
||||
tokio::spawn(async move {
|
||||
if let Ok(mut agent) = connect_agent_socket().await {
|
||||
let mut remote = channel.into_stream();
|
||||
let _ = tokio::io::copy_bidirectional(&mut remote, &mut agent).await;
|
||||
}
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
trait Stream: AsyncRead + AsyncWrite + Unpin + Send {}
|
||||
impl<T: AsyncRead + AsyncWrite + Unpin + Send> Stream for T {}
|
||||
|
||||
/// The local agent's socket: `SSH_AUTH_SOCK` on Unix, the OpenSSH named pipe on Windows.
|
||||
async fn connect_agent_socket() -> std::io::Result<Box<dyn Stream>> {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let p = std::env::var("SSH_AUTH_SOCK").map_err(|_| std::io::Error::other("SSH_AUTH_SOCK not set — SSH agent unavailable"))?;
|
||||
Ok(Box::new(tokio::net::UnixStream::connect(p).await?))
|
||||
}
|
||||
#[cfg(windows)]
|
||||
{
|
||||
let p = std::env::var("SSH_AUTH_SOCK").unwrap_or_else(|_| r"\\.\pipe\openssh-ssh-agent".into());
|
||||
Ok(Box::new(tokio::net::windows::named_pipe::ClientOptions::new().open(p)?))
|
||||
}
|
||||
}
|
||||
|
||||
type Agent = russh::keys::agent::client::AgentClient<Box<dyn russh::keys::agent::client::AgentStream + Send + Unpin>>;
|
||||
|
||||
async fn connect_agent() -> Result<Agent> {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
Ok(russh::keys::agent::client::AgentClient::connect_env().await.map_err(|e| format!("SSH agent unavailable: {e}"))?.dynamic())
|
||||
}
|
||||
#[cfg(windows)]
|
||||
{
|
||||
if let Ok(a) = russh::keys::agent::client::AgentClient::connect_named_pipe(r"\\.\pipe\openssh-ssh-agent").await {
|
||||
return Ok(a.dynamic());
|
||||
}
|
||||
Ok(russh::keys::agent::client::AgentClient::connect_pageant().await.map_err(|e| format!("SSH agent unavailable: {e}"))?.dynamic())
|
||||
}
|
||||
}
|
||||
|
||||
/// Profile env vars plus a UTF-8 LC_CTYPE when the profile sets no locale (Go's SSHEnvVars).
|
||||
fn ssh_env_vars(env: &std::collections::BTreeMap<String, String>) -> std::collections::BTreeMap<String, String> {
|
||||
let mut out: std::collections::BTreeMap<String, String> =
|
||||
env.iter().filter(|(k, _)| !k.trim().is_empty()).map(|(k, v)| (k.trim().to_string(), v.clone())).collect();
|
||||
if !["LC_ALL", "LC_CTYPE", "LANG"].iter().any(|k| out.contains_key(*k)) {
|
||||
out.insert("LC_CTYPE".into(), if cfg!(target_os = "macos") { "en_US.UTF-8" } else { "C.UTF-8" }.into());
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn is_totp_prompt(q: &str) -> bool {
|
||||
let q = q.to_lowercase();
|
||||
[
|
||||
"verification code", "otp", "one-time", "one time", "two-factor", "two factor", "2fa", "totp", "authenticator",
|
||||
"security code", "6-digit", "6 digit",
|
||||
]
|
||||
.iter()
|
||||
.any(|k| q.contains(k))
|
||||
}
|
||||
|
||||
/// RFC 6238 code (SHA-1, 6 digits, 30 s) for a base32 secret, normalised as pquerna/otp does.
|
||||
pub fn totp_code(secret: &str) -> Result<String> {
|
||||
totp_at(secret, std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs())
|
||||
}
|
||||
|
||||
fn totp_at(secret: &str, unix: u64) -> Result<String> {
|
||||
use hmac::{Hmac, KeyInit, Mac};
|
||||
let key = base32_decode(secret).ok_or("invalid TOTP secret (not base32)")?;
|
||||
let mut mac = Hmac::<sha1::Sha1>::new_from_slice(&key).map_err(|e| e.to_string())?;
|
||||
mac.update(&(unix / 30).to_be_bytes());
|
||||
let h = mac.finalize().into_bytes();
|
||||
let o = (h[19] & 0xf) as usize;
|
||||
let bin = u32::from_be_bytes([h[o] & 0x7f, h[o + 1], h[o + 2], h[o + 3]]);
|
||||
Ok(format!("{:06}", bin % 1_000_000))
|
||||
}
|
||||
|
||||
fn base32_decode(s: &str) -> Option<Vec<u8>> {
|
||||
let (mut out, mut buf, mut bits) = (vec![], 0u64, 0u32);
|
||||
for c in s.chars().filter(|c| !c.is_whitespace() && *c != '=') {
|
||||
let v = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567".find(c.to_ascii_uppercase())? as u64;
|
||||
buf = (buf << 5) | v;
|
||||
bits += 5;
|
||||
if bits >= 8 {
|
||||
bits -= 8;
|
||||
out.push((buf >> bits) as u8);
|
||||
}
|
||||
}
|
||||
(!out.is_empty()).then_some(out)
|
||||
}
|
||||
|
||||
/// Reads an OpenSSH/PEM private key, asking for a passphrase only if it's encrypted.
|
||||
pub fn load_private_key(path: &str, passphrase: Option<&SecretFn>) -> Result<russh::keys::PrivateKey> {
|
||||
let meta = std::fs::metadata(path).map_err(|e| format!("unable to stat key file: {e}"))?;
|
||||
if meta.is_dir() {
|
||||
return Err("key path is a directory".into());
|
||||
}
|
||||
if meta.len() > MAX_PRIVATE_KEY_BYTES {
|
||||
return Err(format!("key file is too large: {} bytes exceeds {MAX_PRIVATE_KEY_BYTES} bytes", meta.len()).into());
|
||||
}
|
||||
let data = std::fs::read_to_string(path).map_err(|e| format!("unable to read key file: {e}"))?;
|
||||
match russh::keys::decode_secret_key(&data, None) {
|
||||
Ok(k) => Ok(k),
|
||||
Err(first) => {
|
||||
let Some(get) = passphrase else {
|
||||
return Err(format!("key is encrypted and no passphrase callback provided: {first}").into());
|
||||
};
|
||||
let pass = get().map_err(|e| format!("failed to get passphrase: {e}"))?;
|
||||
russh::keys::decode_secret_key(&data, Some(&pass)).map_err(|e| format!("failed to parse encrypted key: {e}").into())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
|
||||
tokio::task::spawn_blocking(f).await.expect("blocking task panicked")
|
||||
}
|
||||
|
||||
async fn dial(sc: &SshConfig, host: &str, port: u16, timeout: Duration) -> Result<tokio::net::TcpStream> {
|
||||
let fut = async {
|
||||
match sc.proxy_type.to_lowercase().as_str() {
|
||||
"socks5" => socks5(sc, host, port).await,
|
||||
"http" => http_connect(sc, host, port).await,
|
||||
_ => Ok(tokio::net::TcpStream::connect((host, port)).await?),
|
||||
}
|
||||
};
|
||||
tokio::time::timeout(timeout, fut).await.map_err(|_| format!("dial tcp {host}:{port}: i/o timeout"))?
|
||||
}
|
||||
|
||||
async fn socks5(sc: &SshConfig, host: &str, port: u16) -> Result<tokio::net::TcpStream> {
|
||||
let pport = if sc.proxy_port == 0 { 1080 } else { sc.proxy_port as u16 };
|
||||
let mut s = tokio::net::TcpStream::connect((sc.proxy_host.as_str(), pport)).await.map_err(|e| format!("SOCKS5 proxy setup failed: {e}"))?;
|
||||
let auth = !sc.proxy_username.is_empty();
|
||||
s.write_all(if auth { &[5, 2, 0, 2] } else { &[5, 1, 0] }).await?;
|
||||
let mut r = [0u8; 2];
|
||||
s.read_exact(&mut r).await?;
|
||||
match r[1] {
|
||||
0 => {}
|
||||
2 if auth => {
|
||||
let mut m = vec![1, sc.proxy_username.len() as u8];
|
||||
m.extend(sc.proxy_username.as_bytes());
|
||||
m.push(sc.proxy_password.len() as u8);
|
||||
m.extend(sc.proxy_password.as_bytes());
|
||||
s.write_all(&m).await?;
|
||||
s.read_exact(&mut r).await?;
|
||||
if r[1] != 0 {
|
||||
return Err("SOCKS5 proxy: authentication failed".into());
|
||||
}
|
||||
}
|
||||
_ => return Err("SOCKS5 proxy: no acceptable authentication method".into()),
|
||||
}
|
||||
let mut req = vec![5, 1, 0, 3, host.len() as u8];
|
||||
req.extend(host.as_bytes());
|
||||
req.extend(port.to_be_bytes());
|
||||
s.write_all(&req).await?;
|
||||
let mut head = [0u8; 4];
|
||||
s.read_exact(&mut head).await?;
|
||||
if head[1] != 0 {
|
||||
return Err(format!("SOCKS5 proxy connection to {host}:{port} failed: reply code {}", head[1]).into());
|
||||
}
|
||||
let skip = match head[3] {
|
||||
1 => 4,
|
||||
4 => 16,
|
||||
3 => s.read_u8().await? as usize,
|
||||
_ => 0,
|
||||
} + 2;
|
||||
let mut rest = vec![0u8; skip];
|
||||
s.read_exact(&mut rest).await?;
|
||||
Ok(s)
|
||||
}
|
||||
|
||||
async fn http_connect(sc: &SshConfig, host: &str, port: u16) -> Result<tokio::net::TcpStream> {
|
||||
let pport = if sc.proxy_port == 0 { 8080 } else { sc.proxy_port as u16 };
|
||||
let mut s = tokio::net::TcpStream::connect((sc.proxy_host.as_str(), pport)).await.map_err(|e| format!("HTTP proxy connection failed: {e}"))?;
|
||||
let mut req = format!("CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n");
|
||||
if !sc.proxy_username.is_empty() {
|
||||
let cred = base64::engine::general_purpose::STANDARD.encode(format!("{}:{}", sc.proxy_username, sc.proxy_password));
|
||||
req += &format!("Proxy-Authorization: Basic {cred}\r\n");
|
||||
}
|
||||
req += "\r\n";
|
||||
s.write_all(req.as_bytes()).await?;
|
||||
// Read the response head byte by byte so no tunnelled bytes are consumed.
|
||||
let mut head = Vec::new();
|
||||
while !head.ends_with(b"\r\n\r\n") {
|
||||
head.push(s.read_u8().await.map_err(|e| format!("HTTP CONNECT response read failed: {e}"))?);
|
||||
if head.len() > 16 * 1024 {
|
||||
return Err("HTTP CONNECT response too large".into());
|
||||
}
|
||||
}
|
||||
let status = String::from_utf8_lossy(&head).lines().next().unwrap_or("").to_string();
|
||||
if status.split_whitespace().nth(1) != Some("200") {
|
||||
return Err(format!("HTTP CONNECT proxy returned {status}").into());
|
||||
}
|
||||
Ok(s)
|
||||
}
|
||||
|
||||
fn config(sc: &SshConfig) -> Arc<client::Config> {
|
||||
let mut c = client::Config::default();
|
||||
if sc.keepalive_interval > 0 {
|
||||
c.keepalive_interval = Some(Duration::from_secs(sc.keepalive_interval as u64));
|
||||
c.keepalive_max = sc.keepalive_count_max.max(1) as usize;
|
||||
}
|
||||
c.preferred.compression = if sc.compression {
|
||||
vec![russh::compression::ZLIB_LEGACY, russh::compression::ZLIB, russh::compression::NONE].into()
|
||||
} else {
|
||||
vec![russh::compression::NONE].into()
|
||||
};
|
||||
c.nodelay = true;
|
||||
Arc::new(c)
|
||||
}
|
||||
|
||||
struct Auth<'a> {
|
||||
user: &'a str,
|
||||
methods: Vec<String>,
|
||||
agent: bool,
|
||||
key: Option<russh::keys::PrivateKey>,
|
||||
password: Option<SecretFn>,
|
||||
totp_secret: String,
|
||||
}
|
||||
|
||||
async fn authenticate(h: &mut Handle<Handler>, a: Auth<'_>) -> Result<()> {
|
||||
if a.agent
|
||||
&& let Ok(mut agent) = connect_agent().await
|
||||
&& let Ok(ids) = agent.request_identities().await
|
||||
{
|
||||
for id in ids {
|
||||
let key = id.public_key().into_owned();
|
||||
let hash = if key.algorithm().is_rsa() { h.best_supported_rsa_hash().await.ok().flatten().flatten() } else { None };
|
||||
if let Ok(r) = h.authenticate_publickey_with(a.user, key, hash, &mut agent).await
|
||||
&& r.success()
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
for m in &a.methods {
|
||||
match m.as_str() {
|
||||
"publickey" => {
|
||||
if let Some(k) = &a.key {
|
||||
let hash = if k.algorithm().is_rsa() { h.best_supported_rsa_hash().await?.flatten() } else { None };
|
||||
if h.authenticate_publickey(a.user, PrivateKeyWithHashAlg::new(Arc::new(k.clone()), hash)).await?.success() {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
"password" => {
|
||||
if let Some(get) = a.password.clone() {
|
||||
let pw = blocking(move || get()).await?;
|
||||
if h.authenticate_password(a.user, pw).await?.success() {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
"keyboard-interactive" => {
|
||||
let Some(get) = a.password.clone() else { continue };
|
||||
let mut resp = h.authenticate_keyboard_interactive_start(a.user, None).await?;
|
||||
loop {
|
||||
match resp {
|
||||
client::KeyboardInteractiveAuthResponse::Success => return Ok(()),
|
||||
client::KeyboardInteractiveAuthResponse::Failure { .. } => break,
|
||||
client::KeyboardInteractiveAuthResponse::InfoRequest { prompts, .. } => {
|
||||
let mut answers = vec![];
|
||||
for p in prompts {
|
||||
if !a.totp_secret.is_empty() && is_totp_prompt(&p.prompt) {
|
||||
answers.push(totp_code(&a.totp_secret).map_err(|e| format!("failed to generate TOTP code: {e}"))?);
|
||||
} else {
|
||||
let get = get.clone();
|
||||
answers.push(blocking(move || get()).await?);
|
||||
}
|
||||
}
|
||||
resp = h.authenticate_keyboard_interactive_respond(answers).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Err("ssh: handshake failed: ssh: unable to authenticate, no supported methods remain".into())
|
||||
}
|
||||
|
||||
enum Cmd {
|
||||
Data(Vec<u8>),
|
||||
Resize(u32, u32),
|
||||
}
|
||||
|
||||
struct Forward {
|
||||
rule: PortForward,
|
||||
tasks: Arc<Mutex<Vec<tokio::task::AbortHandle>>>,
|
||||
}
|
||||
|
||||
pub struct SshClient {
|
||||
profile: Profile,
|
||||
cb: Callbacks,
|
||||
shared: Arc<Shared>,
|
||||
handle: Mutex<Option<Arc<Handle<Handler>>>>,
|
||||
/// Earlier hops of a jump chain; kept alive for the session's lifetime.
|
||||
hops: Mutex<Vec<Arc<Handle<Handler>>>>,
|
||||
tx: Mutex<Option<mpsc::UnboundedSender<Cmd>>>,
|
||||
forwards: Mutex<HashMap<String, Forward>>,
|
||||
sftp: tokio::sync::Mutex<Option<Arc<russh_sftp::client::SftpSession>>>,
|
||||
closed: std::sync::atomic::AtomicBool,
|
||||
}
|
||||
|
||||
impl SshClient {
|
||||
pub fn new(profile: &Profile, cb: Callbacks) -> Arc<SshClient> {
|
||||
Arc::new(SshClient {
|
||||
profile: profile.clone(),
|
||||
cb,
|
||||
shared: Arc::default(),
|
||||
handle: Mutex::new(None),
|
||||
hops: Mutex::new(vec![]),
|
||||
tx: Mutex::new(None),
|
||||
forwards: Mutex::new(HashMap::new()),
|
||||
sftp: tokio::sync::Mutex::new(None),
|
||||
closed: false.into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn sc(&self) -> SshConfig {
|
||||
self.profile.ssh.clone().unwrap_or_else(SshConfig::defaults)
|
||||
}
|
||||
|
||||
/// 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 })
|
||||
}
|
||||
|
||||
async fn handshake(
|
||||
&self,
|
||||
stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static,
|
||||
host: &str,
|
||||
port: u16,
|
||||
check: &HostKeyCheck,
|
||||
) -> Result<Handle<Handler>> {
|
||||
let handler = Handler { host: host.into(), port, check: check.clone(), shared: self.shared.clone() };
|
||||
let timeout = Duration::from_secs(self.profile.connect_timeout.max(1) as u64);
|
||||
match tokio::time::timeout(timeout, client::connect_stream(config(&self.sc()), stream, handler)).await {
|
||||
Err(_) => Err(format!("SSH handshake with {host}:{port} timed out").into()),
|
||||
Ok(Ok(h)) => Ok(h),
|
||||
Ok(Err(e)) => match self.shared.rejection.lock().unwrap().take() {
|
||||
Some(why) => Err(why.into()),
|
||||
None => Err(format!("SSH handshake failed: {e}").into()),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
async fn connect_async(self: &Arc<Self>, check: HostKeyCheck, password: Option<SecretFn>, passphrase: Option<SecretFn>) -> Result<()> {
|
||||
let sc = self.sc();
|
||||
let p = &self.profile;
|
||||
let port = if p.port == 0 { 22 } else { p.port as u16 };
|
||||
let timeout = Duration::from_secs(if p.connect_timeout == 0 { 30 } else { p.connect_timeout as u64 });
|
||||
let methods = if sc.auth_methods.is_empty() { SshConfig::defaults().auth_methods } else { sc.auth_methods.clone() };
|
||||
let key = if methods.iter().any(|m| m == "publickey") && !sc.private_key_path.is_empty() {
|
||||
let (path, pp) = (sc.private_key_path.clone(), passphrase.clone());
|
||||
Some(blocking(move || load_private_key(&path, pp.as_ref())).await.map_err(|e| format!("failed to build auth methods: failed to parse private key {}: {e}", sc.private_key_path))?)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let mut handle = if sc.jump_hosts.is_empty() {
|
||||
let tcp = dial(&sc, &p.host, port, timeout).await.map_err(|e| format!("SSH connection failed: {e}"))?;
|
||||
self.handshake(tcp, &p.host, port, &check).await?
|
||||
} else {
|
||||
let (prev, h) = self.via_jumps(&sc, &check, passphrase.clone(), timeout).await?;
|
||||
let ch = prev.channel_open_direct_tcpip(p.host.clone(), port as u32, "127.0.0.1", 0).await.map_err(|e| format!("failed to dial target {}:{port} through jump hosts: {e}", p.host))?;
|
||||
self.hops.lock().unwrap().extend(h);
|
||||
self.hops.lock().unwrap().push(prev);
|
||||
self.handshake(ch.into_stream(), &p.host, port, &check).await?
|
||||
};
|
||||
|
||||
authenticate(
|
||||
&mut handle,
|
||||
Auth { user: &p.username, methods, agent: sc.agent_auth, key, password, totp_secret: sc.totp_secret.clone() },
|
||||
)
|
||||
.await?;
|
||||
let handle = Arc::new(handle);
|
||||
|
||||
let ch = handle.channel_open_session().await.map_err(|e| format!("failed to create SSH session: {e}"))?;
|
||||
for (k, v) in ssh_env_vars(&sc.env_vars) {
|
||||
let _ = ch.set_env(false, k, v).await;
|
||||
}
|
||||
let term = if sc.terminal_type.is_empty() { "xterm-256color" } else { &sc.terminal_type };
|
||||
let modes = [(Pty::ECHO, 1), (Pty::TTY_OP_ISPEED, 115200), (Pty::TTY_OP_OSPEED, 115200), (Pty::IUTF8, 1)];
|
||||
ch.request_pty(true, term, 80, 24, 0, 0, &modes).await.map_err(|e| format!("PTY request failed: {e}"))?;
|
||||
if sc.agent_forwarding {
|
||||
self.shared.agent_forwarding.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
if let Err(e) = ch.agent_forward(false).await {
|
||||
(self.cb.on_error)(format!("agent forwarding: {e}"));
|
||||
}
|
||||
}
|
||||
if sc.startup_command.is_empty() {
|
||||
ch.request_shell(true).await.map_err(|e| format!("failed to start shell: {e}"))?;
|
||||
} else {
|
||||
ch.exec(true, sc.startup_command.as_bytes()).await.map_err(|e| format!("failed to start command: {e}"))?;
|
||||
}
|
||||
|
||||
let (mut read, write) = ch.split();
|
||||
let (tx, mut rx) = mpsc::unbounded_channel::<Cmd>();
|
||||
*self.tx.lock().unwrap() = Some(tx);
|
||||
*self.handle.lock().unwrap() = Some(handle);
|
||||
tokio::spawn(async move {
|
||||
while let Some(cmd) = rx.recv().await {
|
||||
let r = match cmd {
|
||||
Cmd::Data(d) => write.data_bytes(d).await,
|
||||
Cmd::Resize(c, r) => write.window_change(c, r, 0, 0).await,
|
||||
};
|
||||
if r.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
let me = self.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = read.wait().await {
|
||||
match msg {
|
||||
ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => (me.cb.on_data)(&data),
|
||||
ChannelMsg::Eof | ChannelMsg::Close => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
let me2 = me.clone();
|
||||
let _ = tokio::task::spawn_blocking(move || me2.close()).await;
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Authenticates through each jump host in turn; returns the last hop and the ones before it.
|
||||
async fn via_jumps(&self, sc: &SshConfig, check: &HostKeyCheck, passphrase: Option<SecretFn>, timeout: Duration) -> Result<(Arc<Handle<Handler>>, Vec<Arc<Handle<Handler>>>)> {
|
||||
let mut chain: Vec<Arc<Handle<Handler>>> = vec![];
|
||||
for (i, j) in sc.jump_hosts.iter().enumerate() {
|
||||
let JumpHost { host, port, username, private_key_path, .. } = j;
|
||||
let jport = if *port == 0 { 22 } else { *port as u16 };
|
||||
let key = if private_key_path.is_empty() {
|
||||
None
|
||||
} else {
|
||||
let (path, pp) = (private_key_path.clone(), passphrase.clone());
|
||||
Some(blocking(move || load_private_key(&path, pp.as_ref())).await.map_err(|e| format!("jump host {i} private key error: {e}"))?)
|
||||
};
|
||||
let mut h = match chain.last() {
|
||||
None => {
|
||||
let tcp = tokio::time::timeout(timeout, tokio::net::TcpStream::connect((host.as_str(), jport)))
|
||||
.await
|
||||
.map_err(|_| format!("failed to connect to jump host {i} ({host}:{jport}): timeout"))?
|
||||
.map_err(|e| format!("failed to connect to jump host {i} ({host}:{jport}): {e}"))?;
|
||||
self.handshake(tcp, host, jport, check).await?
|
||||
}
|
||||
Some(prev) => {
|
||||
let ch = prev.channel_open_direct_tcpip(host.clone(), jport as u32, "127.0.0.1", 0).await.map_err(|e| format!("failed to dial through jump host {i} ({host}:{jport}): {e}"))?;
|
||||
self.handshake(ch.into_stream(), host, jport, check).await?
|
||||
}
|
||||
};
|
||||
authenticate(
|
||||
&mut h,
|
||||
Auth { user: username, methods: vec!["publickey".into()], agent: key.is_none() || sc.agent_auth, key, password: None, totp_secret: String::new() },
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("failed to establish SSH through jump host {i}: {e}"))?;
|
||||
chain.push(Arc::new(h));
|
||||
}
|
||||
let last = chain.pop().expect("at least one jump host");
|
||||
Ok((last, chain))
|
||||
}
|
||||
|
||||
fn handle(&self) -> Result<Arc<Handle<Handler>>> {
|
||||
self.handle.lock().unwrap().clone().ok_or_else(|| "not connected".into())
|
||||
}
|
||||
|
||||
pub fn write(&self, data: &[u8]) -> Result<()> {
|
||||
let tx = self.tx.lock().unwrap();
|
||||
tx.as_ref().ok_or("not connected")?.send(Cmd::Data(data.to_vec())).map_err(|_| "not connected".into())
|
||||
}
|
||||
|
||||
pub fn resize(&self, cols: u32, rows: u32) -> Result<()> {
|
||||
let tx = self.tx.lock().unwrap();
|
||||
tx.as_ref().ok_or("not connected")?.send(Cmd::Resize(cols, rows)).map_err(|_| "not connected".into())
|
||||
}
|
||||
|
||||
/// Round trip of a keepalive, in milliseconds.
|
||||
pub fn ping(&self) -> Result<i64> {
|
||||
let h = self.handle()?;
|
||||
let start = Instant::now();
|
||||
rt().block_on(h.send_ping())?;
|
||||
Ok(start.elapsed().as_millis() as i64)
|
||||
}
|
||||
|
||||
/// Runs a command on a separate channel and returns trimmed stdout.
|
||||
pub fn run_command(&self, command: &str) -> Result<String> {
|
||||
let h = self.handle()?;
|
||||
let command = command.to_string();
|
||||
rt().block_on(async move {
|
||||
let mut ch = h.channel_open_session().await.map_err(|e| format!("failed to open SSH session: {e}"))?;
|
||||
ch.exec(true, command.as_bytes()).await?;
|
||||
let mut out = vec![];
|
||||
while let Some(m) = ch.wait().await {
|
||||
match m {
|
||||
ChannelMsg::Data { data } => out.extend_from_slice(&data),
|
||||
ChannelMsg::Eof | ChannelMsg::Close => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(String::from_utf8_lossy(&out).trim().to_string())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn is_connected(&self) -> bool {
|
||||
self.handle.lock().unwrap().as_ref().is_some_and(|h| !h.is_closed())
|
||||
}
|
||||
|
||||
/// Idempotent; fires on_close once.
|
||||
pub fn close(&self) {
|
||||
if self.closed.swap(true, std::sync::atomic::Ordering::SeqCst) {
|
||||
return;
|
||||
}
|
||||
self.stop_all_forwards();
|
||||
*self.tx.lock().unwrap() = None;
|
||||
let h = self.handle.lock().unwrap().take();
|
||||
let hops = std::mem::take(&mut *self.hops.lock().unwrap());
|
||||
rt().spawn(async move {
|
||||
if let Some(h) = h {
|
||||
let _ = h.disconnect(russh::Disconnect::ByApplication, "", "en").await;
|
||||
}
|
||||
for h in hops.into_iter().rev() {
|
||||
let _ = h.disconnect(russh::Disconnect::ByApplication, "", "en").await;
|
||||
}
|
||||
});
|
||||
(self.cb.on_close)();
|
||||
}
|
||||
|
||||
// --- port forwarding -------------------------------------------------------------------
|
||||
|
||||
pub fn start_forward(&self, rule: PortForward) -> Result<()> {
|
||||
if self.forwards.lock().unwrap().contains_key(&rule.id) {
|
||||
return Err(format!("forward {:?} is already active", rule.id).into());
|
||||
}
|
||||
let h = self.handle()?;
|
||||
let tasks: Arc<Mutex<Vec<tokio::task::AbortHandle>>> = Arc::default();
|
||||
match rule.direction {
|
||||
ForwardDirection::Local => {
|
||||
let bind = if rule.bind_host.is_empty() { "127.0.0.1".to_string() } else { rule.bind_host.clone() };
|
||||
let addr = format!("{bind}:{}", rule.bind_port);
|
||||
let listener = rt()
|
||||
.block_on(tokio::net::TcpListener::bind(addr.clone()))
|
||||
.map_err(|e| format!("local forward: listen on {addr} failed: {e}"))?;
|
||||
let (dhost, dport) = (rule.dest_host.clone(), rule.dest_port as u32);
|
||||
let t2 = tasks.clone();
|
||||
let task = rt().spawn(async move {
|
||||
while let Ok((mut sock, peer)) = listener.accept().await {
|
||||
let (h, dhost) = (h.clone(), dhost.clone());
|
||||
let conn = tokio::spawn(async move {
|
||||
if let Ok(ch) = h.channel_open_direct_tcpip(dhost, dport, peer.ip().to_string(), peer.port() as u32).await {
|
||||
let mut remote = ch.into_stream();
|
||||
let _ = tokio::io::copy_bidirectional(&mut sock, &mut remote).await;
|
||||
}
|
||||
});
|
||||
t2.lock().unwrap().push(conn.abort_handle());
|
||||
}
|
||||
});
|
||||
tasks.lock().unwrap().push(task.abort_handle());
|
||||
}
|
||||
ForwardDirection::Remote => {
|
||||
let bind = if rule.bind_host.is_empty() { "0.0.0.0".to_string() } else { rule.bind_host.clone() };
|
||||
let port = rt()
|
||||
.block_on(h.tcpip_forward(bind.clone(), rule.bind_port as u32))
|
||||
.map_err(|e| format!("remote forward: server listen on {bind}:{} failed: {e}", rule.bind_port))?;
|
||||
let port = if rule.bind_port == 0 { port } else { rule.bind_port as u32 };
|
||||
self.shared.remote_targets.lock().unwrap().insert((bind, port), (rule.dest_host.clone(), rule.dest_port as u16));
|
||||
}
|
||||
}
|
||||
self.forwards.lock().unwrap().insert(rule.id.clone(), Forward { rule, tasks });
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn stop_forward(&self, id: &str) -> Result<()> {
|
||||
let f = self.forwards.lock().unwrap().remove(id).ok_or_else(|| format!("forward {id:?} not found"))?;
|
||||
self.shutdown_forward(f);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn shutdown_forward(&self, f: Forward) {
|
||||
for t in f.tasks.lock().unwrap().drain(..) {
|
||||
t.abort();
|
||||
}
|
||||
if f.rule.direction == ForwardDirection::Remote {
|
||||
let bind = if f.rule.bind_host.is_empty() { "0.0.0.0".to_string() } else { f.rule.bind_host.clone() };
|
||||
self.shared.remote_targets.lock().unwrap().retain(|(a, p), _| !(a == &bind && *p == f.rule.bind_port as u32));
|
||||
if let Ok(h) = self.handle() {
|
||||
let port = f.rule.bind_port as u32;
|
||||
rt().spawn(async move {
|
||||
let _ = h.cancel_tcpip_forward(bind, port).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn stop_all_forwards(&self) {
|
||||
let all: Vec<Forward> = self.forwards.lock().unwrap().drain().map(|(_, f)| f).collect();
|
||||
for f in all {
|
||||
self.shutdown_forward(f);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn active_forwards(&self) -> Vec<PortForward> {
|
||||
self.forwards.lock().unwrap().values().map(|f| f.rule.clone()).collect()
|
||||
}
|
||||
|
||||
// --- SFTP -------------------------------------------------------------------------------
|
||||
|
||||
/// The SFTP subsystem, opened on first use.
|
||||
pub fn sftp(&self) -> Result<Arc<russh_sftp::client::SftpSession>> {
|
||||
rt().block_on(async {
|
||||
let mut slot = self.sftp.lock().await;
|
||||
if let Some(s) = slot.as_ref() {
|
||||
return Ok(s.clone());
|
||||
}
|
||||
let h = self.handle()?;
|
||||
let ch = h.channel_open_session().await?;
|
||||
ch.request_subsystem(true, "sftp").await?;
|
||||
let s = Arc::new(
|
||||
russh_sftp::client::SftpSession::new(ch.into_stream()).await.map_err(|e| format!("sftpclient: failed to start SFTP subsystem: {e}"))?,
|
||||
);
|
||||
*slot = Some(s.clone());
|
||||
Ok(s)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn totp_matches_rfc_vector() {
|
||||
// RFC 6238 SHA-1 secret "12345678901234567890" is GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ in base32.
|
||||
let secret = "gezd gnbv gy3t qojq gezd gnbv gy3t qojq";
|
||||
assert_eq!(totp_at(secret, 59).unwrap(), "287082");
|
||||
assert_eq!(totp_at(secret, 1111111109).unwrap(), "081804");
|
||||
assert!(totp_code("JBSWY3DPEHPK3PXP").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn totp_prompt_detection() {
|
||||
assert!(is_totp_prompt("Verification code: "));
|
||||
assert!(is_totp_prompt("Enter your 6-digit OTP"));
|
||||
assert!(!is_totp_prompt("Password: "));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,262 @@
|
||||
//! End-to-end SSH tests against a throwaway OpenSSH server run as the current user. Skipped
|
||||
//! (passing) when `sshd`/`ssh-keygen` aren't installed.
|
||||
|
||||
use nexus_core::clients::Callbacks;
|
||||
use nexus_core::clients::sftp;
|
||||
use nexus_core::clients::ssh::{HostKeyCheck, SshClient};
|
||||
use nexus_core::models::{ForwardDirection, JumpHost, PortForward, Profile, Protocol};
|
||||
use std::io::{Read, Write};
|
||||
use std::net::{TcpListener, TcpStream};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
struct Server {
|
||||
dir: PathBuf,
|
||||
port: u16,
|
||||
child: Child,
|
||||
agent: Option<Child>,
|
||||
}
|
||||
|
||||
impl Drop for Server {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.child.kill();
|
||||
if let Some(a) = &mut self.agent {
|
||||
let _ = a.kill();
|
||||
}
|
||||
let _ = std::fs::remove_dir_all(&self.dir);
|
||||
}
|
||||
}
|
||||
|
||||
fn which(p: &str) -> Option<PathBuf> {
|
||||
["/usr/sbin", "/usr/bin", "/usr/local/sbin", "/usr/local/bin"].iter().map(|d| Path::new(d).join(p)).find(|p| p.exists())
|
||||
}
|
||||
|
||||
fn free_port() -> u16 {
|
||||
TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
|
||||
}
|
||||
|
||||
fn start(name: &str) -> Option<Server> {
|
||||
let sshd = which("sshd")?;
|
||||
let keygen = which("ssh-keygen")?;
|
||||
let sftp_server = ["/usr/lib/openssh/sftp-server", "/usr/libexec/openssh/sftp-server", "/usr/libexec/sftp-server"]
|
||||
.iter()
|
||||
.find(|p| Path::new(p).exists())?;
|
||||
let dir = std::env::temp_dir().join(format!("nexus-sshd-{name}-{}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
for k in ["host", "client"] {
|
||||
Command::new(&keygen).args(["-q", "-t", "ed25519", "-N", "", "-f"]).arg(dir.join(k)).status().unwrap();
|
||||
}
|
||||
std::fs::copy(dir.join("client.pub"), dir.join("authorized_keys")).unwrap();
|
||||
let port = free_port();
|
||||
let cfg = format!(
|
||||
"Port {port}\nListenAddress 127.0.0.1\nHostKey {d}/host\nAuthorizedKeysFile {d}/authorized_keys\nPidFile {d}/pid\n\
|
||||
StrictModes no\nUsePAM no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nPubkeyAuthentication yes\n\
|
||||
AllowTcpForwarding yes\nAllowAgentForwarding yes\nSubsystem sftp {sftp_server}\n",
|
||||
d = dir.display()
|
||||
);
|
||||
std::fs::write(dir.join("sshd_config"), cfg).unwrap();
|
||||
let child = Command::new(sshd).args(["-D", "-e", "-f"]).arg(dir.join("sshd_config")).spawn().ok()?;
|
||||
let start = Instant::now();
|
||||
while TcpStream::connect(("127.0.0.1", port)).is_err() {
|
||||
if start.elapsed() > Duration::from_secs(5) {
|
||||
return None;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
Some(Server { dir, port, child, agent: None })
|
||||
}
|
||||
|
||||
fn profile(s: &Server) -> Profile {
|
||||
let mut p = Profile::new(Protocol::Ssh);
|
||||
p.host = "127.0.0.1".into();
|
||||
p.port = s.port as i64;
|
||||
p.username = std::env::var("USER").unwrap_or_else(|_| "root".into());
|
||||
let c = p.ssh.as_mut().unwrap();
|
||||
c.auth_methods = vec!["publickey".into()];
|
||||
c.private_key_path = s.dir.join("client").to_string_lossy().into();
|
||||
// Never start the user's login shell: profiles can auto-attach tmux/byobu to their real session.
|
||||
c.startup_command = "/bin/sh -i".into();
|
||||
p
|
||||
}
|
||||
|
||||
type Seen = Arc<Mutex<Vec<u8>>>;
|
||||
|
||||
fn callbacks() -> (Callbacks, Seen) {
|
||||
let seen: Seen = Arc::default();
|
||||
let s2 = seen.clone();
|
||||
let mut cb = Callbacks::noop();
|
||||
cb.on_data = Arc::new(move |d| s2.lock().unwrap().extend_from_slice(d));
|
||||
(cb, seen)
|
||||
}
|
||||
|
||||
fn accept_all() -> (HostKeyCheck, Arc<Mutex<Vec<String>>>) {
|
||||
let fps: Arc<Mutex<Vec<String>>> = Arc::default();
|
||||
let f2 = fps.clone();
|
||||
(Arc::new(move |i| {
|
||||
f2.lock().unwrap().push(format!("{} {}", i.key_type, i.fingerprint));
|
||||
Ok(())
|
||||
}), fps)
|
||||
}
|
||||
|
||||
fn wait_for(seen: &Seen, needle: &str) -> bool {
|
||||
let start = Instant::now();
|
||||
while start.elapsed() < Duration::from_secs(10) {
|
||||
if String::from_utf8_lossy(&seen.lock().unwrap()).contains(needle) {
|
||||
return true;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn echo_server() -> u16 {
|
||||
let l = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let port = l.local_addr().unwrap().port();
|
||||
std::thread::spawn(move || {
|
||||
for s in l.incoming().flatten() {
|
||||
std::thread::spawn(move || {
|
||||
let mut s = s;
|
||||
let mut buf = [0u8; 1024];
|
||||
while let Ok(n) = s.read(&mut buf) {
|
||||
if n == 0 || s.write_all(&buf[..n]).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
port
|
||||
}
|
||||
|
||||
fn echo_through(port: u16) -> String {
|
||||
let start = Instant::now();
|
||||
loop {
|
||||
if let Ok(mut s) = TcpStream::connect(("127.0.0.1", port)) {
|
||||
s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
|
||||
s.write_all(b"ping").unwrap();
|
||||
let mut buf = [0u8; 4];
|
||||
if s.read_exact(&mut buf).is_ok() {
|
||||
return String::from_utf8_lossy(&buf).into();
|
||||
}
|
||||
}
|
||||
if start.elapsed() > Duration::from_secs(5) {
|
||||
return String::new();
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shell_exec_forwards_sftp() {
|
||||
let Some(s) = start("main") else { return eprintln!("sshd not available; skipping") };
|
||||
let (cb, seen) = callbacks();
|
||||
let (check, fps) = accept_all();
|
||||
let c = SshClient::new(&profile(&s), cb);
|
||||
c.connect(check, None, None).unwrap();
|
||||
assert!(fps.lock().unwrap()[0].starts_with("ssh-ed25519 SHA256:"));
|
||||
|
||||
c.resize(120, 40).unwrap();
|
||||
c.write(b"echo nexus-$((40+2)) $COLUMNS\n").unwrap();
|
||||
assert!(wait_for(&seen, "nexus-42"), "shell output: {}", String::from_utf8_lossy(&seen.lock().unwrap()));
|
||||
assert_eq!(c.run_command("echo hi").unwrap(), "hi");
|
||||
assert!(c.ping().unwrap() >= 0);
|
||||
|
||||
let echo = echo_server();
|
||||
let lport = free_port();
|
||||
let rule = PortForward { id: "l".into(), direction: ForwardDirection::Local, bind_port: lport as i64, dest_host: "127.0.0.1".into(), dest_port: echo as i64, enabled: true, ..Default::default() };
|
||||
c.start_forward(rule).unwrap();
|
||||
assert_eq!(echo_through(lport), "ping");
|
||||
let rport = free_port();
|
||||
let rule = PortForward { id: "r".into(), direction: ForwardDirection::Remote, bind_host: "127.0.0.1".into(), bind_port: rport as i64, dest_host: "127.0.0.1".into(), dest_port: echo as i64, enabled: true, ..Default::default() };
|
||||
c.start_forward(rule).unwrap();
|
||||
assert_eq!(echo_through(rport), "ping");
|
||||
assert_eq!(c.active_forwards().len(), 2);
|
||||
c.stop_forward("l").unwrap();
|
||||
let start = Instant::now();
|
||||
while TcpStream::connect(("127.0.0.1", lport)).is_ok() {
|
||||
assert!(start.elapsed() < Duration::from_secs(2), "listener still open after stop");
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
|
||||
let sf = c.sftp().unwrap();
|
||||
let local = s.dir.join("up.txt");
|
||||
std::fs::write(&local, b"sftp payload").unwrap();
|
||||
let remote = s.dir.join("remote.txt").to_string_lossy().to_string();
|
||||
sftp::upload(&sf, &local, &remote, &|_| {}).unwrap();
|
||||
let names: Vec<String> = sftp::list(&sf, &s.dir.to_string_lossy()).unwrap().into_iter().map(|e| e.name).collect();
|
||||
assert!(names.contains(&"remote.txt".to_string()));
|
||||
let back = s.dir.join("back.txt");
|
||||
sftp::download(&sf, &remote, &back, &|_| {}).unwrap();
|
||||
assert_eq!(std::fs::read(back).unwrap(), b"sftp payload");
|
||||
assert_eq!(sftp::stat(&sf, &remote).unwrap().size, 12);
|
||||
c.close();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn close_fires_once_and_remote_exit_closes() {
|
||||
let Some(s) = start("close") else { return };
|
||||
let (mut cb, _) = callbacks();
|
||||
let n = Arc::new(Mutex::new(0));
|
||||
let n2 = n.clone();
|
||||
cb.on_close = Arc::new(move || *n2.lock().unwrap() += 1);
|
||||
let c = SshClient::new(&profile(&s), cb);
|
||||
c.connect(accept_all().0, None, None).unwrap();
|
||||
c.write(b"exit\n").unwrap();
|
||||
let start = Instant::now();
|
||||
while *n.lock().unwrap() == 0 && start.elapsed() < Duration::from_secs(10) {
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
c.close();
|
||||
assert_eq!(*n.lock().unwrap(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejected_host_key_reports_reason() {
|
||||
let Some(s) = start("reject") else { return };
|
||||
let c = SshClient::new(&profile(&s), Callbacks::noop());
|
||||
let err = c.connect(Arc::new(|_| Err("host key rejected by user".into())), None, None).unwrap_err();
|
||||
assert!(err.to_string().contains("rejected by user"), "{err}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jump_host_chain() {
|
||||
let Some(s) = start("jump") else { return };
|
||||
let mut p = profile(&s);
|
||||
let key = p.ssh.as_ref().unwrap().private_key_path.clone();
|
||||
p.ssh.as_mut().unwrap().jump_hosts = vec![JumpHost { host: "127.0.0.1".into(), port: s.port as i64, username: p.username.clone(), private_key_path: key, ..Default::default() }];
|
||||
let (check, fps) = accept_all();
|
||||
let c = SshClient::new(&p, Callbacks::noop());
|
||||
c.connect(check, None, None).unwrap();
|
||||
assert_eq!(fps.lock().unwrap().len(), 2, "jump and target both verified");
|
||||
assert_eq!(c.run_command("echo via-jump").unwrap(), "via-jump");
|
||||
c.close();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_auth_and_forwarding() {
|
||||
let Some(mut s) = start("agent") else { return };
|
||||
let (Some(agent), Some(add)) = (which("ssh-agent"), which("ssh-add")) else { return };
|
||||
let sock = s.dir.join("agent.sock");
|
||||
s.agent = Some(Command::new(agent).arg("-D").arg("-a").arg(&sock).spawn().unwrap());
|
||||
let start = Instant::now();
|
||||
while !sock.exists() && start.elapsed() < Duration::from_secs(5) {
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
assert!(Command::new(add).arg(s.dir.join("client")).env("SSH_AUTH_SOCK", &sock).status().unwrap().success());
|
||||
// SAFETY: tests in this file don't read SSH_AUTH_SOCK concurrently except through this client.
|
||||
unsafe { std::env::set_var("SSH_AUTH_SOCK", &sock) };
|
||||
let mut p = profile(&s);
|
||||
let c = p.ssh.as_mut().unwrap();
|
||||
c.private_key_path.clear();
|
||||
c.agent_auth = true;
|
||||
c.agent_forwarding = true;
|
||||
let (cb, seen) = callbacks();
|
||||
let cl = SshClient::new(&p, cb);
|
||||
cl.connect(accept_all().0, None, None).unwrap();
|
||||
cl.write(b"ssh-add -l >/dev/null && echo agent-$((1+1))-ok\n").unwrap();
|
||||
assert!(wait_for(&seen, "agent-2-ok"), "forwarded agent not usable: {}", String::from_utf8_lossy(&seen.lock().unwrap()));
|
||||
cl.close();
|
||||
}
|
||||
Reference in New Issue
Block a user