//! 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()); } }