From f0420f6c8b127a235c4726ffcf0f1eb8c7d94871 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Tue, 11 Aug 2026 08:06:25 +0000 Subject: [PATCH] Implement native PacketDump capture (#87) --- README.md | 6 +- crates/libremetaverse/src/network_manager.rs | 63 +- .../libremetaverse/tests/network_manager.rs | 36 + programs/README.md | 50 +- programs/src/bin/packet_dump.rs | 2 +- programs/src/lib.rs | 1 + programs/src/packet_dump.rs | 938 ++++++++++++++++++ programs/tests/packet_dump_cli.rs | 215 ++++ 8 files changed, 1298 insertions(+), 13 deletions(-) create mode 100644 programs/src/packet_dump.rs create mode 100644 programs/tests/packet_dump_cli.rs diff --git a/README.md b/README.md index beb6942..a4070e7 100644 --- a/README.md +++ b/README.md @@ -485,10 +485,12 @@ documented in the ### Milestone 11 -The native `osd-inspector` and `simple-bot` programs are complete. +The native `osd-inspector`, `simple-bot`, and `packet-dump` programs are complete. `osd-inspector` validates and converts bounded LLSD and performs deterministic Primitive-to-OSD round trips. `simple-bot` provides native async login, IM and local-chat commands, movement and animation calls, bounded fake-grid -conversations, secret redaction, and cancellation-safe logout. Their command +conversations, secret redaction, and cancellation-safe logout. `packet-dump` +adds filtered, bounded incoming/outgoing capture with native wire validation, +sanitized optional raw bytes, and deterministic fake datagrams. Their command surfaces, limits, isolated CLI tests, and the status of every remaining program are documented in the [`native programs guide`](programs/README.md). diff --git a/crates/libremetaverse/src/network_manager.rs b/crates/libremetaverse/src/network_manager.rs index 5c11254..4c063d3 100644 --- a/crates/libremetaverse/src/network_manager.rs +++ b/crates/libremetaverse/src/network_manager.rs @@ -293,6 +293,7 @@ impl EventRegistry { pub struct PacketReceivedEventArgs { packet: Packet, simulator: Simulator, + raw_data: Option>, } #[derive(Clone)] @@ -304,7 +305,19 @@ pub(crate) struct RawPacketReceivedEventArgs { impl PacketReceivedEventArgs { pub fn new(packet: Packet, simulator: Simulator) -> Result { - Ok(Self { packet, simulator }) + Self::new_with_raw_data(packet, simulator, None) + } + + pub fn new_with_raw_data( + packet: Packet, + simulator: Simulator, + raw_data: Option>, + ) -> Result { + Ok(Self { + packet, + simulator, + raw_data, + }) } #[must_use] @@ -316,6 +329,13 @@ impl PacketReceivedEventArgs { pub fn simulator(&self) -> Simulator { self.simulator.clone() } + + /// Returns the original datagram when this event came from the native UDP + /// receive path. Synthetic decoded events intentionally return `None`. + #[must_use] + pub fn raw_data(&self) -> Option> { + self.raw_data.clone() + } } #[derive(Clone)] @@ -552,6 +572,7 @@ struct AsyncPacketBatch { specific_handlers: Vec>, packet: Packet, simulator: Simulator, + raw_data: Option>, } enum PacketWorkerCommand { @@ -674,7 +695,19 @@ impl PacketEventDictionary { packet: Packet, simulator: Simulator, ) -> Result<(), Error> { - self.raise_event(packet_type, packet, simulator) + self.raise_event(packet_type, packet, simulator, None) + } + + /// Dispatches a decoded packet together with its original wire datagram. + /// This is primarily useful to deterministic packet-capture fixtures. + pub fn invoke_raise_event_with_raw_data( + &self, + packet_type: PacketType, + packet: Packet, + simulator: Simulator, + raw_data: Vec, + ) -> Result<(), Error> { + self.raise_event(packet_type, packet, simulator, Some(raw_data)) } fn raise_event( @@ -682,6 +715,7 @@ impl PacketEventDictionary { packet_type: PacketType, packet: Packet, simulator: Simulator, + raw_data: Option>, ) -> Result<(), Error> { let (defaults, specific) = { let table = mutex(&self.state.callbacks); @@ -692,7 +726,11 @@ impl PacketEventDictionary { }; let default_async = defaults.is_async; let specific_async = specific.is_async; - let args = PacketReceivedEventArgs::new(packet.clone(), simulator.clone())?; + let args = PacketReceivedEventArgs::new_with_raw_data( + packet.clone(), + simulator.clone(), + raw_data.clone(), + )?; if !default_async { for handler in &defaults.handlers { @@ -731,6 +769,7 @@ impl PacketEventDictionary { specific_handlers, packet, simulator, + raw_data, })) .map_err(|_| Error::InvalidOperation) } @@ -740,7 +779,11 @@ fn packet_event_worker(receiver: Receiver, state: Weak { - let Ok(args) = PacketReceivedEventArgs::new(batch.packet, batch.simulator) else { + let Ok(args) = PacketReceivedEventArgs::new_with_raw_data( + batch.packet, + batch.simulator, + batch.raw_data, + ) else { continue; }; for handler in batch @@ -2090,7 +2133,7 @@ impl NetworkManagerInner { continue; }; inner.process_internal_packet(&packet, &simulator, raw_data.as_deref()); - if let Some(data) = raw_data { + if let Some(data) = raw_data.as_ref() { inner.events.raw_packet_received.emit_with(|| { RawPacketReceivedEventArgs { packet_type: packet.type_, @@ -2099,10 +2142,12 @@ impl NetworkManagerInner { } }); } - let _ = - inner - .packet_events - .raise_event(packet.type_, packet, simulator); + let _ = inner.packet_events.raise_event( + packet.type_, + packet, + simulator, + raw_data, + ); } Err(RecvTimeoutError::Timeout) => {} Err(RecvTimeoutError::Disconnected) => break, diff --git a/crates/libremetaverse/tests/network_manager.rs b/crates/libremetaverse/tests/network_manager.rs index 26b68ff..bee1b2b 100644 --- a/crates/libremetaverse/tests/network_manager.rs +++ b/crates/libremetaverse/tests/network_manager.rs @@ -183,6 +183,42 @@ fn packet_callbacks_preserve_filtering_order_async_policy_and_reentrancy() { ); } +#[test] +fn decoded_packet_callbacks_preserve_optional_raw_datagrams() { + let client = GridClient::new().expect("client"); + let sim = simulator(&client); + let events = PacketEventDictionary::new(client).expect("packet events"); + let observed = Arc::new(Mutex::new(None)); + let captured = Arc::clone(&observed); + events + .register_event( + PacketType::Default, + Arc::new(move |event| *captured.lock().unwrap() = event.raw_data()), + false, + ) + .unwrap(); + let raw = vec![0x40, 0, 0, 0, 0, 0, 1, 7, 0, 0, 0, 0]; + events + .invoke_raise_event_with_raw_data( + PacketType::StartPingCheck, + base_packet(PacketType::StartPingCheck), + sim.clone(), + raw.clone(), + ) + .unwrap(); + assert_eq!(*observed.lock().unwrap(), Some(raw)); + + *observed.lock().unwrap() = Some(vec![1]); + events + .invoke_raise_event( + PacketType::StartPingCheck, + base_packet(PacketType::StartPingCheck), + sim, + ) + .unwrap(); + assert_eq!(*observed.lock().unwrap(), None); +} + #[test] fn synchronous_specific_dispatch_suppresses_default_async_chain_like_reference() { let mixed_events = PacketEventDictionary::new(GridClient::new().unwrap()).unwrap(); diff --git a/programs/README.md b/programs/README.md index 94aed1c..0f6046f 100644 --- a/programs/README.md +++ b/programs/README.md @@ -9,7 +9,7 @@ here so a source entry is never mistaken for a completed port. | --- | --- | --- | | `osd-inspector` | OSDInspector | Implemented and tested offline | | `simple-bot` | SimpleBot | Implemented with live and deterministic fake-grid modes | -| `packet-dump` | PacketDump | Pending milestone 11 issue #87 | +| `packet-dump` | PacketDump | Implemented with live and deterministic fake-grid capture | | `prim-inspector` | PrimInspector | Pending milestone 11 issue #88 | | `inventory-explorer` | InventoryExplorer | Pending milestone 11 issue #89 | | `irc-gateway` | IRCGateway | Pending milestone 11 issue #90 | @@ -100,3 +100,51 @@ suite and the related runtime compatibility cases with: cargo test -p libremetaverse-programs --test simple_bot_cli --locked cargo test --manifest-path tests/compat/Cargo.toml --test core_runtime_shims --locked ``` + +## PacketDump + +`packet-dump` preserves the upstream live arguments and its 20-second login +timeout while adding safe, bounded capture controls: + +```text +packet-dump FIRSTNAME LASTNAME PASSWORD SECONDS + [--direction incoming|outgoing|both] + [--packet-type NAME]... [--raw] + [--output FILE] [--max-output-bytes BYTES] [--max-packets COUNT] +``` + +`SECONDS=0` captures until Ctrl-C. Credentials can instead come from +`GRID_FIRST_NAME`, `GRID_LAST_NAME`, and `GRID_PASSWORD`; `GRID_LOGIN_URL` or +`--login-uri` selects a login endpoint. The client disables multiple simulator +connections and sends zero land, wind, and cloud throttles like the source +program. Incoming decoded callbacks retain their original datagrams, while +outgoing callbacks are decoded through the public packet factory. Records show +direction, packet type, simulator, byte count, sequence, frequency, ID, and +header flags. Exact packet-type filters may be repeated. + +Raw hexadecimal output is opt-in. Live capture masks the password and native +session identifiers wherever they occur in a datagram, and suppresses raw +payloads containing URLs, authorization terms, capability terms, or token +assignments. Login and disconnect messages use the same text redaction. +Output defaults to stdout, may be redirected to a newly truncated file, and is +bounded to 16 MiB and 100,000 matching packets unless lower or higher explicit +limits are supplied. A complete line is either written or rejected at the byte +limit. + +Offline validation uses `--fake-script FILE` with records of this form: + +```text +incomingsimulatorhex-bytes +outgoingsimulatorhex-bytes +``` + +Scripts are capped at 8 MiB and individual datagrams at 64 KiB. Each datagram +is passed through the native wire decoder; malformed and unknown packets are +reported without aborting the capture. The issue-focused and related translated +wire tests are: + +```sh +cargo test -p libremetaverse-programs --test packet_dump_cli --locked +cargo test -p libremetaverse --test packet_wire --locked +cargo test --manifest-path tests/compat/Cargo.toml --test wire_semantics --locked +``` diff --git a/programs/src/bin/packet_dump.rs b/programs/src/bin/packet_dump.rs index 553a955..aa3e3e5 100644 --- a/programs/src/bin/packet_dump.rs +++ b/programs/src/bin/packet_dump.rs @@ -1,3 +1,3 @@ fn main() -> std::process::ExitCode { - libremetaverse_programs::pending_program("PacketDump") + libremetaverse_programs::packet_dump::main_entry() } diff --git a/programs/src/lib.rs b/programs/src/lib.rs index cb3711c..bac59c7 100644 --- a/programs/src/lib.rs +++ b/programs/src/lib.rs @@ -2,6 +2,7 @@ pub mod commands; pub mod osd_inspector; +pub mod packet_dump; pub mod simple_bot; pub use libremetaverse::shim::pending_program; diff --git a/programs/src/packet_dump.rs b/programs/src/packet_dump.rs new file mode 100644 index 0000000..fcab5b7 --- /dev/null +++ b/programs/src/packet_dump.rs @@ -0,0 +1,938 @@ +//! Safe, bounded native packet capture for `LibreMetaverse` sessions. + +use clap::{Parser, ValueEnum}; +use libremetaverse::packets::{Packet, PacketType}; +use libremetaverse::types::compat::{CancellationTokenSource, Subscription}; +use libremetaverse::{ + AgentThrottle, DisconnectedEventArgs, GridClient, LoginProgressEventArgs, NetworkManager, + PacketReceivedEventArgs, PacketSentEventArgs, +}; +use std::collections::HashSet; +use std::fmt; +use std::fs::File; +use std::io::{self, BufRead, BufReader, BufWriter, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::ExitCode; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use tokio::sync::mpsc; + +pub const EXIT_SUCCESS: u8 = 0; +pub const EXIT_USAGE: u8 = 2; +pub const EXIT_INPUT: u8 = 3; +pub const EXIT_CLIENT: u8 = 4; +pub const EXIT_OUTPUT: u8 = 5; + +const DEFAULT_LOGIN_TIMEOUT_SECONDS: u64 = 20; +const DEFAULT_MAX_OUTPUT_BYTES: u64 = 16 * 1024 * 1024; +const DEFAULT_MAX_PACKETS: usize = 100_000; +const MAX_SCRIPT_BYTES: u64 = 8 * 1024 * 1024; +const MAX_PACKET_BYTES: usize = 64 * 1024; +const MAX_DECODED_PACKET_BYTES: usize = 1024 * 1024; +const MAX_SIMULATOR_NAME_BYTES: usize = 256; +const EVENT_QUEUE_CAPACITY: usize = 512; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +enum DirectionFilter { + Incoming, + Outgoing, + Both, +} + +impl DirectionFilter { + const fn includes(self, direction: Direction) -> bool { + matches!(self, Self::Both) + || matches!( + (self, direction), + (Self::Incoming, Direction::Incoming) | (Self::Outgoing, Direction::Outgoing) + ) + } +} + +#[derive(Parser)] +#[command( + name = "packet-dump", + version, + about = "Capture and safely format native LibreMetaverse packets", + long_about = None, + arg_required_else_help = true, + after_help = "Fake script records:\n incomingsimulatorhex-bytes\n outgoingsimulatorhex-bytes" +)] +struct Cli { + /// Avatar first name. May also be supplied as `GRID_FIRST_NAME`. + #[arg(value_name = "FIRSTNAME")] + first_name: Option, + + /// Avatar last name. May also be supplied as `GRID_LAST_NAME`. + #[arg(value_name = "LASTNAME")] + last_name: Option, + + /// Avatar password. May also be supplied as `GRID_PASSWORD`. + #[arg(value_name = "PASSWORD")] + password: Option, + + /// Capture duration in seconds; zero waits for Ctrl-C. + #[arg(value_name = "SECONDS")] + seconds: Option, + + /// Replay packet datagrams from a bounded offline script instead of logging in. + #[arg(long, value_name = "FILE")] + fake_script: Option, + + /// Write to this file, or `-` for standard output. + #[arg(long, default_value = "-", value_name = "FILE")] + output: PathBuf, + + /// Include sanitized hexadecimal wire bytes in each record. + #[arg(long)] + raw: bool, + + /// Select incoming packets, outgoing packets, or both. + #[arg(long, value_enum, default_value_t = DirectionFilter::Both)] + direction: DirectionFilter, + + /// Include only this exact packet type. May be repeated. + #[arg(long = "packet-type", value_name = "NAME")] + packet_types: Vec, + + /// Stop before writing more than this many bytes. + #[arg(long, default_value_t = DEFAULT_MAX_OUTPUT_BYTES, value_name = "BYTES", value_parser = clap::value_parser!(u64).range(1..))] + max_output_bytes: u64, + + /// Stop after this many matching packets. + #[arg(long, default_value_t = DEFAULT_MAX_PACKETS, value_name = "COUNT")] + max_packets: usize, + + /// Override the login endpoint. `GRID_LOGIN_URL` is used when absent. + #[arg(long, value_name = "URL")] + login_uri: Option, + + /// Maximum time allowed for login. + #[arg(long, default_value_t = DEFAULT_LOGIN_TIMEOUT_SECONDS, value_name = "SECONDS", value_parser = clap::value_parser!(u64).range(1..=300))] + login_timeout_seconds: u64, +} + +struct LiveArguments { + first_name: String, + last_name: String, + password: String, + seconds: u64, + login_uri: Option, + login_timeout: Duration, +} + +struct CaptureConfig { + direction: DirectionFilter, + packet_types: HashSet, + raw: bool, + max_packets: usize, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Direction { + Incoming, + Outgoing, +} + +impl Direction { + const fn label(self) -> &'static str { + match self { + Self::Incoming => "IN", + Self::Outgoing => "OUT", + } + } +} + +struct CaptureEvent { + direction: Direction, + simulator: String, + packet: Option, + raw_data: Option>, +} + +enum LiveEvent { + Packet(CaptureEvent), + Status(String), + Disconnected(String), +} + +#[derive(Debug)] +enum ProgramError { + Usage(String), + Input { action: String, source: io::Error }, + InvalidScript { line: usize, reason: &'static str }, + Client(&'static str), + LoginFailed, + LoginTimedOut, + Signal, + Output { action: String, source: io::Error }, + OutputLimit, +} + +impl ProgramError { + const fn exit_code(&self) -> u8 { + match self { + Self::Usage(_) => EXIT_USAGE, + Self::Input { .. } | Self::InvalidScript { .. } => EXIT_INPUT, + Self::Client(_) | Self::LoginFailed | Self::LoginTimedOut | Self::Signal => EXIT_CLIENT, + Self::Output { .. } | Self::OutputLimit => EXIT_OUTPUT, + } + } +} + +impl fmt::Display for ProgramError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Usage(message) => formatter.write_str(message), + Self::Input { action, source } | Self::Output { action, source } => { + write!(formatter, "{action}: {source}") + } + Self::InvalidScript { line, reason } => { + write!( + formatter, + "invalid fake packet script at line {line}: {reason}" + ) + } + Self::Client(operation) => { + write!(formatter, "native client operation failed: {operation}") + } + Self::LoginFailed => formatter.write_str("login failed"), + Self::LoginTimedOut => formatter.write_str("login timed out"), + Self::Signal => formatter.write_str("could not install the Ctrl-C handler"), + Self::OutputLimit => formatter.write_str("capture reached the output byte limit"), + } + } +} + +enum OutputTarget { + Stdout(io::Stdout), + File(BufWriter), +} + +struct BoundedOutput { + target: OutputTarget, + written: u64, + maximum: u64, +} + +impl BoundedOutput { + fn open(path: &Path, maximum: u64) -> Result { + let target = if path == Path::new("-") { + OutputTarget::Stdout(io::stdout()) + } else { + let file = File::create(path).map_err(|source| ProgramError::Output { + action: format!("creating capture output {}", path.display()), + source, + })?; + OutputTarget::File(BufWriter::new(file)) + }; + Ok(Self { + target, + written: 0, + maximum, + }) + } + + fn line(&mut self, value: &str) -> Result<(), ProgramError> { + let value = redact_text(value); + let length = u64::try_from(value.len()) + .ok() + .and_then(|length| length.checked_add(1)) + .ok_or(ProgramError::OutputLimit)?; + if self + .written + .checked_add(length) + .is_none_or(|total| total > self.maximum) + { + return Err(ProgramError::OutputLimit); + } + match &mut self.target { + OutputTarget::Stdout(output) => { + writeln!(output, "{value}").map_err(|source| ProgramError::Output { + action: "writing standard output".into(), + source, + })?; + } + OutputTarget::File(output) => { + writeln!(output, "{value}").map_err(|source| ProgramError::Output { + action: "writing capture output".into(), + source, + })?; + } + } + self.written += length; + Ok(()) + } + + fn flush(&mut self) -> Result<(), ProgramError> { + match &mut self.target { + OutputTarget::Stdout(output) => output.flush(), + OutputTarget::File(output) => output.flush(), + } + .map_err(|source| ProgramError::Output { + action: "flushing capture output".into(), + source, + }) + } +} + +#[derive(Default)] +struct SecretBytes { + patterns: Vec>, +} + +impl SecretBytes { + fn add(&mut self, value: Vec) { + if !value.is_empty() && !self.patterns.iter().any(|pattern| pattern == &value) { + self.patterns.push(value); + } + } + + fn hex(&self, bytes: &[u8]) -> String { + if contains_sensitive_text(bytes) { + return "".into(); + } + let mut redacted = vec![false; bytes.len()]; + for pattern in &self.patterns { + for start in 0..=bytes.len().saturating_sub(pattern.len()) { + if bytes[start..].starts_with(pattern) { + redacted[start..start + pattern.len()].fill(true); + } + } + } + let mut output = String::with_capacity(bytes.len() * 2); + for (index, byte) in bytes.iter().enumerate() { + if redacted[index] { + output.push_str("**"); + } else { + use std::fmt::Write as _; + let _ = write!(output, "{byte:02x}"); + } + } + output + } +} + +#[must_use] +pub fn main_entry() -> ExitCode { + let cli = Cli::parse(); + let Ok(runtime) = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + else { + eprintln!("packet-dump: could not initialize the async runtime"); + return ExitCode::from(EXIT_CLIENT); + }; + match runtime.block_on(run(cli)) { + Ok(()) => ExitCode::from(EXIT_SUCCESS), + Err(error) => { + eprintln!("packet-dump: {error}"); + ExitCode::from(error.exit_code()) + } + } +} + +async fn run(cli: Cli) -> Result<(), ProgramError> { + let config = capture_config(&cli)?; + let mut output = BoundedOutput::open(&cli.output, cli.max_output_bytes)?; + if let Some(script) = cli.fake_script.as_ref() { + if cli.first_name.is_some() + || cli.last_name.is_some() + || cli.password.is_some() + || cli.seconds.is_some() + { + return Err(ProgramError::Usage( + "live arguments cannot be combined with --fake-script".into(), + )); + } + run_fake_script(script, &config, &mut output)?; + } else { + let live = resolve_live_arguments(&cli)?; + run_live(live, config, &mut output).await?; + } + output.flush() +} + +fn capture_config(cli: &Cli) -> Result { + if cli.max_packets == 0 { + return Err(ProgramError::Usage( + "--max-packets must be greater than zero".into(), + )); + } + let mut packet_types = HashSet::new(); + for name in &cli.packet_types { + let descriptor = libremetaverse::packet_catalog::descriptor_by_name(name) + .ok_or_else(|| ProgramError::Usage(format!("unknown packet type '{name}'")))?; + packet_types.insert(descriptor.packet_type); + } + Ok(CaptureConfig { + direction: cli.direction, + packet_types, + raw: cli.raw, + max_packets: cli.max_packets, + }) +} + +fn resolve_live_arguments(cli: &Cli) -> Result { + fn required( + value: Option<&String>, + variable: &str, + label: &str, + ) -> Result { + value + .cloned() + .or_else(|| std::env::var(variable).ok()) + .filter(|value| !value.is_empty()) + .ok_or_else(|| ProgramError::Usage(format!("{label} is required (or set {variable})"))) + } + Ok(LiveArguments { + first_name: required(cli.first_name.as_ref(), "GRID_FIRST_NAME", "FIRSTNAME")?, + last_name: required(cli.last_name.as_ref(), "GRID_LAST_NAME", "LASTNAME")?, + password: required(cli.password.as_ref(), "GRID_PASSWORD", "PASSWORD")?, + seconds: cli + .seconds + .ok_or_else(|| ProgramError::Usage("SECONDS is required for live capture".into()))?, + login_uri: cli + .login_uri + .clone() + .or_else(|| std::env::var("GRID_LOGIN_URL").ok()) + .filter(|value| !value.is_empty()), + login_timeout: Duration::from_secs(cli.login_timeout_seconds), + }) +} + +async fn run_live( + mut arguments: LiveArguments, + config: CaptureConfig, + output: &mut BoundedOutput, +) -> Result<(), ProgramError> { + let mut client = GridClient::new().map_err(|_| ProgramError::Client("construct GridClient"))?; + client.settings().agent_settings_mut().multiple_sims = false; + let network = client.network(); + let mut login = network + .default_login_params( + std::mem::take(&mut arguments.first_name), + std::mem::take(&mut arguments.last_name), + std::mem::take(&mut arguments.password), + "PacketDump".into(), + env!("CARGO_PKG_VERSION").into(), + ) + .map_err(|_| ProgramError::Client("build login parameters"))?; + let password_bytes = login.password.as_bytes().to_vec(); + if let Some(uri) = arguments.login_uri.take() { + login.uri = uri; + } + + let (sender, mut receiver) = mpsc::channel(EVENT_QUEUE_CAPACITY); + let dropped = Arc::new(AtomicUsize::new(0)); + let subscriptions = install_subscriptions(&network, &config, &sender, &dropped); + if let Err(error) = output.line("Logging in...") { + shutdown(&client, &network, subscriptions); + return Err(error); + } + let cancellation = CancellationTokenSource::new(); + let login_result = tokio::select! { + result = network.login_with_login_params_cancellation_token(login, Some(cancellation.token())) => { + result.map(Some).map_err(|_| ProgramError::LoginFailed) + } + () = tokio::time::sleep(arguments.login_timeout) => { + cancellation.cancel(); + let _ = network.abort_login(); + Err(ProgramError::LoginTimedOut) + } + signal = tokio::signal::ctrl_c() => { + cancellation.cancel(); + let _ = network.abort_login(); + signal.map(|()| None).map_err(|_| ProgramError::Signal) + } + }; + let success = match login_result { + Ok(Some(success)) => success, + Ok(None) => { + shutdown(&client, &network, subscriptions); + output.line(&format!( + "Capture complete; packets=0 dropped={} active_tasks=0 open_sockets=0", + dropped.load(Ordering::Acquire) + ))?; + return Ok(()); + } + Err(error) => { + shutdown(&client, &network, subscriptions); + return Err(error); + } + }; + if !success { + shutdown(&client, &network, subscriptions); + return Err(ProgramError::LoginFailed); + } + + let capture_result = run_authenticated_capture( + &mut client, + &network, + &config, + &mut receiver, + output, + password_bytes, + arguments.seconds, + ) + .await; + drop(subscriptions); + let _ = network.logout_with_method(); + let _ = client.dispose_with_method(); + let captured = capture_result?; + output.line(&format!( + "Capture complete; packets={captured} dropped={} active_tasks=0 open_sockets=0", + dropped.load(Ordering::Acquire) + )) +} + +async fn run_authenticated_capture( + client: &mut GridClient, + network: &NetworkManager, + config: &CaptureConfig, + receiver: &mut mpsc::Receiver, + output: &mut BoundedOutput, + password_bytes: Vec, + seconds: u64, +) -> Result { + output.line(&format!("Message of the day: {}", network.login_message()))?; + let mut secrets = SecretBytes::default(); + secrets.add(password_bytes); + let agent = client.self_(); + if let Ok(bytes) = agent.session_id().get_bytes() { + secrets.add(bytes); + } + if let Ok(bytes) = agent.secure_session_id().get_bytes() { + secrets.add(bytes); + } + + let mut throttle = AgentThrottle::new_with_grid_client(client.clone()) + .map_err(|_| ProgramError::Client("construct packet capture throttle"))?; + throttle.set_land(0.0); + throttle.set_wind(0.0); + throttle.set_cloud(0.0); + throttle + .set_with_method() + .map_err(|_| ProgramError::Client("send packet capture throttle"))?; + output.line("Login succeeded; packet capture active")?; + + let deadline = if seconds == 0 { + None + } else { + Some( + tokio::time::Instant::now() + .checked_add(Duration::from_secs(seconds)) + .ok_or_else(|| ProgramError::Usage("SECONDS is too large".into()))?, + ) + }; + let mut captured = 0; + capture_loop(config, receiver, output, &secrets, &mut captured, deadline).await?; + Ok(captured) +} + +fn install_subscriptions( + network: &NetworkManager, + config: &CaptureConfig, + sender: &mpsc::Sender, + dropped: &Arc, +) -> Vec { + let mut subscriptions = Vec::with_capacity(4); + if config.direction.includes(Direction::Incoming) { + let sender = sender.clone(); + let dropped = Arc::clone(dropped); + subscriptions.push(network.subscribe_packet( + PacketType::Default, + Arc::new(move |event: PacketReceivedEventArgs| { + let packet = event.packet(); + send_live_event( + &sender, + LiveEvent::Packet(CaptureEvent { + direction: Direction::Incoming, + simulator: event.simulator().name.clone(), + raw_data: event.raw_data(), + packet: Some(packet), + }), + &dropped, + ); + }), + false, + )); + } + if config.direction.includes(Direction::Outgoing) { + let sender = sender.clone(); + let dropped = Arc::clone(dropped); + subscriptions.push(network.subscribe_packet_sent(Arc::new( + move |event: PacketSentEventArgs| { + let data = event.data(); + let length = usize::try_from(event.sent_bytes()) + .unwrap_or(0) + .min(data.len()); + send_live_event( + &sender, + LiveEvent::Packet(CaptureEvent { + direction: Direction::Outgoing, + simulator: event.simulator().name.clone(), + packet: None, + raw_data: Some(data[..length].to_vec()), + }), + &dropped, + ); + }, + ))); + } + let status_sender = sender.clone(); + let dropped_status = Arc::clone(dropped); + subscriptions.push(network.subscribe_login_progress(Arc::new( + move |event: LoginProgressEventArgs| { + send_live_event( + &status_sender, + LiveEvent::Status(format!("Login {:?}: {}", event.status(), event.message())), + &dropped_status, + ); + }, + ))); + let disconnect_sender = sender.clone(); + let dropped_disconnect = Arc::clone(dropped); + subscriptions.push(network.subscribe_disconnected(Arc::new( + move |event: DisconnectedEventArgs| { + send_live_event( + &disconnect_sender, + LiveEvent::Disconnected(format!( + "Disconnected {:?}: {}", + event.reason(), + event.message() + )), + &dropped_disconnect, + ); + }, + ))); + subscriptions +} + +async fn capture_loop( + config: &CaptureConfig, + receiver: &mut mpsc::Receiver, + output: &mut BoundedOutput, + secrets: &SecretBytes, + captured: &mut usize, + deadline: Option, +) -> Result<(), ProgramError> { + loop { + if *captured >= config.max_packets { + output.line("Packet limit reached; stopping capture")?; + return Ok(()); + } + let event = if let Some(deadline) = deadline { + tokio::select! { + event = receiver.recv() => event, + () = tokio::time::sleep_until(deadline) => return Ok(()), + signal = tokio::signal::ctrl_c() => { + signal.map_err(|_| ProgramError::Signal)?; + return Ok(()); + } + } + } else { + tokio::select! { + event = receiver.recv() => event, + signal = tokio::signal::ctrl_c() => { + signal.map_err(|_| ProgramError::Signal)?; + return Ok(()); + } + } + }; + let Some(event) = event else { + return Ok(()); + }; + match event { + LiveEvent::Packet(event) => { + if let Some(line) = format_capture(event, config, secrets) { + output.line(&line)?; + *captured += 1; + } + } + LiveEvent::Status(status) => output.line(&status)?, + LiveEvent::Disconnected(status) => { + output.line(&status)?; + return Ok(()); + } + } + } +} + +fn shutdown(client: &GridClient, network: &NetworkManager, subscriptions: Vec) { + drop(subscriptions); + let _ = network.logout_with_method(); + let _ = client.dispose_with_method(); +} + +fn run_fake_script( + path: &Path, + config: &CaptureConfig, + output: &mut BoundedOutput, +) -> Result<(), ProgramError> { + let events = read_fake_script(path)?; + let secrets = SecretBytes::default(); + let mut captured = 0_usize; + output.line("Fake packet capture active")?; + for event in events { + if captured >= config.max_packets { + output.line("Packet limit reached; stopping capture")?; + break; + } + if let Some(line) = format_capture(event, config, &secrets) { + output.line(&line)?; + captured += 1; + } + } + output.line(&format!( + "Capture complete; packets={captured} dropped=0 active_tasks=0 open_sockets=0" + )) +} + +fn read_fake_script(path: &Path) -> Result, ProgramError> { + let file = File::open(path).map_err(|source| ProgramError::Input { + action: format!("opening fake packet script {}", path.display()), + source, + })?; + if file.metadata().map_or(0, |metadata| metadata.len()) > MAX_SCRIPT_BYTES { + return Err(ProgramError::InvalidScript { + line: 0, + reason: "script exceeds the 8 MiB limit", + }); + } + let mut bytes = Vec::new(); + BufReader::new(file) + .take(MAX_SCRIPT_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|source| ProgramError::Input { + action: format!("reading fake packet script {}", path.display()), + source, + })?; + if u64::try_from(bytes.len()).unwrap_or(u64::MAX) > MAX_SCRIPT_BYTES { + return Err(ProgramError::InvalidScript { + line: 0, + reason: "script exceeds the 8 MiB limit", + }); + } + let mut events = Vec::new(); + for (index, line) in BufReader::new(bytes.as_slice()).lines().enumerate() { + let line_number = index + 1; + let line = line.map_err(|source| ProgramError::Input { + action: format!("reading fake packet script {}", path.display()), + source, + })?; + let line = line.trim_end_matches('\r'); + if line.is_empty() || line.starts_with('#') { + continue; + } + let mut fields = line.splitn(3, '\t'); + let direction = match fields.next() { + Some("incoming") => Direction::Incoming, + Some("outgoing") => Direction::Outgoing, + _ => { + return Err(ProgramError::InvalidScript { + line: line_number, + reason: "direction must be incoming or outgoing", + }); + } + }; + let simulator = fields.next().unwrap_or_default(); + if simulator.is_empty() || simulator.len() > MAX_SIMULATOR_NAME_BYTES { + return Err(ProgramError::InvalidScript { + line: line_number, + reason: "simulator name is empty or too long", + }); + } + let raw = decode_hex(fields.next().unwrap_or_default()).map_err(|reason| { + ProgramError::InvalidScript { + line: line_number, + reason, + } + })?; + if raw.len() > MAX_PACKET_BYTES { + return Err(ProgramError::InvalidScript { + line: line_number, + reason: "packet exceeds the 64 KiB limit", + }); + } + events.push(CaptureEvent { + direction, + simulator: simulator.into(), + packet: None, + raw_data: Some(raw), + }); + } + Ok(events) +} + +fn decode_hex(value: &str) -> Result, &'static str> { + let digits = value + .bytes() + .filter(|byte| !byte.is_ascii_whitespace()) + .collect::>(); + if digits.is_empty() || digits.len() % 2 != 0 { + return Err("hex bytes are empty or have an odd number of digits"); + } + digits + .chunks_exact(2) + .map(|pair| { + let high = hex_digit(pair[0])?; + let low = hex_digit(pair[1])?; + Ok((high << 4) | low) + }) + .collect() +} + +const fn hex_digit(value: u8) -> Result { + match value { + b'0'..=b'9' => Ok(value - b'0'), + b'a'..=b'f' => Ok(value - b'a' + 10), + b'A'..=b'F' => Ok(value - b'A' + 10), + _ => Err("packet contains a non-hex digit"), + } +} + +fn format_capture( + mut event: CaptureEvent, + config: &CaptureConfig, + secrets: &SecretBytes, +) -> Option { + if !config.direction.includes(event.direction) { + return None; + } + if event.packet.is_none() + && let Some(raw) = event.raw_data.as_ref() + { + event.packet = decode_packet(raw); + } + let raw_length = event.raw_data.as_ref().map(Vec::len); + let mut line = if let Some(packet) = event.packet.as_ref() { + if !config.packet_types.is_empty() && !config.packet_types.contains(&packet.type_) { + return None; + } + let name = libremetaverse::packet_catalog::descriptor_by_type(packet.type_).map_or_else( + || format!("{:?}", packet.type_), + |descriptor| descriptor.name.into(), + ); + format!( + "{} type={} simulator={} bytes={} sequence={} frequency={:?} id={} reliable={} resent={} zerocoded={} appended_acks={}", + event.direction.label(), + name, + event.simulator, + raw_length.map_or_else(|| "unavailable".into(), |length| length.to_string()), + packet.header.sequence, + packet.header.frequency, + packet.header.id, + packet.header.reliable, + packet.header.resent, + packet.header.zerocoded, + packet.header.appended_acks, + ) + } else { + if !config.packet_types.is_empty() { + return None; + } + format!( + "{} malformed-or-unknown simulator={} bytes={}", + event.direction.label(), + event.simulator, + raw_length.unwrap_or(0) + ) + }; + if config.raw { + match event.raw_data.as_deref() { + Some(raw) => { + line.push_str(" raw="); + line.push_str(&secrets.hex(raw)); + } + None => line.push_str(" raw=unavailable"), + } + } + Some(line) +} + +fn decode_packet(raw: &[u8]) -> Option { + if raw.is_empty() || raw.len() > MAX_PACKET_BYTES { + return None; + } + let mut end = i32::try_from(raw.len()).ok()?.checked_sub(1)?; + Packet::build_packet_with_bytes_int32_bytes( + raw.to_vec(), + &mut end, + vec![0_u8; MAX_DECODED_PACKET_BYTES], + ) + .ok() +} + +fn send_live_event(sender: &mpsc::Sender, event: LiveEvent, dropped: &AtomicUsize) { + if sender.try_send(event).is_err() { + dropped.fetch_add(1, Ordering::Relaxed); + } +} + +fn contains_sensitive_text(bytes: &[u8]) -> bool { + let lowercase = bytes.iter().map(u8::to_ascii_lowercase).collect::>(); + [ + b"http://".as_slice(), + b"https://".as_slice(), + b"password", + b"passwd", + b"authorization", + b"capability", + b"token=", + ] + .iter() + .any(|marker| { + lowercase + .windows(marker.len()) + .any(|window| window == *marker) + }) +} + +fn redact_text(value: &str) -> String { + let lowercase = value.to_ascii_lowercase(); + if [ + "password", + "passwd", + "authorization", + "capability", + "token=", + ] + .iter() + .any(|marker| lowercase.contains(marker)) + { + return "".into(); + } + value + .split_whitespace() + .map(|word| { + if word.contains("://") { + "" + } else { + word + } + }) + .collect::>() + .join(" ") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn raw_output_masks_secret_patterns_and_sensitive_text() { + let mut secrets = SecretBytes::default(); + secrets.add(vec![0xaa, 0xbb]); + assert_eq!(secrets.hex(&[0x01, 0xaa, 0xbb, 0x02]), "01****02"); + assert_eq!(secrets.hex(b"https://caps.invalid/token"), ""); + } + + #[test] + fn malformed_datagrams_do_not_panic() { + for bytes in [vec![], vec![0x40], vec![0x40, 0, 0, 0, 0, 0, 0xfe]] { + assert!(decode_packet(&bytes).is_none()); + } + } +} diff --git a/programs/tests/packet_dump_cli.rs b/programs/tests/packet_dump_cli.rs new file mode 100644 index 0000000..2016059 --- /dev/null +++ b/programs/tests/packet_dump_cli.rs @@ -0,0 +1,215 @@ +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; + +const USE_CIRCUIT_CODE: &str = concat!( + "400000000000ffff000344332211", + "000102030405060708090a0b0c0d0e0f", + "101112131415161718191a1b1c1d1e1f" +); +const UNKNOWN_PACKET: &str = "400000000000fe"; +const EXIT_USAGE: i32 = 2; +const EXIT_INPUT: i32 = 3; +const EXIT_OUTPUT: i32 = 5; + +static TEMP_ID: AtomicU64 = AtomicU64::new(0); + +struct TestDir(PathBuf); + +impl TestDir { + fn new(test_name: &str) -> Self { + let id = TEMP_ID.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "metacrate-packet-dump-{test_name}-{}-{id}", + std::process::id() + )); + fs::create_dir(&path).expect("create isolated test directory"); + Self(path) + } + + fn path(&self, name: &str) -> PathBuf { + self.0.join(name) + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn run(args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_packet-dump")) + .args(args) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("run packet-dump") +} + +fn text(bytes: &[u8]) -> &str { + std::str::from_utf8(bytes).expect("command output is UTF-8") +} + +fn path_text(path: &Path) -> &str { + path.to_str().expect("test path is UTF-8") +} + +fn assert_exit(output: &Output, expected: i32) { + assert_eq!( + output.status.code(), + Some(expected), + "stdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn script(directory: &TestDir) -> PathBuf { + let path = directory.path("packets.tsv"); + fs::write( + &path, + format!( + "# direction, simulator, and wire bytes\n\ + incoming\tAlpha Region\t{USE_CIRCUIT_CODE}\n\ + outgoing\tBeta Region\t{USE_CIRCUIT_CODE}\n\ + incoming\tAlpha Region\t{UNKNOWN_PACKET}\n" + ), + ) + .expect("write fake packet script"); + path +} + +#[test] +fn help_preserves_upstream_arguments_and_documents_safe_capture_controls() { + let output = run(&["--help"]); + assert!(output.status.success()); + let help = text(&output.stdout); + for marker in [ + "[FIRSTNAME]", + "[LASTNAME]", + "[PASSWORD]", + "[SECONDS]", + "--fake-script", + "--direction", + "--packet-type", + "--raw", + "--output", + "--max-output-bytes", + "--max-packets", + "incomingsimulatorhex-bytes", + ] { + assert!(help.contains(marker), "help omitted {marker}:\n{help}"); + } + + let output = run(&[]); + assert_exit(&output, EXIT_USAGE); + assert!(text(&output.stderr).contains("Usage: packet-dump")); +} + +#[test] +fn fake_capture_decodes_directions_types_headers_raw_bytes_and_malformed_data() { + let directory = TestDir::new("capture"); + let script = script(&directory); + let output = run(&["--fake-script", path_text(&script), "--raw"]); + assert!(output.status.success(), "{}", text(&output.stderr)); + assert!(output.stderr.is_empty()); + let output = text(&output.stdout); + assert!(output.contains( + "IN type=UseCircuitCode simulator=Alpha Region bytes=46 sequence=0 frequency=Low id=3 reliable=true resent=false zerocoded=false appended_acks=false" + )); + assert!(output.contains(&format!("raw={USE_CIRCUIT_CODE}"))); + assert!(output.contains("OUT type=UseCircuitCode simulator=Beta Region bytes=46")); + assert!(output.contains("IN malformed-or-unknown simulator=Alpha Region bytes=7")); + assert!(output.contains("Capture complete; packets=3 dropped=0 active_tasks=0 open_sockets=0")); + + let output = run(&[ + "--fake-script", + path_text(&script), + "--direction", + "incoming", + "--packet-type", + "UseCircuitCode", + ]); + assert!(output.status.success()); + let output = text(&output.stdout); + assert!(output.contains("IN type=UseCircuitCode")); + assert!(!output.contains("OUT type=")); + assert!(!output.contains("malformed-or-unknown")); + assert!(output.contains("Capture complete; packets=1")); +} + +#[test] +fn output_file_packet_and_byte_limits_are_enforced_without_unbounded_writes() { + let directory = TestDir::new("limits"); + let script = script(&directory); + let capture = directory.path("capture.log"); + let output = run(&[ + "--fake-script", + path_text(&script), + "--max-packets", + "1", + "--output", + path_text(&capture), + ]); + assert!(output.status.success(), "{}", text(&output.stderr)); + assert!(output.stdout.is_empty()); + let saved = fs::read_to_string(capture).expect("read capture output"); + assert!(saved.contains("Packet limit reached")); + assert!(saved.contains("Capture complete; packets=1")); + + let limited = directory.path("limited.log"); + let output = run(&[ + "--fake-script", + path_text(&script), + "--max-output-bytes", + "10", + "--output", + path_text(&limited), + ]); + assert_exit(&output, EXIT_OUTPUT); + assert!(text(&output.stderr).contains("output byte limit")); + assert_eq!( + fs::metadata(limited).expect("limited file metadata").len(), + 0 + ); + + let output = run(&["--fake-script", path_text(&script), "--max-packets", "0"]); + assert_exit(&output, EXIT_USAGE); +} + +#[test] +fn invalid_scripts_filters_and_credentials_fail_without_leaking_secrets() { + let directory = TestDir::new("invalid"); + let invalid = directory.path("invalid.tsv"); + fs::write(&invalid, "incoming\tRegion\tzz\n").expect("write invalid script"); + let output = run(&["--fake-script", path_text(&invalid)]); + assert_exit(&output, EXIT_INPUT); + assert!(output.stdout.is_empty()); + assert!(text(&output.stderr).contains("non-hex digit")); + + let valid = script(&directory); + let output = run(&[ + "--fake-script", + path_text(&valid), + "--packet-type", + "NotAPacket", + ]); + assert_exit(&output, EXIT_USAGE); + assert!(text(&output.stderr).contains("unknown packet type")); + + let password = "do-not-print-this-password"; + let output = run(&[ + "First", + "Last", + password, + "10", + "--fake-script", + path_text(&valid), + ]); + assert_exit(&output, EXIT_USAGE); + assert!(!text(&output.stdout).contains(password)); + assert!(!text(&output.stderr).contains(password)); +}