rns_net/interface/
weave.rs

1//! Weave Device Control Link (WDCL) framing and interface state.
2
3use std::collections::{HashMap, VecDeque};
4use std::io::{self, Read, Write};
5use std::os::fd::AsRawFd;
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::sync::{Arc, Mutex};
8use std::thread;
9use std::time::{Duration, Instant};
10
11use rns_core::transport::types::{InterfaceId, InterfaceInfo};
12use rns_crypto::ed25519::{Ed25519PrivateKey, Ed25519PublicKey};
13use rns_crypto::sha256::sha256;
14use rns_crypto::OsRng;
15
16use super::{InterfaceConfigData, InterfaceFactory, StartContext, StartResult, Writer};
17use crate::event::{
18    DynamicInterfaceRegistration, Event, EventSender, InterfaceTelemetry, LoopContext, LoopSource,
19    SourceStatus,
20};
21use crate::interface::device_writer::DeviceWriter;
22use crate::serial::{Parity, SerialConfig, SerialPort};
23
24pub const WDCL_T_DISCOVER: u8 = 0x00;
25pub const WDCL_T_CONNECT: u8 = 0x01;
26pub const WDCL_T_CMD: u8 = 0x02;
27pub const WDCL_T_LOG: u8 = 0x03;
28pub const WDCL_T_DISP: u8 = 0x04;
29pub const WDCL_T_ENDPOINT_PKT: u8 = 0x05;
30pub const WDCL_T_ENCAP_PROTO: u8 = 0x06;
31pub const WDCL_BROADCAST: [u8; 4] = [0xff; 4];
32
33pub const WDCL_CMD_ENDPOINT_PKT: u16 = 0x0001;
34pub const WDCL_CMD_ENDPOINTS_LIST: u16 = 0x0100;
35pub const WDCL_CMD_REMOTE_DISPLAY: u16 = 0x0a00;
36pub const WDCL_CMD_REMOTE_INPUT: u16 = 0x0a01;
37
38pub const ET_BOARD_INIT: u16 = 0x0003;
39pub const ET_PROTO_WDCL_CONNECTION: u16 = 0x3002;
40pub const ET_PROTO_WDCL_HOST_ENDPOINT: u16 = 0x3003;
41pub const ET_PROTO_WEAVE_EP_ALIVE: u16 = 0x3102;
42pub const ET_PROTO_WEAVE_EP_TIMEOUT: u16 = 0x3103;
43pub const ET_PROTO_WEAVE_EP_VIA: u16 = 0x3104;
44pub const ET_STAT_CPU: u16 = 0xe003;
45pub const ET_STAT_MEMORY: u16 = 0xe005;
46
47pub const WEAVE_HW_MTU: u32 = 1024;
48pub const WEAVE_DEFAULT_BITRATE: u64 = 250_000;
49pub const WEAVE_PEERING_TIMEOUT: f64 = 20.0;
50pub const WEAVE_DUPLICATE_TTL: f64 = 0.75;
51pub const WEAVE_DUPLICATE_CAPACITY: usize = 48;
52
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub struct WdclFrame {
55    pub destination: [u8; 4],
56    pub frame_type: u8,
57    pub payload: Vec<u8>,
58}
59
60impl WdclFrame {
61    pub fn encode(&self) -> Vec<u8> {
62        let mut output = Vec::with_capacity(5 + self.payload.len());
63        output.extend_from_slice(&self.destination);
64        output.push(self.frame_type);
65        output.extend_from_slice(&self.payload);
66        output
67    }
68
69    pub fn decode(data: &[u8]) -> Option<Self> {
70        if data.len() < 5 {
71            return None;
72        }
73        Some(Self {
74            destination: data[..4].try_into().ok()?,
75            frame_type: data[4],
76            payload: data[5..].to_vec(),
77        })
78    }
79
80    pub fn command(destination: [u8; 4], command: u16, data: &[u8]) -> Self {
81        let mut payload = Vec::with_capacity(2 + data.len());
82        payload.extend_from_slice(&command.to_be_bytes());
83        payload.extend_from_slice(data);
84        Self {
85            destination,
86            frame_type: WDCL_T_CMD,
87            payload,
88        }
89    }
90
91    pub fn endpoint_packet(destination: [u8; 4], endpoint: [u8; 8], packet: &[u8]) -> Self {
92        let mut data = Vec::with_capacity(8 + packet.len());
93        data.extend_from_slice(&endpoint);
94        data.extend_from_slice(packet);
95        Self::command(destination, WDCL_CMD_ENDPOINT_PKT, &data)
96    }
97}
98
99#[derive(Debug, Clone, PartialEq)]
100pub struct WeaveLogFrame {
101    pub timestamp: f64,
102    pub level: u8,
103    pub event: u16,
104    pub data: Vec<u8>,
105}
106
107#[derive(Debug, Clone, PartialEq)]
108pub enum WeaveInput {
109    Discovery {
110        switch_id: [u8; 4],
111        signing_key: [u8; 32],
112    },
113    EndpointPacket {
114        source: [u8; 8],
115        packet: Vec<u8>,
116    },
117    Log(WeaveLogFrame),
118    Display {
119        offset: usize,
120        total: usize,
121        fragment: Vec<u8>,
122    },
123    Unknown(WdclFrame),
124}
125
126/// Validate and classify a device frame. Discovery signatures cover the
127/// destination/signed switch ID exactly as in Reticulum 1.3.8.
128pub fn parse_device_frame(data: &[u8], host_switch_id: [u8; 4]) -> Option<WeaveInput> {
129    let frame = WdclFrame::decode(data)?;
130    match frame.frame_type {
131        WDCL_T_DISCOVER if frame.payload.len() == 96 => {
132            let key: [u8; 32] = frame.payload[..32].try_into().ok()?;
133            let signature: [u8; 64] = frame.payload[32..].try_into().ok()?;
134            if !Ed25519PublicKey::from_bytes(&key)
135                .ok()?
136                .verify(&signature, &frame.destination)
137            {
138                return None;
139            }
140            Some(WeaveInput::Discovery {
141                switch_id: key[28..].try_into().ok()?,
142                signing_key: key,
143            })
144        }
145        WDCL_T_ENDPOINT_PKT if frame.destination == host_switch_id && frame.payload.len() > 8 => {
146            let split = frame.payload.len() - 8;
147            Some(WeaveInput::EndpointPacket {
148                source: frame.payload[split..].try_into().ok()?,
149                packet: frame.payload[..split].to_vec(),
150            })
151        }
152        WDCL_T_LOG if frame.payload.len() >= 9 => {
153            // Firmware prepends a flow byte to the eight-byte log header.
154            let data = &frame.payload[1..];
155            Some(WeaveInput::Log(WeaveLogFrame {
156                timestamp: u32::from_be_bytes(data[..4].try_into().ok()?) as f64 / 1000.0,
157                level: data[4],
158                event: u16::from_be_bytes(data[5..7].try_into().ok()?),
159                data: data[7..].to_vec(),
160            }))
161        }
162        WDCL_T_DISP if frame.payload.len() >= 9 => Some(WeaveInput::Display {
163            offset: u32::from_be_bytes(frame.payload[1..5].try_into().ok()?) as usize,
164            total: u32::from_be_bytes(frame.payload[5..9].try_into().ok()?) as usize,
165            fragment: frame.payload[9..].to_vec(),
166        }),
167        _ => Some(WeaveInput::Unknown(frame)),
168    }
169}
170
171#[derive(Debug, Clone)]
172pub struct WeavePeerState {
173    pub endpoint_id: [u8; 8],
174    pub via_switch_id: Option<[u8; 4]>,
175    pub last_heard: f64,
176}
177
178#[derive(Debug, Clone, Default)]
179pub struct WeaveState {
180    pub connected: bool,
181    pub switch_id: Option<[u8; 4]>,
182    pub endpoint_id: Option<[u8; 8]>,
183    pub cpu_load: Option<u8>,
184    pub mem_load: Option<f64>,
185    pub peers: HashMap<[u8; 8], WeavePeerState>,
186    pub logs: VecDeque<WeaveLogFrame>,
187    pub display: Vec<u8>,
188    pub unknown_events: VecDeque<WeaveLogFrame>,
189    duplicates: VecDeque<([u8; 32], f64)>,
190}
191
192impl WeaveState {
193    pub fn accept_packet(&mut self, packet: &[u8], now: f64) -> bool {
194        while self
195            .duplicates
196            .front()
197            .is_some_and(|entry| now >= entry.1 + WEAVE_DUPLICATE_TTL)
198        {
199            self.duplicates.pop_front();
200        }
201        let hash = sha256(packet);
202        if self.duplicates.iter().any(|entry| entry.0 == hash) {
203            return false;
204        }
205        if self.duplicates.len() == WEAVE_DUPLICATE_CAPACITY {
206            self.duplicates.pop_front();
207        }
208        self.duplicates.push_back((hash, now));
209        true
210    }
211
212    pub fn handle_log(&mut self, frame: WeaveLogFrame, now: f64) {
213        if self.logs.len() == 1024 {
214            self.logs.pop_front();
215        }
216        self.logs.push_back(frame.clone());
217        match frame.event {
218            ET_BOARD_INIT => {}
219            ET_PROTO_WDCL_CONNECTION => self.connected = true,
220            ET_PROTO_WDCL_HOST_ENDPOINT if frame.data.len() == 8 => {
221                self.endpoint_id = frame.data.as_slice().try_into().ok();
222            }
223            ET_PROTO_WEAVE_EP_ALIVE if frame.data.len() == 8 => {
224                let endpoint_id: [u8; 8] = frame.data.as_slice().try_into().unwrap();
225                self.peers
226                    .entry(endpoint_id)
227                    .and_modify(|peer| peer.last_heard = now)
228                    .or_insert(WeavePeerState {
229                        endpoint_id,
230                        via_switch_id: None,
231                        last_heard: now,
232                    });
233            }
234            ET_PROTO_WEAVE_EP_TIMEOUT if frame.data.len() == 8 => {
235                if let Ok(endpoint) = <[u8; 8]>::try_from(frame.data.as_slice()) {
236                    self.peers.remove(&endpoint);
237                }
238            }
239            ET_PROTO_WEAVE_EP_VIA if frame.data.len() == 12 => {
240                let endpoint: [u8; 8] = frame.data[..8].try_into().unwrap();
241                let via: [u8; 4] = frame.data[8..].try_into().unwrap();
242                if let Some(peer) = self.peers.get_mut(&endpoint) {
243                    peer.via_switch_id = Some(via);
244                }
245            }
246            ET_STAT_CPU if !frame.data.is_empty() => self.cpu_load = Some(frame.data[0]),
247            ET_STAT_MEMORY if frame.data.len() == 8 => {
248                let free = u32::from_be_bytes(frame.data[..4].try_into().unwrap()) as f64;
249                let total = u32::from_be_bytes(frame.data[4..].try_into().unwrap()) as f64;
250                self.mem_load = (total > 0.0).then_some(((total - free) / total) * 100.0);
251            }
252            _ => self.unknown_events.push_back(frame),
253        }
254    }
255
256    pub fn expire_peers(&mut self, now: f64) -> Vec<[u8; 8]> {
257        let expired: Vec<_> = self
258            .peers
259            .iter()
260            .filter_map(|(id, peer)| {
261                (now >= peer.last_heard + WEAVE_PEERING_TIMEOUT).then_some(*id)
262            })
263            .collect();
264        for id in &expired {
265            self.peers.remove(id);
266        }
267        expired
268    }
269
270    pub fn apply_display(&mut self, offset: usize, total: usize, fragment: &[u8]) -> bool {
271        if total > 128 * 64 / 8 || offset > total || fragment.len() > total - offset {
272            return false;
273        }
274        self.display.resize(total, 0);
275        self.display[offset..offset + fragment.len()].copy_from_slice(fragment);
276        offset + fragment.len() == total
277    }
278}
279
280#[derive(Debug, Clone)]
281pub struct WeaveConfig {
282    pub name: String,
283    pub port: String,
284    pub configured_bitrate: u64,
285    pub interface_id: InterfaceId,
286    pub state: Arc<Mutex<WeaveState>>,
287}
288
289struct WeaveSession {
290    config: WeaveConfig,
291    tx: EventSender,
292    next_dynamic_id: Arc<AtomicU64>,
293    mode: u8,
294    gravity: i64,
295    recursive_prs: bool,
296    announces_from_internal: bool,
297    announces_to_internal: Option<bool>,
298    ingress_control: rns_core::transport::types::IngressControlConfig,
299    ifac: Option<crate::ifac::IfacState>,
300    host_identity: Arc<Ed25519PrivateKey>,
301    host_public: [u8; 32],
302    host_switch_id: [u8; 4],
303    writer: Arc<Mutex<Option<DeviceWriter<std::fs::File>>>>,
304    remote_switch_id: Arc<Mutex<Option<[u8; 4]>>>,
305    peer_interfaces: Arc<Mutex<HashMap<[u8; 8], InterfaceId>>>,
306}
307
308impl WeaveSession {
309    fn new(config: WeaveConfig, ctx: &StartContext) -> Arc<Self> {
310        let identity = Ed25519PrivateKey::generate(&mut OsRng);
311        let host_public = identity.public_key().public_bytes();
312        Arc::new(Self {
313            config,
314            tx: ctx.tx.clone(),
315            next_dynamic_id: Arc::clone(&ctx.next_dynamic_id),
316            mode: ctx.mode,
317            gravity: ctx.gravity,
318            recursive_prs: ctx.recursive_prs,
319            announces_from_internal: ctx.announces_from_internal,
320            announces_to_internal: ctx.announces_to_internal,
321            ingress_control: ctx.ingress_control,
322            ifac: ctx.ifac.clone(),
323            host_identity: Arc::new(identity),
324            host_public,
325            host_switch_id: host_public[28..].try_into().unwrap(),
326            writer: Arc::new(Mutex::new(None)),
327            remote_switch_id: Arc::new(Mutex::new(None)),
328            peer_interfaces: Arc::new(Mutex::new(HashMap::new())),
329        })
330    }
331
332    fn spawn(self: &Arc<Self>) -> io::Result<()> {
333        let session = Arc::clone(self);
334        if crate::engine::FOLD_IO {
335            thread::Builder::new()
336                .name(format!("weave-connect-{}", self.config.interface_id.0))
337                .spawn(move || session.connect_folded(false))?;
338            return Ok(());
339        }
340        thread::Builder::new()
341            .name(format!("weave-session-{}", self.config.interface_id.0))
342            .spawn(move || session.reconnect_loop())?;
343        Ok(())
344    }
345
346    fn reconnect_loop(self: Arc<Self>) {
347        loop {
348            if let Err(error) = self.run_connection() {
349                log::warn!(
350                    "[{}] Weave connection failed: {}; retrying in five seconds",
351                    self.config.name,
352                    error
353                );
354            }
355            self.disconnect_all();
356            thread::sleep(Duration::from_secs(5));
357        }
358    }
359
360    fn run_connection(self: &Arc<Self>) -> io::Result<()> {
361        let port = SerialPort::open(&SerialConfig {
362            path: self.config.port.clone(),
363            baud: 3_000_000,
364            data_bits: 8,
365            parity: Parity::None,
366            stop_bits: 1,
367        })?;
368        let mut reader = port.reader()?;
369        *super::lock_or_recover(&self.writer, "weave writer") =
370            Some(DeviceWriter::new(port.writer()?));
371        *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") = None;
372        {
373            let mut state = super::lock_or_recover(&self.config.state, "weave state");
374            state.connected = false;
375            state.switch_id = None;
376        }
377
378        self.send_frame(&WdclFrame {
379            destination: WDCL_BROADCAST,
380            frame_type: WDCL_T_DISCOVER,
381            payload: self.host_switch_id.to_vec(),
382        })?;
383
384        let handshake_deadline = Instant::now() + Duration::from_secs(2);
385        let mut decoder = crate::hdlc::Decoder::with_limits(5, 128 * 1024);
386        let mut buffer = [0u8; 32 * 1024];
387        let mut last_peer_expiry = Instant::now();
388        loop {
389            let mut pollfd = libc::pollfd {
390                fd: reader.as_raw_fd(),
391                events: libc::POLLIN,
392                revents: 0,
393            };
394            let ready = unsafe { libc::poll(&mut pollfd, 1, 250) };
395            if ready < 0 {
396                return Err(io::Error::last_os_error());
397            }
398            if ready > 0 {
399                let read = reader.read(&mut buffer)?;
400                if read == 0 {
401                    return Err(io::Error::new(
402                        io::ErrorKind::UnexpectedEof,
403                        "serial port closed",
404                    ));
405                }
406                for frame in decoder.feed(&buffer[..read]) {
407                    self.process_frame(&frame)?;
408                }
409            }
410
411            let connected = super::lock_or_recover(&self.config.state, "weave state").connected;
412            if !connected && Instant::now() >= handshake_deadline {
413                return Err(io::Error::new(
414                    io::ErrorKind::TimedOut,
415                    "WDCL connection handshake timed out",
416                ));
417            }
418            if last_peer_expiry.elapsed() >= Duration::from_secs(1) {
419                self.expire_peers(crate::time::now());
420                last_peer_expiry = Instant::now();
421            }
422        }
423    }
424
425    /// Queue an event for the driver: from the driver thread itself when
426    /// the device is read there (see [`crate::engine::FOLD_IO`]).
427    fn emit(&self, event: Event) {
428        if crate::engine::FOLD_IO {
429            let _ = self.tx.push_from_loop([event]);
430        } else {
431            let _ = self.tx.send(event);
432        }
433    }
434
435    /// Open the device and serve it on the driver loop, retrying every
436    /// five seconds (and first, when `wait_first`). Runs on a thread of its
437    /// own, which exits once the device is served.
438    fn connect_folded(self: Arc<Self>, wait_first: bool) {
439        let mut wait = wait_first;
440        loop {
441            if wait {
442                thread::sleep(Duration::from_secs(5));
443            }
444            wait = true;
445            match self.open_folded() {
446                Ok(()) => return,
447                Err(error) => log::warn!(
448                    "[{}] Weave connection failed: {}; retrying in five seconds",
449                    self.config.name,
450                    error
451                ),
452            }
453        }
454    }
455
456    fn open_folded(self: &Arc<Self>) -> io::Result<()> {
457        let port = SerialPort::open(&SerialConfig {
458            path: self.config.port.clone(),
459            baud: 3_000_000,
460            data_bits: 8,
461            parity: Parity::None,
462            stop_bits: 1,
463        })?;
464        let reader = port.reader()?;
465        *super::lock_or_recover(&self.writer, "weave writer") = Some(DeviceWriter::driven(
466            port.writer()?,
467            Arc::clone(self.tx.poller()),
468            super::loop_writer::write_key(self.config.interface_id),
469            &self.config.name,
470        ));
471        *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") = None;
472        {
473            let mut state = super::lock_or_recover(&self.config.state, "weave state");
474            state.connected = false;
475            state.switch_id = None;
476        }
477        // Reader and writer share one open file: this makes both
478        // non-blocking.
479        super::stream_read::set_nonblocking(reader.as_raw_fd())?;
480        self.send_frame(&WdclFrame {
481            destination: WDCL_BROADCAST,
482            frame_type: WDCL_T_DISCOVER,
483            payload: self.host_switch_id.to_vec(),
484        })?;
485        self.tx.register_source(
486            super::loop_writer::read_key(self.config.interface_id),
487            Box::new(WeaveSource {
488                reader,
489                decoder: crate::hdlc::Decoder::with_limits(5, 128 * 1024),
490                handshake_deadline: Instant::now() + Duration::from_secs(2),
491                session: Arc::clone(self),
492            }),
493        )
494    }
495
496    /// The device was lost: bring everything down and reopen it.
497    fn connection_lost(self: &Arc<Self>, error: &io::Error) {
498        log::warn!(
499            "[{}] Weave connection failed: {}; retrying in five seconds",
500            self.config.name,
501            error
502        );
503        self.disconnect_all();
504        let session = Arc::clone(self);
505        let spawned = thread::Builder::new()
506            .name(format!("weave-connect-{}", self.config.interface_id.0))
507            .spawn(move || session.connect_folded(true));
508        if let Err(e) = spawned {
509            log::error!(
510                "[{}] cannot start reconnecting, giving up: {}",
511                self.config.name,
512                e
513            );
514        }
515    }
516
517    fn send_frame(&self, frame: &WdclFrame) -> io::Result<()> {
518        let encoded = crate::hdlc::frame(&frame.encode());
519        let mut writer = super::lock_or_recover(&self.writer, "weave writer");
520        writer
521            .as_mut()
522            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "Weave is offline"))?
523            .write_all(&encoded)
524    }
525
526    fn process_frame(self: &Arc<Self>, raw: &[u8]) -> io::Result<()> {
527        let Some(input) = parse_device_frame(raw, self.host_switch_id) else {
528            return Ok(());
529        };
530        match input {
531            WeaveInput::Discovery {
532                switch_id,
533                signing_key: _,
534            } => {
535                *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") =
536                    Some(switch_id);
537                super::lock_or_recover(&self.config.state, "weave state").switch_id =
538                    Some(switch_id);
539                let mut payload = self.host_public.to_vec();
540                payload.extend_from_slice(&self.host_identity.sign(&switch_id));
541                self.send_frame(&WdclFrame {
542                    destination: switch_id,
543                    frame_type: WDCL_T_CONNECT,
544                    payload,
545                })?;
546            }
547            WeaveInput::EndpointPacket { source, packet } => {
548                let now = crate::time::now();
549                {
550                    let mut state = super::lock_or_recover(&self.config.state, "weave state");
551                    state
552                        .peers
553                        .entry(source)
554                        .and_modify(|peer| peer.last_heard = now)
555                        .or_insert(WeavePeerState {
556                            endpoint_id: source,
557                            via_switch_id: None,
558                            last_heard: now,
559                        });
560                    if !state.accept_packet(&packet, now) {
561                        return Ok(());
562                    }
563                }
564                let interface_id = self.ensure_peer(source, now);
565                self.emit(Event::Frame {
566                    interface_id,
567                    data: packet.into(),
568                    rssi: None,
569                    snr: None,
570                });
571            }
572            WeaveInput::Log(frame) => self.process_log(frame),
573            WeaveInput::Display {
574                offset,
575                total,
576                fragment,
577            } => {
578                super::lock_or_recover(&self.config.state, "weave state")
579                    .apply_display(offset, total, &fragment);
580            }
581            WeaveInput::Unknown(_) => {}
582        }
583        Ok(())
584    }
585
586    fn process_log(self: &Arc<Self>, frame: WeaveLogFrame) {
587        let now = crate::time::now();
588        let event = frame.event;
589        let data = frame.data.clone();
590        super::lock_or_recover(&self.config.state, "weave state").handle_log(frame, now);
591
592        match event {
593            ET_PROTO_WDCL_CONNECTION => {
594                self.emit(Event::InterfaceUp(self.config.interface_id, None, None));
595                self.publish_parent_telemetry();
596            }
597            ET_PROTO_WEAVE_EP_ALIVE if data.len() == 8 => {
598                let endpoint: [u8; 8] = data.as_slice().try_into().unwrap();
599                self.ensure_peer(endpoint, now);
600            }
601            ET_PROTO_WEAVE_EP_TIMEOUT if data.len() == 8 => {
602                let endpoint: [u8; 8] = data.as_slice().try_into().unwrap();
603                self.remove_peer(endpoint);
604            }
605            ET_PROTO_WEAVE_EP_VIA if data.len() == 12 => {
606                let endpoint: [u8; 8] = data[..8].try_into().unwrap();
607                let via: [u8; 4] = data[8..].try_into().unwrap();
608                if let Some(interface_id) =
609                    super::lock_or_recover(&self.peer_interfaces, "weave peer interfaces")
610                        .get(&endpoint)
611                        .copied()
612                {
613                    self.emit(Event::InterfaceTelemetry {
614                        interface_id,
615                        telemetry: Box::new(InterfaceTelemetry {
616                            via_switch_id: Some(via),
617                            ..Default::default()
618                        }),
619                    });
620                }
621            }
622            ET_PROTO_WDCL_HOST_ENDPOINT | ET_STAT_CPU | ET_STAT_MEMORY => {
623                self.publish_parent_telemetry()
624            }
625            _ => {}
626        }
627    }
628
629    fn ensure_peer(self: &Arc<Self>, endpoint: [u8; 8], now: f64) -> InterfaceId {
630        if let Some(id) = super::lock_or_recover(&self.peer_interfaces, "weave peers")
631            .get(&endpoint)
632            .copied()
633        {
634            return id;
635        }
636        let id = InterfaceId(self.next_dynamic_id.fetch_add(1, Ordering::Relaxed));
637        super::lock_or_recover(&self.peer_interfaces, "weave peers").insert(endpoint, id);
638        let via_switch_id = super::lock_or_recover(&self.config.state, "weave state")
639            .peers
640            .get(&endpoint)
641            .and_then(|peer| peer.via_switch_id);
642        let name = format!("WeaveInterfacePeer[{}]", hex(&endpoint));
643        let info = InterfaceInfo {
644            stamp_costs: Default::default(),
645            id,
646            name,
647            mode: self.mode,
648            gravity: self.gravity,
649            recursive_prs: self.recursive_prs,
650            announces_from_internal: self.announces_from_internal,
651            announces_to_internal: self.announces_to_internal,
652            out_capable: true,
653            in_capable: true,
654            bitrate: Some(self.config.configured_bitrate),
655            airtime_profile: None,
656            announce_rate_target: None,
657            announce_rate_grace: 0,
658            announce_rate_penalty: 0.0,
659            announce_cap: rns_core::constants::ANNOUNCE_CAP,
660            is_local_client: false,
661            wants_tunnel: false,
662            tunnel_id: None,
663            mtu: WEAVE_HW_MTU,
664            ia_freq: 0.0,
665            ip_freq: 0.0,
666            op_freq: 0.0,
667            op_samples: 0,
668            started: now,
669            ingress_control: self.ingress_control,
670        };
671        self.emit(Event::DynamicInterfaceUp {
672            id,
673            writer: Box::new(WeavePeerWriter {
674                session: Arc::clone(self),
675                endpoint,
676            }),
677            registration: Box::new(DynamicInterfaceRegistration {
678                info,
679                interface_type: "WeaveInterfacePeer".into(),
680                parent_id: self.config.interface_id,
681                telemetry: InterfaceTelemetry {
682                    via_switch_id,
683                    ..Default::default()
684                },
685                ifac: self.ifac.clone(),
686            }),
687        });
688        self.publish_parent_telemetry();
689        id
690    }
691
692    fn remove_peer(&self, endpoint: [u8; 8]) {
693        if let Some(id) =
694            super::lock_or_recover(&self.peer_interfaces, "weave peers").remove(&endpoint)
695        {
696            super::lock_or_recover(&self.config.state, "weave state")
697                .peers
698                .remove(&endpoint);
699            self.emit(Event::InterfaceDown(id));
700            self.publish_parent_telemetry();
701        }
702    }
703
704    fn expire_peers(&self, now: f64) {
705        let expired = super::lock_or_recover(&self.config.state, "weave state").expire_peers(now);
706        for endpoint in &expired {
707            if let Some(id) =
708                super::lock_or_recover(&self.peer_interfaces, "weave peers").remove(endpoint)
709            {
710                self.emit(Event::InterfaceDown(id));
711            }
712        }
713        if !super::lock_or_recover(&self.peer_interfaces, "weave peers").is_empty()
714            || !expired.is_empty()
715        {
716            self.publish_parent_telemetry();
717        }
718    }
719
720    fn publish_parent_telemetry(&self) {
721        let state = super::lock_or_recover(&self.config.state, "weave state");
722        self.emit(Event::InterfaceTelemetry {
723            interface_id: self.config.interface_id,
724            telemetry: Box::new(InterfaceTelemetry {
725                cpu_load: state.cpu_load.map(f64::from),
726                mem_load: state.mem_load,
727                switch_id: state.switch_id,
728                endpoint_id: state.endpoint_id,
729                via_switch_id: None,
730                peers: Some(state.peers.len()),
731                ..Default::default()
732            }),
733        });
734    }
735
736    fn disconnect_all(&self) {
737        *super::lock_or_recover(&self.writer, "weave writer") = None;
738        *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") = None;
739        {
740            let mut state = super::lock_or_recover(&self.config.state, "weave state");
741            state.connected = false;
742            state.switch_id = None;
743            state.peers.clear();
744        }
745        let peers: Vec<_> = super::lock_or_recover(&self.peer_interfaces, "weave peers")
746            .drain()
747            .map(|(_, id)| id)
748            .collect();
749        for id in peers {
750            self.emit(Event::InterfaceDown(id));
751        }
752        self.emit(Event::InterfaceDown(self.config.interface_id));
753        self.publish_parent_telemetry();
754    }
755}
756
757struct WeavePeerWriter {
758    session: Arc<WeaveSession>,
759    endpoint: [u8; 8],
760}
761
762impl Writer for WeavePeerWriter {
763    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
764        let remote = super::lock_or_recover(&self.session.remote_switch_id, "weave remote switch")
765            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "Weave is offline"))?;
766        self.session
767            .send_frame(&WdclFrame::endpoint_packet(remote, self.endpoint, data))
768    }
769
770    fn enqueue_confirmed(
771        &mut self,
772        data: &[u8],
773        completion: &mut Option<crate::link_send::Completion>,
774    ) -> io::Result<()> {
775        // Driver-run (see `sends_inline`): a full device buffer returns
776        // WouldBlock with the completion untouched, and the driver retries.
777        self.send_frame(data)?;
778        if let Some(completion) = completion.take() {
779            completion.finish(Ok(()));
780        }
781        Ok(())
782    }
783
784    fn sends_inline(&self) -> bool {
785        // With interface I/O folded the shared device writer never blocks.
786        crate::engine::FOLD_IO
787    }
788
789    fn flush_pending(&mut self) {
790        if let Some(writer) = super::lock_or_recover(&self.session.writer, "weave writer").as_mut()
791        {
792            writer.drain();
793        }
794    }
795}
796
797/// A Weave device, read on the driver thread (see
798/// [`WeaveSession::open_folded`]).
799struct WeaveSource {
800    reader: std::fs::File,
801    decoder: crate::hdlc::Decoder,
802    handshake_deadline: Instant,
803    session: Arc<WeaveSession>,
804}
805
806impl WeaveSource {
807    fn connected(&self) -> bool {
808        super::lock_or_recover(&self.session.config.state, "weave state").connected
809    }
810
811    /// Fail the handshake once its deadline passed unanswered.
812    fn check_handshake(&self) -> io::Result<()> {
813        if !self.connected() && Instant::now() >= self.handshake_deadline {
814            return Err(io::Error::new(
815                io::ErrorKind::TimedOut,
816                "WDCL connection handshake timed out",
817            ));
818        }
819        Ok(())
820    }
821
822    fn read(&mut self, cx: &mut LoopContext<'_>) -> io::Result<SourceStatus> {
823        let mut read = 0;
824        while read < super::stream_read::STREAM_READ_BUDGET {
825            if !self.session.tx.inbound_data_has_room() {
826                return Ok(SourceStatus::Suspend);
827            }
828            // SAFETY: `cx.buf` is a valid writable buffer for the call; the
829            // device is non-blocking.
830            let n = unsafe {
831                libc::read(
832                    self.reader.as_raw_fd(),
833                    cx.buf.as_mut_ptr().cast(),
834                    cx.buf.len(),
835                )
836            };
837            if n < 0 {
838                let error = io::Error::last_os_error();
839                match error.kind() {
840                    io::ErrorKind::WouldBlock => break,
841                    io::ErrorKind::Interrupted => continue,
842                    _ => return Err(error),
843                }
844            }
845            if n == 0 {
846                return Err(io::Error::new(
847                    io::ErrorKind::UnexpectedEof,
848                    "serial port closed",
849                ));
850            }
851            let n = n as usize;
852            read += n;
853            for frame in self.decoder.feed(&cx.buf[..n]) {
854                self.session.process_frame(&frame)?;
855            }
856            if n < cx.buf.len() {
857                break;
858            }
859        }
860        self.check_handshake()?;
861        Ok(SourceStatus::Rearm)
862    }
863}
864
865impl LoopSource for WeaveSource {
866    fn fd(&self) -> std::os::fd::RawFd {
867        self.reader.as_raw_fd()
868    }
869
870    fn on_ready(&mut self, cx: &mut LoopContext<'_>) -> SourceStatus {
871        match self.read(cx) {
872            Ok(status) => status,
873            Err(error) => {
874                self.session.connection_lost(&error);
875                SourceStatus::Remove
876            }
877        }
878    }
879
880    fn deadline(&self) -> Option<Instant> {
881        // The handshake's deadline until it completes, then the next peer
882        // to time out, if any.
883        if !self.connected() {
884            return Some(self.handshake_deadline);
885        }
886        let state = super::lock_or_recover(&self.session.config.state, "weave state");
887        let oldest = state
888            .peers
889            .values()
890            .map(|peer| peer.last_heard)
891            .min_by(f64::total_cmp)?;
892        let wait = (oldest + WEAVE_PEERING_TIMEOUT - crate::time::now()).max(0.0);
893        Some(Instant::now() + Duration::from_secs_f64(wait))
894    }
895
896    fn on_timer(&mut self, _cx: &mut LoopContext<'_>) -> SourceStatus {
897        if let Err(error) = self.check_handshake() {
898            self.session.connection_lost(&error);
899            return SourceStatus::Remove;
900        }
901        self.session.expire_peers(crate::time::now());
902        SourceStatus::Rearm
903    }
904}
905
906fn hex(bytes: &[u8]) -> String {
907    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
908}
909
910pub struct WeaveFactory;
911/// The parent interface's writer: it transmits nothing, but the driver
912/// reports the device writable under the parent's id, so it writes out
913/// what the peers' writers left buffered.
914struct ParentWriter {
915    session: Arc<WeaveSession>,
916}
917impl Writer for ParentWriter {
918    fn send_frame_confirmed(&mut self, _data: &[u8]) -> io::Result<()> {
919        Err(io::Error::new(
920            io::ErrorKind::Unsupported,
921            "Weave parent is not a transmitting peer",
922        ))
923    }
924    fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
925        Ok(())
926    }
927
928    fn flush_pending(&mut self) {
929        if let Some(writer) = super::lock_or_recover(&self.session.writer, "weave writer").as_mut()
930        {
931            writer.drain();
932        }
933    }
934
935    fn sends_inline(&self) -> bool {
936        // Nothing to send, so no worker thread is needed when the driver
937        // runs device I/O.
938        crate::engine::FOLD_IO
939    }
940}
941
942impl InterfaceFactory for WeaveFactory {
943    fn type_name(&self) -> &str {
944        "WeaveInterface"
945    }
946    fn default_ifac_size(&self) -> usize {
947        16
948    }
949
950    fn parse_config(
951        &self,
952        name: &str,
953        id: InterfaceId,
954        params: &HashMap<String, String>,
955    ) -> Result<Box<dyn InterfaceConfigData>, String> {
956        let port = params
957            .get("port")
958            .cloned()
959            .ok_or_else(|| "WeaveInterface requires 'port'".to_string())?;
960        let configured_bitrate = params
961            .get("configured_bitrate")
962            .and_then(|v| v.parse().ok())
963            .unwrap_or(WEAVE_DEFAULT_BITRATE);
964        Ok(Box::new(WeaveConfig {
965            name: name.into(),
966            port,
967            configured_bitrate,
968            interface_id: id,
969            state: Arc::new(Mutex::new(WeaveState::default())),
970        }))
971    }
972
973    fn start(
974        &self,
975        config: Box<dyn InterfaceConfigData>,
976        ctx: StartContext,
977    ) -> io::Result<StartResult> {
978        let config = *config
979            .into_any()
980            .downcast::<WeaveConfig>()
981            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "wrong Weave config type"))?;
982        let session = WeaveSession::new(config.clone(), &ctx);
983        session.spawn()?;
984        let info = InterfaceInfo {
985            stamp_costs: Default::default(),
986            id: config.interface_id,
987            name: config.name.clone(),
988            mode: ctx.mode,
989            gravity: ctx.gravity,
990            recursive_prs: ctx.recursive_prs,
991            announces_from_internal: ctx.announces_from_internal,
992            announces_to_internal: ctx.announces_to_internal,
993            out_capable: false,
994            in_capable: true,
995            bitrate: Some(config.configured_bitrate),
996            airtime_profile: None,
997            announce_rate_target: None,
998            announce_rate_grace: 0,
999            announce_rate_penalty: 0.0,
1000            announce_cap: rns_core::constants::ANNOUNCE_CAP,
1001            is_local_client: false,
1002            wants_tunnel: false,
1003            tunnel_id: None,
1004            mtu: WEAVE_HW_MTU,
1005            ia_freq: 0.0,
1006            ip_freq: 0.0,
1007            op_freq: 0.0,
1008            op_samples: 0,
1009            started: crate::time::now(),
1010            ingress_control: ctx.ingress_control,
1011        };
1012        Ok(StartResult::Simple {
1013            control: None,
1014            id: config.interface_id,
1015            info,
1016            writer: Box::new(ParentWriter { session }),
1017            interface_type_name: "WeaveInterface".into(),
1018        })
1019    }
1020}
1021
1022#[cfg(test)]
1023mod tests {
1024    use super::*;
1025    use rns_crypto::ed25519::Ed25519PrivateKey;
1026    use rns_crypto::FixedRng;
1027
1028    fn log_frame(host: [u8; 4], event: u16, data: &[u8]) -> Vec<u8> {
1029        let mut payload = vec![0, 0, 0, 0, 1, 2];
1030        payload.extend_from_slice(&event.to_be_bytes());
1031        payload.extend_from_slice(data);
1032        WdclFrame {
1033            destination: host,
1034            frame_type: WDCL_T_LOG,
1035            payload,
1036        }
1037        .encode()
1038    }
1039
1040    #[test]
1041    fn wdcl_fragmented_hdlc_and_discovery_signature() {
1042        let private = Ed25519PrivateKey::generate(&mut FixedRng::new(&[7; 64]));
1043        let key = private.public_key().public_bytes();
1044        let signed_id = [1, 2, 3, 4];
1045        let mut payload = key.to_vec();
1046        payload.extend_from_slice(&private.sign(&signed_id));
1047        let frame = WdclFrame {
1048            destination: signed_id,
1049            frame_type: WDCL_T_DISCOVER,
1050            payload,
1051        }
1052        .encode();
1053        assert!(matches!(
1054            parse_device_frame(&frame, [9; 4]),
1055            Some(WeaveInput::Discovery { .. })
1056        ));
1057
1058        let wire = crate::hdlc::frame(&frame);
1059        let mut decoder = crate::hdlc::Decoder::with_min_frame_size(5);
1060        assert!(decoder.feed(&wire[..3]).is_empty());
1061        assert_eq!(decoder.feed(&wire[3..]), vec![frame]);
1062    }
1063
1064    #[test]
1065    fn board_init_unknown_and_duplicate_state_are_preserved() {
1066        let mut state = WeaveState::default();
1067        state.handle_log(
1068            WeaveLogFrame {
1069                timestamp: 1.0,
1070                level: 1,
1071                event: ET_BOARD_INIT,
1072                data: vec![1],
1073            },
1074            1.0,
1075        );
1076        state.handle_log(
1077            WeaveLogFrame {
1078                timestamp: 2.0,
1079                level: 1,
1080                event: 0x7777,
1081                data: vec![2],
1082            },
1083            2.0,
1084        );
1085        assert_eq!(state.logs.len(), 2);
1086        assert_eq!(state.unknown_events.len(), 1);
1087        assert!(state.accept_packet(b"packet", 3.0));
1088        assert!(!state.accept_packet(b"packet", 3.1));
1089        assert!(state.accept_packet(b"packet", 4.0));
1090    }
1091
1092    #[test]
1093    fn simulated_session_authenticates_and_manages_endpoint_lifecycle() {
1094        let (tx, rx) = crate::event::channel();
1095        let config = WeaveConfig {
1096            name: "weave-test".into(),
1097            port: "/dev/null".into(),
1098            configured_bitrate: WEAVE_DEFAULT_BITRATE,
1099            interface_id: InterfaceId(55),
1100            state: Arc::new(Mutex::new(WeaveState::default())),
1101        };
1102        let context = StartContext {
1103            tx,
1104            next_dynamic_id: Arc::new(AtomicU64::new(10_000)),
1105            mode: rns_core::constants::MODE_GATEWAY,
1106            gravity: 0,
1107            recursive_prs: true,
1108            announces_from_internal: false,
1109            announces_to_internal: None,
1110            ingress_control: rns_core::transport::types::IngressControlConfig::enabled(),
1111            stamp_costs: Default::default(),
1112            ifac: None,
1113            underlay_mark: None,
1114        };
1115        let session = WeaveSession::new(config, &context);
1116        *super::super::lock_or_recover(&session.writer, "test writer") =
1117            Some(DeviceWriter::new(tempfile::tempfile().unwrap()));
1118
1119        let remote = Ed25519PrivateKey::generate(&mut FixedRng::new(&[0x44; 64]));
1120        let remote_public = remote.public_key().public_bytes();
1121        let mut discovery_payload = remote_public.to_vec();
1122        discovery_payload.extend_from_slice(&remote.sign(&session.host_switch_id));
1123        session
1124            .process_frame(
1125                &WdclFrame {
1126                    destination: session.host_switch_id,
1127                    frame_type: WDCL_T_DISCOVER,
1128                    payload: discovery_payload,
1129                }
1130                .encode(),
1131            )
1132            .unwrap();
1133        assert_eq!(
1134            *super::super::lock_or_recover(&session.remote_switch_id, "test remote"),
1135            Some(remote_public[28..].try_into().unwrap())
1136        );
1137
1138        session
1139            .process_frame(&log_frame(
1140                session.host_switch_id,
1141                ET_PROTO_WDCL_CONNECTION,
1142                &[1],
1143            ))
1144            .unwrap();
1145        assert!(matches!(
1146            rx.recv_timeout(Duration::from_secs(1)).unwrap(),
1147            Event::InterfaceUp(InterfaceId(55), None, None)
1148        ));
1149
1150        let endpoint = [0x77; 8];
1151        session
1152            .process_frame(&log_frame(
1153                session.host_switch_id,
1154                ET_PROTO_WEAVE_EP_ALIVE,
1155                &endpoint,
1156            ))
1157            .unwrap();
1158        let peer_id = loop {
1159            if let Event::DynamicInterfaceUp {
1160                id, registration, ..
1161            } = rx.recv_timeout(Duration::from_secs(1)).unwrap()
1162            {
1163                assert_eq!(registration.interface_type, "WeaveInterfacePeer");
1164                assert_eq!(registration.parent_id, InterfaceId(55));
1165                break id;
1166            }
1167        };
1168
1169        let mut endpoint_payload = b"rns-packet".to_vec();
1170        endpoint_payload.extend_from_slice(&endpoint);
1171        let endpoint_frame = WdclFrame {
1172            destination: session.host_switch_id,
1173            frame_type: WDCL_T_ENDPOINT_PKT,
1174            payload: endpoint_payload,
1175        }
1176        .encode();
1177        session.process_frame(&endpoint_frame).unwrap();
1178        assert!(loop {
1179            if let Event::Frame {
1180                interface_id, data, ..
1181            } = rx.recv_timeout(Duration::from_secs(1)).unwrap()
1182            {
1183                break interface_id == peer_id && *data == b"rns-packet"[..];
1184            }
1185        });
1186        session.process_frame(&endpoint_frame).unwrap();
1187        assert!(rx.recv_timeout(Duration::from_millis(50)).is_err());
1188
1189        session.expire_peers(crate::time::now() + WEAVE_PEERING_TIMEOUT + 1.0);
1190        assert!(loop {
1191            if let Event::InterfaceDown(id) = rx.recv_timeout(Duration::from_secs(1)).unwrap() {
1192                break id == peer_id;
1193            }
1194        });
1195    }
1196
1197    #[test]
1198    fn folded_weave_device_runs_on_the_driver_loop() {
1199        use std::os::fd::FromRawFd;
1200        if !crate::engine::FOLD_IO {
1201            return;
1202        }
1203        let tempdir = tempfile::tempdir().unwrap();
1204        let port_path = tempdir.path().join("weave-port");
1205        let (master_fd, slave_fd) = crate::serial::open_pty_pair().unwrap();
1206        std::os::unix::fs::symlink(
1207            std::fs::read_link(format!("/proc/self/fd/{slave_fd}")).unwrap(),
1208            &port_path,
1209        )
1210        .unwrap();
1211        let mut device = unsafe { std::fs::File::from_raw_fd(master_fd) };
1212        let _slave = unsafe { std::fs::File::from_raw_fd(slave_fd) };
1213
1214        let (tx, rx) = crate::event::channel();
1215        let config = WeaveConfig {
1216            name: "weave-folded".into(),
1217            port: port_path.display().to_string(),
1218            configured_bitrate: WEAVE_DEFAULT_BITRATE,
1219            interface_id: InterfaceId(56),
1220            state: Arc::new(Mutex::new(WeaveState::default())),
1221        };
1222        let context = StartContext {
1223            tx: tx.clone(),
1224            next_dynamic_id: Arc::new(AtomicU64::new(11_000)),
1225            mode: rns_core::constants::MODE_GATEWAY,
1226            gravity: 0,
1227            recursive_prs: true,
1228            announces_from_internal: false,
1229            announces_to_internal: None,
1230            ingress_control: rns_core::transport::types::IngressControlConfig::enabled(),
1231            stamp_costs: Default::default(),
1232            ifac: None,
1233            underlay_mark: None,
1234        };
1235        let session = WeaveSession::new(config, &context);
1236        session.open_folded().unwrap();
1237        let host = session.host_switch_id;
1238
1239        // The device reads the host's discovery frames, and answers.
1240        let mut decoder = crate::hdlc::Decoder::with_limits(5, 128 * 1024);
1241        let mut device_reads = |device: &mut std::fs::File| -> Vec<Vec<u8>> {
1242            let mut buf = [0u8; 4096];
1243            let mut pfd = libc::pollfd {
1244                fd: device.as_raw_fd(),
1245                events: libc::POLLIN,
1246                revents: 0,
1247            };
1248            if unsafe { libc::poll(&mut pfd, 1, 2000) } <= 0 {
1249                return Vec::new();
1250            }
1251            let n = device.read(&mut buf).unwrap();
1252            decoder.feed(&buf[..n])
1253        };
1254        let discover = WdclFrame::decode(&device_reads(&mut device)[0]).unwrap();
1255        assert_eq!(discover.frame_type, WDCL_T_DISCOVER);
1256        let remote = Ed25519PrivateKey::generate(&mut FixedRng::new(&[0x45; 64]));
1257        let remote_public = remote.public_key().public_bytes();
1258        let mut payload = remote_public.to_vec();
1259        payload.extend_from_slice(&remote.sign(&host));
1260        let mut wire = crate::hdlc::frame(
1261            &WdclFrame {
1262                destination: host,
1263                frame_type: WDCL_T_DISCOVER,
1264                payload,
1265            }
1266            .encode(),
1267        );
1268        wire.extend_from_slice(&crate::hdlc::frame(&log_frame(
1269            host,
1270            ET_PROTO_WDCL_CONNECTION,
1271            &[1],
1272        )));
1273        let endpoint = [0x78; 8];
1274        wire.extend_from_slice(&crate::hdlc::frame(&log_frame(
1275            host,
1276            ET_PROTO_WEAVE_EP_ALIVE,
1277            &endpoint,
1278        )));
1279        device.write_all(&wire).unwrap();
1280
1281        // Read by the driver loop: the parent comes up, then the peer.
1282        let mut peer = None;
1283        let deadline = Instant::now() + Duration::from_secs(3);
1284        let mut parent_up = false;
1285        while peer.is_none() && Instant::now() < deadline {
1286            match rx.recv_timeout(Duration::from_millis(200)) {
1287                Ok(Event::InterfaceUp(InterfaceId(56), None, None)) => parent_up = true,
1288                Ok(Event::DynamicInterfaceUp { id, writer, .. }) => peer = Some((id, writer)),
1289                _ => {}
1290            }
1291        }
1292        assert!(parent_up);
1293        let (peer_id, writer) = peer.expect("peer came up");
1294        assert!(writer.sends_inline());
1295        let (mut writer, _) = crate::interface::wrap_async_writer(writer, peer_id, "weave", tx, 16);
1296
1297        // The host answered the discovery with a connect request.
1298        let connect = device_reads(&mut device)
1299            .into_iter()
1300            .map(|raw| WdclFrame::decode(&raw).unwrap())
1301            .find(|frame| frame.frame_type == WDCL_T_CONNECT)
1302            .expect("connect request");
1303        assert_eq!(connect.destination, remote_public[28..]);
1304
1305        // Outbound through the peer's writer, inbound from the peer.
1306        writer.send_frame(b"to the peer").unwrap();
1307        writer.flush_pending();
1308        let sent = WdclFrame::decode(&device_reads(&mut device)[0]).unwrap();
1309        assert_eq!(
1310            sent,
1311            WdclFrame::endpoint_packet(
1312                remote_public[28..].try_into().unwrap(),
1313                endpoint,
1314                b"to the peer"
1315            )
1316        );
1317        let mut endpoint_payload = b"from the peer".to_vec();
1318        endpoint_payload.extend_from_slice(&endpoint);
1319        device
1320            .write_all(&crate::hdlc::frame(
1321                &WdclFrame {
1322                    destination: host,
1323                    frame_type: WDCL_T_ENDPOINT_PKT,
1324                    payload: endpoint_payload,
1325                }
1326                .encode(),
1327            ))
1328            .unwrap();
1329        let received = loop {
1330            match rx.recv_timeout(Duration::from_secs(2)).unwrap() {
1331                Event::Frame {
1332                    interface_id, data, ..
1333                } => break (interface_id, data),
1334                _ => continue,
1335            }
1336        };
1337        assert_eq!(received.0, peer_id);
1338        assert_eq!(&*received.1, b"from the peer");
1339
1340        // Losing the device brings the peer and the parent down.
1341        drop(device);
1342        let mut down = Vec::new();
1343        while let Ok(event) = rx.recv_timeout(Duration::from_millis(500)) {
1344            if let Event::InterfaceDown(id) = event {
1345                down.push(id);
1346            }
1347        }
1348        assert!(down.contains(&peer_id));
1349        assert!(down.contains(&InterfaceId(56)));
1350    }
1351}
rns_net/interface/weave.rs

rns_net/interface/
weave.rs
