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 rpc_port = shared_config.rpc_port;
179        let mut node = RnsNode::connect_shared(shared_config, Box::new(callbacks))
180            .map_err(|_| RemoteManagementError::ConnectShared)?;
181        // Upstream rnstatus and rnpath register both types, so a shared
182        // instance filtering announces still answers their path requests.
183        node.use_shared_instance_rpc(&config_dir, rpc_port);
184        node.wants_destination_type("rnstransport.remote.management");
185        node.wants_destination_type("rnstransport.info.blackhole");
186
187        Ok(Self {
188            node,
189            management_identity,
190            timeout,
191            link_rx,
192            response_rx,
193            closed_rx,
194            management_link: Arc::new(Mutex::new(None)),
195            blackhole_link: Arc::new(Mutex::new(None)),
196        })
197    }
198
199    pub fn status(
200        &mut self,
201        transport_identity_hash: [u8; 16],
202        include_link_count: bool,
203    ) -> Result<RemoteStatus, RemoteManagementError> {
204        self.status_with_profiling(transport_identity_hash, include_link_count, false)
205    }
206
207    pub fn status_with_profiling(
208        &mut self,
209        transport_identity_hash: [u8; 16],
210        include_link_count: bool,
211        include_profiling: bool,
212    ) -> Result<RemoteStatus, RemoteManagementError> {
213        let data = msgpack::pack(&Value::Array(vec![
214            Value::Bool(include_link_count),
215            Value::Bool(include_profiling),
216        ]));
217        let response = self.request_management(transport_identity_hash, "/status", &data)?;
218        decode_status_response(&response)
219    }
220
221    pub fn path_table(
222        &mut self,
223        transport_identity_hash: [u8; 16],
224        destination_filter: Option<[u8; 16]>,
225        max_hops: Option<u8>,
226    ) -> Result<PickleValue, RemoteManagementError> {
227        let mut request = vec![Value::Str("table".into())];
228        if destination_filter.is_some() || max_hops.is_some() {
229            request.push(match destination_filter {
230                Some(hash) => Value::Bin(hash.to_vec()),
231                None => Value::Nil,
232            });
233        }
234        if let Some(hops) = max_hops {
235            request.push(Value::UInt(hops as u64));
236        }
237        let data = msgpack::pack(&Value::Array(request));
238        let response = self.request_management(transport_identity_hash, "/path", &data)?;
239        let value = msgpack::unpack_exact(&response)
240            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
241        Ok(msgpack_to_pickle(&value))
242    }
243
244    pub fn rate_table(
245        &mut self,
246        transport_identity_hash: [u8; 16],
247        destination_filter: Option<[u8; 16]>,
248    ) -> Result<PickleValue, RemoteManagementError> {
249        let mut request = vec![Value::Str("rates".into())];
250        if let Some(hash) = destination_filter {
251            request.push(Value::Bin(hash.to_vec()));
252        }
253        let data = msgpack::pack(&Value::Array(request));
254        let response = self.request_management(transport_identity_hash, "/path", &data)?;
255        let value = msgpack::unpack_exact(&response)
256            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
257        Ok(msgpack_to_pickle(&value))
258    }
259
260    pub fn published_blackhole_list(
261        &mut self,
262        transport_identity_hash: [u8; 16],
263    ) -> Result<PickleValue, RemoteManagementError> {
264        let dest_hash = crate::management::blackhole_dest_hash(&transport_identity_hash);
265        let response = self.request(
266            RemotePurpose::Blackhole,
267            dest_hash,
268            None,
269            "/list",
270            &[],
271            false,
272        )?;
273        let value = msgpack::unpack_exact(&response)
274            .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
275        Ok(blackhole_map_to_list(&value))
276    }
277
278    fn request_management(
279        &mut self,
280        transport_identity_hash: [u8; 16],
281        path: &str,
282        data: &[u8],
283    ) -> Result<Vec<u8>, RemoteManagementError> {
284        let dest_hash = crate::management::management_dest_hash(&transport_identity_hash);
285        let prv_key = self
286            .management_identity
287            .as_ref()
288            .and_then(|id| id.get_private_key())
289            .ok_or(RemoteManagementError::MissingIdentity)?;
290        self.request(
291            RemotePurpose::Management,
292            dest_hash,
293            Some(prv_key),
294            path,
295            data,
296            true,
297        )
298    }
299
300    fn request(
301        &mut self,
302        purpose: RemotePurpose,
303        dest_hash: [u8; 16],
304        identity_prv_key: Option<[u8; 64]>,
305        path: &str,
306        data: &[u8],
307        identify_on_new_link: bool,
308    ) -> Result<Vec<u8>, RemoteManagementError> {
309        let link_slot = match purpose {
310            RemotePurpose::Management => Arc::clone(&self.management_link),
311            RemotePurpose::Blackhole => Arc::clone(&self.blackhole_link),
312        };
313
314        if let Some(link_id) = *lock_link(&link_slot) {
315            if self.node.send_request(link_id, path, data).is_ok() {
316                return self.wait_for_response(link_id, path);
317            }
318            *lock_link(&link_slot) = None;
319        }
320
321        let announced = self.wait_for_destination(dest_hash)?;
322        let sig_pub: [u8; 32] = announced.public_key[32..64]
323            .try_into()
324            .expect("slice length checked");
325        let link_id = self
326            .node
327            .create_link(dest_hash, sig_pub)
328            .map_err(|_| RemoteManagementError::LinkTimeout(dest_hash))?;
329        self.wait_for_link_established(link_id, dest_hash)?;
330
331        if identify_on_new_link {
332            let prv_key = identity_prv_key.ok_or(RemoteManagementError::MissingIdentity)?;
333            self.node
334                .identify_on_link(link_id, prv_key)
335                .map_err(|_| RemoteManagementError::SendFailed)?;
336            std::thread::sleep(Duration::from_millis(200));
337        }
338
339        *lock_link(&link_slot) = Some(link_id);
340        self.node
341            .send_request(link_id, path, data)
342            .map_err(|_| RemoteManagementError::SendFailed)?;
343        self.wait_for_response(link_id, path)
344    }
345
346    fn wait_for_destination(
347        &self,
348        dest_hash: [u8; 16],
349    ) -> Result<AnnouncedIdentity, RemoteManagementError> {
350        let dest = DestHash(dest_hash);
351        let deadline = Instant::now() + self.timeout;
352        let _ = self.node.request_path(&dest);
353
354        loop {
355            if let Ok(Some(announced)) = self.node.recall_identity(&dest) {
356                return Ok(announced);
357            }
358            if self.node.has_path(&dest).unwrap_or(false) {
359                if let Ok(Some(announced)) = self.node.recall_identity(&dest) {
360                    return Ok(announced);
361                }
362            }
363            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
364                return Err(RemoteManagementError::RecallTimeout(dest_hash));
365            };
366            if remaining.is_zero() {
367                return Err(RemoteManagementError::PathTimeout(dest_hash));
368            }
369            std::thread::sleep(remaining.min(Duration::from_millis(100)));
370        }
371    }
372
373    fn wait_for_link_established(
374        &self,
375        expected_link_id: [u8; 16],
376        dest_hash: [u8; 16],
377    ) -> Result<(), RemoteManagementError> {
378        let deadline = Instant::now() + self.timeout;
379        loop {
380            if let Ok(closed) = self.closed_rx.try_recv() {
381                if closed.0 == expected_link_id {
382                    return Err(RemoteManagementError::LinkClosed);
383                }
384            }
385            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
386                return Err(RemoteManagementError::LinkTimeout(dest_hash));
387            };
388            match self
389                .link_rx
390                .recv_timeout(remaining.min(Duration::from_millis(50)))
391            {
392                Ok(link_id) if link_id.0 == expected_link_id => return Ok(()),
393                Ok(_) => continue,
394                Err(mpsc::RecvTimeoutError::Timeout) => continue,
395                Err(mpsc::RecvTimeoutError::Disconnected) => {
396                    return Err(RemoteManagementError::LinkTimeout(dest_hash));
397                }
398            }
399        }
400    }
401
402    fn wait_for_response(
403        &self,
404        link_id: [u8; 16],
405        path: &str,
406    ) -> Result<Vec<u8>, RemoteManagementError> {
407        let deadline = Instant::now() + self.timeout;
408        loop {
409            if let Ok(closed) = self.closed_rx.try_recv() {
410                if closed.0 == link_id {
411                    return Err(RemoteManagementError::LinkClosed);
412                }
413            }
414            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
415                return Err(RemoteManagementError::RequestTimeout(path.into()));
416            };
417            match self
418                .response_rx
419                .recv_timeout(remaining.min(Duration::from_millis(50)))
420            {
421                Ok((resp_link_id, _request_id, data)) if resp_link_id.0 == link_id => {
422                    return Ok(data);
423                }
424                Ok(_) => continue,
425                Err(mpsc::RecvTimeoutError::Timeout) => continue,
426                Err(mpsc::RecvTimeoutError::Disconnected) => {
427                    return Err(RemoteManagementError::RequestTimeout(path.into()));
428                }
429            }
430        }
431    }
432}
433
434pub fn parse_transport_identity_hash(s: &str) -> Result<[u8; 16], RemoteManagementError> {
435    let trimmed = s.trim();
436    if trimmed.len() != 32 {
437        return Err(RemoteManagementError::InvalidHash(s.into()));
438    }
439    let mut out = [0u8; 16];
440    for i in 0..16 {
441        out[i] = u8::from_str_radix(&trimmed[i * 2..i * 2 + 2], 16)
442            .map_err(|_| RemoteManagementError::InvalidHash(s.into()))?;
443    }
444    Ok(out)
445}
446
447pub fn msgpack_to_pickle(value: &Value) -> PickleValue {
448    match value {
449        Value::Nil => PickleValue::None,
450        Value::Bool(v) => PickleValue::Bool(*v),
451        Value::UInt(v) => PickleValue::Int((*v).min(i64::MAX as u64) as i64),
452        Value::Int(v) => PickleValue::Int(*v),
453        Value::Float(v) => PickleValue::Float(*v),
454        Value::Bin(v) => PickleValue::Bytes(v.clone()),
455        Value::Str(v) => PickleValue::String(v.clone()),
456        Value::Array(items) => PickleValue::List(items.iter().map(msgpack_to_pickle).collect()),
457        Value::Map(entries) => PickleValue::Dict(
458            entries
459                .iter()
460                .map(|(k, v)| (msgpack_to_pickle(k), msgpack_to_pickle(v)))
461                .collect(),
462        ),
463    }
464}
465
466fn decode_status_response(data: &[u8]) -> Result<RemoteStatus, RemoteManagementError> {
467    let value = msgpack::unpack_exact(data)
468        .map_err(|e| RemoteManagementError::MalformedResponse(e.to_string()))?;
469    let arr = value
470        .as_array()
471        .ok_or_else(|| RemoteManagementError::MalformedResponse("expected status array".into()))?;
472    let stats = arr
473        .first()
474        .ok_or_else(|| RemoteManagementError::MalformedResponse("missing status dict".into()))?;
475    let mut stats = msgpack_to_pickle(stats);
476    normalize_remote_status(&mut stats);
477    let link_count = arr.get(1).and_then(|v| v.as_integer());
478    let profiling_results = arr
479        .get(2)
480        .filter(|value| !matches!(value, Value::Nil))
481        .map(msgpack_to_pickle);
482    Ok(RemoteStatus {
483        stats,
484        link_count,
485        profiling_results,
486    })
487}
488
489fn normalize_remote_status(value: &mut PickleValue) {
490    let PickleValue::Dict(entries) = value else {
491        return;
492    };
493    if !entries
494        .iter()
495        .any(|(k, _)| string_key(k) == Some("transport_enabled"))
496    {
497        entries.push((
498            PickleValue::String("transport_enabled".into()),
499            PickleValue::Bool(true),
500        ));
501    }
502    if let Some((_, PickleValue::List(ifaces))) = entries
503        .iter_mut()
504        .find(|(k, _)| string_key(k) == Some("interfaces"))
505    {
506        for iface in ifaces {
507            normalize_remote_interface(iface);
508        }
509    }
510}
511
512fn normalize_remote_interface(value: &mut PickleValue) {
513    let PickleValue::Dict(entries) = value else {
514        return;
515    };
516    copy_key(entries, "incoming_announce_freq", "ia_freq");
517    copy_key(entries, "outgoing_announce_freq", "oa_freq");
518    copy_key(entries, "incoming_path_request_freq", "ip_freq");
519    copy_key(entries, "outgoing_path_request_freq", "op_freq");
520}
521
522fn copy_key(entries: &mut Vec<(PickleValue, PickleValue)>, from: &str, to: &str) {
523    if entries.iter().any(|(k, _)| string_key(k) == Some(to)) {
524        return;
525    }
526    if let Some((_, value)) = entries.iter().find(|(k, _)| string_key(k) == Some(from)) {
527        entries.push((PickleValue::String(to.into()), value.clone()));
528    }
529}
530
531fn blackhole_map_to_list(value: &Value) -> PickleValue {
532    let Some(entries) = value.as_map() else {
533        return msgpack_to_pickle(value);
534    };
535    let mut list = Vec::new();
536    for (key, info) in entries {
537        let Some(identity_hash) = key.as_bin() else {
538            continue;
539        };
540        let mut item = vec![(
541            PickleValue::String("identity_hash".into()),
542            PickleValue::Bytes(identity_hash.to_vec()),
543        )];
544        if let Some(info_entries) = info.as_map() {
545            for (k, v) in info_entries {
546                if let Some(key) = k.as_str() {
547                    let out_key = if key == "created" { "created" } else { key };
548                    item.push((PickleValue::String(out_key.into()), msgpack_to_pickle(v)));
549                }
550            }
551        }
552        list.push(PickleValue::Dict(item));
553    }
554    PickleValue::List(list)
555}
556
557fn lock_link<'a>(
558    link: &'a Arc<Mutex<Option<[u8; 16]>>>,
559) -> std::sync::MutexGuard<'a, Option<[u8; 16]>> {
560    match link.lock() {
561        Ok(guard) => guard,
562        Err(poisoned) => poisoned.into_inner(),
563    }
564}
565
566fn string_key(value: &PickleValue) -> Option<&str> {
567    match value {
568        PickleValue::String(s) => Some(s),
569        _ => None,
570    }
571}
572
573fn hex(bytes: &[u8]) -> String {
574    let mut out = String::with_capacity(bytes.len() * 2);
575    for byte in bytes {
576        use std::fmt::Write;
577        let _ = write!(&mut out, "{byte:02x}");
578    }
579    out
580}
581
582#[cfg(test)]
583mod tests {
584    use super::*;
585
586    #[test]
587    fn parse_transport_identity_hash_accepts_32_hex_chars() {
588        let hash = parse_transport_identity_hash("00112233445566778899aabbccddeeff").unwrap();
589        assert_eq!(hash[0], 0x00);
590        assert_eq!(hash[15], 0xff);
591    }
592
593    #[test]
594    fn parse_transport_identity_hash_rejects_invalid_input() {
595        assert!(parse_transport_identity_hash("short").is_err());
596        assert!(parse_transport_identity_hash("00112233445566778899aabbccddeeg").is_err());
597    }
598
599    #[test]
600    fn status_response_normalizes_renderer_keys() {
601        let status = Value::Map(vec![
602            (
603                Value::Str("interfaces".into()),
604                Value::Array(vec![Value::Map(vec![
605                    (Value::Str("name".into()), Value::Str("if0".into())),
606                    (
607                        Value::Str("incoming_announce_freq".into()),
608                        Value::Float(1.5),
609                    ),
610                ])]),
611            ),
612            (Value::Str("rxb".into()), Value::UInt(1)),
613            (Value::Str("txb".into()), Value::UInt(2)),
614        ]);
615        let data = msgpack::pack(&Value::Array(vec![status, Value::UInt(7)]));
616        let decoded = decode_status_response(&data).unwrap();
617        assert_eq!(decoded.link_count, Some(7));
618        assert_eq!(
619            decoded
620                .stats
621                .get("transport_enabled")
622                .and_then(|v| v.as_bool()),
623            Some(true)
624        );
625        let iface = &decoded
626            .stats
627            .get("interfaces")
628            .and_then(|v| v.as_list())
629            .unwrap()[0];
630        assert_eq!(iface.get("ia_freq").and_then(|v| v.as_float()), Some(1.5));
631    }
632
633    #[test]
634    fn status_response_decodes_profiling_results() {
635        let status = Value::Map(vec![]);
636        let profiling = Value::Map(vec![(
637            Value::Str("entry110.remote".into()),
638            Value::Map(vec![(Value::Str("count".into()), Value::UInt(2))]),
639        )]);
640        let data = msgpack::pack(&Value::Array(vec![status, Value::UInt(7), profiling]));
641        let decoded = decode_status_response(&data).unwrap();
642        assert_eq!(decoded.link_count, Some(7));
643        assert_eq!(
644            decoded
645                .profiling_results
646                .as_ref()
647                .and_then(|results| results.get("entry110.remote"))
648                .and_then(|sample| sample.get("count"))
649                .and_then(PickleValue::as_int),
650            Some(2)
651        );
652    }
653
654    #[test]
655    fn blackhole_map_converts_to_renderer_list() {
656        let value = Value::Map(vec![(
657            Value::Bin(vec![0x11; 16]),
658            Value::Map(vec![
659                (Value::Str("expires".into()), Value::Float(123.0)),
660                (Value::Str("reason".into()), Value::Str("test".into())),
661            ]),
662        )]);
663        let list = blackhole_map_to_list(&value);
664        let items = list.as_list().unwrap();
665        assert_eq!(items.len(), 1);
666        assert_eq!(
667            items[0]
668                .get("identity_hash")
669                .and_then(|v| v.as_bytes())
670                .unwrap(),
671            &[0x11; 16]
672        );
673    }
674}
rns_net/remote_management.rs

rns_net/
remote_management.rs
