rns_net/
remote_management.rs

1//! Client-side remote management helpers.
2//!
3//! These helpers connect to a local shared instance, derive upstream-compatible
4//! remote management destinations from a remote transport identity hash, and
5//! query the remote node over a Reticulum link.
6
7use std::fmt;
8use std::path::Path;
9use std::sync::mpsc;
10use std::sync::{Arc, Mutex};
11use std::time::{Duration, Instant};
12
13use rns_core::msgpack::{self, Value};
14use rns_core::types::{DestHash, LinkId};
15use rns_crypto::identity::Identity;
16
17use crate::destination::AnnouncedIdentity;
18use crate::pickle::PickleValue;
19use crate::shared_client::SharedClientConfig;
20use crate::{Callbacks, RnsNode, TeardownReason};
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23enum RemotePurpose {
24    Management,
25    Blackhole,
26}
27
28#[derive(Debug, Clone)]
29pub enum RemoteManagementError {
30    InvalidHash(String),
31    MissingIdentity,
32    IdentityLoad(String),
33    Config(String),
34    ConnectShared,
35    PathTimeout([u8; 16]),
36    RecallTimeout([u8; 16]),
37    LinkTimeout([u8; 16]),
38    RequestTimeout(String),
39    LinkClosed,
40    SendFailed,
41    MalformedResponse(String),
42    Unsupported(String),
43}
44
45impl fmt::Display for RemoteManagementError {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        match self {
48            Self::InvalidHash(s) => {
49                write!(
50                    f,
51                    "invalid transport identity hash: {s} (expected 32 hex chars)"
52                )
53            }
54            Self::MissingIdentity => {
55                write!(
56                    f,
57                    "remote management requires an identity file; use -i PATH"
58                )
59            }
60            Self::IdentityLoad(e) => write!(f, "could not load management identity: {e}"),
61            Self::Config(e) => write!(f, "could not load local Reticulum config: {e}"),
62            Self::ConnectShared => {
63                write!(f, "could not connect to local shared Reticulum instance")
64            }
65            Self::PathTimeout(hash) => write!(f, "timed out waiting for path to {}", hex(hash)),
66            Self::RecallTimeout(hash) => {
67                write!(f, "timed out recalling identity for {}", hex(hash))
68            }
69            Self::LinkTimeout(hash) => write!(f, "timed out establishing link to {}", hex(hash)),
70            Self::RequestTimeout(path) => write!(f, "remote request to {path} timed out"),
71            Self::LinkClosed => write!(f, "remote link closed before request completed"),
72            Self::SendFailed => write!(f, "could not send remote management request"),
73            Self::MalformedResponse(e) => write!(f, "malformed remote response: {e}"),
74            Self::Unsupported(e) => write!(f, "{e}"),
75        }
76    }
77}
78
79impl std::error::Error for RemoteManagementError {}
80
81#[derive(Debug, Clone)]
82pub struct RemoteStatus {
83    pub stats: PickleValue,
84    pub link_count: Option<i64>,
85    pub profiling_results: Option<PickleValue>,
86}
87
88struct RemoteCallbacks {
89    link_established_tx: mpsc::Sender<LinkId>,
90    response_tx: mpsc::Sender<(LinkId, [u8; 16], Vec<u8>)>,
91    link_closed_tx: mpsc::Sender<LinkId>,
92}
93
94impl Callbacks for RemoteCallbacks {
95    fn on_announce(&mut self, _announced: AnnouncedIdentity) {}
96
97    fn on_path_updated(&mut self, _dest_hash: DestHash, _hops: u8) {}
98
99    fn on_local_delivery(
100        &mut self,
101        _dest_hash: DestHash,
102        _raw: Vec<u8>,
103        _packet_hash: crate::PacketHash,
104    ) {
105    }
106
107    fn on_link_established(
108        &mut self,
109        link_id: LinkId,
110        _dest_hash: DestHash,
111        _rtt: f64,
112        _is_initiator: bool,
113    ) {
114        let _ = self.link_established_tx.send(link_id);
115    }
116
117    fn on_response(&mut self, link_id: LinkId, request_id: [u8; 16], data: Vec<u8>) {
118        let _ = self.response_tx.send((link_id, request_id, data));
119    }
120
121    fn on_link_closed(&mut self, link_id: LinkId, reason: Option<TeardownReason>) {
122        if reason == Some(TeardownReason::InitiatorClosed) {
123            return;
124        }
125        let _ = self.link_closed_tx.send(link_id);
126    }
127}
128
129pub struct RemoteManagementClient {
130    node: RnsNode,
131    management_identity: Option<Identity>,
132    timeout: Duration,
133    link_rx: mpsc::Receiver<LinkId>,
134    response_rx: mpsc::Receiver<(LinkId, [u8; 16], Vec<u8>)>,
135    closed_rx: mpsc::Receiver<LinkId>,
136    management_link: Arc<Mutex<Option<[u8; 16]>>>,
137    blackhole_link: Arc<Mutex<Option<[u8; 16]>>>,
138}
139
140impl RemoteManagementClient {
141    pub fn connect(
142        config_path: Option<&Path>,
143        management_identity_path: Option<&Path>,
144        timeout: Duration,
145    ) -> Result<Self, RemoteManagementError> {
146        let management_identity = match management_identity_path {
147            Some(path) => Some(
148                crate::storage::load_identity(path)
149                    .map_err(|e| RemoteManagementError::IdentityLoad(e.to_string()))?,
150            ),
151            None => None,
152        };
153
154        let (link_tx, link_rx) = mpsc::channel();
155        let (response_tx, response_rx) = mpsc::channel();
156        let (closed_tx, closed_rx) = mpsc::channel();
157        let callbacks = RemoteCallbacks {
158            link_established_tx: link_tx,
159            response_tx,
160            link_closed_tx: closed_tx,
161        };
162
163        let config_dir = crate::storage::resolve_config_dir(config_path);
164        let config_file = config_dir.join("config");
165        let rns_config = if config_file.exists() {
166            crate::config::parse_file(&config_file)
167                .map_err(|e| RemoteManagementError::Config(e.to_string()))?
168        } else {
169            crate::config::parse("").map_err(|e| RemoteManagementError::Config(e.to_string()))?
170        };
171
172        let shared_config = SharedClientConfig {
173            instance_name: rns_config.reticulum.instance_name.clone(),
174            port: rns_config.reticulum.shared_instance_port,
175            rpc_port: rns_config.reticulum.instance_control_port,
176        };
177
178        let node = RnsNode::connect_shared(shared_config, Box::new(callbacks))
179            .map_err(|_| RemoteManagementError::ConnectShared)?;
180
181        Ok(Self {
182            node,
183            management_identity,
184            timeout,
185            link_rx,
186            response_rx,
187            closed_rx,
188            management_link: Arc::new(Mutex::new(None)),
189            blackhole_link: Arc::new(Mutex::new(None)),
190        })
191    }
192
193    pub fn status(
194        &mut self,
195        transport_identity_hash: [u8; 16],
196        include_link_count: bool,
197    ) -> Result<RemoteStatus, RemoteManagementError> {
198        self.status_with_profiling(transport_identity_hash, include_link_count, false)
199    }
200
201    pub fn status_with_profiling(
202        &mut self,
203        transport_identity_hash: [u8; 16],
204        include_link_count: bool,
205        include_profiling: bool,
206    ) -> Result<RemoteStatus, RemoteManagementError> {
207        let data = msgpack::pack(&Value::Array(vec![
208            Value::Bool(include_link_count),
209            Value::Bool(include_profiling),
210        ]));
211        let response = self.request_management(transport_identity_hash, "/status", &data)?;
212        decode_status_response(&response)
213    }
214
215    pub fn path_table(
216        &mut self,
217        transport_identity_hash: [u8; 16],
218        destination_filter: Option<[u8; 16]>,
219        max_hops: Option<u8>,
220    ) -> Result<PickleValue, RemoteManagementError> {
221        let mut request = vec![Value::Str("table".into())];
222        if destination_filter.is_some() || max_hops.is_some() {
223            request.push(match destination_filter {
224                Some(hash) => Value::Bin(hash.to_vec()),
225                None => Value::Nil,
226            });
227        }
228        if let Some(hops) = max_hops {
229            request.push(Value::UInt(hops as u64));
230        }
231        let data = msgpack::pack(&Value::Array(request));
232        let response = self.request_management(transport_identity_hash, "/path", &data)?;
233        let value = msgpack::unpack_exact(&response)
234            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
235        Ok(msgpack_to_pickle(&value))
236    }
237
238    pub fn rate_table(
239        &mut self,
240        transport_identity_hash: [u8; 16],
241        destination_filter: Option<[u8; 16]>,
242    ) -> Result<PickleValue, RemoteManagementError> {
243        let mut request = vec![Value::Str("rates".into())];
244        if let Some(hash) = destination_filter {
245            request.push(Value::Bin(hash.to_vec()));
246        }
247        let data = msgpack::pack(&Value::Array(request));
248        let response = self.request_management(transport_identity_hash, "/path", &data)?;
249        let value = msgpack::unpack_exact(&response)
250            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
251        Ok(msgpack_to_pickle(&value))
252    }
253
254    pub fn published_blackhole_list(
255        &mut self,
256        transport_identity_hash: [u8; 16],
257    ) -> Result<PickleValue, RemoteManagementError> {
258        let dest_hash = crate::management::blackhole_dest_hash(&transport_identity_hash);
259        let response = self.request(
260            RemotePurpose::Blackhole,
261            dest_hash,
262            None,
263            "/list",
264            &[],
265            false,
266        )?;
267        let value = msgpack::unpack_exact(&response)
268            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
269        Ok(blackhole_map_to_list(&value))
270    }
271
272    fn request_management(
273        &mut self,
274        transport_identity_hash: [u8; 16],
275        path: &str,
276        data: &[u8],
277    ) -> Result<Vec<u8>, RemoteManagementError> {
278        let dest_hash = crate::management::management_dest_hash(&transport_identity_hash);
279        let prv_key = self
280            .management_identity
281            .as_ref()
282            .and_then(|id| id.get_private_key())
283            .ok_or(RemoteManagementError::MissingIdentity)?;
284        self.request(
285            RemotePurpose::Management,
286            dest_hash,
287            Some(prv_key),
288            path,
289            data,
290            true,
291        )
292    }
293
294    fn request(
295        &mut self,
296        purpose: RemotePurpose,
297        dest_hash: [u8; 16],
298        identity_prv_key: Option<[u8; 64]>,
299        path: &str,
300        data: &[u8],
301        identify_on_new_link: bool,
302    ) -> Result<Vec<u8>, RemoteManagementError> {
303        let link_slot = match purpose {
304            RemotePurpose::Management => Arc::clone(&self.management_link),
305            RemotePurpose::Blackhole => Arc::clone(&self.blackhole_link),
306        };
307
308        if let Some(link_id) = *lock_link(&link_slot) {
309            if self.node.send_request(link_id, path, data).is_ok() {
310                return self.wait_for_response(link_id, path);
311            }
312            *lock_link(&link_slot) = None;
313        }
314
315        let announced = self.wait_for_destination(dest_hash)?;
316        let sig_pub: [u8; 32] = announced.public_key[32..64]
317            .try_into()
318            .expect("slice length checked");
319        let link_id = self
320            .node
321            .create_link(dest_hash, sig_pub)
322            .map_err(|_| RemoteManagementError::LinkTimeout(dest_hash))?;
323        self.wait_for_link_established(link_id, dest_hash)?;
324
325        if identify_on_new_link {
326            let prv_key = identity_prv_key.ok_or(RemoteManagementError::MissingIdentity)?;
327            self.node
328                .identify_on_link(link_id, prv_key)
329                .map_err(|_| RemoteManagementError::SendFailed)?;
330            std::thread::sleep(Duration::from_millis(200));
331        }
332
333        *lock_link(&link_slot) = Some(link_id);
334        self.node
335            .send_request(link_id, path, data)
336            .map_err(|_| RemoteManagementError::SendFailed)?;
337        self.wait_for_response(link_id, path)
338    }
339
340    fn wait_for_destination(
341        &self,
342        dest_hash: [u8; 16],
343    ) -> Result<AnnouncedIdentity, RemoteManagementError> {
344        let dest = DestHash(dest_hash);
345        let deadline = Instant::now() + self.timeout;
346        let _ = self.node.request_path(&dest);
347
348        loop {
349            if let Ok(Some(announced)) = self.node.recall_identity(&dest) {
350                return Ok(announced);
351            }
352            if self.node.has_path(&dest).unwrap_or(false) {
353                if let Ok(Some(announced)) = self.node.recall_identity(&dest) {
354                    return Ok(announced);
355                }
356            }
357            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
358                return Err(RemoteManagementError::RecallTimeout(dest_hash));
359            };
360            if remaining.is_zero() {
361                return Err(RemoteManagementError::PathTimeout(dest_hash));
362            }
363            std::thread::sleep(remaining.min(Duration::from_millis(100)));
364        }
365    }
366
367    fn wait_for_link_established(
368        &self,
369        expected_link_id: [u8; 16],
370        dest_hash: [u8; 16],
371    ) -> Result<(), RemoteManagementError> {
372        let deadline = Instant::now() + self.timeout;
373        loop {
374            if let Ok(closed) = self.closed_rx.try_recv() {
375                if closed.0 == expected_link_id {
376                    return Err(RemoteManagementError::LinkClosed);
377                }
378            }
379            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
380                return Err(RemoteManagementError::LinkTimeout(dest_hash));
381            };
382            match self
383                .link_rx
384                .recv_timeout(remaining.min(Duration::from_millis(50)))
385            {
386                Ok(link_id) if link_id.0 == expected_link_id => return Ok(()),
387                Ok(_) => continue,
388                Err(mpsc::RecvTimeoutError::Timeout) => continue,
389                Err(mpsc::RecvTimeoutError::Disconnected) => {
390                    return Err(RemoteManagementError::LinkTimeout(dest_hash));
391                }
392            }
393        }
394    }
395
396    fn wait_for_response(
397        &self,
398        link_id: [u8; 16],
399        path: &str,
400    ) -> Result<Vec<u8>, RemoteManagementError> {
401        let deadline = Instant::now() + self.timeout;
402        loop {
403            if let Ok(closed) = self.closed_rx.try_recv() {
404                if closed.0 == link_id {
405                    return Err(RemoteManagementError::LinkClosed);
406                }
407            }
408            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
409                return Err(RemoteManagementError::RequestTimeout(path.into()));
410            };
411            match self
412                .response_rx
413                .recv_timeout(remaining.min(Duration::from_millis(50)))
414            {
415                Ok((resp_link_id, _request_id, data)) if resp_link_id.0 == link_id => {
416                    return Ok(data);
417                }
418                Ok(_) => continue,
419                Err(mpsc::RecvTimeoutError::Timeout) => continue,
420                Err(mpsc::RecvTimeoutError::Disconnected) => {
421                    return Err(RemoteManagementError::RequestTimeout(path.into()));
422                }
423            }
424        }
425    }
426}
427
428pub fn parse_transport_identity_hash(s: &str) -> Result<[u8; 16], RemoteManagementError> {
429    let trimmed = s.trim();
430    if trimmed.len() != 32 {
431        return Err(RemoteManagementError::InvalidHash(s.into()));
432    }
433    let mut out = [0u8; 16];
434    for i in 0..16 {
435        out[i] = u8::from_str_radix(&trimmed[i * 2..i * 2 + 2], 16)
436            .map_err(|_| RemoteManagementError::InvalidHash(s.into()))?;
437    }
438    Ok(out)
439}
440
441pub fn msgpack_to_pickle(value: &Value) -> PickleValue {
442    match value {
443        Value::Nil => PickleValue::None,
444        Value::Bool(v) => PickleValue::Bool(*v),
445        Value::UInt(v) => PickleValue::Int((*v).min(i64::MAX as u64) as i64),
446        Value::Int(v) => PickleValue::Int(*v),
447        Value::Float(v) => PickleValue::Float(*v),
448        Value::Bin(v) => PickleValue::Bytes(v.clone()),
449        Value::Str(v) => PickleValue::String(v.clone()),
450        Value::Array(items) => PickleValue::List(items.iter().map(msgpack_to_pickle).collect()),
451        Value::Map(entries) => PickleValue::Dict(
452            entries
453                .iter()
454                .map(|(k, v)| (msgpack_to_pickle(k), msgpack_to_pickle(v)))
455                .collect(),
456        ),
457    }
458}
459
460fn decode_status_response(data: &[u8]) -> Result<RemoteStatus, RemoteManagementError> {
461    let value = msgpack::unpack_exact(data)
462        .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
463    let arr = value
464        .as_array()
465        .ok_or_else(|| RemoteManagementError::MalformedResponse("expected status array".into()))?;
466    let stats = arr
467        .first()
468        .ok_or_else(|| RemoteManagementError::MalformedResponse("missing status dict".into()))?;
469    let mut stats = msgpack_to_pickle(stats);
470    normalize_remote_status(&mut stats);
471    let link_count = arr.get(1).and_then(|v| v.as_integer());
472    let profiling_results = arr
473        .get(2)
474        .filter(|value| !matches!(value, Value::Nil))
475        .map(msgpack_to_pickle);
476    Ok(RemoteStatus {
477        stats,
478        link_count,
479        profiling_results,
480    })
481}
482
483fn normalize_remote_status(value: &mut PickleValue) {
484    let PickleValue::Dict(entries) = value else {
485        return;
486    };
487    if !entries
488        .iter()
489        .any(|(k, _)| string_key(k) == Some("transport_enabled"))
490    {
491        entries.push((
492            PickleValue::String("transport_enabled".into()),
493            PickleValue::Bool(true),
494        ));
495    }
496    if let Some((_, PickleValue::List(ifaces))) = entries
497        .iter_mut()
498        .find(|(k, _)| string_key(k) == Some("interfaces"))
499    {
500        for iface in ifaces {
501            normalize_remote_interface(iface);
502        }
503    }
504}
505
506fn normalize_remote_interface(value: &mut PickleValue) {
507    let PickleValue::Dict(entries) = value else {
508        return;
509    };
510    copy_key(entries, "incoming_announce_freq", "ia_freq");
511    copy_key(entries, "outgoing_announce_freq", "oa_freq");
512    copy_key(entries, "incoming_path_request_freq", "ip_freq");
513    copy_key(entries, "outgoing_path_request_freq", "op_freq");
514}
515
516fn copy_key(entries: &mut Vec<(PickleValue, PickleValue)>, from: &str, to: &str) {
517    if entries.iter().any(|(k, _)| string_key(k) == Some(to)) {
518        return;
519    }
520    if let Some((_, value)) = entries.iter().find(|(k, _)| string_key(k) == Some(from)) {
521        entries.push((PickleValue::String(to.into()), value.clone()));
522    }
523}
524
525fn blackhole_map_to_list(value: &Value) -> PickleValue {
526    let Some(entries) = value.as_map() else {
527        return msgpack_to_pickle(value);
528    };
529    let mut list = Vec::new();
530    for (key, info) in entries {
531        let Some(identity_hash) = key.as_bin() else {
532            continue;
533        };
534        let mut item = vec![(
535            PickleValue::String("identity_hash".into()),
536            PickleValue::Bytes(identity_hash.to_vec()),
537        )];
538        if let Some(info_entries) = info.as_map() {
539            for (k, v) in info_entries {
540                if let Some(key) = k.as_str() {
541                    let out_key = if key == "created" { "created" } else { key };
542                    item.push((PickleValue::String(out_key.into()), msgpack_to_pickle(v)));
543                }
544            }
545        }
546        list.push(PickleValue::Dict(item));
547    }
548    PickleValue::List(list)
549}
550
551fn lock_link<'a>(
552    link: &'a Arc<Mutex<Option<[u8; 16]>>>,
553) -> std::sync::MutexGuard<'a, Option<[u8; 16]>> {
554    match link.lock() {
555        Ok(guard) => guard,
556        Err(poisoned) => poisoned.into_inner(),
557    }
558}
559
560fn string_key(value: &PickleValue) -> Option<&str> {
561    match value {
562        PickleValue::String(s) => Some(s),
563        _ => None,
564    }
565}
566
567fn hex(bytes: &[u8]) -> String {
568    let mut out = String::with_capacity(bytes.len() * 2);
569    for byte in bytes {
570        use std::fmt::Write;
571        let _ = write!(&mut out, "{byte:02x}");
572    }
573    out
574}
575
576#[cfg(test)]
577mod tests {
578    use super::*;
579
580    #[test]
581    fn parse_transport_identity_hash_accepts_32_hex_chars() {
582        let hash = parse_transport_identity_hash("00112233445566778899aabbccddeeff").unwrap();
583        assert_eq!(hash[0], 0x00);
584        assert_eq!(hash[15], 0xff);
585    }
586
587    #[test]
588    fn parse_transport_identity_hash_rejects_invalid_input() {
589        assert!(parse_transport_identity_hash("short").is_err());
590        assert!(parse_transport_identity_hash("00112233445566778899aabbccddeeg").is_err());
591    }
592
593    #[test]
594    fn status_response_normalizes_renderer_keys() {
595        let status = Value::Map(vec![
596            (
597                Value::Str("interfaces".into()),
598                Value::Array(vec![Value::Map(vec![
599                    (Value::Str("name".into()), Value::Str("if0".into())),
600                    (
601                        Value::Str("incoming_announce_freq".into()),
602                        Value::Float(1.5),
603                    ),
604                ])]),
605            ),
606            (Value::Str("rxb".into()), Value::UInt(1)),
607            (Value::Str("txb".into()), Value::UInt(2)),
608        ]);
609        let data = msgpack::pack(&Value::Array(vec![status, Value::UInt(7)]));
610        let decoded = decode_status_response(&data).unwrap();
611        assert_eq!(decoded.link_count, Some(7));
612        assert_eq!(
613            decoded
614                .stats
615                .get("transport_enabled")
616                .and_then(|v| v.as_bool()),
617            Some(true)
618        );
619        let iface = &decoded
620            .stats
621            .get("interfaces")
622            .and_then(|v| v.as_list())
623            .unwrap()[0];
624        assert_eq!(iface.get("ia_freq").and_then(|v| v.as_float()), Some(1.5));
625    }
626
627    #[test]
628    fn status_response_decodes_profiling_results() {
629        let status = Value::Map(vec![]);
630        let profiling = Value::Map(vec![(
631            Value::Str("entry110.remote".into()),
632            Value::Map(vec![(Value::Str("count".into()), Value::UInt(2))]),
633        )]);
634        let data = msgpack::pack(&Value::Array(vec![status, Value::UInt(7), profiling]));
635        let decoded = decode_status_response(&data).unwrap();
636        assert_eq!(decoded.link_count, Some(7));
637        assert_eq!(
638            decoded
639                .profiling_results
640                .as_ref()
641                .and_then(|results| results.get("entry110.remote"))
642                .and_then(|sample| sample.get("count"))
643                .and_then(PickleValue::as_int),
644            Some(2)
645        );
646    }
647
648    #[test]
649    fn blackhole_map_converts_to_renderer_list() {
650        let value = Value::Map(vec![(
651            Value::Bin(vec![0x11; 16]),
652            Value::Map(vec![
653                (Value::Str("expires".into()), Value::Float(123.0)),
654                (Value::Str("reason".into()), Value::Str("test".into())),
655            ]),
656        )]);
657        let list = blackhole_map_to_list(&value);
658        let items = list.as_list().unwrap();
659        assert_eq!(items.len(), 1);
660        assert_eq!(
661            items[0]
662                .get("identity_hash")
663                .and_then(|v| v.as_bytes())
664                .unwrap(),
665            &[0x11; 16]
666        );
667    }
668}
rns_net/remote_management.rs

rns_net/
remote_management.rs
