umsh_node/
peer.rs

1use alloc::boxed::Box;
2
3use umsh_core::{MicSize, PublicKey};
4use umsh_mac::SendOptions;
5
6/// MIC size callers should normally ask for on a ping.
7///
8/// A ping frame is otherwise nearly half MIC: with a 2-byte nonce the whole
9/// frame is 34 bytes, 16 of them the authenticator. Dropping to 8 takes a
10/// quarter off the airtime while leaving forgery resistance at 2^-64 — far
11/// out of reach at LoRa packet rates, and the echo payload is a random nonce
12/// and filler that is worth nothing to forge.
13///
14/// Not applied inside [`PeerConnection::ping`], which honours whatever it is
15/// given: a caller measuring how 16-byte-MIC traffic fares should ping with a
16/// 16-byte MIC.
17pub const PING_MIC_SIZE: MicSize = MicSize::Mic8;
18
19use crate::node::{LocalNode, NodeError, Subscription, SubscriptionHandle};
20use crate::receive::ReceivedPacketRef;
21use crate::ticket::SendProgressTicket;
22use crate::transport::Transport;
23
24/// Relationship with one remote peer, bound to a transport context.
25///
26/// Generic over `T: Transport` — works with `LocalNode` (unicast) or
27/// `BoundChannel` (blind unicast) identically.
28#[derive(Clone)]
29pub struct PeerConnection<T: Transport> {
30    transport: T,
31    peer: PublicKey,
32}
33
34impl<T: Transport> PeerConnection<T> {
35    /// Create a new peer connection.
36    pub(crate) fn new(transport: T, peer: PublicKey) -> Self {
37        Self { transport, peer }
38    }
39
40    /// The remote peer's public key.
41    pub fn peer(&self) -> &PublicKey {
42        &self.peer
43    }
44
45    /// Send a raw payload to this peer (delegates to `transport.send()`).
46    pub async fn send(
47        &self,
48        payload: &[u8],
49        options: &SendOptions,
50    ) -> Result<SendProgressTicket, T::Error> {
51        self.transport.send(&self.peer, payload, options).await
52    }
53}
54
55impl<M: crate::mac::MacBackend> PeerConnection<LocalNode<M>> {
56    fn add_receive_handler<F>(&self, handler: F) -> SubscriptionHandle
57    where
58        F: FnMut(&ReceivedPacketRef<'_>) -> bool + 'static,
59    {
60        self.transport
61            .state()
62            .borrow_mut()
63            .peer_subscriptions_mut(self.peer)
64            .receive_handlers
65            .insert(Box::new(handler))
66    }
67
68    pub fn on_receive<F>(&self, handler: F) -> Subscription
69    where
70        F: FnMut(&ReceivedPacketRef<'_>) -> bool + 'static,
71    {
72        let handle = self.add_receive_handler(handler);
73        let state = self.transport.state().clone();
74        let peer = self.peer;
75        Subscription::new(move || {
76            let mut state = state.borrow_mut();
77            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
78                return false;
79            };
80            entry.receive_handlers.remove(handle)
81        })
82    }
83
84    fn add_ack_received_handler<F>(&self, handler: F) -> SubscriptionHandle
85    where
86        F: FnMut(crate::SendToken) + 'static,
87    {
88        self.transport
89            .state()
90            .borrow_mut()
91            .peer_subscriptions_mut(self.peer)
92            .ack_received_handlers
93            .insert(Box::new(handler))
94    }
95
96    pub fn on_ack_received<F>(&self, handler: F) -> Subscription
97    where
98        F: FnMut(crate::SendToken) + 'static,
99    {
100        let handle = self.add_ack_received_handler(handler);
101        let state = self.transport.state().clone();
102        let peer = self.peer;
103        Subscription::new(move || {
104            let mut state = state.borrow_mut();
105            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
106                return false;
107            };
108            entry.ack_received_handlers.remove(handle)
109        })
110    }
111
112    fn add_ack_timeout_handler<F>(&self, handler: F) -> SubscriptionHandle
113    where
114        F: FnMut(crate::SendToken) + 'static,
115    {
116        self.transport
117            .state()
118            .borrow_mut()
119            .peer_subscriptions_mut(self.peer)
120            .ack_timeout_handlers
121            .insert(Box::new(handler))
122    }
123
124    pub fn on_ack_timeout<F>(&self, handler: F) -> Subscription
125    where
126        F: FnMut(crate::SendToken) + 'static,
127    {
128        let handle = self.add_ack_timeout_handler(handler);
129        let state = self.transport.state().clone();
130        let peer = self.peer;
131        Subscription::new(move || {
132            let mut state = state.borrow_mut();
133            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
134                return false;
135            };
136            entry.ack_timeout_handlers.remove(handle)
137        })
138    }
139
140    pub fn on_pfs_established<F>(&self, handler: F) -> Subscription
141    where
142        F: FnMut() + 'static,
143    {
144        let handle = self
145            .transport
146            .state()
147            .borrow_mut()
148            .peer_subscriptions_mut(self.peer)
149            .pfs_established_handlers
150            .insert(Box::new(handler));
151        let state = self.transport.state().clone();
152        let peer = self.peer;
153        Subscription::new(move || {
154            let mut state = state.borrow_mut();
155            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
156                return false;
157            };
158            entry.pfs_established_handlers.remove(handle)
159        })
160    }
161
162    pub fn on_pfs_ended<F>(&self, handler: F) -> Subscription
163    where
164        F: FnMut() + 'static,
165    {
166        let handle = self
167            .transport
168            .state()
169            .borrow_mut()
170            .peer_subscriptions_mut(self.peer)
171            .pfs_ended_handlers
172            .insert(Box::new(handler));
173        let state = self.transport.state().clone();
174        let peer = self.peer;
175        Subscription::new(move || {
176            let mut state = state.borrow_mut();
177            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
178                return false;
179            };
180            entry.pfs_ended_handlers.remove(handle)
181        })
182    }
183
184    pub fn on_pong<F>(&self, handler: F) -> Subscription
185    where
186        F: FnMut(u64) + 'static,
187    {
188        let handle = self
189            .transport
190            .state()
191            .borrow_mut()
192            .peer_subscriptions_mut(self.peer)
193            .pong_handlers
194            .insert(Box::new(handler));
195        let state = self.transport.state().clone();
196        let peer = self.peer;
197        Subscription::new(move || {
198            let mut state = state.borrow_mut();
199            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
200                return false;
201            };
202            entry.pong_handlers.remove(handle)
203        })
204    }
205
206    pub fn on_ping_timeout<F>(&self, handler: F) -> Subscription
207    where
208        F: FnMut() + 'static,
209    {
210        let handle = self
211            .transport
212            .state()
213            .borrow_mut()
214            .peer_subscriptions_mut(self.peer)
215            .ping_timeout_handlers
216            .insert(Box::new(handler));
217        let state = self.transport.state().clone();
218        let peer = self.peer;
219        Subscription::new(move || {
220            let mut state = state.borrow_mut();
221            let Some(entry) = state.find_peer_subscriptions_mut(peer) else {
222                return false;
223            };
224            entry.ping_timeout_handlers.remove(handle)
225        })
226    }
227
228    /// Send an echo request and record it as pending until the matching echo
229    /// response arrives or `timeout_ms` elapses.
230    ///
231    /// `options` is honoured as given, so a ping travels the same way the
232    /// traffic it is measuring would: same MIC size, same encryption, same
233    /// routing and region options. Only `ack_requested` is overridden — the
234    /// echo response *is* the acknowledgement, so asking for a MAC ack as
235    /// well would put a second frame on the air for nothing.
236    ///
237    /// `extra_bytes` pads the payload beyond the 2-byte nonce. Padding
238    /// matters for the same reason the MIC size does: packet error rate rises
239    /// with frame length, so a minimal ping reports a link as usable when a
240    /// full-size message would not get through. See [`PING_MIC_SIZE`] for the
241    /// MIC size callers should normally ask for.
242    pub async fn ping(
243        &self,
244        extra_bytes: usize,
245        options: &SendOptions,
246        timeout_ms: u64,
247    ) -> Result<crate::ticket::SendProgressTicket, NodeError<M>> {
248        // Generate a 2-byte nonce via the MAC RNG.
249        let mut nonce_bytes = [0u8; 2];
250        self.transport.fill_random(&mut nonce_bytes).await;
251        let nonce = u16::from_be_bytes(nonce_bytes);
252
253        // Build data: [nonce_hi, nonce_lo, 0xA5, 0xA5, ...] capped at 60 bytes.
254        let total = (2 + extra_bytes).min(60);
255        let mut data = alloc::vec![0xA5u8; total];
256        data[0] = nonce_bytes[0];
257        data[1] = nonce_bytes[1];
258
259        // Encode the outbound payload: PayloadType::MacCommand byte followed by
260        // the encoded command body. The receiver's MAC dispatches on payload[0]
261        // and the EchoRequest auto-reply in the coordinator depends on this
262        // framing.
263        let cmd = crate::mac_command::MacCommand::EchoRequest { data: &data };
264        let mut buf = [0u8; 128];
265        buf[0] = umsh_core::PayloadType::MacCommand as u8;
266        let n = crate::mac_command::encode(&cmd, &mut buf[1..])?;
267        let n = n + 1;
268
269        // Record the pending ping BEFORE sending (avoid race if response is very fast).
270        let sent_at_ms = self.transport.now_ms().await;
271        self.transport
272            .record_ping(nonce, self.peer, sent_at_ms, sent_at_ms + timeout_ms);
273
274        // The caller's options carry through unchanged: a ping that is
275        // secured, sized or routed differently from real traffic measures a
276        // link the real traffic will not use. Only the ack request is
277        // dropped, because the EchoResponse already serves as one.
278        let opts = options.clone().with_ack_requested(false);
279        self.send(&buf[..n], &opts).await
280    }
281
282    /// Solicit this peer's current identity by sending a targeted MAC
283    /// Identity Request (command 1). Because the request is a unicast to a
284    /// specific peer, no filter options are needed — filters exist only to
285    /// narrow a broadcast solicitation. A random NONCE is included so the
286    /// peer echoes it in its identity response, matching the responder's
287    /// correlation contract.
288    ///
289    /// The response arrives asynchronously as a `PayloadType::NodeIdentity`
290    /// frame on the normal receive path, not as the return value here.
291    pub async fn request_identity(
292        &self,
293        options: &SendOptions,
294    ) -> Result<SendProgressTicket, NodeError<M>> {
295        let mut nonce_bytes = [0u8; 4];
296        self.transport.fill_random(&mut nonce_bytes).await;
297        let nonce = u32::from_be_bytes(nonce_bytes);
298
299        let opts_block = crate::mac_command::IdentityRequestBuilder::new()
300            .nonce(nonce)?
301            .build();
302        let cmd = crate::mac_command::MacCommand::IdentityRequest {
303            options: &opts_block,
304        };
305        let mut buf = [0u8; 128];
306        buf[0] = umsh_core::PayloadType::MacCommand as u8;
307        let n = crate::mac_command::encode(&cmd, &mut buf[1..])? + 1;
308
309        self.send(&buf[..n], options).await
310    }
311
312    /// The route the MAC has cached for this peer, if any.
313    pub async fn route(&self) -> Option<umsh_mac::CachedRoute> {
314        self.transport.peer_route(&self.peer).await
315    }
316
317    /// Forget this peer's cached route, returning whether one was held.
318    pub async fn clear_route(&self) -> bool {
319        self.transport.clear_peer_route(&self.peer).await
320    }
321
322    #[cfg(feature = "software-crypto")]
323    pub async fn request_pfs(
324        &self,
325        duration_minutes: u16,
326        options: &SendOptions,
327    ) -> Result<SendProgressTicket, NodeError<M>> {
328        self.transport
329            .request_pfs(&self.peer, duration_minutes, options)
330            .await
331    }
332
333    #[cfg(feature = "software-crypto")]
334    pub async fn end_pfs(&self, options: &SendOptions) -> Result<(), NodeError<M>> {
335        self.transport.end_pfs(&self.peer, options).await
336    }
337
338    #[cfg(feature = "software-crypto")]
339    pub async fn pfs_status(&self) -> Result<crate::node::PfsStatus, NodeError<M>> {
340        self.transport.pfs_status(&self.peer).await
341    }
342}
umsh_node/peer.rs

umsh_node/
peer.rs
