diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 768a4dd..b841210 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -2335,6 +2335,16 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "io-kit-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617ee6cf8e3f66f3b4ea67a4058564628cde41901316e19f559e14c7c72c5e7b" +dependencies = [ + "core-foundation-sys", + "mach2", +] + [[package]] name = "itertools" version = "0.15.0" @@ -2583,6 +2593,15 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "mach2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d640282b302c0bb0a2a8e0233ead9035e3bed871f0b7e81fe4a1ec829765db44" +dependencies = [ + "libc", +] + [[package]] name = "md5" version = "0.8.1" @@ -2782,6 +2801,7 @@ dependencies = [ "chrono", "hmac 0.13.0", "keyring", + "libc", "portable-pty", "rand 0.8.8", "rusqlite", @@ -2789,6 +2809,7 @@ dependencies = [ "russh-sftp", "serde", "serde_json", + "serialport", "sha1 0.11.0", "sha2 0.11.0", "tokio", @@ -2796,6 +2817,17 @@ dependencies = [ "uuid", ] +[[package]] +name = "nix" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b" +dependencies = [ + "bitflags 1.3.2", + "cfg-if", + "libc", +] + [[package]] name = "nix" version = "0.28.0" @@ -4423,6 +4455,24 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "serialport" +version = "4.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba5f8f29aa20853c4e3e85a33ec580eb66be1f057142e77a333834a318bacf2" +dependencies = [ + "bitflags 2.13.2", + "cfg-if", + "core-foundation 0.10.1", + "core-foundation-sys", + "io-kit-sys", + "mach2", + "nix 0.26.4", + "scopeguard", + "unescaper", + "windows-sys 0.52.0", +] + [[package]] name = "sha1" version = "0.10.7" @@ -5018,6 +5068,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "unescaper" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7285e83a80ce76f5e7bce79fa41f68d78ba62d1003cf27bf748ab24413808cf4" +dependencies = [ + "thiserror 2.0.21", +] + [[package]] name = "unicode-general-category" version = "1.1.0" diff --git a/rust/core/Cargo.toml b/rust/core/Cargo.toml index 689e533..dd28272 100644 --- a/rust/core/Cargo.toml +++ b/rust/core/Cargo.toml @@ -22,3 +22,7 @@ sha2 = "0.11.0" hmac = "0.13.0" sha1 = "0.11.0" portable-pty = "0.9.0" +serialport = { version = "4", default-features = false } + +[target.'cfg(target_os = "linux")'.dev-dependencies] +libc = "0.2" diff --git a/rust/core/src/clients/mod.rs b/rust/core/src/clients/mod.rs index 48e5987..4442451 100644 --- a/rust/core/src/clients/mod.rs +++ b/rust/core/src/clients/mod.rs @@ -36,3 +36,122 @@ impl Callbacks { Callbacks { on_data: Arc::new(|_| {}), on_error: Arc::new(|_| {}), on_close: Arc::new(|| {}) } } } + +pub mod raw; +pub mod serial; +pub mod telnet; + +/// Called with protocol negotiation debug messages (Go's `onDebug`). +pub type DebugFn = Arc; + +/// TCP dial shared by the Telnet and RAW clients: Go's `net.DialTimeout` with a 30 s default. +/// Returns the stream and the `host:port` string used in error messages. +pub(crate) fn dial(host: &str, port: i64, timeout_secs: i64) -> (String, std::io::Result) { + use std::net::{TcpStream, ToSocketAddrs}; + let timeout = std::time::Duration::from_secs(if timeout_secs > 0 { timeout_secs as u64 } else { 30 }); + // Go's net.JoinHostPort brackets IPv6 literals. + let addr = if host.contains(':') { format!("[{host}]:{port}") } else { format!("{host}:{port}") }; + let result = (|| { + let mut last = std::io::Error::new(std::io::ErrorKind::NotFound, "no addresses found"); + for sa in addr.to_socket_addrs()? { + match TcpStream::connect_timeout(&sa, timeout) { + Ok(s) => return Ok(s), + Err(e) => last = e, + } + } + Err(last) + })(); + (addr, result) +} + +/// The TCP connection lifecycle shared by the Telnet and RAW clients: one writer-side stream behind +/// a lock, a reader thread that feeds `on_read`, and Go's error/close semantics (a read error or EOF +/// reports `read error: …` then closes; `close` is idempotent and fires `on_close` once). +pub(crate) struct TcpConn { + stream: std::sync::Mutex>, + on_error: ErrorFn, + on_close: CloseFn, +} + +impl TcpConn { + pub(crate) fn new(on_error: ErrorFn, on_close: CloseFn) -> TcpConn { + TcpConn { stream: std::sync::Mutex::new(None), on_error, on_close } + } + + pub(crate) fn connect( + self: &Arc, + stream: std::net::TcpStream, + mut on_read: impl FnMut(&[u8]) + Send + 'static, + ) -> crate::Result<()> { + let mut reader = stream.try_clone()?; + { + let mut guard = self.stream.lock().unwrap(); + if guard.is_some() { + return Err("already connected".into()); + } + *guard = Some(stream); + } + let me = self.clone(); + std::thread::spawn(move || { + use std::io::Read; + let mut buf = [0u8; 4096]; + loop { + let err = match reader.read(&mut buf) { + Ok(0) => "EOF".to_string(), + Ok(n) => { + on_read(&buf[..n]); + continue; + } + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) => e.to_string(), + }; + // A local close also ends the read; Go reported that as an error too, we don't. + if me.is_connected() { + (me.on_error)(format!("read error: {err}")); + me.close(); + } + return; + } + }); + Ok(()) + } + + pub(crate) fn write(&self, data: &[u8]) -> crate::Result<()> { + use std::io::Write; + let guard = self.stream.lock().unwrap(); + let mut s = guard.as_ref().ok_or("not connected")?; + s.write_all(data)?; + Ok(()) + } + + pub(crate) fn is_connected(&self) -> bool { + self.stream.lock().unwrap().is_some() + } + + pub(crate) fn close(&self) { + let taken = self.stream.lock().unwrap().take(); + if let Some(s) = taken { + let _ = s.shutdown(std::net::Shutdown::Both); + (self.on_close)(); + } + } +} + +/// True for the encoding names Go mapped to ISO 8859-1. +pub(crate) fn is_latin1(encoding: &str) -> bool { + matches!(encoding.trim().to_ascii_lowercase().as_str(), "iso-8859-1" | "latin1" | "latin-1") +} + +/// UTF-8 → ISO 8859-1. Like Go's charmap encoder, input that isn't valid UTF-8 or holds a +/// character above U+00FF is returned unchanged. +pub(crate) fn utf8_to_latin1(data: &[u8]) -> Vec { + std::str::from_utf8(data) + .ok() + .and_then(|s| s.chars().map(|c| u8::try_from(c).ok()).collect()) + .unwrap_or_else(|| data.to_vec()) +} + +/// ISO 8859-1 → UTF-8 (every byte maps to the code point of the same value). +pub(crate) fn latin1_to_utf8(data: &[u8]) -> String { + data.iter().map(|&b| b as char).collect() +} diff --git a/rust/core/src/clients/raw.rs b/rust/core/src/clients/raw.rs new file mode 100644 index 0000000..a003afa --- /dev/null +++ b/rust/core/src/clients/raw.rs @@ -0,0 +1,431 @@ +//! RAW TCP client (port of Go `internal/rawclient`): binary-safe send/receive, a capped history +//! of sent ("send") and received ("recv") frames, JSON capture, and the hex/escape encoders the +//! RAW view reuses. + +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; + +use base64::Engine; +use serde::Serialize; + +use super::{CloseFn, ErrorFn, TcpConn}; +use crate::models::{LineEnding, Profile, Time}; + +pub const MAX_HISTORY_ENTRIES: usize = 10_000; +pub const MAX_HISTORY_BYTES: usize = 10 * 1024 * 1024; + +pub const SEND: &str = "send"; +pub const RECV: &str = "recv"; + +/// `(data, direction, timestamp in unix ms)`; direction is [`SEND`] or [`RECV`]. +pub type RawDataFn = Arc; + +/// One sent or received frame. Serialises like Go's (RFC 3339 timestamp, base64 data). +#[derive(Clone, Debug, PartialEq, Serialize)] +pub struct HistoryEntry { + pub timestamp: Time, + pub direction: String, + #[serde(serialize_with = "ser_b64")] + pub data: Vec, +} + +fn ser_b64(d: &[u8], s: S) -> Result { + s.serialize_str(&base64::engine::general_purpose::STANDARD.encode(d)) +} + +pub struct RawClient { + inner: Arc, +} + +struct Inner { + conn: Arc, + host: String, + port: i64, + connect_timeout: i64, + line_ending: LineEnding, + on_data: RawDataFn, + history: Mutex, +} + +#[derive(Default)] +struct History { + entries: VecDeque, + bytes: usize, +} + +impl RawClient { + pub fn new(profile: &Profile, on_data: RawDataFn, on_error: ErrorFn, on_close: CloseFn) -> RawClient { + RawClient { + inner: Arc::new(Inner { + conn: Arc::new(TcpConn::new(on_error, on_close)), + host: profile.host.clone(), + port: profile.port, + connect_timeout: profile.connect_timeout, + line_ending: profile.raw.as_ref().map(|r| r.line_ending).unwrap_or_default(), + on_data, + history: Mutex::default(), + }), + } + } + + pub fn connect(&self) -> crate::Result<()> { + let inner = &self.inner; + if inner.conn.is_connected() { + return Err("already connected".into()); + } + if inner.port == 0 { + return Err("port must be specified for raw TCP connections".into()); + } + let (addr, res) = super::dial(&inner.host, inner.port, inner.connect_timeout); + let stream = res.map_err(|e| format!("raw TCP connection to {addr} failed: {e}"))?; + let me = inner.clone(); + inner.conn.connect(stream, move |d| me.record(RECV, d)) + } + + /// Sends bytes as-is, then records them and reports them through `on_data` as "send". + pub fn write(&self, data: &[u8]) -> crate::Result<()> { + self.inner.conn.write(data)?; + self.inner.record(SEND, data); + Ok(()) + } + + /// Sends text with the configured line ending appended (unless it already ends with it). + pub fn write_text(&self, text: &str) -> crate::Result<()> { + self.write(&apply_line_ending(text, self.inner.line_ending)) + } + + /// Sends hex such as "48656c6c6f", "48 65 6c" or "0x48 0x65". + pub fn write_hex(&self, hex: &str) -> crate::Result<()> { + let data = parse_hex_string(hex).map_err(|e| format!("invalid hex input: {e}"))?; + self.write(&data) + } + + /// Sends text after processing C-style escapes (see [`parse_escaped`]). + pub fn write_escaped(&self, text: &str) -> crate::Result<()> { + let data = parse_escaped(text).map_err(|e| format!("escape processing failed: {e}"))?; + self.write(&data) + } + + pub fn close(&self) { + self.inner.conn.close(); + } + + pub fn is_connected(&self) -> bool { + self.inner.conn.is_connected() + } + + pub fn history(&self) -> Vec { + self.inner.history.lock().unwrap().entries.iter().cloned().collect() + } + + pub fn clear_history(&self) { + *self.inner.history.lock().unwrap() = History::default(); + } + + /// Writes the history to `path` as indented JSON (mode 0600 on Unix), as Go did. + pub fn save_capture(&self, path: &std::path::Path) -> crate::Result<()> { + let json = serde_json::to_vec_pretty(&self.history()).map_err(|e| format!("failed to marshal capture data: {e}"))?; + let mut opts = std::fs::OpenOptions::new(); + opts.write(true).create(true).truncate(true); + #[cfg(unix)] + std::os::unix::fs::OpenOptionsExt::mode(&mut opts, 0o600); + std::io::Write::write_all(&mut opts.open(path)?, &json)?; + Ok(()) + } +} + +impl Inner { + fn record(&self, direction: &str, data: &[u8]) { + let timestamp = crate::models::now(); + let ms = timestamp.timestamp_millis(); + self.append_history(HistoryEntry { timestamp, direction: direction.into(), data: data.to_vec() }); + (self.on_data)(data, direction, ms); + } + + fn append_history(&self, mut entry: HistoryEntry) { + if entry.data.len() > MAX_HISTORY_BYTES { + entry.data.drain(..entry.data.len() - MAX_HISTORY_BYTES); + } + let mut h = self.history.lock().unwrap(); + h.bytes += entry.data.len(); + h.entries.push_back(entry); + while h.entries.len() > MAX_HISTORY_ENTRIES || h.bytes > MAX_HISTORY_BYTES { + let Some(old) = h.entries.pop_front() else { break }; + h.bytes -= old.data.len(); + } + } +} + +fn apply_line_ending(text: &str, le: LineEnding) -> Vec { + let suffix = le.bytes(); + let mut out = text.as_bytes().to_vec(); + if !suffix.is_empty() && !out.ends_with(suffix) { + out.extend_from_slice(suffix); + } + out +} + +/// Parses hex with optional whitespace and 0x/0X prefixes into bytes. +pub fn parse_hex_string(s: &str) -> crate::Result> { + let s = s.trim(); + if s.is_empty() { + return Err("empty hex string".into()); + } + let s = s.replace("0x", "").replace("0X", ""); + let mut digits = Vec::new(); + for c in s.chars().filter(|c| !c.is_whitespace()) { + let d = c.to_digit(16).ok_or_else(|| format!("invalid character in hex string: {c:?}"))?; + digits.push(d as u8); + } + if digits.len() % 2 != 0 { + return Err("hex string has odd number of digits".into()); + } + Ok(digits.chunks(2).map(|p| (p[0] << 4) | p[1]).collect()) +} + +/// Space-separated lowercase hex, e.g. "48 65 6c 6c 6f". +pub fn format_hex(data: &[u8]) -> String { + data.iter().map(|b| format!("{b:02x}")).collect::>().join(" ") +} + +/// Classic 16-bytes-per-line hex dump with offset, hex and ASCII columns. +pub fn format_hex_dump(data: &[u8]) -> String { + use std::fmt::Write; + let mut sb = String::new(); + for (line, chunk) in data.chunks(16).enumerate() { + let _ = write!(sb, "{:08x} ", line * 16); + for i in 0..16 { + if i == 8 { + sb.push(' '); + } + match chunk.get(i) { + Some(b) => { + let _ = write!(sb, "{b:02x} "); + } + None => sb.push_str(" "), + } + } + sb.push_str(" |"); + sb.extend(chunk.iter().map(|&b| if (0x20..=0x7e).contains(&b) { b as char } else { '.' })); + sb.extend(std::iter::repeat_n(' ', 16 - chunk.len())); + sb.push_str("|\n"); + } + sb +} + +/// Processes C-style escapes: `\r \n \t \\ \0 \xNN` (and `\XNN`); anything else is an error. +pub fn parse_escaped(s: &str) -> crate::Result> { + let b = s.as_bytes(); + let mut out = Vec::with_capacity(b.len()); + let mut i = 0; + while i < b.len() { + if b[i] != b'\\' { + out.push(b[i]); + i += 1; + continue; + } + let Some(&next) = b.get(i + 1) else { + return Err(format!("trailing backslash at position {i}").into()); + }; + let simple = match next { + b'r' => Some(b'\r'), + b'n' => Some(b'\n'), + b't' => Some(b'\t'), + b'\\' => Some(b'\\'), + b'0' => Some(0), + _ => None, + }; + if let Some(v) = simple { + out.push(v); + i += 2; + } else if next == b'x' || next == b'X' { + if i + 3 >= b.len() { + return Err(format!("incomplete \\x escape at position {i}").into()); + } + let hex = &b[i + 2..i + 4]; + let v = std::str::from_utf8(hex) + .ok() + .filter(|h| h.bytes().all(|c| c.is_ascii_hexdigit())) + .and_then(|h| u8::from_str_radix(h, 16).ok()) + .ok_or_else(|| format!("invalid \\x escape at position {i}: {}", String::from_utf8_lossy(hex)))?; + out.push(v); + i += 4; + } else { + let c = s[i + 1..].chars().next().unwrap_or('?'); + return Err(format!("unknown escape sequence \\{c} at position {i}").into()); + } + } + Ok(out) +} + +/// Bytes → display text: "ascii" (non-printables except \n \r \t become '.'), "iso-8859-1" / +/// "latin1" / "latin-1", otherwise UTF-8 with invalid bytes replaced by U+FFFD. +pub fn to_display_encoding(data: &[u8], encoding: &str) -> String { + match encoding.trim().to_ascii_lowercase().as_str() { + "ascii" => data + .iter() + .map(|&b| if (0x20..=0x7e).contains(&b) || matches!(b, b'\n' | b'\r' | b'\t') { b as char } else { '.' }) + .collect(), + e if super::is_latin1(e) => super::latin1_to_utf8(data), + _ => String::from_utf8_lossy(data).into_owned(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::mpsc; + use std::time::Duration; + + fn client() -> RawClient { + RawClient::new(&Profile::new(crate::models::Protocol::Raw), Arc::new(|_, _, _| {}), Arc::new(|_| {}), Arc::new(|| {})) + } + + #[test] + fn parse_hex_string_cases() { + let ok: [(&str, &[u8]); 5] = [ + ("48656c6c6f", b"Hello"), + ("48 65 6c 6c 6f", b"Hello"), + ("0x48 0x65 0x6c 0x6c 0x6f", b"Hello"), + ("0000", &[0, 0]), + ("ff ff", &[0xff, 0xff]), + ]; + for (input, want) in ok { + assert_eq!(parse_hex_string(input).unwrap(), want, "{input:?}"); + } + for input in ["", "ZZZZ", "abc"] { + assert!(parse_hex_string(input).is_err(), "{input:?}"); + } + } + + #[test] + fn parse_escaped_cases() { + let ok: [(&str, &[u8]); 9] = [ + ("hello", b"hello"), + (r"\n", b"\n"), + (r"\r\n", b"\r\n"), + (r"\t", b"\t"), + (r"\0", &[0]), + (r"\xff", &[0xff]), + (r"\\", b"\\"), + (r"AT\r\n", b"AT\r\n"), + (r"\x00", &[0]), + ]; + for (input, want) in ok { + assert_eq!(parse_escaped(input).unwrap(), want, "{input:?}"); + } + for input in [r"hello\", r"\q", r"\x0", r"\x+f"] { + assert!(parse_escaped(input).is_err(), "{input:?}"); + } + } + + #[test] + fn format_hex_dump_cases() { + assert_eq!(format_hex_dump(&[]), ""); + let got = format_hex_dump(b"Hi"); + assert!(got.contains("48 69") && got.contains("|Hi"), "{got}"); + let got = format_hex_dump(b"ABCDEFGHIJKLMNOPQRST"); + let lines: Vec<_> = got.trim_end_matches('\n').split('\n').collect(); + assert_eq!(lines.len(), 2); + assert!(lines[0].starts_with("00000000") && lines[1].starts_with("00000010")); + assert_eq!(lines[0], "00000000 41 42 43 44 45 46 47 48 49 4a 4b 4c 4d 4e 4f 50 |ABCDEFGHIJKLMNOP|"); + assert_eq!(lines[1].len(), lines[0].len()); + } + + #[test] + fn format_hex_cases() { + assert_eq!(format_hex(&[]), ""); + assert_eq!(format_hex(&[0x42]), "42"); + assert_eq!(format_hex(b"Hello"), "48 65 6c 6c 6f"); + } + + #[test] + fn to_display_encoding_cases() { + assert_eq!(to_display_encoding(b"Hello, World!", "utf-8"), "Hello, World!"); + assert_eq!(to_display_encoding(&[0x80, 0x41], "ascii"), ".A"); + assert_eq!(to_display_encoding(&[0xe9], "latin-1"), "é"); + assert_eq!(to_display_encoding(&[0xe9], "iso-8859-1"), "é"); + } + + #[test] + fn history_is_capped_by_entry_count() { + let c = client(); + for i in 0..MAX_HISTORY_ENTRIES + 25 { + c.inner.record(RECV, &[i as u8]); + } + assert_eq!(c.history().len(), MAX_HISTORY_ENTRIES); + } + + #[test] + fn history_is_capped_by_bytes() { + let c = client(); + let chunk = vec![b'a'; 1024]; + for _ in 0..MAX_HISTORY_BYTES / chunk.len() + 10 { + c.inner.record(RECV, &chunk); + } + let total: usize = c.history().iter().map(|e| e.data.len()).sum(); + assert!(total <= MAX_HISTORY_BYTES); + c.clear_history(); + assert!(c.history().is_empty()); + } + + #[test] + fn line_ending_is_appended_once() { + assert_eq!(apply_line_ending("AT", LineEnding::Crlf), b"AT\r\n"); + assert_eq!(apply_line_ending("AT\r\n", LineEnding::Crlf), b"AT\r\n"); + assert_eq!(apply_line_ending("AT", LineEnding::Cr), b"AT\r"); + assert_eq!(apply_line_ending("AT", LineEnding::None), b"AT"); + } + + #[test] + fn send_receive_history_and_capture() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let profile = Profile { host: "127.0.0.1".into(), port: listener.local_addr().unwrap().port() as i64, ..Profile::new(crate::models::Protocol::Raw) }; + let (tx, rx) = mpsc::channel::<(Vec, String, i64)>(); + let (ctx, crx) = mpsc::channel::<()>(); + let c = RawClient::new( + &profile, + Arc::new(move |d, dir, ts| tx.send((d.to_vec(), dir.to_string(), ts)).unwrap()), + Arc::new(|_| {}), + Arc::new(move || ctx.send(()).unwrap()), + ); + assert!(c.write(b"x").is_err()); + c.connect().unwrap(); + let (mut srv, _) = listener.accept().unwrap(); + srv.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + + c.write_text("AT").unwrap(); + c.write_hex("01 ff").unwrap(); + c.write_escaped(r"\x00z").unwrap(); + assert!(c.write_hex("zz").unwrap_err().to_string().starts_with("invalid hex input")); + let mut buf = [0u8; 8]; + srv.read_exact(&mut buf).unwrap(); + assert_eq!(&buf, b"AT\r\n\x01\xff\x00z"); + + srv.write_all(b"OK").unwrap(); + let mut events = Vec::new(); + while events.iter().map(|(d, dir, _): &(Vec, String, i64)| if dir == RECV { d.len() } else { 0 }).sum::() < 2 { + events.push(rx.recv_timeout(Duration::from_secs(5)).unwrap()); + } + assert_eq!(events[0].1, SEND); + assert!(events[0].2 > 1_600_000_000_000); + let hist = c.history(); + assert_eq!(hist.iter().filter(|e| e.direction == SEND).count(), 3); + assert_eq!(hist.last().unwrap().direction, RECV); + + let path = std::env::temp_dir().join(format!("nexus-raw-capture-{}.json", std::process::id())); + c.save_capture(&path).unwrap(); + let v: serde_json::Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + std::fs::remove_file(&path).unwrap(); + assert_eq!(v[0]["direction"], "send"); + assert_eq!(v[0]["data"], "QVQNCg=="); // "AT\r\n" + assert!(v[0]["timestamp"].as_str().unwrap().contains('T')); + + c.close(); + c.close(); + crx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(crx.recv_timeout(Duration::from_millis(100)).is_err()); + } +} diff --git a/rust/core/src/clients/serial.rs b/rust/core/src/clients/serial.rs new file mode 100644 index 0000000..977f925 --- /dev/null +++ b/rust/core/src/clients/serial.rs @@ -0,0 +1,368 @@ +//! Serial client (new protocol, no Go original) on the `serialport` crate. The device path is +//! `Profile::host`, line settings come from `Profile::serial`. Same conventions as the other +//! clients: blocking `connect`, an own reader thread, `on_close` exactly once (unplug included). + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serialport::{DataBits, FlowControl, Parity, SerialPort, StopBits}; + +use super::Callbacks; +use crate::models::{self, LineEnding, Profile, SerialConfig}; + +/// The reader polls this often, so `close` stops it within this long. +const READ_POLL: Duration = Duration::from_millis(100); +const WRITE_TIMEOUT: Duration = Duration::from_secs(5); +const BREAK_DURATION: Duration = Duration::from_millis(250); + +type OpenPort = (Box, Arc); + +pub struct SerialClient { + inner: Arc, +} + +struct Inner { + path: String, + cfg: SerialConfig, + cb: Callbacks, + /// The writer handle plus this connection's stop flag (the reader owns a clone of the port). + port: Mutex>, +} + +/// Device paths of the serial ports on this machine (empty if enumeration fails). +pub fn list_ports() -> Vec { + serialport::available_ports().map(|v| v.into_iter().map(|p| p.port_name).collect()).unwrap_or_default() +} + +impl SerialClient { + pub fn new(profile: &Profile, cb: Callbacks) -> SerialClient { + SerialClient { + inner: Arc::new(Inner { + path: profile.host.trim().to_string(), + cfg: profile.serial.clone().unwrap_or_default(), + cb, + port: Mutex::new(None), + }), + } + } + + /// Opens and configures the port, applies DTR/RTS, and starts the reader thread. + pub fn connect(&self) -> crate::Result<()> { + let inner = &self.inner; + if inner.port.lock().unwrap().is_some() { + return Err("already connected".into()); + } + if inner.path.is_empty() { + return Err("no serial device specified".into()); + } + let c = &inner.cfg; + let mut port = serialport::new(&inner.path, c.baud_rate) + .data_bits(match c.data_bits { + 5 => DataBits::Five, + 6 => DataBits::Six, + 7 => DataBits::Seven, + _ => DataBits::Eight, + }) + .parity(match c.parity { + models::Parity::None => Parity::None, + models::Parity::Odd => Parity::Odd, + models::Parity::Even => Parity::Even, + }) + .stop_bits(if c.stop_bits == 2 { StopBits::Two } else { StopBits::One }) + .flow_control(match c.flow_control { + models::FlowControl::None => FlowControl::None, + models::FlowControl::Xonxoff => FlowControl::Software, + models::FlowControl::Rtscts => FlowControl::Hardware, + }) + .timeout(WRITE_TIMEOUT) + .open() + .map_err(|e| open_error(&inner.path, &e))?; + // Opening asserts both lines on most OSes, so set them either way. Ptys and some adapters + // have no modem lines; that's not fatal. + let _ = port.write_data_terminal_ready(c.dtr_on_connect); + let _ = port.write_request_to_send(c.rts_on_connect); + + let mut reader = port.try_clone()?; + reader.set_timeout(READ_POLL)?; + let stop = Arc::new(AtomicBool::new(false)); + { + let mut guard = inner.port.lock().unwrap(); + if guard.is_some() { + return Err("already connected".into()); + } + *guard = Some((port, stop.clone())); + } + let me = inner.clone(); + std::thread::spawn(move || me.read_loop(reader, stop)); + Ok(()) + } + + /// Sends user input: the Enter key's `\r` becomes the configured line ending, text is encoded, + /// and with local echo on the input is fed back through `on_data` (Enter shown as CRLF). + pub fn write(&self, data: &[u8]) -> crate::Result<()> { + let out = translate_enter(data, self.inner.cfg.line_ending); + let out = if super::is_latin1(&self.inner.cfg.encoding) { super::utf8_to_latin1(&out) } else { out }; + self.with_port(|p| Ok(p.write_all(&out)?))?; + if self.inner.cfg.local_echo { + (self.inner.cb.on_data)(&translate_enter(data, LineEnding::Crlf)); + } + Ok(()) + } + + pub fn set_lines(&self, dtr: bool, rts: bool) -> crate::Result<()> { + self.with_port(|p| { + p.write_data_terminal_ready(dtr)?; + p.write_request_to_send(rts)?; + Ok(()) + }) + } + + /// Holds the line in break for ~250 ms (blocks the caller, and other writes, meanwhile). + pub fn send_break(&self) -> crate::Result<()> { + self.with_port(|p| { + p.set_break()?; + std::thread::sleep(BREAK_DURATION); + p.clear_break()?; + Ok(()) + }) + } + + pub fn close(&self) { + self.inner.close(); + } + + pub fn is_connected(&self) -> bool { + self.inner.port.lock().unwrap().is_some() + } + + fn with_port(&self, f: impl FnOnce(&mut Box) -> crate::Result<()>) -> crate::Result<()> { + let mut guard = self.inner.port.lock().unwrap(); + let (port, _) = guard.as_mut().ok_or("not connected")?; + f(port) + } +} + +impl Inner { + fn read_loop(&self, mut reader: Box, stop: Arc) { + let mut buf = [0u8; 4096]; + let latin1 = super::is_latin1(&self.cfg.encoding); + while !stop.load(Ordering::SeqCst) { + let err = match reader.read(&mut buf) { + Ok(0) => "device disconnected".to_string(), + Ok(n) if latin1 => { + (self.cb.on_data)(super::latin1_to_utf8(&buf[..n]).as_bytes()); + continue; + } + Ok(n) => { + (self.cb.on_data)(&buf[..n]); + continue; + } + Err(e) if matches!(e.kind(), std::io::ErrorKind::TimedOut | std::io::ErrorKind::Interrupted) => continue, + Err(e) => format!("device disconnected: {e}"), + }; + if !stop.load(Ordering::SeqCst) { + (self.cb.on_error)(format!("read error: {err}")); + self.close(); + } + return; + } + } + + fn close(&self) { + let taken = self.port.lock().unwrap().take(); + if let Some((port, stop)) = taken { + stop.store(true, Ordering::SeqCst); + drop(port); + (self.cb.on_close)(); + } + } +} + +/// Replaces each `\r` (the Enter key) with `le`; `None` leaves it alone. +fn translate_enter(data: &[u8], le: LineEnding) -> Vec { + if le == LineEnding::None { + return data.to_vec(); + } + let mut out = Vec::with_capacity(data.len() + 4); + for &b in data { + if b == b'\r' { out.extend_from_slice(le.bytes()) } else { out.push(b) } + } + out +} + +fn open_error(path: &str, e: &serialport::Error) -> crate::Error { + let desc = e.description.to_lowercase(); + let busy = ["busy", "lock", "access is denied", "in use"].iter().any(|s| desc.contains(s)); + let hint = match e.kind { + serialport::ErrorKind::NoDevice if busy => " (the port is in use by another program)", + serialport::ErrorKind::Io(std::io::ErrorKind::PermissionDenied) => { + " (permission denied: add your user to the dialout (Linux) / uucp group, then log in again)" + } + _ => "", + }; + format!("cannot open serial port {path}: {}{hint}", e.description).into() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + #[test] + fn enter_translation() { + assert_eq!(translate_enter(b"AT\r", LineEnding::Crlf), b"AT\r\n"); + assert_eq!(translate_enter(b"AT\r", LineEnding::Lf), b"AT\n"); + assert_eq!(translate_enter(b"AT\r", LineEnding::Cr), b"AT\r"); + assert_eq!(translate_enter(b"AT\r", LineEnding::None), b"AT\r"); + } + + #[test] + fn error_hints() { + let perm = serialport::Error::new(serialport::ErrorKind::Io(std::io::ErrorKind::PermissionDenied), "Permission denied"); + assert!(open_error("/dev/ttyUSB0", &perm).to_string().contains("dialout")); + let busy = serialport::Error::new(serialport::ErrorKind::NoDevice, "Device or resource busy"); + assert!(open_error("/dev/ttyUSB0", &busy).to_string().contains("in use by another program")); + let missing = serialport::Error::new(serialport::ErrorKind::Io(std::io::ErrorKind::NotFound), "No such file or directory"); + assert_eq!(open_error("/dev/x", &missing).to_string(), "cannot open serial port /dev/x: No such file or directory"); + } + + #[test] + fn not_connected() { + let c = SerialClient::new(&Profile::new(models::Protocol::Serial), Callbacks::noop()); + assert!(c.write(b"x").is_err()); + assert_eq!(c.connect().unwrap_err().to_string(), "no serial device specified"); + let _ = list_ports(); + } + + /// A pty pair: the client opens the slave as its "device", the test plays the far end on the master. + #[cfg(target_os = "linux")] + mod pty { + use super::*; + use std::fs::File; + use std::io::Write; + use std::os::fd::{AsRawFd, FromRawFd}; + + struct Pty { + master: File, + slave: File, + path: String, + } + + fn open_pty() -> Pty { + let (mut m, mut s) = (0, 0); + let mut name = [0 as libc::c_char; 128]; + let rc = unsafe { libc::openpty(&mut m, &mut s, name.as_mut_ptr(), std::ptr::null(), std::ptr::null()) }; + assert_eq!(rc, 0); + let path = unsafe { std::ffi::CStr::from_ptr(name.as_ptr()) }.to_string_lossy().into_owned(); + Pty { master: unsafe { File::from_raw_fd(m) }, slave: unsafe { File::from_raw_fd(s) }, path } + } + + /// Reads from the master until something arrives or `timeout` passes. + fn read_master(master: &File, timeout: Duration) -> Vec { + let mut pfd = libc::pollfd { fd: master.as_raw_fd(), events: libc::POLLIN, revents: 0 }; + if unsafe { libc::poll(&mut pfd, 1, timeout.as_millis() as i32) } != 1 { + return Vec::new(); + } + let mut buf = [0u8; 256]; + let n = std::io::Read::read(&mut &*master, &mut buf).unwrap(); + buf[..n].to_vec() + } + + fn master_n(master: &File, n: usize) -> Vec { + let mut got = Vec::new(); + while got.len() < n { + let chunk = read_master(master, Duration::from_secs(5)); + assert!(!chunk.is_empty(), "timed out"); + got.extend(chunk); + } + got + } + + fn recv_n(rx: &mpsc::Receiver>, n: usize) -> Vec { + let mut got = Vec::new(); + while got.len() < n { + got.extend(rx.recv_timeout(Duration::from_secs(5)).expect("timed out")); + } + got + } + + fn client(pty: &Pty, cfg: SerialConfig) -> (SerialClient, mpsc::Receiver>, mpsc::Receiver, mpsc::Receiver<()>) { + let (dtx, drx) = mpsc::channel(); + let (etx, erx) = mpsc::channel(); + let (ctx, crx) = mpsc::channel(); + let cb = Callbacks { + on_data: Arc::new(move |d| { let _ = dtx.send(d.to_vec()); }), + on_error: Arc::new(move |e| { let _ = etx.send(e); }), + on_close: Arc::new(move || { let _ = ctx.send(()); }), + }; + let profile = Profile { host: pty.path.clone(), serial: Some(cfg), ..Profile::new(models::Protocol::Serial) }; + (SerialClient::new(&profile, cb), drx, erx, crx) + } + + #[test] + fn data_echo_line_endings_and_close() { + let mut pty = open_pty(); + let cfg = SerialConfig { local_echo: true, line_ending: LineEnding::Crlf, ..Default::default() }; + let (c, data, errors, closed) = client(&pty, cfg); + c.connect().unwrap(); + assert!(c.connect().is_err()); + + c.write(b"AT\r").unwrap(); + assert_eq!(master_n(&pty.master, 4), b"AT\r\n"); + assert_eq!(recv_n(&data, 4), b"AT\r\n"); // local echo + + pty.master.write_all(b"OK\r\n").unwrap(); + assert_eq!(recv_n(&data, 4), b"OK\r\n"); + + c.close(); + c.close(); + closed.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(closed.recv_timeout(Duration::from_millis(300)).is_err()); + assert!(errors.try_recv().is_err()); + assert!(c.write(b"x").is_err()); + } + + #[test] + fn latin1_encoding() { + let mut pty = open_pty(); + let cfg = SerialConfig { encoding: "latin1".into(), line_ending: LineEnding::Lf, ..Default::default() }; + let (c, data, _e, _c) = client(&pty, cfg); + c.connect().unwrap(); + c.write("é\r".as_bytes()).unwrap(); + assert_eq!(master_n(&pty.master, 2), [0xe9, b'\n']); + pty.master.write_all(&[0xe9]).unwrap(); + assert_eq!(recv_n(&data, 2), "é".as_bytes()); + } + + #[test] + fn xon_xoff_flow_control() { + let mut pty = open_pty(); + let cfg = SerialConfig { flow_control: models::FlowControl::Xonxoff, line_ending: LineEnding::None, ..Default::default() }; + let (c, _d, _e, _c) = client(&pty, cfg); + c.connect().unwrap(); + pty.master.write_all(&[0x13]).unwrap(); // XOFF + std::thread::sleep(Duration::from_millis(100)); + let c = Arc::new(c); + let w = c.clone(); + let writer = std::thread::spawn(move || w.write(b"x")); + assert!(read_master(&pty.master, Duration::from_millis(300)).is_empty(), "output not paused by XOFF"); + pty.master.write_all(&[0x11]).unwrap(); // XON + assert_eq!(master_n(&pty.master, 1), b"x"); + writer.join().unwrap().unwrap(); + } + + #[test] + fn unplug_reports_error_and_closes() { + let pty = open_pty(); + let (c, _d, errors, closed) = client(&pty, SerialConfig::default()); + c.connect().unwrap(); + drop(pty.master); + drop(pty.slave); + let err = errors.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(err.starts_with("read error: device disconnected"), "{err}"); + closed.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(!c.is_connected()); + } + } +} diff --git a/rust/core/src/clients/telnet.rs b/rust/core/src/clients/telnet.rs new file mode 100644 index 0000000..5ef5cac --- /dev/null +++ b/rust/core/src/clients/telnet.rs @@ -0,0 +1,434 @@ +//! Telnet client (port of Go `internal/telnetclient`): RFC 854/855 option negotiation (ECHO, SGA, +//! TTYPE, NAWS; everything else refused), IAC escaping, line-ending conversion, ISO 8859-1 output +//! encoding and BRK. Go read but never acted on `lineMode`, and neither do we. + +use std::sync::{Arc, Mutex}; + +use super::{Callbacks, DataFn, DebugFn, TcpConn}; +use crate::models::{LineEnding, Profile, TelnetConfig}; + +pub const IAC: u8 = 255; +pub const DONT: u8 = 254; +pub const DO: u8 = 253; +pub const WONT: u8 = 252; +pub const WILL: u8 = 251; +pub const SB: u8 = 250; +pub const SE: u8 = 240; +pub const BRK: u8 = 243; + +pub const OPT_ECHO: u8 = 1; +pub const OPT_SGA: u8 = 3; +pub const OPT_TERM_TYPE: u8 = 24; +pub const OPT_NAWS: u8 = 31; + +const SUB_IS: u8 = 0; +const SUB_SEND: u8 = 1; + +/// An unterminated IAC sequence longer than this is dropped rather than buffered forever. +const MAX_PENDING: usize = 8192; + +pub struct TelnetClient { + inner: Arc, +} + +struct Inner { + conn: Arc, + host: String, + port: i64, + connect_timeout: i64, + cfg: TelnetConfig, + encoding: String, + on_data: DataFn, + on_debug: Option, + state: Mutex, +} + +struct State { + local_echo: bool, + cols: u16, + rows: u16, +} + +impl TelnetClient { + /// `on_debug` receives negotiation messages ("received DO 31", "sending WILL 31", …), always, + /// as in Go; the profile's `negotiationDebug` is for the caller to honour. + pub fn new(profile: &Profile, cb: Callbacks, on_debug: Option) -> TelnetClient { + let cfg = profile.telnet.clone().unwrap_or_default(); + let encoding = if cfg.encoding.is_empty() { "utf-8".to_string() } else { cfg.encoding.clone() }; + let state = State { local_echo: cfg.local_echo, cols: 80, rows: 24 }; + TelnetClient { + inner: Arc::new(Inner { + conn: Arc::new(TcpConn::new(cb.on_error, cb.on_close)), + host: profile.host.clone(), + port: if profile.port == 0 { 23 } else { profile.port }, + connect_timeout: profile.connect_timeout, + cfg, + encoding, + on_data: cb.on_data, + on_debug, + state: Mutex::new(state), + }), + } + } + + pub fn connect(&self) -> crate::Result<()> { + let inner = &self.inner; + if inner.conn.is_connected() { + return Err("already connected".into()); + } + let (addr, res) = super::dial(&inner.host, inner.port, inner.connect_timeout); + let stream = res.map_err(|e| format!("Telnet connection to {addr} failed: {e}"))?; + let me = inner.clone(); + let mut pending = Vec::new(); + inner.conn.connect(stream, move |chunk| me.process(&mut pending, chunk)) + } + + /// Sends user input: line endings converted, encoded, literal 0xFF doubled. + pub fn write(&self, data: &[u8]) -> crate::Result<()> { + let converted = convert_line_ending(data, self.inner.cfg.line_ending); + let encoded = encode_outgoing(&converted, &self.inner.encoding); + self.inner.conn.write(&escape_iac(&encoded)) + } + + /// Records the size and sends a NAWS update (sent even if the server never asked for NAWS, as in Go). + pub fn resize(&self, cols: u16, rows: u16) -> crate::Result<()> { + if !self.inner.conn.is_connected() { + return Err("not connected".into()); + } + { + let mut st = self.inner.state.lock().unwrap(); + st.cols = cols; + st.rows = rows; + } + self.inner.send_naws(cols, rows) + } + + /// Sends IAC BRK. + pub fn send_break(&self) -> crate::Result<()> { + self.inner.conn.write(&[IAC, BRK]) + } + + pub fn close(&self) { + self.inner.conn.close(); + } + + pub fn is_connected(&self) -> bool { + self.inner.conn.is_connected() + } + + /// The negotiated local-echo state: starts at the profile's `localEcho`, cleared by the + /// server's WILL ECHO and set by WONT ECHO. Like Go, the client itself never echoes. + pub fn local_echo(&self) -> bool { + self.inner.state.lock().unwrap().local_echo + } +} + +impl Inner { + /// Splits a received chunk into data and IAC sequences. Unlike Go, a sequence cut off at the + /// end of a read is kept for the next read instead of dropped. + fn process(&self, pending: &mut Vec, chunk: &[u8]) { + pending.extend_from_slice(chunk); + let mut data = Vec::new(); + let mut i = 0; + while i < pending.len() { + if pending[i] != IAC { + data.push(pending[i]); + i += 1; + continue; + } + if !data.is_empty() { + (self.on_data)(&data); + data.clear(); + } + let consumed = self.handle_iac(&pending[i..]); + if consumed == 0 { + break; + } + i += consumed; + } + if !data.is_empty() { + (self.on_data)(&data); + } + pending.drain(..i); + if pending.len() > MAX_PENDING { + pending.clear(); + } + } + + /// Handles the IAC sequence at `data[0]`; returns bytes consumed, 0 if incomplete. + fn handle_iac(&self, data: &[u8]) -> usize { + if data.len() < 2 { + return 0; + } + match data[1] { + IAC => { + (self.on_data)(&[IAC]); + 2 + } + DO | DONT | WILL | WONT => { + let Some(&opt) = data.get(2) else { return 0 }; + match data[1] { + DO => self.handle_do(opt), + DONT => self.handle_dont(opt), + WILL => self.handle_will(opt), + _ => self.handle_wont(opt), + } + 3 + } + SB => match (2..data.len().saturating_sub(1)).find(|&i| data[i] == IAC && data[i + 1] == SE) { + Some(end) => { + self.process_subnegotiation(&data[2..end]); + end + 2 + } + None => 0, + }, + cmd => { + self.debug(format!("received IAC command: {cmd}")); + 2 + } + } + } + + fn handle_do(&self, opt: u8) { + self.debug(format!("received DO {opt}")); + match opt { + OPT_TERM_TYPE | OPT_SGA => self.negotiate(WILL, opt), + OPT_NAWS => { + self.negotiate(WILL, opt); + let (cols, rows) = { + let st = self.state.lock().unwrap(); + (st.cols, st.rows) + }; + let _ = self.send_naws(cols, rows); + } + _ => self.negotiate(WONT, opt), + } + } + + fn handle_dont(&self, opt: u8) { + self.debug(format!("received DONT {opt}")); + self.negotiate(WONT, opt); + } + + fn handle_will(&self, opt: u8) { + self.debug(format!("received WILL {opt}")); + match opt { + OPT_ECHO => { + self.negotiate(DO, opt); + self.state.lock().unwrap().local_echo = false; + } + OPT_SGA => self.negotiate(DO, opt), + _ => self.negotiate(DONT, opt), + } + } + + fn handle_wont(&self, opt: u8) { + self.debug(format!("received WONT {opt}")); + if opt == OPT_ECHO { + self.state.lock().unwrap().local_echo = true; + } + self.negotiate(DONT, opt); + } + + fn process_subnegotiation(&self, payload: &[u8]) { + let Some(&opt) = payload.first() else { return }; + self.debug(format!("sub-negotiation for option {opt}")); + if opt == OPT_TERM_TYPE && payload.get(1) == Some(&SUB_SEND) { + let tt = if self.cfg.terminal_type.is_empty() { "xterm-256color" } else { &self.cfg.terminal_type }; + self.debug(format!("sending terminal type: {tt}")); + let mut resp = vec![IAC, SB, OPT_TERM_TYPE, SUB_IS]; + resp.extend_from_slice(tt.to_uppercase().as_bytes()); + resp.extend_from_slice(&[IAC, SE]); + let _ = self.conn.write(&resp); + } + } + + fn negotiate(&self, cmd: u8, opt: u8) { + self.debug(format!("sending {} {opt}", command_name(cmd))); + let _ = self.conn.write(&[IAC, cmd, opt]); + } + + fn send_naws(&self, cols: u16, rows: u16) -> crate::Result<()> { + self.debug(format!("sending NAWS: {cols}x{rows}")); + let [c0, c1] = cols.to_be_bytes(); + let [r0, r1] = rows.to_be_bytes(); + self.conn.write(&[IAC, SB, OPT_NAWS, c0, c1, r0, r1, IAC, SE]) + } + + fn debug(&self, msg: String) { + if let Some(f) = &self.on_debug { + f(msg); + } + } +} + +fn command_name(cmd: u8) -> String { + match cmd { + WILL => "WILL".into(), + WONT => "WONT".into(), + DO => "DO".into(), + DONT => "DONT".into(), + _ => format!("CMD({cmd})"), + } +} + +/// Normalises CRLF/CR/LF to the configured ending (`None` leaves data untouched). +fn convert_line_ending(data: &[u8], le: LineEnding) -> Vec { + if le == LineEnding::None { + return data.to_vec(); + } + let mut out = Vec::with_capacity(data.len() + 8); + let mut i = 0; + while i < data.len() { + match data[i] { + b'\r' => { + if data.get(i + 1) == Some(&b'\n') { + i += 1; + } + out.extend_from_slice(le.bytes()); + } + b'\n' => out.extend_from_slice(le.bytes()), + b => out.push(b), + } + i += 1; + } + out +} + +/// Doubles literal 0xFF bytes. +fn escape_iac(data: &[u8]) -> Vec { + let mut out = Vec::with_capacity(data.len()); + for &b in data { + out.push(b); + if b == IAC { + out.push(IAC); + } + } + out +} + +/// UTF-8 → ISO 8859-1 for the Latin-1 encodings; everything else is sent as-is. Unlike Go this +/// runs before IAC escaping, so a Latin-1 'ÿ' (0xFF) is escaped correctly. +fn encode_outgoing(data: &[u8], encoding: &str) -> Vec { + if super::is_latin1(encoding) { super::utf8_to_latin1(data) } else { data.to_vec() } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::mpsc; + use std::time::Duration; + + #[test] + fn line_endings() { + let cases = [ + (LineEnding::Crlf, "a\nb\r\nc\rd", "a\r\nb\r\nc\r\nd"), + (LineEnding::Lf, "a\r\nb\rc\n", "a\nb\nc\n"), + (LineEnding::Cr, "a\r\nb\n", "a\rb\r"), + (LineEnding::None, "a\r\nb\n", "a\r\nb\n"), + ]; + for (le, input, want) in cases { + assert_eq!(convert_line_ending(input.as_bytes(), le), want.as_bytes(), "{le:?}"); + } + } + + #[test] + fn iac_escaping_and_latin1() { + assert_eq!(escape_iac(&[1, 0xff, 2]), vec![1, 0xff, 0xff, 2]); + assert_eq!(encode_outgoing("é".as_bytes(), "Latin-1"), vec![0xe9]); + assert_eq!(encode_outgoing("€".as_bytes(), "latin1"), "€".as_bytes()); // unencodable: unchanged + assert_eq!(encode_outgoing("é".as_bytes(), "utf-8"), "é".as_bytes()); + } + + fn read_n(s: &mut std::net::TcpStream, n: usize) -> Vec { + let mut buf = vec![0; n]; + s.read_exact(&mut buf).unwrap(); + buf + } + + #[test] + fn negotiates_and_frames_data() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let profile = Profile { host: "127.0.0.1".into(), port: listener.local_addr().unwrap().port() as i64, ..Profile::new(crate::models::Protocol::Telnet) }; + let (dtx, drx) = mpsc::channel::>(); + let (etx, erx) = mpsc::channel::(); + let (ctx, crx) = mpsc::channel::<()>(); + let debug = Arc::new(Mutex::new(Vec::::new())); + let dbg = debug.clone(); + let cb = Callbacks { + on_data: Arc::new(move |d| dtx.send(d.to_vec()).unwrap()), + on_error: Arc::new(move |e| etx.send(e).unwrap()), + on_close: Arc::new(move || ctx.send(()).unwrap()), + }; + let client = TelnetClient::new(&profile, cb, Some(Arc::new(move |m| dbg.lock().unwrap().push(m)))); + client.connect().unwrap(); + let (mut srv, _) = listener.accept().unwrap(); + srv.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + + srv.write_all(&[IAC, DO, OPT_TERM_TYPE, IAC, DO, OPT_NAWS, IAC, WILL, OPT_ECHO, IAC, WILL, OPT_SGA, IAC, DO, 99]) + .unwrap(); + let mut want = vec![IAC, WILL, OPT_TERM_TYPE, IAC, WILL, OPT_NAWS, IAC, SB, OPT_NAWS, 0, 80, 0, 24, IAC, SE]; + want.extend_from_slice(&[IAC, DO, OPT_ECHO, IAC, DO, OPT_SGA, IAC, WONT, 99]); + assert_eq!(read_n(&mut srv, want.len()), want); + assert!(!client.local_echo()); + + // Data with an escaped 0xFF, then a TTYPE SEND split across two writes. + srv.write_all(b"hi").unwrap(); + srv.write_all(&[IAC, IAC, IAC, SB, OPT_TERM_TYPE]).unwrap(); + std::thread::sleep(Duration::from_millis(50)); + srv.write_all(&[SUB_SEND, IAC, SE]).unwrap(); + let mut want = vec![IAC, SB, OPT_TERM_TYPE, SUB_IS]; + want.extend_from_slice(b"XTERM-256COLOR"); + want.extend_from_slice(&[IAC, SE]); + assert_eq!(read_n(&mut srv, want.len()), want); + let mut got = Vec::new(); + while got.len() < 3 { + got.extend(drx.recv_timeout(Duration::from_secs(5)).unwrap()); + } + assert_eq!(got, b"hi\xff"); + + srv.write_all(&[IAC, WONT, OPT_ECHO]).unwrap(); + assert_eq!(read_n(&mut srv, 3), [IAC, DONT, OPT_ECHO]); + assert!(client.local_echo()); + + client.resize(132, 50).unwrap(); + assert_eq!(read_n(&mut srv, 9), [IAC, SB, OPT_NAWS, 0, 132, 0, 50, IAC, SE]); + client.send_break().unwrap(); + assert_eq!(read_n(&mut srv, 2), [IAC, BRK]); + client.write(b"ls\n\xff").unwrap(); + assert_eq!(read_n(&mut srv, 6), b"ls\r\n\xff\xff"); + + let log = debug.lock().unwrap().clone(); + for m in ["received DO 24", "sending WILL 31", "sending NAWS: 80x24", "sending WONT 99", "sending terminal type: xterm-256color"] { + assert!(log.iter().any(|l| l == m), "missing debug {m:?} in {log:?}"); + } + + // Remote close: Go reports "read error: EOF" then closes, exactly once. + drop(srv); + assert_eq!(erx.recv_timeout(Duration::from_secs(5)).unwrap(), "read error: EOF"); + crx.recv_timeout(Duration::from_secs(5)).unwrap(); + client.close(); + assert!(crx.recv_timeout(Duration::from_millis(100)).is_err()); + assert!(client.write(b"x").is_err()); + } + + #[test] + fn local_close_fires_once_without_error() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let profile = Profile { host: "127.0.0.1".into(), port: listener.local_addr().unwrap().port() as i64, ..Profile::new(crate::models::Protocol::Telnet) }; + let (etx, erx) = mpsc::channel::(); + let (ctx, crx) = mpsc::channel::<()>(); + let cb = Callbacks { on_error: Arc::new(move |e| etx.send(e).unwrap()), on_close: Arc::new(move || ctx.send(()).unwrap()), ..Callbacks::noop() }; + let client = TelnetClient::new(&profile, cb, None); + client.connect().unwrap(); + assert!(client.connect().is_err()); + let _srv = listener.accept().unwrap(); + client.close(); + client.close(); + crx.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(crx.recv_timeout(Duration::from_millis(100)).is_err()); + assert!(erx.recv_timeout(Duration::from_millis(100)).is_err()); + } +}