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

rns_net/interface/
weave.rs
