From 8fb96275cc7a35d22879115adb592a8f9100c299 Mon Sep 17 00:00:00 2001 From: Joey Stanford Date: Thu, 2 Jul 2026 13:29:22 -0600 Subject: [PATCH] feat: wire packet tap for sniffer/debug consumers Add PacketTapEvent/PacketTapDirection, TransportMessage::SetPacketTap, RX/TX emission in the transport actor, and ReticulumHandle::register_packet_tap for embedders that need wire-level observation (e.g. mesh-client sidecar sniffer). --- crates/rns-runtime/src/reticulum.rs | 11 ++++ crates/rns-transport/src/actor/inbound.rs | 9 +++ crates/rns-transport/src/actor/mod.rs | 50 +++++++++++++++- crates/rns-transport/src/messages.rs | 71 +++++++++++++++++++++++ 4 files changed, 139 insertions(+), 2 deletions(-) diff --git a/crates/rns-runtime/src/reticulum.rs b/crates/rns-runtime/src/reticulum.rs index ec0a40b7..2418a175 100644 --- a/crates/rns-runtime/src/reticulum.rs +++ b/crates/rns-runtime/src/reticulum.rs @@ -1477,6 +1477,17 @@ impl ReticulumHandle { result } + /// Register a wire packet tap on the local transport actor (sniffer/debug). + pub async fn register_packet_tap( + &self, + tap_tx: tokio::sync::broadcast::Sender, + ) { + let _ = self + .transport_tx + .send(TransportMessage::SetPacketTap { tap_tx }) + .await; + } + /// Recall the identity and latest announce metadata for `destination_hash`. /// /// This reads this process' live, validated replicated announce cache in diff --git a/crates/rns-transport/src/actor/inbound.rs b/crates/rns-transport/src/actor/inbound.rs index 9a487250..9ec63061 100644 --- a/crates/rns-transport/src/actor/inbound.rs +++ b/crates/rns-transport/src/actor/inbound.rs @@ -41,6 +41,15 @@ impl TransportActor { packet.raw.clone() }; + self.emit_packet_tap( + crate::messages::PacketTapDirection::Rx, + packet.interface_id, + &raw, + packet.rssi, + packet.snr, + packet.q, + ); + let (mut parsed, data_offset) = match rns_wire::header::PacketHeader::unpack(&raw) { Ok((header, offset)) => (header, offset), Err(e) => { diff --git a/crates/rns-transport/src/actor/mod.rs b/crates/rns-transport/src/actor/mod.rs index 4aba1c10..c6c594bc 100644 --- a/crates/rns-transport/src/actor/mod.rs +++ b/crates/rns-transport/src/actor/mod.rs @@ -245,6 +245,9 @@ pub struct TransportActor { /// background switches the maintenance tick to the long interval so the /// actor stops burning CPU (and battery) while the app is suspended. pub is_foreground: Arc, + + /// Optional wire packet tap — emits RX/TX frames for sniffer UIs. + packet_tap: Option>, } /// Cached announce metadata for diagnostics + CacheRequest replay. Raw @@ -411,6 +414,7 @@ impl TransportActor { announce_handlers: Vec::new(), next_announce_handler_id: 0, is_foreground: Arc::new(AtomicBool::new(true)), + packet_tap: None, }; (actor, tx) @@ -974,6 +978,9 @@ impl TransportActor { debug!(dest = hex::encode(dest), "registered path waiter"); } } + TransportMessage::SetPacketTap { tap_tx } => { + self.packet_tap = Some(tap_tx); + } TransportMessage::Shutdown => unreachable!(), } } @@ -1447,6 +1454,35 @@ impl TransportActor { /// /// `try_send` failures: `Full` bumps `tx_drops`; `Closed` auto-deregisters /// (zombie interface — receiver dropped without DeregisterInterface). + fn emit_packet_tap( + &self, + direction: crate::messages::PacketTapDirection, + interface_id: InterfaceId, + raw: &[u8], + rssi: Option, + snr: Option, + q: Option, + ) { + let Some(tap) = self.packet_tap.as_ref() else { + return; + }; + let interface_name = self + .interfaces + .get(&interface_id) + .map(|e| e.name.clone()) + .unwrap_or_else(|| format!("interface_{interface_id}")); + let event = crate::messages::PacketTapEvent::from_wire( + direction, + interface_id, + interface_name, + raw, + rssi, + snr, + q, + ); + let _ = tap.send(event); + } + #[tracing::instrument( level = "trace", name = "actor.send_to_interface", @@ -1468,8 +1504,18 @@ impl TransportActor { } else { Bytes::copy_from_slice(raw) }; - match entry.tx.try_send(data) { - Ok(()) => InterfaceSendOutcome::Sent, + match entry.tx.try_send(data.clone()) { + Ok(()) => { + self.emit_packet_tap( + crate::messages::PacketTapDirection::Tx, + id, + &data, + None, + None, + None, + ); + InterfaceSendOutcome::Sent + } Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => InterfaceSendOutcome::Full, Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => InterfaceSendOutcome::Closed, } diff --git a/crates/rns-transport/src/messages.rs b/crates/rns-transport/src/messages.rs index 7adbf81b..a96aae4d 100644 --- a/crates/rns-transport/src/messages.rs +++ b/crates/rns-transport/src/messages.rs @@ -419,6 +419,71 @@ pub struct PathRequestOptions { pub recursive: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PacketTapDirection { + Rx, + Tx, +} + +/// Wire-level packet observation event for debug/sniffer UIs. +#[derive(Debug, Clone)] +pub struct PacketTapEvent { + pub direction: PacketTapDirection, + pub interface_id: InterfaceId, + pub interface_name: String, + pub raw: Vec, + pub rssi: Option, + pub snr: Option, + pub q: Option, + pub packet_type: Option, + pub header_type: Option, + pub destination_hash: Option<[u8; 16]>, + pub transport_type: Option, + pub context: Option, +} + +const PACKET_TAP_RAW_CAP: usize = 4096; + +impl PacketTapEvent { + pub fn from_wire( + direction: PacketTapDirection, + interface_id: InterfaceId, + interface_name: String, + raw: &[u8], + rssi: Option, + snr: Option, + q: Option, + ) -> Self { + let capped = if raw.len() > PACKET_TAP_RAW_CAP { + raw[..PACKET_TAP_RAW_CAP].to_vec() + } else { + raw.to_vec() + }; + let mut event = Self { + direction, + interface_id, + interface_name, + raw: capped, + rssi, + snr, + q, + packet_type: None, + header_type: None, + destination_hash: None, + transport_type: None, + context: None, + }; + if let Ok((header, _)) = rns_wire::header::PacketHeader::unpack(&event.raw) { + event.packet_type = Some(format!("{:?}", header.flags.packet_type)); + event.header_type = Some(format!("{:?}", header.flags.header_type)); + event.destination_hash = Some(header.destination_hash); + event.transport_type = Some(format!("{:?}", header.flags.transport_type)); + event.context = Some(format!("{:?}", header.context)); + } + event + } +} + /// Every mutation of transport state enters through this enum — the actor /// dispatches on the variant, so adding a new operation is a matter of adding /// a variant and a match arm rather than exposing a new lock or shared type. @@ -618,6 +683,10 @@ pub enum TransportMessage { dest: [u8; 16], reply: tokio::sync::oneshot::Sender, }, + /// Optional wire packet tap for debug/sniffer consumers. + SetPacketTap { + tap_tx: tokio::sync::broadcast::Sender, + }, Shutdown, } @@ -661,6 +730,7 @@ pub fn msg_variant_name(msg: &TransportMessage) -> &'static str { TransportMessage::RegisterLink { .. } => "RegisterLink", TransportMessage::ActivateLink { .. } => "ActivateLink", TransportMessage::AwaitPath { .. } => "AwaitPath", + TransportMessage::SetPacketTap { .. } => "SetPacketTap", TransportMessage::Shutdown => "Shutdown", } } @@ -1161,6 +1231,7 @@ impl std::fmt::Debug for TransportMessage { Self::AwaitPath { dest, .. } => { f.debug_struct("AwaitPath").field("dest", dest).finish() } + Self::SetPacketTap { .. } => f.debug_struct("SetPacketTap").finish(), Self::Shutdown => write!(f, "Shutdown"), } }