Something went wrong. Try again.
TRAMP-style remote editing for nushell: an ssh agent and a nu plugin, in Rust
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769//! RPC backend — communicates with a remote `tramp-agent` process via//! MsgPack-RPC over piped stdin/stdout (through an SSH session).//!//! This backend replaces the shell-parsing approach used by [`SshBackend`]//! with native RPC calls to the agent, providing://!//! - **Lower latency**: no shell startup overhead per operation//! - **Binary safety**: file contents use MsgPack's native `bin` type//! - **Batch operations**: multiple ops in a single round-trip//! - **Richer metadata**: native `lstat()` syscalls instead of parsing//! `stat --format=…` text output//!//! The backend is created by the deployment module after successfully//! uploading and starting the agent on the remote host.
use async_trait::async_trait;use bytes::Bytes;use std::sync::Arc;use std::time::{Duration, UNIX_EPOCH};
use super::rpc_client::{ RpcClient, get_array, get_bin, get_bool, get_i64, get_str, get_u64, make_params, val_bin, val_bool, val_str, val_str_array, val_u64,};use super::{ Backend, DirEntry, EntryKind, ExecResult, Metadata, PtyHandle, PtyReadResult, WatchInfo, WatchNotification,};use crate::errors::{TrampError, TrampResult};
// ---------------------------------------------------------------------------// RPC Backend// ---------------------------------------------------------------------------
/// A backend that delegates all operations to a running `tramp-agent` via/// MsgPack-RPC.////// The RPC client communicates over the agent's stdin/stdout, which are/// piped through the SSH connection. The `RpcBackend` owns the client and/// the description string for display purposes.pub struct RpcBackend<R, W> { /// The RPC client communicating with the remote agent. client: Arc<RpcClient<R, W>>, /// Human-readable description for `tramp connections` output. host: String,}
impl<R, W> RpcBackend<R, W>where R: tokio::io::AsyncRead + Unpin + Send + 'static, W: tokio::io::AsyncWrite + Unpin + Send + 'static,{ /// Create a new RPC backend wrapping an already-started agent. pub fn new(client: RpcClient<R, W>, host: String) -> Self { Self { client: Arc::new(client), host, } }
/// Get a reference to the underlying RPC client. pub fn client(&self) -> &RpcClient<R, W> { &self.client }}
// ---------------------------------------------------------------------------// Backend trait implementation// ---------------------------------------------------------------------------
#[async_trait]impl<R, W> Backend for RpcBackend<R, W>where R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static, W: tokio::io::AsyncWrite + Unpin + Send + Sync + 'static,{ async fn read(&self, path: &str) -> TrampResult<Bytes> { let params = make_params(vec![("path", val_str(path))]); let result = self.client.call("file.read", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("file.read: expected map result".into()))?;
let data = get_bin(map, "data").ok_or_else(|| { TrampError::Internal("file.read: missing 'data' field in response".into()) })?;
Ok(Bytes::copy_from_slice(data)) }
async fn write(&self, path: &str, data: Bytes) -> TrampResult<()> { let params = make_params(vec![("path", val_str(path)), ("data", val_bin(&data))]); let _result = self.client.call("file.write", params).await?; Ok(()) }
async fn list(&self, path: &str) -> TrampResult<Vec<DirEntry>> { let params = make_params(vec![("path", val_str(path))]); let result = self.client.call("dir.list", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("dir.list: expected map result".into()))?;
let entries_val = get_array(map, "entries").ok_or_else(|| { TrampError::Internal("dir.list: missing 'entries' field in response".into()) })?;
let mut entries = Vec::with_capacity(entries_val.len()); for entry_val in entries_val { let entry_map = match entry_val.as_map() { Some(m) => m, None => continue, };
// Skip entries that had errors on the remote side. if get_str(entry_map, "error").is_some() { continue; }
let name = get_str(entry_map, "name").unwrap_or("").to_string(); let kind = parse_kind(get_str(entry_map, "kind").unwrap_or("file")); let size = get_u64(entry_map, "size"); let modified = get_u64(entry_map, "modified_ns").map(|ns| UNIX_EPOCH + Duration::from_nanos(ns)); let permissions = get_u64(entry_map, "permissions").map(|p| p as u32); let nlinks = get_u64(entry_map, "nlinks"); let inode = get_u64(entry_map, "inode"); let owner = get_str(entry_map, "owner").map(|s| s.to_string()); let group = get_str(entry_map, "group").map(|s| s.to_string()); let symlink_target = get_str(entry_map, "symlink_target").map(|s| s.to_string());
entries.push(DirEntry { name, kind, size, modified, permissions, nlinks, inode, owner, group, symlink_target, }); }
Ok(entries) }
async fn stat(&self, path: &str) -> TrampResult<Metadata> { let params = make_params(vec![("path", val_str(path))]); let result = self.client.call("file.stat", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("file.stat: expected map result".into()))?;
parse_stat_map(map) }
async fn exec(&self, cmd: &str, args: &[&str]) -> TrampResult<ExecResult> { let mut param_entries = vec![("command", val_str(cmd))];
// Build the args array. We hold onto the Value so the borrow // into `param_entries` remains valid. let args_val = val_str_array(args); param_entries.push(("args", args_val));
let params = make_params(param_entries); let result = self.client.call("process.run", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("process.run: expected map result".into()))?;
let exit_code = get_i64(map, "exit_code").unwrap_or(-1) as i32;
// stdout / stderr can be either binary or string in the response. let stdout = extract_bytes(map, "stdout"); let stderr = extract_bytes(map, "stderr");
Ok(ExecResult { stdout, stderr, exit_code, }) }
async fn delete(&self, path: &str) -> TrampResult<()> { let params = make_params(vec![("path", val_str(path))]); let _result = self.client.call("file.delete", params).await?; Ok(()) }
async fn check(&self) -> TrampResult<()> { self.client.ping().await }
fn description(&self) -> String { format!("rpc:{}", self.host) }
// -- Watch operations (RPC agent supports these natively) -----------------
async fn watch_add(&self, path: &str, recursive: bool) -> TrampResult<()> { let params = make_params(vec![ ("path", val_str(path)), ("recursive", val_bool(recursive)), ]); let _result = self.client.call("watch.add", params).await?; Ok(()) }
async fn watch_remove(&self, path: &str) -> TrampResult<()> { let params = make_params(vec![("path", val_str(path))]); let _result = self.client.call("watch.remove", params).await?; Ok(()) }
async fn watch_list(&self) -> TrampResult<Vec<WatchInfo>> { let params = make_params(vec![]); let result = self.client.call("watch.list", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("watch.list: expected map result".into()))?;
let watches_val = get_array(map, "watches").ok_or_else(|| { TrampError::Internal("watch.list: missing 'watches' field in response".into()) })?;
let mut watches = Vec::with_capacity(watches_val.len()); for entry_val in watches_val { let entry_map = match entry_val.as_map() { Some(m) => m, None => continue, }; let path = get_str(entry_map, "path").unwrap_or("").to_string(); let recursive = get_bool(entry_map, "recursive").unwrap_or(false); watches.push(WatchInfo { path, recursive }); }
Ok(watches) }
async fn watch_poll(&self) -> TrampResult<Vec<WatchNotification>> { let notifications = self.client.drain_notifications().await;
let mut result = Vec::with_capacity(notifications.len()); for notif in ¬ifications { if notif.method != "fs.changed" { continue; } let map = match notif.params.as_map() { Some(m) => m, None => continue, }; let paths = get_array(map, "paths") .map(|arr| { arr.iter() .filter_map(|v| v.as_str().map(|s| s.to_string())) .collect() }) .unwrap_or_default(); let kind = get_str(map, "kind").unwrap_or("unknown").to_string(); result.push(WatchNotification { paths, kind }); }
Ok(result) }
fn supports_watch(&self) -> bool { true }
// ----------------------------------------------------------------------- // Chunked / streaming I/O // -----------------------------------------------------------------------
async fn file_size(&self, path: &str) -> TrampResult<u64> { let params = make_params(vec![("path", val_str(path))]); let result = self.client.call("file.size", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("file.size: expected map result".into()))?;
let size = get_u64(map, "size").ok_or_else(|| { TrampError::Internal("file.size: missing 'size' field in response".into()) })?;
Ok(size) }
async fn read_range( &self, path: &str, offset: u64, length: u64, ) -> TrampResult<(Vec<u8>, bool)> { let params = make_params(vec![ ("path", val_str(path)), ("offset", val_u64(offset)), ("length", val_u64(length)), ]); let result = self.client.call("file.read_range", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("file.read_range: expected map result".into()))?;
let data = get_bin(map, "data") .ok_or_else(|| { TrampError::Internal("file.read_range: missing 'data' field in response".into()) })? .to_vec();
let eof = get_bool(map, "eof").unwrap_or(true);
Ok((data, eof)) }
async fn write_range( &self, path: &str, offset: u64, data: Bytes, truncate: bool, ) -> TrampResult<u64> { let mut kv = vec![ ("path", val_str(path)), ("offset", val_u64(offset)), ("data", val_bin(&data)), ]; if truncate { kv.push(("truncate", val_u64(1))); } let params = make_params(kv); let result = self.client.call("file.write_range", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("file.write_range: expected map result".into()))?;
let written = get_u64(map, "written").unwrap_or(data.len() as u64); Ok(written) }
fn supports_streaming(&self) -> bool { true }
// -- PTY operations ---------------------------------------------------
async fn pty_start( &self, program: &str, args: &[&str], rows: u16, cols: u16, ) -> TrampResult<PtyHandle> { let params = make_params(vec![ ("program", val_str(program)), ("args", val_str_array(args)), ("rows", val_u64(rows as u64)), ("cols", val_u64(cols as u64)), ]); let result = self.client.call("process.start_pty", params).await?;
let map = result .as_map() .ok_or_else(|| TrampError::Internal("process.start_pty: expected map result".into()))?;
let handle = get_u64(map, "handle") .ok_or_else(|| TrampError::Internal("process.start_pty: missing handle".into()))?; let pid = get_u64(map, "pid").unwrap_or(0);
Ok(PtyHandle { handle, pid }) }
async fn pty_read(&self, handle: u64) -> TrampResult<PtyReadResult> { let params = make_params(vec![("handle", val_u64(handle))]); let result = self.client.call("process.read", params).await?;
let map = result.as_map().ok_or_else(|| { TrampError::Internal("process.read (pty): expected map result".into()) })?;
let data = get_bin(map, "stdout") .map(|b| b.to_vec()) .unwrap_or_default(); let running = get_bool(map, "running").unwrap_or(false); let exit_code = get_i64(map, "exit_code").map(|c| c as i32);
Ok(PtyReadResult { data, running, exit_code, }) }
async fn pty_write(&self, handle: u64, data: Bytes) -> TrampResult<()> { let params = make_params(vec![("handle", val_u64(handle)), ("data", val_bin(&data))]); self.client.call("process.write", params).await?; Ok(()) }
async fn pty_resize(&self, handle: u64, rows: u16, cols: u16) -> TrampResult<()> { let params = make_params(vec![ ("handle", val_u64(handle)), ("rows", val_u64(rows as u64)), ("cols", val_u64(cols as u64)), ]); self.client.call("process.resize", params).await?; Ok(()) }
async fn pty_kill(&self, handle: u64) -> TrampResult<()> { let params = make_params(vec![("handle", val_u64(handle))]); self.client.call("process.kill", params).await?; Ok(()) }
fn supports_pty(&self) -> bool { true }}
// ---------------------------------------------------------------------------// Helpers// ---------------------------------------------------------------------------
/// Parse a `kind` string from the agent into an [`EntryKind`].fn parse_kind(kind: &str) -> EntryKind { match kind { "dir" | "directory" => EntryKind::Dir, "symlink" | "symbolic link" => EntryKind::Symlink, _ => EntryKind::File, }}
/// Parse a stat metadata map from the agent into a [`Metadata`] struct.fn parse_stat_map(map: &[(rmpv::Value, rmpv::Value)]) -> TrampResult<Metadata> { let kind = parse_kind(get_str(map, "kind").unwrap_or("file")); let size = get_u64(map, "size").unwrap_or(0); let modified = get_u64(map, "modified_ns").map(|ns| UNIX_EPOCH + Duration::from_nanos(ns)); let permissions = get_u64(map, "permissions").map(|p| p as u32); let nlinks = get_u64(map, "nlinks"); let inode = get_u64(map, "inode"); let owner = get_str(map, "owner").map(|s| s.to_string()); let group = get_str(map, "group").map(|s| s.to_string()); let symlink_target = get_str(map, "symlink_target").map(|s| s.to_string());
Ok(Metadata { kind, size, modified, permissions, nlinks, inode, owner, group, symlink_target, })}
/// Extract a byte buffer from a MsgPack map field.////// The agent may encode stdout/stderr as either `bin` (binary) or `str`/// (string) depending on the content. This helper handles both.fn extract_bytes(map: &[(rmpv::Value, rmpv::Value)], key: &str) -> Bytes { // Try binary first. if let Some(data) = get_bin(map, key) { return Bytes::copy_from_slice(data); } // Fall back to string. if let Some(s) = get_str(map, key) { return Bytes::copy_from_slice(s.as_bytes()); } Bytes::new()}
// ---------------------------------------------------------------------------// Tests// ---------------------------------------------------------------------------
#[cfg(test)]mod tests { use super::super::rpc_client; use super::*; use rmpv::Value;
#[test] fn parse_kind_variants() { assert_eq!(parse_kind("file"), EntryKind::File); assert_eq!(parse_kind("dir"), EntryKind::Dir); assert_eq!(parse_kind("directory"), EntryKind::Dir); assert_eq!(parse_kind("symlink"), EntryKind::Symlink); assert_eq!(parse_kind("symbolic link"), EntryKind::Symlink); assert_eq!(parse_kind("other"), EntryKind::File); }
#[test] fn parse_stat_map_basic() { let map = vec![ (Value::String("kind".into()), Value::String("file".into())), (Value::String("size".into()), Value::Integer(1024u64.into())), ( Value::String("permissions".into()), Value::Integer(0o644u64.into()), ), (Value::String("nlinks".into()), Value::Integer(1u64.into())), ( Value::String("inode".into()), Value::Integer(12345u64.into()), ), (Value::String("owner".into()), Value::String("root".into())), (Value::String("group".into()), Value::String("root".into())), ];
let meta = parse_stat_map(&map).unwrap(); assert_eq!(meta.kind, EntryKind::File); assert_eq!(meta.size, 1024); assert_eq!(meta.permissions, Some(0o644)); assert_eq!(meta.nlinks, Some(1)); assert_eq!(meta.inode, Some(12345)); assert_eq!(meta.owner.as_deref(), Some("root")); assert_eq!(meta.group.as_deref(), Some("root")); assert!(meta.symlink_target.is_none()); }
#[test] fn parse_stat_map_symlink() { let map = vec![ ( Value::String("kind".into()), Value::String("symlink".into()), ), (Value::String("size".into()), Value::Integer(11u64.into())), ( Value::String("symlink_target".into()), Value::String("/etc/resolv.conf".into()), ), ];
let meta = parse_stat_map(&map).unwrap(); assert_eq!(meta.kind, EntryKind::Symlink); assert_eq!(meta.symlink_target.as_deref(), Some("/etc/resolv.conf")); }
#[test] fn parse_stat_map_with_modified_ns() { let ns: u64 = 1_700_000_000_000_000_000; // ~2023-11-14 let map = vec![ (Value::String("kind".into()), Value::String("file".into())), (Value::String("size".into()), Value::Integer(0u64.into())), ( Value::String("modified_ns".into()), Value::Integer(ns.into()), ), ];
let meta = parse_stat_map(&map).unwrap(); assert!(meta.modified.is_some()); let dur = meta.modified.unwrap().duration_since(UNIX_EPOCH).unwrap(); assert_eq!(dur.as_nanos() as u64, ns); }
#[test] fn extract_bytes_binary() { let map = vec![( Value::String("stdout".into()), Value::Binary(vec![0xDE, 0xAD, 0xBE, 0xEF]), )];
let data = extract_bytes(&map, "stdout"); assert_eq!(&data[..], &[0xDE, 0xAD, 0xBE, 0xEF]); }
#[test] fn extract_bytes_string_fallback() { let map = vec![( Value::String("stderr".into()), Value::String("error message".into()), )];
let data = extract_bytes(&map, "stderr"); assert_eq!(&data[..], b"error message"); }
#[test] fn extract_bytes_missing_key() { let map: Vec<(Value, Value)> = vec![]; let data = extract_bytes(&map, "stdout"); assert!(data.is_empty()); }
/// End-to-end test using a mock reader/writer pair that simulates the /// agent responding to `file.stat`. #[tokio::test] async fn rpc_backend_stat_mock() { // Build a mock response for file.stat. let resp = rpc_client::Response { version: "2.0".into(), id: 1, result: Some(Value::Map(vec![ (Value::String("kind".into()), Value::String("file".into())), (Value::String("size".into()), Value::Integer(42u64.into())), ( Value::String("permissions".into()), Value::Integer(0o755u64.into()), ), (Value::String("nlinks".into()), Value::Integer(1u64.into())), ( Value::String("inode".into()), Value::Integer(9999u64.into()), ), ( Value::String("owner".into()), Value::String("nobody".into()), ), ( Value::String("group".into()), Value::String("nogroup".into()), ), ])), error: None, };
let payload = rmp_serde::to_vec_named(&resp).unwrap(); let len = payload.len() as u32; let mut frame = Vec::new(); frame.extend_from_slice(&len.to_be_bytes()); frame.extend_from_slice(&payload);
let reader = std::io::Cursor::new(frame); let writer = Vec::<u8>::new();
let client = RpcClient::new(reader, writer); let backend = RpcBackend::new(client, "mockhost".into());
let meta = backend.stat("/etc/hosts").await.unwrap(); assert_eq!(meta.kind, EntryKind::File); assert_eq!(meta.size, 42); assert_eq!(meta.permissions, Some(0o755)); assert_eq!(meta.nlinks, Some(1)); assert_eq!(meta.inode, Some(9999)); assert_eq!(meta.owner.as_deref(), Some("nobody")); assert_eq!(meta.group.as_deref(), Some("nogroup")); }
/// End-to-end test using a mock reader/writer pair that simulates the /// agent responding to `dir.list`. #[tokio::test] async fn rpc_backend_list_mock() { let resp = rpc_client::Response { version: "2.0".into(), id: 1, result: Some(Value::Map(vec![( Value::String("entries".into()), Value::Array(vec![ Value::Map(vec![ (Value::String("name".into()), Value::String("hosts".into())), (Value::String("kind".into()), Value::String("file".into())), (Value::String("size".into()), Value::Integer(256u64.into())), ]), Value::Map(vec![ (Value::String("name".into()), Value::String("ssl".into())), (Value::String("kind".into()), Value::String("dir".into())), (Value::String("size".into()), Value::Integer(4096u64.into())), ]), ]), )])), error: None, };
let payload = rmp_serde::to_vec_named(&resp).unwrap(); let len = payload.len() as u32; let mut frame = Vec::new(); frame.extend_from_slice(&len.to_be_bytes()); frame.extend_from_slice(&payload);
let reader = std::io::Cursor::new(frame); let writer = Vec::<u8>::new();
let client = RpcClient::new(reader, writer); let backend = RpcBackend::new(client, "mockhost".into());
let entries = backend.list("/etc").await.unwrap(); assert_eq!(entries.len(), 2); assert_eq!(entries[0].name, "hosts"); assert_eq!(entries[0].kind, EntryKind::File); assert_eq!(entries[0].size, Some(256)); assert_eq!(entries[1].name, "ssl"); assert_eq!(entries[1].kind, EntryKind::Dir); }
/// Test that RPC errors are correctly mapped to TrampError variants. #[tokio::test] async fn rpc_backend_not_found_error() { let resp = rpc_client::Response { version: "2.0".into(), id: 1, result: None, error: Some(rpc_client::RpcErrorData { code: rpc_client::error_code::NOT_FOUND, message: "no such file or directory: /tmp/nope".into(), }), };
let payload = rmp_serde::to_vec_named(&resp).unwrap(); let len = payload.len() as u32; let mut frame = Vec::new(); frame.extend_from_slice(&len.to_be_bytes()); frame.extend_from_slice(&payload);
let reader = std::io::Cursor::new(frame); let writer = Vec::<u8>::new();
let client = RpcClient::new(reader, writer); let backend = RpcBackend::new(client, "mockhost".into());
let err = backend.stat("/tmp/nope").await.unwrap_err(); assert!( matches!(err, TrampError::NotFound(_)), "expected NotFound, got: {err:?}" ); }
/// Test that process.run responses are correctly parsed. #[tokio::test] async fn rpc_backend_exec_mock() { let resp = rpc_client::Response { version: "2.0".into(), id: 1, result: Some(Value::Map(vec![ ( Value::String("exit_code".into()), Value::Integer(0i64.into()), ), ( Value::String("stdout".into()), Value::Binary(b"hello world\n".to_vec()), ), (Value::String("stderr".into()), Value::Binary(vec![])), ])), error: None, };
let payload = rmp_serde::to_vec_named(&resp).unwrap(); let len = payload.len() as u32; let mut frame = Vec::new(); frame.extend_from_slice(&len.to_be_bytes()); frame.extend_from_slice(&payload);
let reader = std::io::Cursor::new(frame); let writer = Vec::<u8>::new();
let client = RpcClient::new(reader, writer); let backend = RpcBackend::new(client, "mockhost".into());
let result = backend.exec("echo", &["hello", "world"]).await.unwrap(); assert_eq!(result.exit_code, 0); assert_eq!(&result.stdout[..], b"hello world\n"); assert!(result.stderr.is_empty()); }}