Implement native PacketDump capture (#87)

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
+incoming<TAB>simulator<TAB>hex-bytes
+outgoing<TAB>simulator<TAB>hex-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
+```

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()
 }

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;

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  incoming<TAB>simulator<TAB>hex-bytes\n  outgoing<TAB>simulator<TAB>hex-bytes"
+)]
+struct Cli {
+    /// Avatar first name. May also be supplied as `GRID_FIRST_NAME`.
+    #[arg(value_name = "FIRSTNAME")]
+    first_name: Option<String>,
+
+    /// Avatar last name. May also be supplied as `GRID_LAST_NAME`.
+    #[arg(value_name = "LASTNAME")]
+    last_name: Option<String>,
+
+    /// Avatar password. May also be supplied as `GRID_PASSWORD`.
+    #[arg(value_name = "PASSWORD")]
+    password: Option<String>,
+
+    /// Capture duration in seconds; zero waits for Ctrl-C.
+    #[arg(value_name = "SECONDS")]
+    seconds: Option<u64>,
+
+    /// Replay packet datagrams from a bounded offline script instead of logging in.
+    #[arg(long, value_name = "FILE")]
+    fake_script: Option<PathBuf>,
+
+    /// 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<String>,
+
+    /// 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<String>,
+
+    /// 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<String>,
+    login_timeout: Duration,
+}
+
+struct CaptureConfig {
+    direction: DirectionFilter,
+    packet_types: HashSet<PacketType>,
+    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<Packet>,
+    raw_data: Option<Vec<u8>>,
+}
+
+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<File>),
+}
+
+struct BoundedOutput {
+    target: OutputTarget,
+    written: u64,
+    maximum: u64,
+}
+
+impl BoundedOutput {
+    fn open(path: &Path, maximum: u64) -> Result<Self, ProgramError> {
+        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<Vec<u8>>,
+}
+
+impl SecretBytes {
+    fn add(&mut self, value: Vec<u8>) {
+        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 "<redacted>".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<CaptureConfig, ProgramError> {
+    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<LiveArguments, ProgramError> {
+    fn required(
+        value: Option<&String>,
+        variable: &str,
+        label: &str,
+    ) -> Result<String, ProgramError> {
+        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<LiveEvent>,
+    output: &mut BoundedOutput,
+    password_bytes: Vec<u8>,
+    seconds: u64,
+) -> Result<usize, ProgramError> {
+    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<LiveEvent>,
+    dropped: &Arc<AtomicUsize>,
+) -> Vec<Subscription> {
+    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<LiveEvent>,
+    output: &mut BoundedOutput,
+    secrets: &SecretBytes,
+    captured: &mut usize,
+    deadline: Option<tokio::time::Instant>,
+) -> 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<Subscription>) {
+    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<Vec<CaptureEvent>, 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<Vec<u8>, &'static str> {
+    let digits = value
+        .bytes()
+        .filter(|byte| !byte.is_ascii_whitespace())
+        .collect::<Vec<_>>();
+    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<u8, &'static str> {
+    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<String> {
+    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<Packet> {
+    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<LiveEvent>, 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::<Vec<_>>();
+    [
+        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 "<redacted>".into();
+    }
+    value
+        .split_whitespace()
+        .map(|word| {
+            if word.contains("://") {
+                "<redacted-url>"
+            } else {
+                word
+            }
+        })
+        .collect::<Vec<_>>()
+        .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"), "<redacted>");
+    }
+
+    #[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());
+        }
+    }
+}

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",
+        "incoming<TAB>simulator<TAB>hex-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));
+}