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::{DynamicInterfaceRegistration, Event, EventSender, InterfaceTelemetry};
18use crate::serial::{Parity, SerialConfig, SerialPort};
19
20pub const WDCL_T_DISCOVER: u8 = 0x00;
21pub const WDCL_T_CONNECT: u8 = 0x01;
22pub const WDCL_T_CMD: u8 = 0x02;
23pub const WDCL_T_LOG: u8 = 0x03;
24pub const WDCL_T_DISP: u8 = 0x04;
25pub const WDCL_T_ENDPOINT_PKT: u8 = 0x05;
26pub const WDCL_T_ENCAP_PROTO: u8 = 0x06;
27pub const WDCL_BROADCAST: [u8; 4] = [0xff; 4];
28
29pub const WDCL_CMD_ENDPOINT_PKT: u16 = 0x0001;
30pub const WDCL_CMD_ENDPOINTS_LIST: u16 = 0x0100;
31pub const WDCL_CMD_REMOTE_DISPLAY: u16 = 0x0a00;
32pub const WDCL_CMD_REMOTE_INPUT: u16 = 0x0a01;
33
34pub const ET_BOARD_INIT: u16 = 0x0003;
35pub const ET_PROTO_WDCL_CONNECTION: u16 = 0x3002;
36pub const ET_PROTO_WDCL_HOST_ENDPOINT: u16 = 0x3003;
37pub const ET_PROTO_WEAVE_EP_ALIVE: u16 = 0x3102;
38pub const ET_PROTO_WEAVE_EP_TIMEOUT: u16 = 0x3103;
39pub const ET_PROTO_WEAVE_EP_VIA: u16 = 0x3104;
40pub const ET_STAT_CPU: u16 = 0xe003;
41pub const ET_STAT_MEMORY: u16 = 0xe005;
42
43pub const WEAVE_HW_MTU: u32 = 1024;
44pub const WEAVE_DEFAULT_BITRATE: u64 = 250_000;
45pub const WEAVE_PEERING_TIMEOUT: f64 = 20.0;
46pub const WEAVE_DUPLICATE_TTL: f64 = 0.75;
47pub const WEAVE_DUPLICATE_CAPACITY: usize = 48;
48
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct WdclFrame {
51    pub destination: [u8; 4],
52    pub frame_type: u8,
53    pub payload: Vec<u8>,
54}
55
56impl WdclFrame {
57    pub fn encode(&self) -> Vec<u8> {
58        let mut output = Vec::with_capacity(5 + self.payload.len());
59        output.extend_from_slice(&self.destination);
60        output.push(self.frame_type);
61        output.extend_from_slice(&self.payload);
62        output
63    }
64
65    pub fn decode(data: &[u8]) -> Option<Self> {
66        if data.len() < 5 {
67            return None;
68        }
69        Some(Self {
70            destination: data[..4].try_into().ok()?,
71            frame_type: data[4],
72            payload: data[5..].to_vec(),
73        })
74    }
75
76    pub fn command(destination: [u8; 4], command: u16, data: &[u8]) -> Self {
77        let mut payload = Vec::with_capacity(2 + data.len());
78        payload.extend_from_slice(&command.to_be_bytes());
79        payload.extend_from_slice(data);
80        Self {
81            destination,
82            frame_type: WDCL_T_CMD,
83            payload,
84        }
85    }
86
87    pub fn endpoint_packet(destination: [u8; 4], endpoint: [u8; 8], packet: &[u8]) -> Self {
88        let mut data = Vec::with_capacity(8 + packet.len());
89        data.extend_from_slice(&endpoint);
90        data.extend_from_slice(packet);
91        Self::command(destination, WDCL_CMD_ENDPOINT_PKT, &data)
92    }
93}
94
95#[derive(Debug, Clone, PartialEq)]
96pub struct WeaveLogFrame {
97    pub timestamp: f64,
98    pub level: u8,
99    pub event: u16,
100    pub data: Vec<u8>,
101}
102
103#[derive(Debug, Clone, PartialEq)]
104pub enum WeaveInput {
105    Discovery {
106        switch_id: [u8; 4],
107        signing_key: [u8; 32],
108    },
109    EndpointPacket {
110        source: [u8; 8],
111        packet: Vec<u8>,
112    },
113    Log(WeaveLogFrame),
114    Display {
115        offset: usize,
116        total: usize,
117        fragment: Vec<u8>,
118    },
119    Unknown(WdclFrame),
120}
121
122/// Validate and classify a device frame. Discovery signatures cover the
123/// destination/signed switch ID exactly as in Reticulum 1.3.8.
124pub fn parse_device_frame(data: &[u8], host_switch_id: [u8; 4]) -> Option<WeaveInput> {
125    let frame = WdclFrame::decode(data)?;
126    match frame.frame_type {
127        WDCL_T_DISCOVER if frame.payload.len() == 96 => {
128            let key: [u8; 32] = frame.payload[..32].try_into().ok()?;
129            let signature: [u8; 64] = frame.payload[32..].try_into().ok()?;
130            if !Ed25519PublicKey::from_bytes(&key).verify(&signature, &frame.destination) {
131                return None;
132            }
133            Some(WeaveInput::Discovery {
134                switch_id: key[28..].try_into().ok()?,
135                signing_key: key,
136            })
137        }
138        WDCL_T_ENDPOINT_PKT if frame.destination == host_switch_id && frame.payload.len() > 8 => {
139            let split = frame.payload.len() - 8;
140            Some(WeaveInput::EndpointPacket {
141                source: frame.payload[split..].try_into().ok()?,
142                packet: frame.payload[..split].to_vec(),
143            })
144        }
145        WDCL_T_LOG if frame.payload.len() >= 9 => {
146            // Firmware prepends a flow byte to the eight-byte log header.
147            let data = &frame.payload[1..];
148            Some(WeaveInput::Log(WeaveLogFrame {
149                timestamp: u32::from_be_bytes(data[..4].try_into().ok()?) as f64 / 1000.0,
150                level: data[4],
151                event: u16::from_be_bytes(data[5..7].try_into().ok()?),
152                data: data[7..].to_vec(),
153            }))
154        }
155        WDCL_T_DISP if frame.payload.len() >= 9 => Some(WeaveInput::Display {
156            offset: u32::from_be_bytes(frame.payload[1..5].try_into().ok()?) as usize,
157            total: u32::from_be_bytes(frame.payload[5..9].try_into().ok()?) as usize,
158            fragment: frame.payload[9..].to_vec(),
159        }),
160        _ => Some(WeaveInput::Unknown(frame)),
161    }
162}
163
164#[derive(Debug, Clone)]
165pub struct WeavePeerState {
166    pub endpoint_id: [u8; 8],
167    pub via_switch_id: Option<[u8; 4]>,
168    pub last_heard: f64,
169}
170
171#[derive(Debug, Clone, Default)]
172pub struct WeaveState {
173    pub connected: bool,
174    pub switch_id: Option<[u8; 4]>,
175    pub endpoint_id: Option<[u8; 8]>,
176    pub cpu_load: Option<u8>,
177    pub mem_load: Option<f64>,
178    pub peers: HashMap<[u8; 8], WeavePeerState>,
179    pub logs: VecDeque<WeaveLogFrame>,
180    pub display: Vec<u8>,
181    pub unknown_events: VecDeque<WeaveLogFrame>,
182    duplicates: VecDeque<([u8; 32], f64)>,
183}
184
185impl WeaveState {
186    pub fn accept_packet(&mut self, packet: &[u8], now: f64) -> bool {
187        while self
188            .duplicates
189            .front()
190            .is_some_and(|entry| now >= entry.1 + WEAVE_DUPLICATE_TTL)
191        {
192            self.duplicates.pop_front();
193        }
194        let hash = sha256(packet);
195        if self.duplicates.iter().any(|entry| entry.0 == hash) {
196            return false;
197        }
198        if self.duplicates.len() == WEAVE_DUPLICATE_CAPACITY {
199            self.duplicates.pop_front();
200        }
201        self.duplicates.push_back((hash, now));
202        true
203    }
204
205    pub fn handle_log(&mut self, frame: WeaveLogFrame, now: f64) {
206        if self.logs.len() == 1024 {
207            self.logs.pop_front();
208        }
209        self.logs.push_back(frame.clone());
210        match frame.event {
211            ET_BOARD_INIT => {}
212            ET_PROTO_WDCL_CONNECTION => self.connected = true,
213            ET_PROTO_WDCL_HOST_ENDPOINT if frame.data.len() == 8 => {
214                self.endpoint_id = frame.data.as_slice().try_into().ok();
215            }
216            ET_PROTO_WEAVE_EP_ALIVE if frame.data.len() == 8 => {
217                let endpoint_id: [u8; 8] = frame.data.as_slice().try_into().unwrap();
218                self.peers
219                    .entry(endpoint_id)
220                    .and_modify(|peer| peer.last_heard = now)
221                    .or_insert(WeavePeerState {
222                        endpoint_id,
223                        via_switch_id: None,
224                        last_heard: now,
225                    });
226            }
227            ET_PROTO_WEAVE_EP_TIMEOUT if frame.data.len() == 8 => {
228                if let Ok(endpoint) = <[u8; 8]>::try_from(frame.data.as_slice()) {
229                    self.peers.remove(&endpoint);
230                }
231            }
232            ET_PROTO_WEAVE_EP_VIA if frame.data.len() == 12 => {
233                let endpoint: [u8; 8] = frame.data[..8].try_into().unwrap();
234                let via: [u8; 4] = frame.data[8..].try_into().unwrap();
235                if let Some(peer) = self.peers.get_mut(&endpoint) {
236                    peer.via_switch_id = Some(via);
237                }
238            }
239            ET_STAT_CPU if !frame.data.is_empty() => self.cpu_load = Some(frame.data[0]),
240            ET_STAT_MEMORY if frame.data.len() == 8 => {
241                let free = u32::from_be_bytes(frame.data[..4].try_into().unwrap()) as f64;
242                let total = u32::from_be_bytes(frame.data[4..].try_into().unwrap()) as f64;
243                self.mem_load = (total > 0.0).then_some(((total - free) / total) * 100.0);
244            }
245            _ => self.unknown_events.push_back(frame),
246        }
247    }
248
249    pub fn expire_peers(&mut self, now: f64) -> Vec<[u8; 8]> {
250        let expired: Vec<_> = self
251            .peers
252            .iter()
253            .filter_map(|(id, peer)| {
254                (now >= peer.last_heard + WEAVE_PEERING_TIMEOUT).then_some(*id)
255            })
256            .collect();
257        for id in &expired {
258            self.peers.remove(id);
259        }
260        expired
261    }
262
263    pub fn apply_display(&mut self, offset: usize, total: usize, fragment: &[u8]) -> bool {
264        if total > 128 * 64 / 8 || offset > total || fragment.len() > total - offset {
265            return false;
266        }
267        self.display.resize(total, 0);
268        self.display[offset..offset + fragment.len()].copy_from_slice(fragment);
269        offset + fragment.len() == total
270    }
271}
272
273#[derive(Debug, Clone)]
274pub struct WeaveConfig {
275    pub name: String,
276    pub port: String,
277    pub configured_bitrate: u64,
278    pub interface_id: InterfaceId,
279    pub state: Arc<Mutex<WeaveState>>,
280}
281
282struct WeaveSession {
283    config: WeaveConfig,
284    tx: EventSender,
285    next_dynamic_id: Arc<AtomicU64>,
286    mode: u8,
287    gravity: i64,
288    recursive_prs: bool,
289    announces_from_internal: bool,
290    announces_to_internal: Option<bool>,
291    ingress_control: rns_core::transport::types::IngressControlConfig,
292    ifac: Option<crate::ifac::IfacState>,
293    host_identity: Arc<Ed25519PrivateKey>,
294    host_public: [u8; 32],
295    host_switch_id: [u8; 4],
296    writer: Arc<Mutex<Option<std::fs::File>>>,
297    remote_switch_id: Arc<Mutex<Option<[u8; 4]>>>,
298    peer_interfaces: Arc<Mutex<HashMap<[u8; 8], InterfaceId>>>,
299}
300
301impl WeaveSession {
302    fn new(config: WeaveConfig, ctx: &StartContext) -> Arc<Self> {
303        let identity = Ed25519PrivateKey::generate(&mut OsRng);
304        let host_public = identity.public_key().public_bytes();
305        Arc::new(Self {
306            config,
307            tx: ctx.tx.clone(),
308            next_dynamic_id: Arc::clone(&ctx.next_dynamic_id),
309            mode: ctx.mode,
310            gravity: ctx.gravity,
311            recursive_prs: ctx.recursive_prs,
312            announces_from_internal: ctx.announces_from_internal,
313            announces_to_internal: ctx.announces_to_internal,
314            ingress_control: ctx.ingress_control,
315            ifac: ctx.ifac.clone(),
316            host_identity: Arc::new(identity),
317            host_public,
318            host_switch_id: host_public[28..].try_into().unwrap(),
319            writer: Arc::new(Mutex::new(None)),
320            remote_switch_id: Arc::new(Mutex::new(None)),
321            peer_interfaces: Arc::new(Mutex::new(HashMap::new())),
322        })
323    }
324
325    fn spawn(self: &Arc<Self>) -> io::Result<()> {
326        let session = Arc::clone(self);
327        thread::Builder::new()
328            .name(format!("weave-session-{}", self.config.interface_id.0))
329            .spawn(move || session.reconnect_loop())?;
330        Ok(())
331    }
332
333    fn reconnect_loop(self: Arc<Self>) {
334        loop {
335            if let Err(error) = self.run_connection() {
336                log::warn!(
337                    "[{}] Weave connection failed: {}; retrying in five seconds",
338                    self.config.name,
339                    error
340                );
341            }
342            self.disconnect_all();
343            thread::sleep(Duration::from_secs(5));
344        }
345    }
346
347    fn run_connection(self: &Arc<Self>) -> io::Result<()> {
348        let port = SerialPort::open(&SerialConfig {
349            path: self.config.port.clone(),
350            baud: 3_000_000,
351            data_bits: 8,
352            parity: Parity::None,
353            stop_bits: 1,
354        })?;
355        let mut reader = port.reader()?;
356        *super::lock_or_recover(&self.writer, "weave writer") = Some(port.writer()?);
357        *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") = None;
358        {
359            let mut state = super::lock_or_recover(&self.config.state, "weave state");
360            state.connected = false;
361            state.switch_id = None;
362        }
363
364        self.send_frame(&WdclFrame {
365            destination: WDCL_BROADCAST,
366            frame_type: WDCL_T_DISCOVER,
367            payload: self.host_switch_id.to_vec(),
368        })?;
369
370        let handshake_deadline = Instant::now() + Duration::from_secs(2);
371        let mut decoder = crate::hdlc::Decoder::with_limits(5, 128 * 1024);
372        let mut buffer = [0u8; 32 * 1024];
373        let mut last_peer_expiry = Instant::now();
374        loop {
375            let mut pollfd = libc::pollfd {
376                fd: reader.as_raw_fd(),
377                events: libc::POLLIN,
378                revents: 0,
379            };
380            let ready = unsafe { libc::poll(&mut pollfd, 1, 250) };
381            if ready < 0 {
382                return Err(io::Error::last_os_error());
383            }
384            if ready > 0 {
385                let read = reader.read(&mut buffer)?;
386                if read == 0 {
387                    return Err(io::Error::new(
388                        io::ErrorKind::UnexpectedEof,
389                        "serial port closed",
390                    ));
391                }
392                for frame in decoder.feed(&buffer[..read]) {
393                    self.process_frame(&frame)?;
394                }
395            }
396
397            let connected = super::lock_or_recover(&self.config.state, "weave state").connected;
398            if !connected && Instant::now() >= handshake_deadline {
399                return Err(io::Error::new(
400                    io::ErrorKind::TimedOut,
401                    "WDCL connection handshake timed out",
402                ));
403            }
404            if last_peer_expiry.elapsed() >= Duration::from_secs(1) {
405                self.expire_peers(crate::time::now());
406                last_peer_expiry = Instant::now();
407            }
408        }
409    }
410
411    fn send_frame(&self, frame: &WdclFrame) -> io::Result<()> {
412        let encoded = crate::hdlc::frame(&frame.encode());
413        let mut writer = super::lock_or_recover(&self.writer, "weave writer");
414        writer
415            .as_mut()
416            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "Weave is offline"))?
417            .write_all(&encoded)
418    }
419
420    fn process_frame(self: &Arc<Self>, raw: &[u8]) -> io::Result<()> {
421        let Some(input) = parse_device_frame(raw, self.host_switch_id) else {
422            return Ok(());
423        };
424        match input {
425            WeaveInput::Discovery {
426                switch_id,
427                signing_key: _,
428            } => {
429                *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") =
430                    Some(switch_id);
431                super::lock_or_recover(&self.config.state, "weave state").switch_id =
432                    Some(switch_id);
433                let mut payload = self.host_public.to_vec();
434                payload.extend_from_slice(&self.host_identity.sign(&switch_id));
435                self.send_frame(&WdclFrame {
436                    destination: switch_id,
437                    frame_type: WDCL_T_CONNECT,
438                    payload,
439                })?;
440            }
441            WeaveInput::EndpointPacket { source, packet } => {
442                let now = crate::time::now();
443                {
444                    let mut state = super::lock_or_recover(&self.config.state, "weave state");
445                    state
446                        .peers
447                        .entry(source)
448                        .and_modify(|peer| peer.last_heard = now)
449                        .or_insert(WeavePeerState {
450                            endpoint_id: source,
451                            via_switch_id: None,
452                            last_heard: now,
453                        });
454                    if !state.accept_packet(&packet, now) {
455                        return Ok(());
456                    }
457                }
458                let interface_id = self.ensure_peer(source, now);
459                let _ = self.tx.send(Event::Frame {
460                    interface_id,
461                    data: packet,
462                    rssi: None,
463                    snr: None,
464                });
465            }
466            WeaveInput::Log(frame) => self.process_log(frame),
467            WeaveInput::Display {
468                offset,
469                total,
470                fragment,
471            } => {
472                super::lock_or_recover(&self.config.state, "weave state")
473                    .apply_display(offset, total, &fragment);
474            }
475            WeaveInput::Unknown(_) => {}
476        }
477        Ok(())
478    }
479
480    fn process_log(self: &Arc<Self>, frame: WeaveLogFrame) {
481        let now = crate::time::now();
482        let event = frame.event;
483        let data = frame.data.clone();
484        super::lock_or_recover(&self.config.state, "weave state").handle_log(frame, now);
485
486        match event {
487            ET_PROTO_WDCL_CONNECTION => {
488                let _ = self
489                    .tx
490                    .send(Event::InterfaceUp(self.config.interface_id, None, None));
491                self.publish_parent_telemetry();
492            }
493            ET_PROTO_WEAVE_EP_ALIVE if data.len() == 8 => {
494                let endpoint: [u8; 8] = data.as_slice().try_into().unwrap();
495                self.ensure_peer(endpoint, now);
496            }
497            ET_PROTO_WEAVE_EP_TIMEOUT if data.len() == 8 => {
498                let endpoint: [u8; 8] = data.as_slice().try_into().unwrap();
499                self.remove_peer(endpoint);
500            }
501            ET_PROTO_WEAVE_EP_VIA if data.len() == 12 => {
502                let endpoint: [u8; 8] = data[..8].try_into().unwrap();
503                let via: [u8; 4] = data[8..].try_into().unwrap();
504                if let Some(interface_id) =
505                    super::lock_or_recover(&self.peer_interfaces, "weave peer interfaces")
506                        .get(&endpoint)
507                        .copied()
508                {
509                    let _ = self.tx.send(Event::InterfaceTelemetry {
510                        interface_id,
511                        telemetry: InterfaceTelemetry {
512                            via_switch_id: Some(via),
513                            ..Default::default()
514                        },
515                    });
516                }
517            }
518            ET_PROTO_WDCL_HOST_ENDPOINT | ET_STAT_CPU | ET_STAT_MEMORY => {
519                self.publish_parent_telemetry()
520            }
521            _ => {}
522        }
523    }
524
525    fn ensure_peer(self: &Arc<Self>, endpoint: [u8; 8], now: f64) -> InterfaceId {
526        if let Some(id) = super::lock_or_recover(&self.peer_interfaces, "weave peers")
527            .get(&endpoint)
528            .copied()
529        {
530            return id;
531        }
532        let id = InterfaceId(self.next_dynamic_id.fetch_add(1, Ordering::Relaxed));
533        super::lock_or_recover(&self.peer_interfaces, "weave peers").insert(endpoint, id);
534        let via_switch_id = super::lock_or_recover(&self.config.state, "weave state")
535            .peers
536            .get(&endpoint)
537            .and_then(|peer| peer.via_switch_id);
538        let name = format!("WeaveInterfacePeer[{}]", hex(&endpoint));
539        let info = InterfaceInfo {
540            id,
541            name,
542            mode: self.mode,
543            gravity: self.gravity,
544            recursive_prs: self.recursive_prs,
545            announces_from_internal: self.announces_from_internal,
546            announces_to_internal: self.announces_to_internal,
547            out_capable: true,
548            in_capable: true,
549            bitrate: Some(self.config.configured_bitrate),
550            airtime_profile: None,
551            announce_rate_target: None,
552            announce_rate_grace: 0,
553            announce_rate_penalty: 0.0,
554            announce_cap: rns_core::constants::ANNOUNCE_CAP,
555            is_local_client: false,
556            wants_tunnel: false,
557            tunnel_id: None,
558            mtu: WEAVE_HW_MTU,
559            ia_freq: 0.0,
560            ip_freq: 0.0,
561            op_freq: 0.0,
562            op_samples: 0,
563            started: now,
564            ingress_control: self.ingress_control,
565        };
566        let _ = self.tx.send(Event::DynamicInterfaceUp {
567            id,
568            writer: Box::new(WeavePeerWriter {
569                session: Arc::clone(self),
570                endpoint,
571            }),
572            registration: DynamicInterfaceRegistration {
573                info,
574                interface_type: "WeaveInterfacePeer".into(),
575                parent_id: self.config.interface_id,
576                telemetry: InterfaceTelemetry {
577                    via_switch_id,
578                    ..Default::default()
579                },
580                ifac: self.ifac.clone(),
581            },
582        });
583        self.publish_parent_telemetry();
584        id
585    }
586
587    fn remove_peer(&self, endpoint: [u8; 8]) {
588        if let Some(id) =
589            super::lock_or_recover(&self.peer_interfaces, "weave peers").remove(&endpoint)
590        {
591            super::lock_or_recover(&self.config.state, "weave state")
592                .peers
593                .remove(&endpoint);
594            let _ = self.tx.send(Event::InterfaceDown(id));
595            self.publish_parent_telemetry();
596        }
597    }
598
599    fn expire_peers(&self, now: f64) {
600        let expired = super::lock_or_recover(&self.config.state, "weave state").expire_peers(now);
601        for endpoint in &expired {
602            if let Some(id) =
603                super::lock_or_recover(&self.peer_interfaces, "weave peers").remove(endpoint)
604            {
605                let _ = self.tx.send(Event::InterfaceDown(id));
606            }
607        }
608        if !super::lock_or_recover(&self.peer_interfaces, "weave peers").is_empty()
609            || !expired.is_empty()
610        {
611            self.publish_parent_telemetry();
612        }
613    }
614
615    fn publish_parent_telemetry(&self) {
616        let state = super::lock_or_recover(&self.config.state, "weave state");
617        let _ = self.tx.send(Event::InterfaceTelemetry {
618            interface_id: self.config.interface_id,
619            telemetry: InterfaceTelemetry {
620                cpu_load: state.cpu_load.map(f64::from),
621                mem_load: state.mem_load,
622                switch_id: state.switch_id,
623                endpoint_id: state.endpoint_id,
624                via_switch_id: None,
625                peers: Some(state.peers.len()),
626            },
627        });
628    }
629
630    fn disconnect_all(&self) {
631        *super::lock_or_recover(&self.writer, "weave writer") = None;
632        *super::lock_or_recover(&self.remote_switch_id, "weave remote switch") = None;
633        {
634            let mut state = super::lock_or_recover(&self.config.state, "weave state");
635            state.connected = false;
636            state.switch_id = None;
637            state.peers.clear();
638        }
639        let peers: Vec<_> = super::lock_or_recover(&self.peer_interfaces, "weave peers")
640            .drain()
641            .map(|(_, id)| id)
642            .collect();
643        for id in peers {
644            let _ = self.tx.send(Event::InterfaceDown(id));
645        }
646        let _ = self.tx.send(Event::InterfaceDown(self.config.interface_id));
647        self.publish_parent_telemetry();
648    }
649}
650
651struct WeavePeerWriter {
652    session: Arc<WeaveSession>,
653    endpoint: [u8; 8],
654}
655
656impl Writer for WeavePeerWriter {
657    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
658        let remote = super::lock_or_recover(&self.session.remote_switch_id, "weave remote switch")
659            .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "Weave is offline"))?;
660        self.session
661            .send_frame(&WdclFrame::endpoint_packet(remote, self.endpoint, data))
662    }
663}
664
665fn hex(bytes: &[u8]) -> String {
666    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
667}
668
669pub struct WeaveFactory;
670struct ParentWriter;
671impl Writer for ParentWriter {
672    fn send_frame_confirmed(&mut self, _data: &[u8]) -> io::Result<()> {
673        Err(io::Error::new(
674            io::ErrorKind::Unsupported,
675            "Weave parent is not a transmitting peer",
676        ))
677    }
678    fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
679        Ok(())
680    }
681}
682
683impl InterfaceFactory for WeaveFactory {
684    fn type_name(&self) -> &str {
685        "WeaveInterface"
686    }
687    fn default_ifac_size(&self) -> usize {
688        16
689    }
690
691    fn parse_config(
692        &self,
693        name: &str,
694        id: InterfaceId,
695        params: &HashMap<String, String>,
696    ) -> Result<Box<dyn InterfaceConfigData>, String> {
697        let port = params
698            .get("port")
699            .cloned()
700            .ok_or_else(|| "WeaveInterface requires 'port'".to_string())?;
701        let configured_bitrate = params
702            .get("configured_bitrate")
703            .and_then(|v| v.parse().ok())
704            .unwrap_or(WEAVE_DEFAULT_BITRATE);
705        Ok(Box::new(WeaveConfig {
706            name: name.into(),
707            port,
708            configured_bitrate,
709            interface_id: id,
710            state: Arc::new(Mutex::new(WeaveState::default())),
711        }))
712    }
713
714    fn start(
715        &self,
716        config: Box<dyn InterfaceConfigData>,
717        ctx: StartContext,
718    ) -> io::Result<StartResult> {
719        let config = *config
720            .into_any()
721            .downcast::<WeaveConfig>()
722            .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "wrong Weave config type"))?;
723        let session = WeaveSession::new(config.clone(), &ctx);
724        session.spawn()?;
725        let info = InterfaceInfo {
726            id: config.interface_id,
727            name: config.name.clone(),
728            mode: ctx.mode,
729            gravity: ctx.gravity,
730            recursive_prs: ctx.recursive_prs,
731            announces_from_internal: ctx.announces_from_internal,
732            announces_to_internal: ctx.announces_to_internal,
733            out_capable: false,
734            in_capable: true,
735            bitrate: Some(config.configured_bitrate),
736            airtime_profile: None,
737            announce_rate_target: None,
738            announce_rate_grace: 0,
739            announce_rate_penalty: 0.0,
740            announce_cap: rns_core::constants::ANNOUNCE_CAP,
741            is_local_client: false,
742            wants_tunnel: false,
743            tunnel_id: None,
744            mtu: WEAVE_HW_MTU,
745            ia_freq: 0.0,
746            ip_freq: 0.0,
747            op_freq: 0.0,
748            op_samples: 0,
749            started: crate::time::now(),
750            ingress_control: ctx.ingress_control,
751        };
752        Ok(StartResult::Simple {
753            control: None,
754            id: config.interface_id,
755            info,
756            writer: Box::new(ParentWriter),
757            interface_type_name: "WeaveInterface".into(),
758        })
759    }
760}
761
762#[cfg(test)]
763mod tests {
764    use super::*;
765    use rns_crypto::ed25519::Ed25519PrivateKey;
766    use rns_crypto::FixedRng;
767
768    fn log_frame(host: [u8; 4], event: u16, data: &[u8]) -> Vec<u8> {
769        let mut payload = vec![0, 0, 0, 0, 1, 2];
770        payload.extend_from_slice(&event.to_be_bytes());
771        payload.extend_from_slice(data);
772        WdclFrame {
773            destination: host,
774            frame_type: WDCL_T_LOG,
775            payload,
776        }
777        .encode()
778    }
779
780    #[test]
781    fn wdcl_fragmented_hdlc_and_discovery_signature() {
782        let private = Ed25519PrivateKey::generate(&mut FixedRng::new(&[7; 64]));
783        let key = private.public_key().public_bytes();
784        let signed_id = [1, 2, 3, 4];
785        let mut payload = key.to_vec();
786        payload.extend_from_slice(&private.sign(&signed_id));
787        let frame = WdclFrame {
788            destination: signed_id,
789            frame_type: WDCL_T_DISCOVER,
790            payload,
791        }
792        .encode();
793        assert!(matches!(
794            parse_device_frame(&frame, [9; 4]),
795            Some(WeaveInput::Discovery { .. })
796        ));
797
798        let wire = crate::hdlc::frame(&frame);
799        let mut decoder = crate::hdlc::Decoder::with_min_frame_size(5);
800        assert!(decoder.feed(&wire[..3]).is_empty());
801        assert_eq!(decoder.feed(&wire[3..]), vec![frame]);
802    }
803
804    #[test]
805    fn board_init_unknown_and_duplicate_state_are_preserved() {
806        let mut state = WeaveState::default();
807        state.handle_log(
808            WeaveLogFrame {
809                timestamp: 1.0,
810                level: 1,
811                event: ET_BOARD_INIT,
812                data: vec![1],
813            },
814            1.0,
815        );
816        state.handle_log(
817            WeaveLogFrame {
818                timestamp: 2.0,
819                level: 1,
820                event: 0x7777,
821                data: vec![2],
822            },
823            2.0,
824        );
825        assert_eq!(state.logs.len(), 2);
826        assert_eq!(state.unknown_events.len(), 1);
827        assert!(state.accept_packet(b"packet", 3.0));
828        assert!(!state.accept_packet(b"packet", 3.1));
829        assert!(state.accept_packet(b"packet", 4.0));
830    }
831
832    #[test]
833    fn simulated_session_authenticates_and_manages_endpoint_lifecycle() {
834        let (tx, rx) = crate::event::channel();
835        let config = WeaveConfig {
836            name: "weave-test".into(),
837            port: "/dev/null".into(),
838            configured_bitrate: WEAVE_DEFAULT_BITRATE,
839            interface_id: InterfaceId(55),
840            state: Arc::new(Mutex::new(WeaveState::default())),
841        };
842        let context = StartContext {
843            tx,
844            next_dynamic_id: Arc::new(AtomicU64::new(10_000)),
845            mode: rns_core::constants::MODE_GATEWAY,
846            gravity: 0,
847            recursive_prs: true,
848            announces_from_internal: false,
849            announces_to_internal: None,
850            ingress_control: rns_core::transport::types::IngressControlConfig::enabled(),
851            ifac: None,
852            underlay_mark: None,
853        };
854        let session = WeaveSession::new(config, &context);
855        *super::super::lock_or_recover(&session.writer, "test writer") =
856            Some(tempfile::tempfile().unwrap());
857
858        let remote = Ed25519PrivateKey::generate(&mut FixedRng::new(&[0x44; 64]));
859        let remote_public = remote.public_key().public_bytes();
860        let mut discovery_payload = remote_public.to_vec();
861        discovery_payload.extend_from_slice(&remote.sign(&session.host_switch_id));
862        session
863            .process_frame(
864                &WdclFrame {
865                    destination: session.host_switch_id,
866                    frame_type: WDCL_T_DISCOVER,
867                    payload: discovery_payload,
868                }
869                .encode(),
870            )
871            .unwrap();
872        assert_eq!(
873            *super::super::lock_or_recover(&session.remote_switch_id, "test remote"),
874            Some(remote_public[28..].try_into().unwrap())
875        );
876
877        session
878            .process_frame(&log_frame(
879                session.host_switch_id,
880                ET_PROTO_WDCL_CONNECTION,
881                &[1],
882            ))
883            .unwrap();
884        assert!(matches!(
885            rx.recv_timeout(Duration::from_secs(1)).unwrap(),
886            Event::InterfaceUp(InterfaceId(55), None, None)
887        ));
888
889        let endpoint = [0x77; 8];
890        session
891            .process_frame(&log_frame(
892                session.host_switch_id,
893                ET_PROTO_WEAVE_EP_ALIVE,
894                &endpoint,
895            ))
896            .unwrap();
897        let peer_id = loop {
898            if let Event::DynamicInterfaceUp {
899                id, registration, ..
900            } = rx.recv_timeout(Duration::from_secs(1)).unwrap()
901            {
902                assert_eq!(registration.interface_type, "WeaveInterfacePeer");
903                assert_eq!(registration.parent_id, InterfaceId(55));
904                break id;
905            }
906        };
907
908        let mut endpoint_payload = b"rns-packet".to_vec();
909        endpoint_payload.extend_from_slice(&endpoint);
910        let endpoint_frame = WdclFrame {
911            destination: session.host_switch_id,
912            frame_type: WDCL_T_ENDPOINT_PKT,
913            payload: endpoint_payload,
914        }
915        .encode();
916        session.process_frame(&endpoint_frame).unwrap();
917        assert!(loop {
918            if let Event::Frame {
919                interface_id, data, ..
920            } = rx.recv_timeout(Duration::from_secs(1)).unwrap()
921            {
922                break interface_id == peer_id && data == b"rns-packet";
923            }
924        });
925        session.process_frame(&endpoint_frame).unwrap();
926        assert!(rx.recv_timeout(Duration::from_millis(50)).is_err());
927
928        session.expire_peers(crate::time::now() + WEAVE_PEERING_TIMEOUT + 1.0);
929        assert!(loop {
930            if let Event::InterfaceDown(id) = rx.recv_timeout(Duration::from_secs(1)).unwrap() {
931                break id == peer_id;
932            }
933        });
934    }
935}
rns_net/interface/weave.rs

rns_net/interface/
weave.rs
