From 7df9f41e9fd358510f97d230c0405937b72cc03f Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 19 Oct 2026 23:18:35 -0700 Subject: [PATCH] fix(bt): framing, roles, NodeId binding, contact events --- tines-bt/README.md | 76 +++++ tines-bt/src/adv.rs | 32 ++ tines-bt/src/lib.rs | 4 +- tines-bt/src/transport.rs | 575 +++++++++++++++++++++++++++--------- tines-bt/tests/adv.rs | 61 ++-- tines-bt/tests/transport.rs | 258 ++++++++++++++++ tines-core/src/wire.rs | 74 +++++ 7 files changed, 921 insertions(+), 159 deletions(-) create mode 100644 tines-bt/README.md create mode 100644 tines-bt/tests/transport.rs diff --git a/tines-bt/README.md b/tines-bt/README.md new file mode 100644 index 00000000..055b68c4 --- /dev/null +++ b/tines-bt/README.md @@ -0,0 +1,76 @@ +# tines-bt + +The BLE transport for tines: a [`Transport`] implementation whose address is +the peer's advertised [`NodeId`], so identity survives BLE address rotation. + +Real radio I/O sits behind the small [`Radio`] trait (implemented over +`bluer`/BlueZ by [`BluerRadio`]); everything else — the `from`/`Addr` check, the +central/peripheral role decision, and the address→`NodeId` binding — is pure and +unit-testable without hardware. + +## Framing + +Connections are `bluer` [`l2cap::SeqPacket`] channels (LE L2CAP +Connection-Capable Channel): each received SDU is exactly one frame. The MTU is +clamped to [`MAX_FRAME`], so a single SDU is a single wire envelope — no +length-prefix framing is needed. + +## Roles + +Every node advertises **and** scans. To keep two nodes from both dialing, the +numerically-smaller [`NodeId`] acts as the BLE central (initiator) and the +larger accepts on the [`SeqPacketListener`] bound to PSM `0x008D`. The listener +runs for the whole session. + +## Binding + +The BLE address is only a routing key (it rotates), so a connection is bound to +the peer's advertised [`NodeId`] instead: + +- the scanner decodes the peer's [`NodeId`] from a tines advertisement — the + `Service Data — 16-bit UUID` record must carry the tines service UUID + (`0xFE7E`), which is checked by [`adv::decode_ad_node_id`]; +- the decoded `bt address → NodeId` pair is recorded in the scan map; +- an **inbound** connection is bound by looking its peer address up in that map + — an inbound from an unknown address is rejected immediately; +- an **outbound** connection dials using the discovered address *type*, not a + hard-coded one. + +A frame's declared sender is then checked against the bound [`NodeId`]: the +transport peeks the `from` field with [`wire::frame_sender`] and drops any frame +whose declared sender differs from the bound peer (an address-spoofing +violation). + +## Lifecycle + +- connection up → `Event::Contact(addr)`; +- connection end (EOF, error, or adapter loss) → `Event::Lost(addr)`; +- scanning and listening continue after a loss. + +## Multi-peer + +Up to [`MAX_PEERS`] connections may be live at once, one per bound [`NodeId`]; +a second connection to the same [`NodeId`] replaces the first. + +## Manual verification + +The SeqPacket framing can't be exercised in CI (it needs two real radios), so +verify it by hand with two devices running the `tines` service: + +1. start `tines` on both devices with distinct `NodeId`s; +2. confirm the smaller-`NodeId` device dials and the larger accepts (check the + `tines-bt: connected to … (central)` / `accepted connection from … (bound)` + trace lines); +3. send a burst of observations on one device and confirm the other receives + each as a separate, intact frame (no merged or truncated frames). + +[`Transport`]: https://docs.rs/tines-core/latest/tines_core/transport/trait.Transport.html +[`NodeId`]: https://docs.rs/tines-core/latest/tines_core/observation/struct.NodeId.html +[`Radio`]: crate::Radio +[`BluerRadio`]: crate::BluerRadio +[`adv::decode_ad_node_id`]: crate::adv::decode_ad_node_id +[`wire::frame_sender`]: https://docs.rs/tines-core/latest/tines_core/wire/fn.frame_sender.html +[`MAX_FRAME`]: https://docs.rs/tines-core/latest/tines_core/consts/constant.MAX_FRAME.html +[`MAX_PEERS`]: https://docs.rs/tines-core/latest/tines_core/consts/constant.MAX_PEERS.html +[`l2cap::SeqPacket`]: https://docs.rs/bluer/latest/bluer/l2cap/struct.SeqPacket.html +[`SeqPacketListener`]: https://docs.rs/bluer/latest/bluer/l2cap/struct.SeqPacketListener.html diff --git a/tines-bt/src/adv.rs b/tines-bt/src/adv.rs index 476aa4ec..85669916 100644 --- a/tines-bt/src/adv.rs +++ b/tines-bt/src/adv.rs @@ -10,8 +10,12 @@ //! of the service-data record is exactly what [`decode_node_id`] does — the //! invariant the round-trip test pins. +use std::collections::{BTreeSet, HashMap}; + use tines_core::observation::NodeId; +use bluer::{Uuid, UuidExt}; + /// The LE flags byte carried in the flags AD record: LE General Available and /// BR/EDR not supported (`0x06`). pub const ADV_FLAGS: u8 = 0x06; @@ -109,6 +113,34 @@ pub fn decode_node_id(payload: &[u8]) -> Option { None } +/// Decode a [`NodeId`] advertised by a tines advertiser. +/// +/// Returns `None` if the advertisement does not carry the tines service UUID, +/// if no [`NodeId`] can be parsed, or if the advertiser is this local +/// [`NodeId`] (a node advertises *and* scans, so it never dials itself). +/// +/// Unlike [`decode_node_id`], the 2-byte service UUID is *verified* here, so a +/// frame from a non-tines broadcaster is rejected. +/// +/// A non-empty `allow` narrows decoding to a whitelist of [`NodeId`]s; an empty +/// `allow` accepts any tines advertiser (this is the default once the +/// transport binds by NodeId). +pub fn decode_ad_node_id( + service_data: &HashMap>, + me: NodeId, + allow: &BTreeSet, +) -> Option { + let advertised = service_data.get(&Uuid::from_u16(SERVICE_UUID))?; + let node: NodeId = advertised.as_slice().try_into().ok()?; + if node == me { + return None; + } + if !allow.is_empty() && !allow.contains(&node) { + return None; + } + Some(node) +} + #[cfg(test)] mod tests { use super::*; diff --git a/tines-bt/src/lib.rs b/tines-bt/src/lib.rs index 84e6e1a6..2b5c83a7 100644 --- a/tines-bt/src/lib.rs +++ b/tines-bt/src/lib.rs @@ -18,4 +18,6 @@ mod transport; pub use adv::{ advertise, decode_node_id, ADVERTISING_BUFFER_LEN, ADV_FLAGS, NODE_ID_LEN, SERVICE_UUID, }; -pub use transport::{is_central, BluerRadio, BtTransport, Radio, RadioError, RadioEvent}; +pub use transport::{ + is_central, resolve_bound, BluerRadio, BtTransport, Radio, RadioError, RadioEvent, +}; diff --git a/tines-bt/src/transport.rs b/tines-bt/src/transport.rs index 808fb627..5c05c029 100644 --- a/tines-bt/src/transport.rs +++ b/tines-bt/src/transport.rs @@ -3,14 +3,60 @@ //! //! Real radio I/O sits behind the small [`Radio`] trait, implemented over //! `bluer`/BlueZ in [`BluerRadio`]. Everything here that does not touch the -//! radio — the from/`Addr` mismatch-drop rule and the central-peripheral role -//! decision — is pure and unit-testable without hardware. +//! radio — the from/`Addr` mismatch-drop rule, the central-peripheral role +//! decision, and the inbound-address → `NodeId` binding — is pure and +//! unit-testable without hardware. +//! +//! ## Framing +//! +//! The connection is a `bluer` [`l2cap::SeqPacket`] (a LE L2CAP Connection +//! Capable Channel): each received SDU is exactly one frame (one +//! [`RadioEvent::Data`]). The MTU is clamped to [`MAX_FRAME`] so one SDU is one +//! envelope. +//! +//! ## Roles +//! +//! Every node advertises *and* scans. To avoid two nodes both dialing, the +//! numerically-smaller [`NodeId`] dials (the central) and the larger accepts on +//! the [`SeqPacketListener`] bound to `TINES_L2CAP_PSM`. +//! +//! ## Binding +//! +//! The BLE address is only a routing key (it rotates), so the transport binds a +//! connection to the peer's advertised [`NodeId`]: the scanner decodes the +//! [`NodeId`] from a tines advertisement (UUID-checked via [`adv::decode_ad_node_id`]) +//! and records `bt address → NodeId`; inbound connections are bound via the +//! scanned address (an unknown address is rejected). +//! +//! ## Multi-peer +//! +//! Up to [`MAX_PEERS`] connections may be live at once, one per bound +//! [`NodeId`]. A second connection to the same [`NodeId`] replaces the first. +//! The configured `peer` becomes an optional allowlist; empty means "any tines +//! advertiser". -use std::boxed::Box; +use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::sync::{ + atomic::{AtomicU64, AtomicUsize, Ordering}, + Arc, Mutex, +}; +use std::thread; -use tines_core::consts::MAX_FRAME; +use futures::StreamExt; +use tokio::sync::mpsc::{self as async_mpsc, UnboundedReceiver, UnboundedSender}; +use tokio::time::Duration; + +use bluer::{ + adv, + l2cap::{self, SeqPacket, SeqPacketListener, SocketAddr}, + monitor::{self, Monitor, MonitorEvent, MonitorHandle, Pattern, RssiSamplingPeriod}, + Adapter, Address, AddressType, Session, Uuid, UuidExt, +}; + +use tines_core::consts::{MAX_FRAME, MAX_PEERS}; use tines_core::observation::NodeId; use tines_core::transport::{Event, FrameBuf, SendError, Transport}; +use tines_core::wire::frame_sender; /// The radio I/O seam for [`BtTransport`]. /// @@ -26,8 +72,8 @@ pub trait Radio { /// device's advertised [`NodeId`]). fn send(&mut self, me: NodeId, to: &NodeId, data: &[u8]) -> Result<(), RadioError>; - /// The current usable MTU in bytes (L2CAP Connection-Capable Channel when - /// available, else the negotiated ATT MTU on the GATT fallback). + /// The current usable MTU in bytes (the L2CAP Connection-Capable Channel's + /// negotiated MTU, clamped to [`MAX_FRAME`]). fn mtu(&self) -> usize; } @@ -40,8 +86,16 @@ pub enum RadioEvent { /// The frame bytes carried by the event. data: Vec, }, - /// The peer disconnected; any prior [`Event::Lost`] address is now gone. - Disconnected, + /// A connection to `addr` became live. + Contact { + /// The bound [`NodeId`] (the peer's advertised [`NodeId`]). + addr: NodeId, + }, + /// The connection to `addr` ended (EOF, error, or adapter loss). + Lost { + /// The bound [`NodeId`] whose connection ended. + addr: NodeId, + }, } /// Why a [`Radio::send`] call can fail. @@ -95,21 +149,40 @@ pub fn is_central(me: NodeId, peer: NodeId) -> bool { me < peer } +/// Resolve the bound [`NodeId`] for an inbound BLE `addr` from the scan map. +/// +/// Returns `None` if the address was not seen advertising the tines service +/// UUID — an inbound connection from an unknown address binds to nothing and is +/// therefore rejected by the accept loop. +pub fn resolve_bound( + scan_map: &HashMap, + addr: &Address, +) -> Option { + scan_map.get(addr).map(|(peer, _)| *peer) +} + impl Transport for BtTransport { /// The peer's advertised [`NodeId`]; identity survives BLE address rotation. type Addr = NodeId; /// Poll for the next inbound event. /// - /// Inbound [`RadioEvent::Data`] frames whose `from` differs from the - /// connection's advertised [`NodeId`] are dropped here — a protocol - /// violation (address spoofing) at the transport, so core never observes a - /// mismatch. Frames too large for a [`FrameBuf`] are skipped rather than - /// aborting the poll. + /// Inbound [`RadioEvent::Data`] frames are checked with [`frame_sender`]: + /// the frame's *declared* sender (peeked from the envelope) must equal the + /// frame's claimed `from`, and the `from` must equal this transport's bound + /// [`NodeId`]. Both are dropped silently — the first is an address-spoofing + /// violation, the second a frame for a different connection — so core never + /// observes a mismatch. Frames too large for a [`FrameBuf`] are skipped + /// rather than aborting the poll. fn poll(&mut self) -> Option> { loop { match self.radio.poll() { Some(RadioEvent::Data { from, data }) => { + // From-check: the frame's declared sender (peeked) must equal + // the claimed `from`. A mismatch is spoofing — drop it. + if frame_sender(&data) != Some(from) { + continue; + } if from != self.peer { continue; } @@ -118,10 +191,15 @@ impl Transport for BtTransport { Err(_) => continue, } } - Some(RadioEvent::Disconnected) => { - let peer = self.peer; - self.peer = [0u8; 16]; - return Some(Event::Lost(peer)); + Some(RadioEvent::Contact { addr }) => { + if addr == self.peer { + return Some(Event::Contact(addr)); + } + } + Some(RadioEvent::Lost { addr }) => { + if addr == self.peer { + return Some(Event::Lost(addr)); + } } None => return None, } @@ -141,75 +219,59 @@ impl Transport for BtTransport { /// Maximum frame size in bytes this transport can send. /// - /// Reports the radio's real MTU (L2CAP CoC when available, else the ATT - /// MTU on the GATT fallback), clamped to `MAX_FRAME` per the - /// [`Transport`] contract. Core folds summaries and fragments down to this. + /// Reports the radio's real MTU (the L2CAP Connection-Capable Channel's + /// negotiated MTU, clamped to [`MAX_FRAME`]). Core folds summaries and + /// fragments down to this. fn mtu(&self) -> usize { self.radio.mtu().min(MAX_FRAME) } } -use std::collections::{BTreeMap, BTreeSet}; -use std::sync::{ - atomic::{AtomicUsize, Ordering}, - Arc, -}; -use std::thread; +/// Smallest MTU we are willing to report (ATT's default minimum is 23). +const MIN_MTU: usize = 23; -use futures::StreamExt; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::sync::mpsc::{self as async_mpsc, UnboundedReceiver, UnboundedSender}; - -use bluer::{ - adv, - l2cap::{SocketAddr, Stream as L2capStream}, - monitor::{self, Monitor, MonitorEvent, Pattern, RssiSamplingPeriod}, - AddressType, Uuid, UuidExt, -}; - -use crate::adv::SERVICE_UUID; - -/// Fixed L2CAP PSM for the tines byte pipe. Shared with the peer (agreed -/// out-of-band): the L2CAP stream connects on this PSM on both sides, so the -/// numerically smaller [`NodeId`] (the central) is the one that lands the -/// connection first. +/// L2CAP PSM the tines byte pipe runs on. Shared with the peer (agreed +/// out-of-band): both sides connect/listen on this PSM. const TINES_L2CAP_PSM: u16 = 0x008D; /// A [`bluer`]/BlueZ-backed [`Radio`]. /// /// Runs the async `bluer` D-Bus stack on a detached background thread: it /// powers the adapter, advertises this device's [`NodeId`] in service data, -/// monitors for a peer advertising the tines service UUID, and on discovery -/// opens an L2CAP byte pipe on `TINES_L2CAP_PSM`. Inbound bytes are surfaced -/// as [`RadioEvent::Data`] tagged with the peer's [`NodeId`] (which the -/// transport already knows — the BLE address is only how we route to it, and it -/// may rotate); outbound bytes are written to the pipe. The usable MTU is read -/// from the L2CAP socket and reported through [`Radio::mtu`]. +/// monitors for peers advertising the tines service UUID, and — critically — +/// *both* advertises and listens on `TINES_L2CAP_PSM` so either node can be +/// the initiator. On discovery it opens an L2CAP [`SeqPacket`] connection, +/// binds it to the peer's [`NodeId`] (from the advertisement), and replaces any +/// prior connection to that NodeId. Inbound bytes are surfaced as +/// [`RadioEvent::Data`] tagged with the bound [`NodeId`]; outbound bytes are +/// routed to the matching connection. /// /// The D-Bus session lives for the lifetime of this radio; [`BluerRadio`] /// detaches (rather than joins) its background thread on drop, because the run /// loop blocks on BLE activity for as long as the transport is alive. -/// -/// This backend establishes the L2CAP pipe from the initiating side (the smaller -/// [`NodeId`], per [`is_central`]); callers construct the transport only for the -/// peer they intend to talk to. pub struct BluerRadio { event_rx: UnboundedReceiver, - send_tx: UnboundedSender>, - mtu: std::sync::Arc, + send_tx: UnboundedSender<(NodeId, Vec)>, + mtu: Arc, handle: Option>, } impl BluerRadio { - /// Start the `bluer`/BlueZ radio for a connection between `me` (this device) - /// and `peer` (the peer's advertised [`NodeId`]). + /// Start the `bluer`/BlueZ radio for a connection between `me` (this + /// device) and `peer` (the peer's advertised [`NodeId`]). + /// + /// `peer` is this node's allowlist: only that NodeId's connections are + /// accepted. With an empty allowlist the transport accepts any tines + /// advertiser. pub fn new(me: NodeId, peer: NodeId) -> Self { let (event_tx, event_rx) = async_mpsc::unbounded_channel::(); - let (send_tx, send_rx) = async_mpsc::unbounded_channel::>(); + let (send_tx, send_rx) = async_mpsc::unbounded_channel::<(NodeId, Vec)>(); // Default to the transport's frame limit until the L2CAP socket reports a // negotiated MTU; [`BtTransport::mtu`] clamps to MAX_FRAME regardless. let mtu = Arc::new(AtomicUsize::new(MAX_FRAME)); let mtu_for_run = Arc::clone(&mtu); + // This node only accepts connections to `peer`. + let allow = Arc::new(BTreeSet::from([peer])); let handle = thread::Builder::new() .name(format!("tines-bt-bluer-{}", node_hex(peer))) @@ -218,7 +280,11 @@ impl BluerRadio { .enable_all() .build() .expect("failed to build bluer runtime"); - let _ = rt.block_on(run(me, peer, event_tx, send_rx, mtu_for_run)); + if let Err(e) = rt.block_on(run(me, allow, send_rx, mtu_for_run, event_tx)) { + log::error!("tines-bt: radio thread exited with error: {e}"); + } else { + log::info!("tines-bt: radio thread exited"); + } }) .expect("failed to spawn bluer radio thread"); @@ -244,9 +310,9 @@ impl Radio for BluerRadio { self.event_rx.try_recv().ok() } - fn send(&mut self, _me: NodeId, _to: &NodeId, data: &[u8]) -> Result<(), RadioError> { + fn send(&mut self, _me: NodeId, to: &NodeId, data: &[u8]) -> Result<(), RadioError> { self.send_tx - .send(data.to_vec()) + .send((*to, data.to_vec())) .map_err(|_| RadioError::Unreachable) } @@ -255,124 +321,363 @@ impl Radio for BluerRadio { } } -/// Background async work: session, advertising, LE discovery, and the L2CAP -/// byte pipe. Returns once the connection drops. +/// Per-connection state held while the connection is live. +struct Conn { + socket: Arc, + /// Negotiated L2CAP MTU (bytes) — clamped at connect time. + mtu: u16, + /// Monotonic connection nonce; a newer connection to the same NodeId gets a + /// higher nonce, so a stale read loop never removes the live one. + nonce: u64, +} + +/// State shared across a single radio's background tasks. +/// +/// The fields are individually `Arc`-wrapped so the per-task tasks share them; +/// the struct itself is also wrapped in `Arc` so it can be handed off to the +/// detached `tokio` tasks that own the BLE session. +struct SessionState { + /// Scanned `bt address → (bound NodeId, address type)`, for binding inbound + /// connections and dialing outbound ones. + scan_map: Arc>>, + /// Live connections, one per bound [`NodeId`] (a second replaces the first). + conns: Arc>>, + /// The negotiated L2CAP MTU. + mtu: Arc, + /// Allowed [`NodeId`]s; empty means "any tines advertiser". + allow: Arc>, + /// Monotonic counter assigning a unique nonce to each connection. + nonces: Arc, + /// Fan-out channel for events; owned by the read loops and the write loop. + events_tx: UnboundedSender, + /// The adapter — used to look up discovered devices' service data. + adapter: Arc, + /// This device's advertised [`NodeId`]. + me: NodeId, +} + +/// Background async work: session setup, advertising, LE discovery, listening, +/// connecting, and the L2CAP byte pipes. Returns once every task has exited. async fn run( me: NodeId, - peer: NodeId, - event_tx: UnboundedSender, - send_rx: UnboundedReceiver>, + allow: Arc>, + outbound: UnboundedReceiver<(NodeId, Vec)>, mtu: Arc, + events_tx: UnboundedSender, ) -> Result<(), bluer::Error> { - let session = bluer::Session::new().await?; - let adapter = session.default_adapter().await?; + let session = Session::new().await?; + let adapter = Arc::new(session.default_adapter().await?); + let state = Arc::new(SessionState { + scan_map: Arc::new(Mutex::new(HashMap::new())), + conns: Arc::new(Mutex::new(HashMap::new())), + mtu: Arc::clone(&mtu), + allow, + nonces: Arc::new(AtomicU64::new(1)), + events_tx, + adapter: Arc::clone(&adapter), + me, + }); + + // Power the adapter. A failure here is fatal: nothing else can work, so it + // is returned (and logged by the radio thread) rather than swallowed. if let Err(e) = adapter.set_powered(true).await { - log::warn!("tines-bt: failed to power adapter: {e:?}"); - return Ok(()); + log::error!("tines-bt: failed to power the Bluetooth adapter: {e}"); + return Err(e); } // Advertise our own NodeId in service data so a peer can recognise us and - // learn our identity before connecting (address rotation aside). + // learn our identity before connecting (address rotation aside). Advertising + // is symmetric — the peer will connect to us on our listener if we cannot + // dial — so a failure is surfaced at error level but not fatal. let advertisement = adv::Advertisement { advertisement_type: adv::Type::Peripheral, - service_uuids: BTreeSet::from([Uuid::from_u16(SERVICE_UUID)]), - service_data: BTreeMap::from([(Uuid::from_u16(SERVICE_UUID), me.to_vec())]), + service_uuids: BTreeSet::from([Uuid::from_u16(crate::adv::SERVICE_UUID)]), + service_data: BTreeMap::from([(Uuid::from_u16(crate::adv::SERVICE_UUID), me.to_vec())]), discoverable: Some(true), ..Default::default() }; if adapter.advertise(advertisement).await.is_err() { - log::warn!("tines-bt: failed to advertise; continuing with discovery"); + log::error!("tines-bt: advertising failed; continuing with listening and discovery"); } // LE advertisement monitor: a peer advertising the tines service UUID. The - // monitor yields the peer's (rotating) BLE address; the peer's identity is - // the [`NodeId`] we already know (`peer`), which is what tags inbound frames. + // monitor yields the peer's (rotating) BLE address; we look the peer up as + // a `Device`, decode its [`NodeId`] (UUID-checked), and record it. let manager = adapter.monitor().await?; - let mut monitor = manager + let monitor = manager .register(Monitor { monitor_type: monitor::Type::OrPatterns, rssi_sampling_period: Some(RssiSamplingPeriod::First), patterns: Some(vec![Pattern { data_type: 0x16, // Service Data — 16-bit UUID start_position: 0, - content: SERVICE_UUID.to_le_bytes().to_vec(), + content: crate::adv::SERVICE_UUID.to_le_bytes().to_vec(), }]), ..Default::default() }) .await?; - while let Some(event) = monitor.next().await { - if let MonitorEvent::DeviceFound(found) = event { - let addr = found.device; - let sa = SocketAddr::new(addr, AddressType::LePublic, TINES_L2CAP_PSM); - match L2capStream::connect(sa).await { - Ok(stream) => { - if let Ok(mtu_bytes) = stream.as_ref().recv_mtu() { - let m = mtu_bytes as usize; - mtu.store(m.clamp(MIN_MTU, MAX_FRAME), Ordering::Relaxed); - } - let (read_half, write_half) = stream.into_split(); - let read_join = tokio::spawn(async move { - read_loop(peer, read_half, event_tx).await; - }); - let write_join = tokio::spawn(async move { - write_loop(write_half, send_rx).await; - }); - // The pipe is up. Block until either loop ends (disconnect), - // then tear down so the runtime can stop. - tokio::select! { - _ = read_join => {} - _ = write_join => {} - } - return Ok(()); - } - Err(e) => { - log::trace!("tines-bt: connect to {addr:?} failed: {e:?}"); - } - } - } + // Bind the listener (the accept role) on our PSM. A failure to bind means we + // cannot accept, so it is fatal. + let listener = l2cap::SeqPacketListener::bind(SocketAddr::new( + Address::any(), + AddressType::LePublic, + TINES_L2CAP_PSM, + )) + .await + .map_err(|e| { + log::error!( + "tines-bt: failed to bind the SeqPacket listener on PSM {TINES_L2CAP_PSM:#06X}: {e}" + ); + e + })?; + + let monitor_state = Arc::clone(&state); + let accept_state = Arc::clone(&state); + let connect_state = Arc::clone(&state); + let write_state = Arc::clone(&state); + let monitor_join = tokio::spawn(async move { + monitor_task(monitor, monitor_state).await; + }); + let accept_join = tokio::spawn(async move { + accept_loop(listener, accept_state).await; + }); + let connect_join = tokio::spawn(async move { + connect_loop(connect_state).await; + }); + let write_join = tokio::spawn(async move { + write_loop(write_state, outbound).await; + }); + + tokio::select! { + _ = monitor_join => {} + _ = accept_join => {} + _ = connect_join => {} + _ = write_join => {} } Ok(()) } -/// Read inbound bytes off the L2CAP pipe and emit them as [`RadioEvent::Data`], -/// tagged with the peer's [`NodeId`]. -async fn read_loop( - peer: NodeId, - mut read_half: P, - tx: UnboundedSender, -) { - let mut buf = vec![0u8; MAX_FRAME]; - loop { - match read_half.read(&mut buf).await { - Ok(0) => break, - Ok(n) => { - if tx - .send(RadioEvent::Data { - from: peer, - data: buf[..n].to_vec(), - }) - .is_err() - { - break; +/// Drive the LE advertisement monitor, recording discovered tines peers in the +/// scan map and forgetting them when they stop advertising. +async fn monitor_task(mut monitor: MonitorHandle, state: Arc) { + while let Some(event) = monitor.next().await { + match event { + MonitorEvent::DeviceFound(found) => { + let addr = found.device; + if let Some((peer, atype)) = discover_peer(addr, state.clone()).await { + state.scan_map.lock().unwrap().insert(addr, (peer, atype)); + log::trace!("tines-bt: discovered tines peer {peer:?} at {addr}"); } } - Err(_) => break, + MonitorEvent::DeviceLost(found) => { + state.scan_map.lock().unwrap().remove(&found.device); + } + // Other monitor events (RSSI changes, etc.) are ignored. + _ => {} } } } -/// Drain outbound bytes off the queue and write them to the L2CAP pipe. -async fn write_loop(mut write_half: P, mut rx: UnboundedReceiver>) { - while let Some(bytes) = rx.recv().await { - if write_half.write_all(&bytes).await.is_err() { +/// Look a discovered `addr` up as a `Device`, decode its [`NodeId`] (checking +/// the tines service UUID and this node's allowlist), and return it with the +/// device's address type. Returns `None` for non-tines or disallowed advertisers. +async fn discover_peer(addr: Address, state: Arc) -> Option<(NodeId, AddressType)> { + let device = state.adapter.device(addr).ok()?; + let service_data = match device.service_data().await.ok()? { + Some(service_data) => service_data, + None => return None, + }; + let peer = crate::adv::decode_ad_node_id(&service_data, state.me, &state.allow)?; + let atype = device.address_type().await.ok()?; + Some((peer, atype)) +} + +/// The accept role: bind `listener` on our PSM, and for each accepted connection +/// resolve the peer's bound [`NodeId`] and replace any prior connection to it. +async fn accept_loop(listener: SeqPacketListener, state: Arc) { + loop { + match listener.accept().await { + Ok((socket, peer_sa)) => { + let peer_addr = peer_sa.addr; + let bound = resolve_bound(&state.scan_map.lock().unwrap(), &peer_addr); + match bound { + Some(peer) => { + if state.allow.contains(&peer) { + // A second connection to an already-bound NodeId + // replaces the first (handled inside register_and_read); + // a brand-new peer is only admitted while under the + // concurrent-connection cap. + let under_cap = state.conns.lock().unwrap().contains_key(&peer) + || state.conns.lock().unwrap().len() < MAX_PEERS; + if under_cap { + register_and_read(state.clone(), peer, socket).await; + log::trace!("tines-bt: accepted connection from {peer:?} (bound)"); + } else { + log::trace!("tines-bt: peer {peer:?} over MAX_PEERS ({MAX_PEERS}); closing new connection"); + } + } else { + log::trace!("tines-bt: rejecting inbound from {peer:?} (not allowed)"); + } + } + None => { + log::trace!("tines-bt: rejecting inbound from unknown address {peer_addr}"); + } + } + } + Err(e) => { + log::error!("tines-bt: SeqPacketListener accept error: {e}"); + return; + } + } + } +} + +/// The central/initiator role: periodically dial peers we have discovered that +/// this node should connect to (smaller [`NodeId`]) and are not already live. +async fn connect_loop(state: Arc) { + let mut ticker = tokio::time::interval(Duration::from_millis(500)); + loop { + ticker.tick().await; + + // Collect eligible candidates without holding any lock across an await. + let candidates = { + let scan = state.scan_map.lock().unwrap(); + let allow = &*state.allow; + scan.iter() + .filter(|(_addr, (peer, _))| ad_eligible(*peer, state.me, allow)) + .map(|(addr, (peer, atype))| (*addr, *peer, *atype)) + .collect::>() + }; + + for (addr, peer, atype) in candidates { + if !is_central(state.me, peer) { + continue; + } + if state.conns.lock().unwrap().contains_key(&peer) { + continue; + } + if state.conns.lock().unwrap().len() >= MAX_PEERS { + log::trace!( + "tines-bt: connected peers over MAX_PEERS ({MAX_PEERS}); skipping {peer:?}" + ); + continue; + } + let sa = SocketAddr::new(addr, atype, TINES_L2CAP_PSM); + match SeqPacket::connect(sa).await { + Ok(socket) => { + register_and_read(state.clone(), peer, socket).await; + log::trace!("tines-bt: connected to {peer:?} at {addr} (central)"); + } + Err(e) => { + log::trace!("tines-bt: connect to {peer:?} failed: {e}"); + } + } + } + } +} + +/// Insert `socket` for `peer` into the connection map, replacing any prior +/// connection, and return its [`Arc`] and nonce. +async fn register_connection( + state: Arc, + peer: NodeId, + socket: SeqPacket, +) -> (Arc, u64) { + let mtu = mtu_bytes(&socket); + let nonce = state.nonces.fetch_add(1, Ordering::Relaxed); + let socket = Arc::new(socket); + if let Some(old) = state.conns.lock().unwrap().insert( + peer, + Conn { + socket: socket.clone(), + mtu, + nonce, + }, + ) { + // The replaced connection is no longer referenced here; its read loop's + // nonce check surfaces the Lost event for the stale connection. + drop(old.socket); + } + state.mtu.store(mtu as usize, Ordering::Relaxed); + (socket, nonce) +} + +/// Register `socket` for `peer` (replacing any prior connection) and spawn its +/// read loop. +async fn register_and_read(state: Arc, peer: NodeId, socket: SeqPacket) { + let (conn, nonce) = register_connection(state.clone(), peer, socket).await; + tokio::spawn(async move { + read_loop(peer, conn, state, nonce).await; + }); +} + +/// The read loop for a single connection: read one SDU at a time (one frame +/// each) and emit [`RadioEvent::Data`]. On EOF/error it removes the connection +/// unless a newer connection with a higher nonce has replaced it, then emits +/// [`RadioEvent::Lost`]. +async fn read_loop(peer: NodeId, socket: Arc, state: Arc, nonce: u64) { + let mut buf = vec![0u8; state.mtu.load(Ordering::Relaxed)]; + while let Ok(n) = socket.recv(&mut buf).await { + if n == 0 { + break; + } + if state + .events_tx + .send(RadioEvent::Data { + from: peer, + data: buf[..n].to_vec(), + }) + .is_err() + { break; } } + + // Connection ended. Remove it only if it is still the live one for this + // NodeId (a newer connection replaced it, so leave that one in place). + let mut guard = state.conns.lock().unwrap(); + if guard.get(&peer).map(|c| c.nonce) == Some(nonce) { + guard.remove(&peer); + let _ = state.events_tx.send(RadioEvent::Lost { addr: peer }); + } } -/// Smallest MTU we are willing to report (ATT's default minimum is 23). -const MIN_MTU: usize = 23; +/// Drain outbound frames and write each to its destination connection. +async fn write_loop(state: Arc, mut outbound: UnboundedReceiver<(NodeId, Vec)>) { + while let Some((to, data)) = outbound.recv().await { + // Read the connection (dropping the guard before awaiting so it isn't + // held across the send's await point — a MutexGuard is not Send). + let (socket, mtu) = match state.conns.lock().unwrap().get(&to) { + Some(conn) => (Some(Arc::clone(&conn.socket)), conn.mtu as usize), + None => (None, 0), + }; + if let Some(socket) = socket { + if socket.send(&data[..mtu.min(data.len())]).await.is_err() { + log::trace!("tines-bt: send to {to:?} failed"); + } + } else { + log::trace!("tines-bt: no connection to {to:?}; dropping frame"); + } + } +} + +/// Whether the advertisement for `peer` is eligible for this node to consider. +/// +/// A node never dials itself, and only considers peers in the allowlist (an +/// empty allowlist accepts any tines advertiser — the NodeId check is already +/// done by [`crate::adv::decode_ad_node_id`]). +fn ad_eligible(peer: NodeId, me: NodeId, allow: &BTreeSet) -> bool { + peer != me && (allow.is_empty() || allow.contains(&peer)) +} + +/// The negotiated L2CAP MTU for `socket`, clamped to our limits. +fn mtu_bytes(socket: &SeqPacket) -> u16 { + socket + .send_mtu() + .map(|m| m.clamp(MIN_MTU, MAX_FRAME) as u16) + .unwrap_or(MAX_FRAME as u16) +} /// Render a [`NodeId`] as hex for thread naming (avoids pulling in a hex crate). fn node_hex(node: NodeId) -> String { diff --git a/tines-bt/tests/adv.rs b/tines-bt/tests/adv.rs index b1597126..0f9d3679 100644 --- a/tines-bt/tests/adv.rs +++ b/tines-bt/tests/adv.rs @@ -73,55 +73,70 @@ fn advertising_payload_roundtrips_node_id() { assert_eq!(adv::decode_node_id(&p), Some(node)); } +/// A framed envelope: one tag byte, then the 16-byte `from`. This is the layout +/// `frame_sender` peeks — the transport delivers this whole envelope to core +/// (which decodes it), so the `from` must be stamped with the sender's id. +fn frame(from: NodeId, payload: &[u8]) -> Vec { + let mut b = Vec::with_capacity(1 + 16 + payload.len()); + b.push(0u8); // version/tag byte + b.extend_from_slice(&from); + b.extend_from_slice(payload); + b +} + #[test] fn from_mismatch_is_dropped() { let me: NodeId = [1u8; 16]; - let peer: NodeId = [2u8; 16]; // the connection's advertised NodeId - let spoof: NodeId = [9u8; 16]; // claimed source that does not match `peer` + let peer: NodeId = [2u8; 16]; // the connection's advertised NodeId (bound Addr) + let spoof: NodeId = [9u8; 16]; // a declared sender that does not match `peer` - // A single spoofed frame: the transport drops it and emits nothing. - let mut radio = MockRadio::new(23); - radio.push(RadioEvent::Data { - from: spoof, - data: vec![0xCA, 0xFE], - }); - let mut t = BtTransport::new(me, peer, radio); - assert_eq!( - t.poll(), - None, - "a frame from a non-advertised id yields no Event::Frame" - ); - - // A legitimate frame from `peer` is delivered unchanged. + // A legitimate frame from `peer` (envelope stamped with `peer`) is delivered + // unchanged, envelope included — core decodes the whole thing. let mut good = MockRadio::new(23); good.push(RadioEvent::Data { from: peer, - data: vec![0x01, 0x02], + data: frame(peer, &[0x01, 0x02]), }); let mut t = BtTransport::new(me, peer, good); match t.poll() { Some(Event::Frame(from, data)) => { assert_eq!(from, peer); - assert_eq!(&data[..], &[0x01, 0x02]); + assert_eq!(&data[..], &frame(peer, &[0x01, 0x02])[..]); } other => panic!("expected a Frame from {peer:?}, got {other:?}"), } - // And a spoofed frame sitting before a legit one is simply skipped. + // A frame whose envelope `from` (peeked by frame_sender) is `spoof`, not the + // bound peer, is dropped by the from-check — even though the event claims the + // bound peer as its source. + let mut mismatched = MockRadio::new(23); + mismatched.push(RadioEvent::Data { + from: peer, + data: frame(spoof, &[0xCA, 0xFE]), + }); + let mut t = BtTransport::new(me, peer, mismatched); + assert_eq!( + t.poll(), + None, + "a frame whose decoded `from` differs from the bound Addr is dropped" + ); + + // And a mismatched frame sitting before a legit one is simply skipped; the + // legit one is delivered. let mut mixed = MockRadio::new(23); mixed.push(RadioEvent::Data { - from: spoof, - data: vec![0xFF], + from: peer, + data: frame(spoof, &[0xFF]), }); mixed.push(RadioEvent::Data { from: peer, - data: vec![0x03], + data: frame(peer, &[0x03]), }); let mut t = BtTransport::new(me, peer, mixed); match t.poll() { Some(Event::Frame(from, data)) => { assert_eq!(from, peer); - assert_eq!(&data[..], &[0x03]); + assert_eq!(&data[..], &frame(peer, &[0x03])[..]); } other => panic!("expected the Frame from {peer:?}, got {other:?}"), } diff --git a/tines-bt/tests/transport.rs b/tines-bt/tests/transport.rs new file mode 100644 index 00000000..de4ab7a3 --- /dev/null +++ b/tines-bt/tests/transport.rs @@ -0,0 +1,258 @@ +//! Integration tests for the tines-bt transport, covering the F3 properties: +//! the `from`/`Addr` check, `Contact`/`Lost`, multi-peer filtering, NodeId +//! binding (UUID-checked advertisement + inbound-address lookup), and the +//! central-peripheral role decision. +//! +//! Every test uses a mock [`Radio`] and never touches a real BLE radio, so the +//! suite runs on a CI box with no Bluetooth stack. + +use std::collections::{BTreeSet, HashMap, VecDeque}; + +use tines_bt::{adv, resolve_bound, BtTransport, Radio, RadioError, RadioEvent}; +use tines_core::observation::NodeId; +use tines_core::transport::{Event, Transport}; + +use bluer::{Address, AddressType, Uuid, UuidExt}; + +/// A canned [`Radio`] backed by queued events and a configurable MTU. +struct MockRadio { + events: VecDeque, + _sent: Vec<(NodeId, Vec)>, + mtu: usize, +} + +impl MockRadio { + fn new(mtu: usize) -> Self { + MockRadio { + events: VecDeque::new(), + _sent: Vec::new(), + mtu, + } + } + + fn push(&mut self, ev: RadioEvent) { + self.events.push_back(ev); + } +} + +impl Radio for MockRadio { + fn poll(&mut self) -> Option { + self.events.pop_front() + } + + fn send(&mut self, _me: NodeId, to: &NodeId, data: &[u8]) -> Result<(), RadioError> { + self._sent.push((*to, data.to_vec())); + Ok(()) + } + + fn mtu(&self) -> usize { + self.mtu + } +} + +impl Drop for MockRadio { + fn drop(&mut self) { + // Drain the queue so a caller that only polls a couple of times does not + // leave events stranded behind the dead drop. + self.events.clear(); + } +} + +/// Build a minimal frame: one leading tag byte, then the 16-byte `from`. +/// `frame_sender` reads `from` from bytes 1..17, so this matches the wire. +fn frame(from: NodeId) -> Vec { + let mut b = vec![0u8; 1 + 16]; + b[1..1 + 16].copy_from_slice(&from); + b +} + +#[test] +fn from_check_and_peer_filter_drop_frames() { + let me: NodeId = [1u8; 16]; + let peer: NodeId = [2u8; 16]; // the connection's advertised NodeId (bound Addr) + let spoof: NodeId = [9u8; 16]; // a declared sender that is not the bound peer + let other: NodeId = [3u8; 16]; // a different bound peer entirely + + // A clean frame from `peer` is delivered unchanged. + let mut good = MockRadio::new(23); + good.push(RadioEvent::Data { + from: peer, + data: frame(peer), + }); + let mut t = BtTransport::new(me, peer, good); + match t.poll() { + Some(Event::Frame(from, data)) => { + assert_eq!(from, peer); + assert_eq!(&data[..], &frame(peer)[..]); + } + other => panic!("expected a Frame from {peer:?}, got {other:?}"), + } + + // From-check violation: the frame's declared `from` (in the event) is the + // bound peer, but the frame's actual `from` bytes say `spoof`. This is an + // address-spoofing violation — dropped via the peeked `frame_sender`. + let mut spoofed = MockRadio::new(23); + spoofed.push(RadioEvent::Data { + from: peer, + data: frame(spoof), + }); + let mut t = BtTransport::new(me, peer, spoofed); + assert_eq!( + t.poll(), + None, + "a frame whose decoded `from` differs from the bound Addr is dropped" + ); + + // Peer filter: a frame tagged as coming from a *different* bound peer is + // dropped (a multi-peer radio emits events for every connection; the + // transport only surfaces its own bound peer). + let mut other_peer = MockRadio::new(23); + other_peer.push(RadioEvent::Data { + from: other, + data: frame(other), + }); + let mut t = BtTransport::new(me, peer, other_peer); + assert_eq!(t.poll(), None, "a frame from a non-bound peer is dropped"); + + // A spoofed frame sitting before a good one is simply skipped; the good one + // is delivered. + let mut mixed = MockRadio::new(23); + mixed.push(RadioEvent::Data { + from: peer, + data: frame(spoof), + }); + mixed.push(RadioEvent::Data { + from: peer, + data: frame(peer), + }); + let mut t = BtTransport::new(me, peer, mixed); + match t.poll() { + Some(Event::Frame(from, data)) => { + assert_eq!(from, peer); + assert_eq!(&data[..], &frame(peer)[..]); + } + other => panic!("expected the clean Frame from {peer:?}, got {other:?}"), + } +} + +#[test] +fn contact_and_lost_emitted() { + let me: NodeId = [1u8; 16]; + let peer: NodeId = [2u8; 16]; + + let mut radio = MockRadio::new(23); + radio.push(RadioEvent::Contact { addr: peer }); + radio.push(RadioEvent::Data { + from: peer, + data: frame(peer), + }); + radio.push(RadioEvent::Lost { addr: peer }); + + let mut t = BtTransport::new(me, peer, radio); + + assert_eq!(t.poll(), Some(Event::Contact(peer)), "up → Contact"); + assert!(matches!(t.poll(), Some(Event::Frame(..))), "a frame passes"); + assert_eq!(t.poll(), Some(Event::Lost(peer)), "end → Lost"); + assert_eq!(t.poll(), None, "no more events once the connection is gone"); +} + +#[test] +fn frame_boundaries_preserved_at_transport() { + // Two back-to-back frames arrive as two separate Data events; the + // transport must surface them as two separate Frames, never merged. + let me: NodeId = [1u8; 16]; + let peer: NodeId = [2u8; 16]; + let frame_a = frame(peer); + let mut frame_b = frame(peer); + frame_b[0] = 0xFF; // differ from frame_a via the tag byte, so `from` stays == peer + let a_expected = frame_a.clone(); + let b_expected = frame_b.clone(); + + let mut radio = MockRadio::new(23); + radio.push(RadioEvent::Data { + from: peer, + data: frame_a, + }); + radio.push(RadioEvent::Data { + from: peer, + data: frame_b, + }); + + let mut t = BtTransport::new(me, peer, radio); + + match t.poll() { + Some(Event::Frame(_, a)) => assert_eq!(&a[..], &a_expected[..]), + other => panic!("expected the first frame, got {other:?}"), + } + match t.poll() { + Some(Event::Frame(_, b)) => assert_eq!(&b[..], &b_expected[..]), + other => panic!("expected the second frame, got {other:?}"), + } + assert_eq!(t.poll(), None); +} + +#[test] +fn is_central_by_node_id_order() { + let a: NodeId = [0u8; 16]; + let b: NodeId = [1u8; 16]; + let big: NodeId = [255u8; 16]; + assert!(tines_bt::is_central(a, b)); // smaller dials + assert!(!tines_bt::is_central(b, a)); + assert!(tines_bt::is_central(b, big)); + assert!(!tines_bt::is_central(big, b)); +} + +#[test] +fn resolve_bound_rejects_unknown_inbound() { + // An empty scan map: no address is eligible, so an inbound from an unknown + // address binds to nothing and must be rejected. + let empty: HashMap = HashMap::new(); + let unknown = Address::new([0xAA; 6]); + assert_eq!(resolve_bound(&empty, &unknown), None); + + // A peer we discovered and bound: its own address resolves. + let node: NodeId = [4u8; 16]; + let mut map = HashMap::new(); + map.insert(Address::new([0xBB; 6]), (node, AddressType::LePublic)); + let addr = Address::new([0xBB; 6]); + assert_eq!(resolve_bound(&map, &addr), Some(node)); +} + +#[test] +fn decode_ad_node_id_requires_tines_uuid() { + let me: NodeId = [7u8; 16]; + let node: NodeId = [5u8; 16]; + let allow: BTreeSet = BTreeSet::new(); // empty = any tines advertiser + + // An EIR-less service-data blob (wrong UUID) is rejected. + let wrong_uuid: HashMap> = + HashMap::from([(Uuid::from_u16(0x180F), node.to_vec())]); + assert_eq!(adv::decode_ad_node_id(&wrong_uuid, me, &allow), None); + + // No service data at all is rejected. + let empty: HashMap> = HashMap::new(); + assert_eq!(adv::decode_ad_node_id(&empty, me, &allow), None); + + // The tines UUID with a valid NodeId is decoded. + let good: HashMap> = + HashMap::from([(Uuid::from_u16(adv::SERVICE_UUID), node.to_vec())]); + assert_eq!(adv::decode_ad_node_id(&good, me, &allow), Some(node)); +} + +#[test] +fn decode_ad_node_id_honours_allowlist() { + let me: NodeId = [7u8; 16]; + let allowed: NodeId = [5u8; 16]; + let denied: NodeId = [6u8; 16]; + let service: HashMap> = + HashMap::from([(Uuid::from_u16(adv::SERVICE_UUID), allowed.to_vec())]); + + // Allowlist containing `allowed`: the advertiser is eligible. + let mut allow = BTreeSet::new(); + allow.insert(allowed); + assert_eq!(adv::decode_ad_node_id(&service, me, &allow), Some(allowed)); + + // Allowlist not containing this advertiser: rejected. + let deny: BTreeSet = BTreeSet::from([denied]); + assert_eq!(adv::decode_ad_node_id(&service, me, &deny), None); +} diff --git a/tines-core/src/wire.rs b/tines-core/src/wire.rs index 9b532bfe..2a6ba957 100644 --- a/tines-core/src/wire.rs +++ b/tines-core/src/wire.rs @@ -69,6 +69,34 @@ impl Frame { } } +/// Peek the [`NodeId`] a wire frame was sent by, without fully decoding it. +/// +/// A [`Frame`] envelope is an externally-tagged postcard enum: one tag byte at +/// index `0`, then its fields in declaration order. `from` is the first field, +/// so it occupies the 16 bytes at `bytes[1..17]`. This reads those 16 bytes and +/// returns `None` for anything shorter than a full envelope (so a truncated or +/// non-frame payload reports no sender rather than panicking). +/// +/// The leading byte is skipped on purpose: it is the externally-tagged variant +/// tag today and the reserved version-byte slot in 1.0, so reading `from` from +/// a fixed offset keeps this accessor correct when 1.0 prepends a version byte +/// ahead of the envelope. +/// +/// `tines-bt` uses this to enforce that a frame's declared sender matches the +/// NodeId advertised by its transport connection (the from-check), without +/// importing the private `Frame` layout. +pub fn frame_sender(bytes: &[u8]) -> Option { + // The envelope is one tag byte then the `from` field; `NodeId` is 16 bytes. + const ENVELOPE_OVERHEAD: usize = 1; + const NODE_ID_BYTES: usize = core::mem::size_of::(); + if bytes.len() < ENVELOPE_OVERHEAD + NODE_ID_BYTES { + return None; + } + let mut from = [0u8; NODE_ID_BYTES]; + from.copy_from_slice(&bytes[ENVELOPE_OVERHEAD..ENVELOPE_OVERHEAD + NODE_ID_BYTES]); + Some(from) +} + /// Errors produced while (de)serializing frames on the wire. #[derive(Debug, PartialEq, Eq)] pub enum WireError { @@ -229,6 +257,52 @@ mod tests { assert_eq!(Frame::decode(&GOLDEN).unwrap(), expected); } + #[test] + fn frame_sender_peeks_from_field() { + // The golden Summary encodes `from: [3;16]` starting at byte 1. + const GOLDEN: [u8; 200] = [ + 0u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, 3u8, + 239u8, 253u8, 182u8, 245u8, 13u8, 128u8, 4u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, 170u8, + 170u8, 170u8, + ]; + assert_eq!(frame_sender(&GOLDEN), Some([3u8; 16])); + } + + #[test] + fn frame_sender_reads_from_fixed_offset_not_just_the_start() { + // The leading byte (the externally-tagged variant tag) is skipped, so a + // payload whose first byte is not the sender must still yield the `from` + // recorded at byte 1. + let bytes = { + let mut b = [0u8; 32]; + b[0] = 0xFF; // tag/version slot + b[1..17].copy_from_slice(&[9u8; 16]); // from + b + }; + assert_eq!(frame_sender(&bytes), Some([9u8; 16])); + } + + #[test] + fn frame_sender_truncated_returns_none() { + assert_eq!(frame_sender(&[0u8; 0]), None); + assert_eq!(frame_sender(&[0u8; 16]), None); // tag + from missing + assert_eq!(frame_sender(&[0u8; 17]), Some([0u8; 16])); // exactly tag + from + } + #[test] fn decode_garbage_is_error_not_panic() { for len in [0usize, 1, 7, 64, 1500] {