1use std::collections::{HashMap, HashSet};
6use std::io::{self, Read};
7use std::path::Path;
8use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
9use std::sync::Arc;
10use std::thread::{self, JoinHandle};
11use std::time::Duration;
12
13use rns_core::transport::announce_verify_queue::OverflowPolicy as AnnounceQueueOverflowPolicy;
14use rns_core::transport::types::TransportConfig;
15use rns_crypto::identity::Identity;
16use rns_crypto::{OsRng, Rng};
17
18use crate::config;
19use crate::driver::{AnnounceRateDefaults, Callbacks, Driver};
20#[cfg(feature = "iface-backbone")]
21use crate::driver::{BackbonePeerPoolCandidateConfig, BackbonePeerPoolSettings};
22use crate::event::{self, Event, EventSender};
23use crate::ifac;
24#[cfg(feature = "iface-auto")]
25use crate::interface::auto::{auto_runtime_handle_from_config, AutoConfig};
26#[cfg(feature = "iface-backbone")]
27use crate::interface::backbone::{
28 client_config_from_mode, client_priority_from_mode, client_runtime_handle_from_mode,
29 peer_state_handle_from_mode, runtime_handle_from_mode, BackboneMode,
30};
31#[cfg(feature = "iface-i2p")]
32use crate::interface::i2p::{i2p_runtime_handle_from_config, I2pConfig};
33#[cfg(feature = "iface-local")]
34use crate::interface::local::LocalServerConfig;
35#[cfg(feature = "iface-pipe")]
36use crate::interface::pipe::{pipe_runtime_handle_from_config, PipeConfig};
37#[cfg(feature = "iface-rnode")]
38use crate::interface::rnode::{rnode_runtime_handle_from_config, RNodeConfig};
39#[cfg(feature = "iface-tcp")]
40use crate::interface::tcp::{tcp_client_runtime_handle_from_config, TcpClientConfig};
41#[cfg(feature = "iface-tcp")]
42use crate::interface::tcp_server::{
43 runtime_handle_from_config as tcp_runtime_handle_from_config, TcpServerConfig,
44};
45#[cfg(feature = "iface-udp")]
46use crate::interface::udp::{udp_runtime_handle_from_config, UdpConfig};
47use crate::interface::{InterfaceEntry, InterfaceStats};
48use crate::storage;
49use crate::time;
50
51static NEXT_RESOURCE_TRANSFER_ID: AtomicU64 = AtomicU64::new(1);
52
53#[cfg(test)]
54const DEFAULT_KNOWN_DESTINATIONS_TTL: Duration = Duration::from_secs(48 * 60 * 60);
55const DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES: usize = 8192;
56
57fn parse_interface_mode(mode: &str) -> u8 {
60 match mode.to_lowercase().as_str() {
61 "full" => rns_core::constants::MODE_FULL,
62 "access_point" | "accesspoint" | "ap" => rns_core::constants::MODE_ACCESS_POINT,
63 "pointtopoint" | "ptp" => rns_core::constants::MODE_POINT_TO_POINT,
64 "roaming" => rns_core::constants::MODE_ROAMING,
65 "boundary" => rns_core::constants::MODE_BOUNDARY,
66 "gateway" | "gw" => rns_core::constants::MODE_GATEWAY,
67 "internal" => rns_core::constants::MODE_INTERNAL,
68 _ => rns_core::constants::MODE_FULL,
69 }
70}
71
72fn parse_optional_interface_mode(mode: &str) -> Option<u8> {
73 match mode.to_lowercase().as_str() {
74 "full" => Some(rns_core::constants::MODE_FULL),
75 "access_point" | "accesspoint" | "ap" => Some(rns_core::constants::MODE_ACCESS_POINT),
76 "pointtopoint" | "ptp" => Some(rns_core::constants::MODE_POINT_TO_POINT),
77 "roaming" => Some(rns_core::constants::MODE_ROAMING),
78 "boundary" => Some(rns_core::constants::MODE_BOUNDARY),
79 "gateway" | "gw" => Some(rns_core::constants::MODE_GATEWAY),
80 "internal" => Some(rns_core::constants::MODE_INTERNAL),
81 _ => None,
82 }
83}
84
85fn parse_interface_gravity(value: Option<&str>, default_gravity: i64) -> i64 {
86 value
87 .and_then(|value| value.parse().ok())
88 .unwrap_or(default_gravity)
89}
90
91fn normalize_discovery_mode(
93 interface_type: &str,
94 configured_mode: u8,
95 discoverable: bool,
96 ignore_config_warnings: bool,
97) -> u8 {
98 if !discoverable
99 || ignore_config_warnings
100 || matches!(
101 configured_mode,
102 rns_core::constants::MODE_ACCESS_POINT
103 | rns_core::constants::MODE_GATEWAY
104 | rns_core::constants::MODE_INTERNAL
105 )
106 {
107 return configured_mode;
108 }
109
110 if matches!(interface_type, "RNodeInterface" | "RNodeMultiInterface") {
111 rns_core::constants::MODE_ACCESS_POINT
112 } else {
113 rns_core::constants::MODE_GATEWAY
114 }
115}
116
117fn default_ingress_control_for_type(
118 iface_type: &str,
119) -> rns_core::transport::types::IngressControlConfig {
120 match iface_type {
121 "AutoInterface" | "BackboneInterface" | "TCPClientInterface" | "TCPServerInterface"
122 | "UDPInterface" | "I2PInterface" => {
123 rns_core::transport::types::IngressControlConfig::enabled()
124 }
125 _ => rns_core::transport::types::IngressControlConfig::disabled(),
126 }
127}
128
129fn parse_ingress_control_config(
130 iface_type: &str,
131 params: &std::collections::HashMap<String, String>,
132) -> Result<rns_core::transport::types::IngressControlConfig, String> {
133 let mut config = default_ingress_control_for_type(iface_type);
134
135 if let Some(v) = params.get("ingress_control") {
136 config.enabled = config::parse_bool_pub(v)
137 .ok_or_else(|| format!("ingress_control must be a boolean, got '{}'", v))?;
138 }
139 if let Some(v) = params.get("ic_max_held_announces") {
140 config.max_held_announces = v
141 .parse::<usize>()
142 .map_err(|_| format!("ic_max_held_announces must be an integer, got '{}'", v))?;
143 }
144 if let Some(v) = params.get("ic_burst_hold") {
145 config.burst_hold = parse_nonnegative_f64("ic_burst_hold", v)?;
146 }
147 if let Some(v) = params.get("ic_burst_freq_new") {
148 config.burst_freq_new = parse_nonnegative_f64("ic_burst_freq_new", v)?;
149 }
150 if let Some(v) = params.get("ic_burst_freq") {
151 config.burst_freq = parse_nonnegative_f64("ic_burst_freq", v)?;
152 }
153 if let Some(v) = params.get("ic_pr_burst_freq_new") {
154 config.pr_burst_freq_new = parse_nonnegative_f64("ic_pr_burst_freq_new", v)?;
155 }
156 if let Some(v) = params.get("ic_pr_burst_freq") {
157 config.pr_burst_freq = parse_nonnegative_f64("ic_pr_burst_freq", v)?;
158 }
159 if let Some(v) = params.get("egress_control") {
160 config.egress_enabled = config::parse_bool_pub(v)
161 .ok_or_else(|| format!("egress_control must be a boolean, got '{}'", v))?;
162 }
163 if let Some(v) = params.get("ec_pr_freq") {
164 config.egress_pr_freq = parse_nonnegative_f64("ec_pr_freq", v)?;
165 }
166 if let Some(v) = params.get("ic_new_time") {
167 config.new_time = parse_nonnegative_f64("ic_new_time", v)?;
168 }
169 if let Some(v) = params.get("ic_burst_penalty") {
170 config.burst_penalty = parse_nonnegative_f64("ic_burst_penalty", v)?;
171 }
172 if let Some(v) = params.get("ic_held_release_interval") {
173 config.held_release_interval = parse_nonnegative_f64("ic_held_release_interval", v)?;
174 }
175
176 Ok(config)
177}
178
179fn parse_nonnegative_f64(key: &str, value: &str) -> Result<f64, String> {
180 let parsed = value
181 .parse::<f64>()
182 .map_err(|_| format!("{} must be numeric, got '{}'", key, value))?;
183 if parsed < 0.0 {
184 return Err(format!("{} must be >= 0, got '{}'", key, value));
185 }
186 Ok(parsed)
187}
188
189fn sanitize_ifac_config_string(kind: &str, value: &str) -> Option<String> {
195 if value == "None" {
196 log::warn!(
197 "Ambiguous IFAC {} \"None\", this value is ignored and an IFAC {} has NOT been set",
198 kind,
199 kind
200 );
201 return None;
202 }
203 if value.is_empty() {
204 return None;
205 }
206 Some(value.to_string())
207}
208
209fn extract_ifac_config(
212 params: &std::collections::HashMap<String, String>,
213 default_size: usize,
214) -> Option<IfacConfig> {
215 let netname = params
216 .get("networkname")
217 .or_else(|| params.get("network_name"))
218 .cloned()
219 .and_then(|value| sanitize_ifac_config_string("network name", &value));
220 let netkey = params
221 .get("passphrase")
222 .or_else(|| params.get("pass_phrase"))
223 .cloned()
224 .and_then(|value| sanitize_ifac_config_string("passphrase", &value));
225
226 if netname.is_none() && netkey.is_none() {
227 return None;
228 }
229
230 let size = params
232 .get("ifac_size")
233 .and_then(|v| v.parse::<usize>().ok())
234 .map(|bits| (bits / 8).max(1))
235 .unwrap_or(default_size);
236
237 Some(IfacConfig {
238 netname,
239 netkey,
240 size,
241 })
242}
243
244fn effective_static_transport_identity(
248 configured: bool,
249 transport_enabled: bool,
250 has_discoverable_interface: bool,
251) -> bool {
252 configured || (!transport_enabled && has_discoverable_interface)
253}
254
255fn extract_discovery_config(
257 iface_name: &str,
258 iface_type: &str,
259 params: &std::collections::HashMap<String, String>,
260) -> Option<crate::discovery::DiscoveryConfig> {
261 let discoverable = params
262 .get("discoverable")
263 .and_then(|v| config::parse_bool_pub(v))
264 .unwrap_or(false);
265 if !discoverable {
266 return None;
267 }
268
269 if iface_type == "TCPClientInterface" {
270 log::error!(
271 "Invalid interface discovery configuration for {}, aborting discovery announce",
272 iface_name
273 );
274 return None;
275 }
276
277 let discovery_name = params
278 .get("discovery_name")
279 .cloned()
280 .unwrap_or_else(|| iface_name.to_string());
281
282 let announce_interval = params
284 .get("announce_interval")
285 .and_then(|v| v.parse::<u64>().ok())
286 .map(|secs| secs.max(300))
287 .unwrap_or(21600);
288
289 let stamp_value = params
290 .get("discovery_stamp_value")
291 .and_then(|v| v.parse::<u8>().ok())
292 .unwrap_or(crate::discovery::DEFAULT_STAMP_VALUE);
293
294 let reachable_on = params.get("reachable_on").cloned();
295
296 let listen_port = params
297 .get("listen_port")
298 .or_else(|| params.get("port"))
299 .and_then(|v| v.parse().ok());
300
301 let location_cmd = params.get("location_cmd").cloned();
302
303 let latitude = params
304 .get("latitude")
305 .or_else(|| params.get("lat"))
306 .and_then(|v| v.parse().ok());
307 let longitude = params
308 .get("longitude")
309 .or_else(|| params.get("lon"))
310 .and_then(|v| v.parse().ok());
311 let height = params.get("height").and_then(|v| v.parse().ok());
312 let operator_lxmf_address = params.get("discovery_lxmf_address").and_then(|value| {
313 let parsed = parse_discovery_lxmf_address(value);
314 if parsed.is_none() {
315 log::error!(
316 "Invalid interface discovery LXMF address for '{}': {}",
317 iface_name,
318 value
319 );
320 }
321 parsed
322 });
323
324 Some(crate::discovery::DiscoveryConfig {
325 discovery_name,
326 announce_interval,
327 stamp_value,
328 reachable_on,
329 interface_type: iface_type.to_string(),
330 listen_port,
331 location_cmd,
332 latitude,
333 longitude,
334 height,
335 operator_lxmf_address,
336 })
337}
338
339fn default_discovery_runtime_config(
340 interface_name: &str,
341 interface_type: &str,
342 listen_port: Option<u16>,
343) -> crate::discovery::DiscoveryConfig {
344 crate::discovery::DiscoveryConfig {
345 discovery_name: interface_name.to_string(),
346 announce_interval: 21600,
347 stamp_value: crate::discovery::DEFAULT_STAMP_VALUE,
348 reachable_on: None,
349 interface_type: interface_type.to_string(),
350 listen_port,
351 location_cmd: None,
352 latitude: None,
353 longitude: None,
354 height: None,
355 operator_lxmf_address: None,
356 }
357}
358
359fn parse_discovery_lxmf_address(value: &str) -> Option<[u8; 16]> {
360 if value.len() != 32 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
361 return None;
362 }
363 let mut address = [0u8; 16];
364 for (index, byte) in address.iter_mut().enumerate() {
365 *byte = u8::from_str_radix(&value[index * 2..index * 2 + 2], 16).ok()?;
366 }
367 Some(address)
368}
369
370fn discovery_runtime_ifac_fields(ifac: Option<&IfacConfig>) -> (Option<String>, Option<String>) {
371 (
372 ifac.and_then(|cfg| cfg.netname.clone()),
373 ifac.and_then(|cfg| cfg.netkey.clone()),
374 )
375}
376
377#[cfg(feature = "iface-backbone")]
378pub(crate) fn backbone_discovery_runtime_from_interface(
379 interface_name: &str,
380 mode: &BackboneMode,
381 discovery: Option<&crate::discovery::DiscoveryConfig>,
382 transport_enabled: bool,
383 ifac: Option<&IfacConfig>,
384) -> Option<crate::driver::BackboneDiscoveryRuntimeHandle> {
385 let config = match mode {
386 BackboneMode::Server(config) => config,
387 BackboneMode::Client(_, _) => return None,
388 };
389
390 let startup_config = discovery.cloned().unwrap_or_else(|| {
391 default_discovery_runtime_config(
392 interface_name,
393 "BackboneInterface",
394 Some(config.listen_port),
395 )
396 });
397 let (ifac_netname, ifac_netkey) = discovery_runtime_ifac_fields(ifac);
398
399 Some(crate::driver::BackboneDiscoveryRuntimeHandle::from_parts(
400 config.name.clone(),
401 startup_config,
402 transport_enabled,
403 ifac_netname,
404 ifac_netkey,
405 discovery.is_some(),
406 ))
407}
408
409#[cfg(feature = "iface-tcp")]
410pub(crate) fn tcp_server_discovery_runtime_from_interface(
411 interface_name: &str,
412 config: &crate::interface::tcp_server::TcpServerConfig,
413 discovery: Option<&crate::discovery::DiscoveryConfig>,
414 transport_enabled: bool,
415 ifac: Option<&IfacConfig>,
416) -> crate::driver::TcpServerDiscoveryRuntimeHandle {
417 let startup_config = discovery.cloned().unwrap_or_else(|| {
418 default_discovery_runtime_config(
419 interface_name,
420 "TCPServerInterface",
421 Some(config.listen_port),
422 )
423 });
424 let (ifac_netname, ifac_netkey) = discovery_runtime_ifac_fields(ifac);
425
426 crate::driver::TcpServerDiscoveryRuntimeHandle::from_parts(
427 config.name.clone(),
428 startup_config,
429 transport_enabled,
430 ifac_netname,
431 ifac_netkey,
432 discovery.is_some(),
433 )
434}
435
436pub(crate) fn ifac_runtime_from_config(
437 ifac: Option<&IfacConfig>,
438 default_size: usize,
439) -> crate::driver::IfacRuntimeConfig {
440 crate::driver::IfacRuntimeConfig::from_parts(
441 ifac.and_then(|cfg| cfg.netname.clone()),
442 ifac.and_then(|cfg| cfg.netkey.clone()),
443 ifac.map(|cfg| cfg.size).unwrap_or(default_size),
444 )
445}
446
447pub(crate) fn discoverable_interface_from_config(
448 interface_name: &str,
449 discovery: &crate::discovery::DiscoveryConfig,
450 transport_enabled: bool,
451 ifac: Option<&IfacConfig>,
452) -> crate::discovery::DiscoverableInterface {
453 crate::discovery::DiscoverableInterface {
454 interface_name: interface_name.to_string(),
455 config: discovery.clone(),
456 transport_enabled,
457 ifac_netname: ifac.and_then(|cfg| cfg.netname.clone()),
458 ifac_netkey: ifac.and_then(|cfg| cfg.netkey.clone()),
459 }
460}
461
462pub(crate) fn derive_ifac_state(
463 ifac: Option<&IfacConfig>,
464 interface_name: &str,
465) -> io::Result<Option<crate::ifac::IfacState>> {
466 let Some(ifac) = ifac else {
467 return Ok(None);
468 };
469 if ifac.netname.is_none() && ifac.netkey.is_none() {
470 return Ok(None);
471 }
472
473 ifac::derive_ifac(ifac.netname.as_deref(), ifac.netkey.as_deref(), ifac.size)
474 .map(Some)
475 .map_err(|err| {
476 io::Error::other(format!(
477 "failed to derive IFAC for {}: {}",
478 interface_name, err
479 ))
480 })
481}
482
483pub(crate) struct StartedInterface<'a> {
484 pub(crate) driver: &'a mut Driver,
485 pub(crate) tx: &'a EventSender,
486 pub(crate) queue_capacity: usize,
487 pub(crate) id: rns_core::transport::types::InterfaceId,
488 pub(crate) info: rns_core::transport::types::InterfaceInfo,
489 pub(crate) writer: Box<dyn crate::interface::Writer>,
490 pub(crate) interface_type_name: String,
491 pub(crate) ifac_state: Option<crate::ifac::IfacState>,
492 pub(crate) ifac_runtime: &'a crate::driver::IfacRuntimeConfig,
493}
494
495pub(crate) fn register_started_interface(params: StartedInterface<'_>) {
496 let StartedInterface {
497 driver,
498 tx,
499 queue_capacity,
500 id,
501 mut info,
502 writer,
503 interface_type_name,
504 ifac_state,
505 ifac_runtime,
506 } = params;
507 driver.apply_announce_rate_defaults(&mut info);
508 driver.apply_ingress_control_defaults(&mut info);
509 let (writer, async_writer_metrics) =
510 crate::interface::wrap_async_writer(writer, id, &info.name, tx.clone(), queue_capacity);
511 driver.register_interface_runtime_defaults(&info);
512 driver.register_interface_ifac_runtime(&info.name, ifac_runtime.clone());
513 driver.engine.register_interface(info.clone());
514 driver.interfaces.insert(
515 id,
516 InterfaceEntry {
517 id,
518 info,
519 writer,
520 async_writer_metrics: Some(async_writer_metrics),
521 enabled: true,
522 online: false,
523 dynamic: false,
524 ifac: ifac_state,
525 stats: InterfaceStats {
526 started: time::now(),
527 ..Default::default()
528 },
529 interface_type: interface_type_name,
530 send_retry_at: None,
531 send_retry_backoff: Duration::ZERO,
532 },
533 );
534}
535
536pub(crate) struct InterfaceManagementConfig {
538 pub(crate) config_file: std::path::PathBuf,
539 pub(crate) storage_dir: std::path::PathBuf,
540 pub(crate) initial_ids: HashMap<String, rns_core::transport::types::InterfaceId>,
541 pub(crate) queue_capacity: usize,
542 pub(crate) transport_enabled: bool,
543 pub(crate) underlay_mark: Option<u32>,
544}
545
546pub struct NodeConfig {
548 pub transport_enabled: bool,
549 pub static_transport_identity: bool,
550 pub local_hops_delta: bool,
551 pub identity: Option<Identity>,
552 pub interfaces: Vec<InterfaceConfig>,
554 pub share_instance: bool,
556 pub instance_name: String,
558 pub shared_instance_port: u16,
560 pub rpc_port: u16,
562 pub cache_dir: Option<std::path::PathBuf>,
564 pub ratchet_store: Option<Arc<dyn storage::RatchetStore>>,
566 pub ratchet_expiry: Duration,
568 pub management: crate::management::ManagementConfig,
570 pub probe_port: Option<u16>,
572 pub probe_addrs: Vec<std::net::SocketAddr>,
574 pub probe_protocol: rns_core::holepunch::ProbeProtocol,
576 pub direct_connect_policy: crate::event::HolePunchPolicy,
578 pub device: Option<String>,
580 pub underlay_mark: Option<u32>,
582 pub hooks: Vec<config::ParsedHook>,
584 pub discover_interfaces: bool,
586 pub autoconnect_interface_mode: Option<u8>,
588 pub autoconnect_interface_gravity: i64,
590 pub autoconnect_announces_to_internal: bool,
592 pub autoconnect_unverified_implementations: bool,
594 pub discovery_required_value: Option<u8>,
596 pub respond_to_probes: bool,
598 pub prefer_shorter_path: bool,
602 pub max_paths_per_destination: usize,
605 pub packet_hashlist_max_entries: usize,
607 pub packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation,
609 pub max_discovery_pr_tags: usize,
611 pub max_path_destinations: usize,
613 pub max_tunnel_destinations_total: usize,
615 pub known_destinations_ttl: Duration,
617 pub known_destinations_max_entries: usize,
619 pub announce_table_ttl: Duration,
621 pub announce_table_max_bytes: usize,
623 pub driver_event_queue_capacity: usize,
625 pub interface_writer_queue_capacity: usize,
627 pub announce_rate_defaults: AnnounceRateDefaults,
629 pub ingress_control_defaults: rns_core::transport::types::IngressControlConfig,
631 #[cfg(feature = "iface-backbone")]
633 pub backbone_peer_pool: Option<BackbonePeerPoolSettings>,
634 pub announce_sig_cache_enabled: bool,
636 pub announce_sig_cache_max_entries: usize,
638 pub announce_sig_cache_ttl: Duration,
640 pub registry: Option<crate::interface::registry::InterfaceRegistry>,
642 pub panic_on_interface_error: bool,
645 #[cfg(feature = "hooks")]
647 pub provider_bridge: Option<crate::provider_bridge::ProviderBridgeConfig>,
648}
649
650struct NodeQueueConfig {
651 rpc_auth_key: Option<[u8; 32]>,
652 announce_max_entries: usize,
653 announce_max_interfaces: usize,
654 announce_max_bytes: usize,
655 announce_ttl_secs: f64,
656 announce_overflow_policy: AnnounceQueueOverflowPolicy,
657 inbound_capacities: crate::event::InboundQueueCapacities,
658 link_mtu_discovery: bool,
659}
660
661impl Default for NodeQueueConfig {
662 fn default() -> Self {
663 Self {
664 rpc_auth_key: None,
665 announce_max_entries: 256,
666 announce_max_interfaces: 1024,
667 announce_max_bytes: 256 * 1024,
668 announce_ttl_secs: 30.0,
669 announce_overflow_policy: AnnounceQueueOverflowPolicy::DropWorst,
670 inbound_capacities: crate::event::InboundQueueCapacities::default(),
671 link_mtu_discovery: true,
672 }
673 }
674}
675
676impl Default for NodeConfig {
677 fn default() -> Self {
678 Self {
679 transport_enabled: false,
680 static_transport_identity: false,
681 local_hops_delta: false,
682 identity: None,
683 interfaces: Vec::new(),
684 share_instance: false,
685 instance_name: "default".into(),
686 shared_instance_port: 37428,
687 rpc_port: 0,
688 cache_dir: None,
689 ratchet_store: None,
690 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
691 management: Default::default(),
692 probe_port: None,
693 probe_addrs: vec![],
694 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
695 direct_connect_policy: crate::event::HolePunchPolicy::AcceptAll,
696 device: None,
697 underlay_mark: None,
698 hooks: Vec::new(),
699 discover_interfaces: false,
700 autoconnect_interface_mode: None,
701 autoconnect_interface_gravity: 0,
702 autoconnect_announces_to_internal: false,
703 autoconnect_unverified_implementations: false,
704 discovery_required_value: None,
705 respond_to_probes: false,
706 prefer_shorter_path: false,
707 max_paths_per_destination: 1,
708 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
709 packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Eager,
710 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
711 max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
712 max_tunnel_destinations_total: usize::MAX,
713 known_destinations_ttl: Duration::from_secs(48 * 60 * 60),
714 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
715 announce_table_ttl: Duration::from_secs(rns_core::constants::ANNOUNCE_TABLE_TTL as u64),
716 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
717 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
718 interface_writer_queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
719 announce_rate_defaults: AnnounceRateDefaults::default(),
720 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(),
721 #[cfg(feature = "iface-backbone")]
722 backbone_peer_pool: None,
723 announce_sig_cache_enabled: true,
724 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
725 announce_sig_cache_ttl: Duration::from_secs(
726 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
727 ),
728 registry: None,
729 panic_on_interface_error: false,
730 #[cfg(feature = "hooks")]
731 provider_bridge: None,
732 }
733 }
734}
735
736pub struct IfacConfig {
738 pub netname: Option<String>,
739 pub netkey: Option<String>,
740 pub size: usize,
741}
742
743pub struct InterfaceConfig {
745 pub name: String,
746 pub type_name: String,
747 pub config_data: Box<dyn crate::interface::InterfaceConfigData>,
748 pub mode: u8,
749 pub gravity: i64,
750 pub recursive_prs: bool,
751 pub announces_from_internal: bool,
752 pub announces_to_internal: Option<bool>,
753 pub ingress_control: rns_core::transport::types::IngressControlConfig,
754 pub ifac: Option<IfacConfig>,
755 pub discovery: Option<crate::discovery::DiscoveryConfig>,
756}
757
758pub(crate) fn parse_managed_interface(
759 iface: &config::ParsedInterface,
760 parsed: &config::RnsConfig,
761 storage_dir: &Path,
762 id: rns_core::transport::types::InterfaceId,
763 registry: &crate::interface::registry::InterfaceRegistry,
764) -> io::Result<InterfaceConfig> {
765 let factory = registry.get(&iface.interface_type).ok_or_else(|| {
766 io::Error::new(
767 io::ErrorKind::InvalidInput,
768 format!("unsupported interface type '{}'", iface.interface_type),
769 )
770 })?;
771 let discoverable = iface
772 .params
773 .get("discoverable")
774 .and_then(|value| config::parse_bool_pub(value))
775 .unwrap_or(false);
776 let ignore_warnings = iface
777 .params
778 .get("ignore_config_warnings")
779 .and_then(|value| config::parse_bool_pub(value))
780 .unwrap_or(false);
781 let mode = normalize_discovery_mode(
782 &iface.interface_type,
783 parse_interface_mode(&iface.mode),
784 discoverable,
785 ignore_warnings,
786 );
787 let ifac = extract_ifac_config(&iface.params, factory.default_ifac_size());
788 let discovery = extract_discovery_config(&iface.name, &iface.interface_type, &iface.params);
789 let ingress_control = parse_ingress_control_config(&iface.interface_type, &iface.params)
790 .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
791 let recursive_prs = iface
792 .params
793 .get("recursive_prs")
794 .and_then(|value| config::parse_bool_pub(value))
795 .unwrap_or(false);
796 let announces_from_internal = iface
797 .params
798 .get("announces_from_internal")
799 .and_then(|value| config::parse_bool_pub(value))
800 .unwrap_or(true);
801 let announces_to_internal = iface
802 .params
803 .get("announces_to_internal")
804 .and_then(|value| config::parse_bool_pub(value));
805 let gravity = parse_interface_gravity(
806 iface.params.get("gravity").map(String::as_str),
807 parsed.reticulum.default_gravity,
808 );
809 let mut params = iface.params.clone();
810 params
811 .entry("storage_dir".to_string())
812 .or_insert_with(|| storage_dir.to_string_lossy().to_string());
813 if let Some(device) = &parsed.reticulum.device {
814 params
815 .entry("device".to_string())
816 .or_insert_with(|| device.clone());
817 }
818 let config_data = factory
819 .parse_config_section(
820 &iface.name,
821 id,
822 crate::interface::ConfigSection {
823 params: ¶ms,
824 children: &iface.subinterfaces,
825 },
826 )
827 .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
828 Ok(InterfaceConfig {
829 name: iface.name.clone(),
830 type_name: iface.interface_type.clone(),
831 config_data,
832 mode,
833 gravity,
834 recursive_prs,
835 announces_from_internal,
836 announces_to_internal,
837 ingress_control,
838 ifac,
839 discovery,
840 })
841}
842
843use crate::event::{QueryRequest, QueryResponse};
844
845#[derive(Debug)]
847pub struct SendError;
848
849impl std::fmt::Display for SendError {
850 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
851 write!(f, "driver shut down")
852 }
853}
854
855impl std::error::Error for SendError {}
856
857#[derive(Debug, Clone, PartialEq, Eq)]
859pub enum ChannelSendError {
860 NotReady,
861 Rejected(String),
862 Unavailable,
863}
864
865impl From<String> for ChannelSendError {
866 fn from(error: String) -> Self {
867 if error == rns_core::channel::ChannelError::NotReady.to_string() {
868 Self::NotReady
869 } else {
870 Self::Rejected(error)
871 }
872 }
873}
874
875impl std::fmt::Display for ChannelSendError {
876 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
877 match self {
878 Self::NotReady => write!(f, "channel is not ready to send"),
879 Self::Rejected(error) => write!(f, "{error}"),
880 Self::Unavailable => write!(f, "driver is stopped or draining"),
881 }
882 }
883}
884
885impl std::error::Error for ChannelSendError {}
886
887pub struct RnsNode {
889 tx: EventSender,
890 driver_handle: Option<JoinHandle<()>>,
891 verify_handle: Option<JoinHandle<()>>,
892 verify_shutdown: Arc<AtomicBool>,
893 persist_handle: Option<JoinHandle<()>>,
894 persist_stop: Option<std::sync::mpsc::Sender<()>>,
895 rpc_server: Option<crate::rpc::RpcServer>,
896 #[cfg(target_os = "linux")]
897 rpc_unix_server: Option<crate::rpc::RpcServer>,
898 tick_interval_ms: Arc<AtomicU64>,
899 #[allow(dead_code)]
900 probe_server: Option<crate::holepunch::probe::ProbeServerHandle>,
901 known_destinations_path: Option<std::path::PathBuf>,
902 transport_state_dir: Option<std::path::PathBuf>,
903 ratchet_store: Option<Arc<dyn storage::RatchetStore>>,
904 ratchet_expiry: Duration,
905}
906
907impl RnsNode {
908 pub fn from_config(
911 config_path: Option<&Path>,
912 callbacks: Box<dyn Callbacks>,
913 ) -> io::Result<Self> {
914 let config_dir = storage::resolve_config_dir(config_path);
915 Self::from_config_dirs(&config_dir, &config_dir, None, false, callbacks)
916 }
917
918 pub fn from_private_config(
926 config_dir: &Path,
927 state_dir: &Path,
928 underlay_mark: Option<u32>,
929 callbacks: Box<dyn Callbacks>,
930 ) -> io::Result<Self> {
931 Self::from_config_dirs(config_dir, state_dir, underlay_mark, true, callbacks)
932 }
933
934 fn from_config_dirs(
935 config_dir: &Path,
936 state_dir: &Path,
937 underlay_mark: Option<u32>,
938 private: bool,
939 callbacks: Box<dyn Callbacks>,
940 ) -> io::Result<Self> {
941 let paths = storage::ensure_storage_dirs(state_dir)?;
942 let known_destinations_path = paths.storage.join("known_destinations");
943 let ratchet_store: Arc<dyn storage::RatchetStore> =
944 Arc::new(storage::FsRatchetStore::new(paths.ratchets.clone()));
945
946 let config_file = config_dir.join("config");
948 let config_content = if config_file.exists() {
949 std::fs::read_to_string(&config_file)?
950 } else {
951 String::new()
952 };
953 let rns_config = config::parse(&config_content)
954 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;
955 let link_mtu_discovery = config::parse_link_mtu_discovery(&config_content)
956 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;
957
958 let identity = if let Some(ref id_path_str) = rns_config.reticulum.network_identity {
960 let id_path = std::path::PathBuf::from(id_path_str);
961 if id_path.exists() {
962 storage::load_identity(&id_path)?
963 } else {
964 let id = Identity::new(&mut OsRng);
965 storage::save_identity(&id, &id_path)?;
966 id
967 }
968 } else {
969 storage::load_or_create_identity(&paths.identities)?
970 };
971 let rpc_auth_key = if !private && rns_config.reticulum.share_instance {
974 let transport_identity_path = paths.storage.join("transport_identity");
975 let rpc_identity = if transport_identity_path.exists() {
976 storage::load_identity(&transport_identity_path)?
977 } else {
978 storage::save_identity(&identity, &transport_identity_path)?;
979 Identity::from_private_key(
980 &identity
981 .get_private_key()
982 .expect("node identity has private key"),
983 )
984 };
985 Some(crate::rpc::derive_auth_key(
986 &rpc_identity
987 .get_private_key()
988 .expect("RPC identity has private key"),
989 ))
990 } else {
991 None
992 };
993
994 let registry = crate::interface::registry::InterfaceRegistry::with_builtins();
996 let mut interface_configs = Vec::new();
997 let mut initial_ids = HashMap::new();
998 let mut next_id_val = 1u64;
999
1000 for iface in &rns_config.interfaces {
1001 if !iface.enabled {
1002 continue;
1003 }
1004
1005 let iface_id = rns_core::transport::types::InterfaceId(next_id_val);
1006 next_id_val += 1;
1007
1008 let factory = match registry.get(&iface.interface_type) {
1009 Some(f) => f,
1010 None => {
1011 if private {
1012 return Err(io::Error::new(
1013 io::ErrorKind::InvalidData,
1014 format!(
1015 "unsupported enabled interface type '{}' for '{}'",
1016 iface.interface_type, iface.name
1017 ),
1018 ));
1019 }
1020 log::warn!(
1021 "Unsupported interface type '{}' for '{}'",
1022 iface.interface_type,
1023 iface.name,
1024 );
1025 continue;
1026 }
1027 };
1028
1029 let mut iface_mode = parse_interface_mode(&iface.mode);
1030
1031 let discoverable = iface
1032 .params
1033 .get("discoverable")
1034 .and_then(|value| config::parse_bool_pub(value))
1035 .unwrap_or(false);
1036 let ignore_config_warnings = iface
1037 .params
1038 .get("ignore_config_warnings")
1039 .and_then(|value| config::parse_bool_pub(value))
1040 .unwrap_or(false);
1041 let normalized_mode = normalize_discovery_mode(
1042 &iface.interface_type,
1043 iface_mode,
1044 discoverable,
1045 ignore_config_warnings,
1046 );
1047 if normalized_mode != iface_mode {
1048 log::info!(
1049 "Interface '{}' has discovery enabled, auto-configuring mode to {}",
1050 iface.name,
1051 if normalized_mode == rns_core::constants::MODE_ACCESS_POINT {
1052 "ACCESS_POINT"
1053 } else {
1054 "GATEWAY"
1055 }
1056 );
1057 iface_mode = normalized_mode;
1058 }
1059
1060 let default_ifac_size = factory.default_ifac_size();
1061 let ifac_config = extract_ifac_config(&iface.params, default_ifac_size);
1062 let discovery_config =
1063 extract_discovery_config(&iface.name, &iface.interface_type, &iface.params);
1064 let ingress_control =
1065 match parse_ingress_control_config(&iface.interface_type, &iface.params) {
1066 Ok(config) => config,
1067 Err(e) => {
1068 if private {
1069 return Err(io::Error::new(
1070 io::ErrorKind::InvalidData,
1071 format!("invalid ingress control for '{}': {e}", iface.name),
1072 ));
1073 }
1074 log::warn!(
1075 "Failed to parse ingress control config for '{}': {}",
1076 iface.name,
1077 e
1078 );
1079 continue;
1080 }
1081 };
1082 let recursive_prs = iface
1083 .params
1084 .get("recursive_prs")
1085 .and_then(|v| config::parse_bool_pub(v))
1086 .unwrap_or(false);
1087 let announces_from_internal = iface
1088 .params
1089 .get("announces_from_internal")
1090 .and_then(|v| config::parse_bool_pub(v))
1091 .unwrap_or(true);
1092 let gravity = parse_interface_gravity(
1093 iface.params.get("gravity").map(String::as_str),
1094 rns_config.reticulum.default_gravity,
1095 );
1096 let announces_to_internal = iface
1097 .params
1098 .get("announces_to_internal")
1099 .and_then(|v| config::parse_bool_pub(v));
1100
1101 let mut params = iface.params.clone();
1103 if !params.contains_key("storage_dir") {
1104 params.insert(
1105 "storage_dir".to_string(),
1106 paths.storage.to_string_lossy().to_string(),
1107 );
1108 }
1109 if let Some(ref device) = rns_config.reticulum.device {
1111 if !params.contains_key("device") {
1112 params.insert("device".to_string(), device.clone());
1113 }
1114 }
1115
1116 let config_data = match factory.parse_config_section(
1117 &iface.name,
1118 iface_id,
1119 crate::interface::ConfigSection {
1120 params: ¶ms,
1121 children: &iface.subinterfaces,
1122 },
1123 ) {
1124 Ok(data) => data,
1125 Err(e) => {
1126 if private {
1127 return Err(io::Error::new(
1128 io::ErrorKind::InvalidData,
1129 format!("invalid interface '{}': {e}", iface.name),
1130 ));
1131 }
1132 log::warn!("Failed to parse config for '{}': {}", iface.name, e);
1133 continue;
1134 }
1135 };
1136
1137 interface_configs.push(InterfaceConfig {
1138 name: iface.name.clone(),
1139 type_name: iface.interface_type.clone(),
1140 config_data,
1141 mode: iface_mode,
1142 gravity,
1143 recursive_prs,
1144 announces_from_internal,
1145 announces_to_internal,
1146 ingress_control,
1147 ifac: ifac_config,
1148 discovery: discovery_config,
1149 });
1150 initial_ids.insert(iface.name.clone(), iface_id);
1151 }
1152
1153 let mut mgmt_allowed = Vec::new();
1155 for hex_hash in &rns_config.reticulum.remote_management_allowed {
1156 if hex_hash.len() == 32 {
1157 if let Ok(bytes) = (0..hex_hash.len())
1158 .step_by(2)
1159 .map(|i| u8::from_str_radix(&hex_hash[i..i + 2], 16))
1160 .collect::<Result<Vec<u8>, _>>()
1161 {
1162 if bytes.len() == 16 {
1163 let mut h = [0u8; 16];
1164 h.copy_from_slice(&bytes);
1165 mgmt_allowed.push(h);
1166 }
1167 } else {
1168 log::warn!("Invalid hex in remote_management_allowed: {}", hex_hash);
1169 }
1170 } else {
1171 log::warn!(
1172 "Invalid entry in remote_management_allowed (expected 32 hex chars, got {}): {}",
1173 hex_hash.len(), hex_hash,
1174 );
1175 }
1176 }
1177
1178 let probe_addrs: Vec<std::net::SocketAddr> = rns_config
1180 .reticulum
1181 .probe_addr
1182 .as_ref()
1183 .map(|s| {
1184 s.split(',')
1185 .filter_map(|entry| {
1186 let trimmed = entry.trim();
1187 if trimmed.is_empty() {
1188 return None;
1189 }
1190 trimmed
1191 .parse::<std::net::SocketAddr>()
1192 .map_err(|e| {
1193 log::warn!("Invalid probe_addr entry '{}': {}", trimmed, e);
1194 e
1195 })
1196 .ok()
1197 })
1198 .collect()
1199 })
1200 .unwrap_or_default();
1201
1202 let probe_protocol = match rns_config
1204 .reticulum
1205 .probe_protocol
1206 .as_deref()
1207 .map(|s| s.to_lowercase())
1208 {
1209 Some(ref s) if s == "stun" => rns_core::holepunch::ProbeProtocol::Stun,
1210 _ => rns_core::holepunch::ProbeProtocol::Rnsp,
1211 };
1212
1213 let known_destinations = match storage::load_known_destinations(&known_destinations_path) {
1214 Ok(destinations) => destinations,
1215 Err(err) if err.kind() == io::ErrorKind::NotFound => Default::default(),
1216 Err(err) => {
1217 log::warn!("failed to load known destinations: {}", err);
1218 Default::default()
1219 }
1220 };
1221 let known_destination_hashes: HashSet<[u8; 16]> =
1222 known_destinations.keys().copied().collect();
1223 match ratchet_store.cleanup(
1224 &known_destination_hashes,
1225 time::now(),
1226 rns_config.reticulum.ratchet_expiry as f64,
1227 ) {
1228 Ok(stats) if stats.processed > 0 => log::debug!(
1229 "Processed {} ratchets, not in use {}, removed {}",
1230 stats.processed,
1231 stats.not_known,
1232 stats.removed
1233 ),
1234 Ok(_) => {}
1235 Err(err) => log::warn!("failed to clean ratchets: {}", err),
1236 }
1237
1238 let interface_management = if private || !rns_config.reticulum.enable_interface_management {
1239 None
1240 } else {
1241 Some(InterfaceManagementConfig {
1242 config_file: config_file.clone(),
1243 storage_dir: paths.storage.clone(),
1244 initial_ids,
1245 queue_capacity: rns_config.reticulum.interface_writer_queue_capacity,
1246 transport_enabled: rns_config.reticulum.enable_transport,
1247 underlay_mark: underlay_mark.or(rns_config.reticulum.underlay_mark),
1248 })
1249 };
1250 let node_config = NodeConfig {
1251 transport_enabled: rns_config.reticulum.enable_transport,
1252 static_transport_identity: rns_config.reticulum.static_transport_identity,
1253 local_hops_delta: rns_config.reticulum.local_hops_delta,
1254 identity: Some(identity),
1255 share_instance: !private && rns_config.reticulum.share_instance,
1256 instance_name: rns_config.reticulum.instance_name.clone(),
1257 shared_instance_port: rns_config.reticulum.shared_instance_port,
1258 rpc_port: rns_config.reticulum.instance_control_port,
1259 cache_dir: Some(paths.cache),
1260 ratchet_store: Some(Arc::clone(&ratchet_store)),
1261 ratchet_expiry: Duration::from_secs(rns_config.reticulum.ratchet_expiry),
1262 management: if private {
1263 crate::management::ManagementConfig::default()
1264 } else {
1265 crate::management::ManagementConfig {
1266 enable_remote_management: rns_config.reticulum.enable_remote_management,
1267 remote_management_allowed: mgmt_allowed,
1268 publish_blackhole: rns_config.reticulum.publish_blackhole,
1269 }
1270 },
1271 probe_port: rns_config.reticulum.probe_port,
1272 probe_addrs,
1273 probe_protocol,
1274 direct_connect_policy: rns_config.reticulum.direct_connect_policy,
1275 device: rns_config.reticulum.device.clone(),
1276 underlay_mark: underlay_mark.or(rns_config.reticulum.underlay_mark),
1277 hooks: if private {
1278 Vec::new()
1279 } else {
1280 rns_config.hooks.clone()
1281 },
1282 discover_interfaces: rns_config.reticulum.discover_interfaces,
1283 autoconnect_interface_mode: rns_config
1284 .reticulum
1285 .autoconnect_interface_mode
1286 .as_deref()
1287 .and_then(parse_optional_interface_mode),
1288 autoconnect_interface_gravity: rns_config.reticulum.autoconnect_interface_gravity,
1289 autoconnect_announces_to_internal: rns_config
1290 .reticulum
1291 .autoconnect_announces_to_internal,
1292 autoconnect_unverified_implementations: rns_config
1293 .reticulum
1294 .autoconnect_unverified_implementations,
1295 discovery_required_value: rns_config.reticulum.required_discovery_value,
1296 respond_to_probes: rns_config.reticulum.respond_to_probes,
1297 prefer_shorter_path: rns_config.reticulum.prefer_shorter_path,
1298 max_paths_per_destination: rns_config.reticulum.max_paths_per_destination,
1299 packet_hashlist_max_entries: rns_config.reticulum.packet_hashlist_max_entries,
1300 packet_hashlist_allocation: rns_config.reticulum.packet_hashlist_allocation,
1301 max_discovery_pr_tags: rns_config.reticulum.max_discovery_pr_tags,
1302 max_path_destinations: rns_config.reticulum.max_path_destinations,
1303 max_tunnel_destinations_total: rns_config.reticulum.max_tunnel_destinations_total,
1304 known_destinations_ttl: Duration::from_secs(
1305 rns_config.reticulum.known_destinations_ttl,
1306 ),
1307 known_destinations_max_entries: rns_config.reticulum.known_destinations_max_entries,
1308 announce_table_ttl: Duration::from_secs(rns_config.reticulum.announce_table_ttl),
1309 announce_table_max_bytes: rns_config.reticulum.announce_table_max_bytes,
1310 driver_event_queue_capacity: rns_config.reticulum.driver_event_queue_capacity,
1311 interface_writer_queue_capacity: rns_config.reticulum.interface_writer_queue_capacity,
1312 announce_rate_defaults: AnnounceRateDefaults {
1313 target: rns_config.reticulum.default_ar_target,
1314 penalty: rns_config.reticulum.default_ar_penalty,
1315 grace: rns_config.reticulum.default_ar_grace,
1316 },
1317 ingress_control_defaults: rns_core::transport::types::IngressControlConfig {
1318 enabled: true,
1319 egress_enabled: rns_config.reticulum.default_egress_control,
1320 max_held_announces: rns_config.reticulum.default_ic_max_held_announces,
1321 burst_freq_new: rns_config.reticulum.default_ic_burst_freq_new,
1322 burst_freq: rns_config.reticulum.default_ic_burst_freq,
1323 pr_burst_freq_new: rns_config.reticulum.default_ic_pr_burst_freq_new,
1324 pr_burst_freq: rns_config.reticulum.default_ic_pr_burst_freq,
1325 egress_pr_freq: rns_config.reticulum.default_ec_pr_freq,
1326 new_time: rns_config.reticulum.default_ic_new_time,
1327 burst_hold: rns_config.reticulum.default_ic_burst_hold,
1328 burst_penalty: rns_config.reticulum.default_ic_burst_penalty,
1329 held_release_interval: rns_config.reticulum.default_ic_held_release_interval,
1330 },
1331 #[cfg(feature = "iface-backbone")]
1332 backbone_peer_pool: if rns_config.reticulum.backbone_peer_pool_max_connected > 0 {
1333 Some(BackbonePeerPoolSettings {
1334 max_connected: rns_config.reticulum.backbone_peer_pool_max_connected,
1335 failure_threshold: rns_config.reticulum.backbone_peer_pool_failure_threshold,
1336 failure_window: Duration::from_secs(
1337 rns_config.reticulum.backbone_peer_pool_failure_window,
1338 ),
1339 cooldown: Duration::from_secs(rns_config.reticulum.backbone_peer_pool_cooldown),
1340 })
1341 } else {
1342 None
1343 },
1344 announce_sig_cache_enabled: rns_config.reticulum.announce_sig_cache_enabled,
1345 announce_sig_cache_max_entries: rns_config.reticulum.announce_sig_cache_max_entries,
1346 announce_sig_cache_ttl: Duration::from_secs(
1347 rns_config.reticulum.announce_sig_cache_ttl,
1348 ),
1349 interfaces: interface_configs,
1350 registry: None,
1351 panic_on_interface_error: private || rns_config.reticulum.panic_on_interface_error,
1352 #[cfg(feature = "hooks")]
1353 provider_bridge: if !private && rns_config.reticulum.provider_bridge {
1354 Some(crate::provider_bridge::ProviderBridgeConfig {
1355 enabled: true,
1356 socket_path: rns_config
1357 .reticulum
1358 .provider_socket_path
1359 .as_ref()
1360 .map(std::path::PathBuf::from)
1361 .unwrap_or_else(|| config_dir.join("provider.sock")),
1362 queue_max_events: rns_config.reticulum.provider_queue_max_events,
1363 queue_max_bytes: rns_config.reticulum.provider_queue_max_bytes,
1364 overflow_policy: match rns_config.reticulum.provider_overflow_policy.as_str() {
1365 "drop_oldest" => crate::provider_bridge::OverflowPolicy::DropOldest,
1366 _ => crate::provider_bridge::OverflowPolicy::DropNewest,
1367 },
1368 node_instance: rns_config.reticulum.instance_name.clone(),
1369 })
1370 } else {
1371 None
1372 },
1373 };
1374
1375 let mut node = Self::start_with_queue_config(
1376 node_config,
1377 callbacks,
1378 NodeQueueConfig {
1379 rpc_auth_key,
1380 announce_max_entries: rns_config.reticulum.announce_queue_max_entries,
1381 announce_max_interfaces: rns_config.reticulum.announce_queue_max_interfaces,
1382 announce_max_bytes: rns_config.reticulum.announce_queue_max_bytes,
1383 announce_ttl_secs: rns_config.reticulum.announce_queue_ttl as f64,
1384 announce_overflow_policy: match rns_config
1385 .reticulum
1386 .announce_queue_overflow_policy
1387 .as_str()
1388 {
1389 "drop_newest" => AnnounceQueueOverflowPolicy::DropNewest,
1390 "drop_oldest" => AnnounceQueueOverflowPolicy::DropOldest,
1391 _ => AnnounceQueueOverflowPolicy::DropWorst,
1392 },
1393 inbound_capacities: rns_config.reticulum.inbound_queue_capacities,
1394 link_mtu_discovery,
1395 },
1396 interface_management,
1397 )?;
1398
1399 node.known_destinations_path = Some(known_destinations_path.clone());
1400 for (dest_hash, known) in known_destinations {
1401 let _ = node.query(QueryRequest::RestoreKnownDestination(
1402 crate::event::KnownDestinationEntry {
1403 dest_hash,
1404 identity_hash: known.identity_hash,
1405 public_key: known.public_key,
1406 app_data: known.app_data,
1407 hops: known.hops,
1408 received_at: known.received_at,
1409 receiving_interface: rns_core::transport::types::InterfaceId(
1410 known.receiving_interface,
1411 ),
1412 was_used: known.was_used,
1413 last_used_at: known.last_used_at,
1414 retained: known.retained,
1415 },
1416 ));
1417 }
1418
1419 Ok(node)
1420 }
1421
1422 pub fn start(config: NodeConfig, callbacks: Box<dyn Callbacks>) -> io::Result<Self> {
1424 crate::interface::validate_underlay_mark(config.underlay_mark)?;
1425 let inbound_queue_capacities = crate::event::InboundQueueCapacities::from_shared_capacity(
1426 config.driver_event_queue_capacity,
1427 );
1428 Self::start_with_queue_config(
1429 config,
1430 callbacks,
1431 NodeQueueConfig {
1432 inbound_capacities: inbound_queue_capacities,
1433 ..NodeQueueConfig::default()
1434 },
1435 None,
1436 )
1437 }
1438
1439 fn start_with_queue_config(
1440 config: NodeConfig,
1441 callbacks: Box<dyn Callbacks>,
1442 queue_config: NodeQueueConfig,
1443 interface_management: Option<InterfaceManagementConfig>,
1444 ) -> io::Result<Self> {
1445 let transport_state_dir = if config.transport_enabled {
1446 config.cache_dir.as_ref().map(|cache_dir| {
1447 cache_dir
1448 .parent()
1449 .unwrap_or_else(|| std::path::Path::new("."))
1450 .join("storage")
1451 })
1452 } else {
1453 None
1454 };
1455 let identity = config.identity.unwrap_or_else(|| Identity::new(&mut OsRng));
1456 let rpc_auth_key = queue_config.rpc_auth_key.unwrap_or_else(|| {
1457 crate::rpc::derive_auth_key(
1458 &identity
1459 .get_private_key()
1460 .expect("node identity has private key"),
1461 )
1462 });
1463 let has_discoverable_interface = config
1464 .interfaces
1465 .iter()
1466 .any(|iface| iface.discovery.is_some());
1467 let static_transport_identity = effective_static_transport_identity(
1468 config.static_transport_identity,
1469 config.transport_enabled,
1470 has_discoverable_interface,
1471 );
1472 if static_transport_identity && !config.static_transport_identity {
1473 log::warn!(
1474 "Discoverable interface was configured, enabling static transport identity on non-transport instance"
1475 );
1476 }
1477 let transport_identity = if config.transport_enabled || static_transport_identity {
1478 Identity::from_private_key(
1479 &identity
1480 .get_private_key()
1481 .expect("node identity must expose a private key"),
1482 )
1483 } else {
1484 Identity::new(&mut OsRng)
1485 };
1486
1487 let local_hops_delta = if config.local_hops_delta {
1488 let mut byte = [0u8; 1];
1489 OsRng.fill_bytes(&mut byte);
1490 byte[0] % 6 + 2
1491 } else {
1492 0
1493 };
1494
1495 let transport_config = TransportConfig {
1496 transport_enabled: config.transport_enabled,
1497 identity_hash: Some(*transport_identity.hash()),
1498 local_hops_delta,
1499 prefer_shorter_path: config.prefer_shorter_path,
1500 max_paths_per_destination: config.max_paths_per_destination,
1501 packet_hashlist_max_entries: config.packet_hashlist_max_entries,
1502 packet_hashlist_allocation: config.packet_hashlist_allocation,
1503 max_discovery_pr_tags: config.max_discovery_pr_tags,
1504 max_path_destinations: config.max_path_destinations,
1505 max_tunnel_destinations_total: config.max_tunnel_destinations_total,
1506 destination_timeout_secs: config.known_destinations_ttl.as_secs_f64(),
1507 announce_table_ttl_secs: config.announce_table_ttl.as_secs_f64(),
1508 announce_table_max_bytes: config.announce_table_max_bytes,
1509 announce_sig_cache_enabled: config.announce_sig_cache_enabled,
1510 announce_sig_cache_max_entries: config.announce_sig_cache_max_entries,
1511 announce_sig_cache_ttl_secs: config.announce_sig_cache_ttl.as_secs_f64(),
1512 announce_queue_max_entries: queue_config.announce_max_entries,
1513 announce_queue_max_interfaces: queue_config.announce_max_interfaces,
1514 };
1515
1516 let (tx, rx) = event::channel_with_queue_capacities(
1517 config.driver_event_queue_capacity,
1518 queue_config.inbound_capacities,
1519 );
1520 let tick_interval_ms = Arc::new(AtomicU64::new(1000));
1521 let mut driver = Driver::new(transport_config, rx, tx.clone(), callbacks);
1522 driver.underlay_mark = config.underlay_mark;
1523 driver
1524 .link_manager
1525 .set_link_mtu_discovery(queue_config.link_mtu_discovery);
1526 driver.set_announce_verify_queue_config(
1527 queue_config.announce_max_entries,
1528 queue_config.announce_max_bytes,
1529 queue_config.announce_ttl_secs,
1530 queue_config.announce_overflow_policy,
1531 );
1532 driver.async_announce_verification = true;
1533 driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
1534 driver.known_destinations_ttl = config.known_destinations_ttl.as_secs_f64();
1535 driver.known_destinations_max_entries = config.known_destinations_max_entries;
1536 driver.ratchet_store = config.ratchet_store.clone();
1537 driver.ratchet_expiry_secs = config.ratchet_expiry.as_secs_f64();
1538 driver.interface_writer_queue_capacity = config.interface_writer_queue_capacity;
1539 driver.set_announce_rate_defaults(config.announce_rate_defaults);
1540 driver.set_ingress_control_defaults(config.ingress_control_defaults);
1541 driver.runtime_config_defaults.known_destinations_ttl =
1542 config.known_destinations_ttl.as_secs_f64();
1543 driver
1544 .holepunch_manager
1545 .set_policy(config.direct_connect_policy);
1546 driver.runtime_config_defaults.direct_connect_policy = config.direct_connect_policy;
1547 #[cfg(feature = "hooks")]
1548 if let Some(provider_config) = config.provider_bridge.clone() {
1549 driver.runtime_config_defaults.provider_queue_max_events =
1550 provider_config.queue_max_events;
1551 driver.runtime_config_defaults.provider_queue_max_bytes =
1552 provider_config.queue_max_bytes;
1553 if provider_config.enabled {
1554 match crate::provider_bridge::ProviderBridge::start(provider_config) {
1555 Ok(bridge) => driver.provider_bridge = Some(bridge),
1556 Err(err) => log::warn!("failed to start provider bridge: {}", err),
1557 }
1558 }
1559 }
1560
1561 if let Some(ref cache_dir) = config.cache_dir {
1563 let announces_dir = cache_dir.join("announces");
1564 let _ = std::fs::create_dir_all(&announces_dir);
1565 driver.announce_cache = Some(crate::announce_cache::AnnounceCache::new(announces_dir));
1566 }
1567
1568 if !config.probe_addrs.is_empty()
1570 || config.device.is_some()
1571 || config.underlay_mark.is_some()
1572 {
1573 driver.set_probe_config(
1574 config.probe_addrs.clone(),
1575 config.probe_protocol,
1576 config.device.clone(),
1577 );
1578 }
1579
1580 let probe_server = if let Some(port) = config.probe_port {
1582 let listen_addr: std::net::SocketAddr = ([0, 0, 0, 0], port).into();
1583 match crate::holepunch::probe::start_probe_server_with_mark(
1584 listen_addr,
1585 config.underlay_mark,
1586 ) {
1587 Ok(handle) => {
1588 log::info!("Probe server started on 0.0.0.0:{}", port);
1589 Some(handle)
1590 }
1591 Err(e) => {
1592 log::error!("Failed to start probe server on port {}: {}", port, e);
1593 None
1594 }
1595 }
1596 } else {
1597 None
1598 };
1599
1600 driver.management_config = config.management.clone();
1602
1603 driver.transport_identity = Some(transport_identity);
1605
1606 #[cfg(feature = "hooks")]
1608 {
1609 for hook_cfg in &config.hooks {
1610 if !hook_cfg.enabled {
1611 continue;
1612 }
1613 let point_idx = match config::parse_hook_point(&hook_cfg.attach_point) {
1614 Some(idx) => idx,
1615 None => {
1616 log::warn!(
1617 "Unknown hook point '{}' for hook '{}'",
1618 hook_cfg.attach_point,
1619 hook_cfg.name,
1620 );
1621 continue;
1622 }
1623 };
1624 let mgr = match driver.hook_manager.as_ref() {
1625 Some(m) => m,
1626 None => {
1627 log::warn!(
1628 "Hook manager not available, skipping hook '{}'",
1629 hook_cfg.name
1630 );
1631 continue;
1632 }
1633 };
1634 let hook_backend = match config::parse_hook_backend(&hook_cfg.hook_type) {
1635 Ok(backend) => backend,
1636 Err(e) => {
1637 log::warn!(
1638 "Invalid hook type '{}' for hook '{}': {}",
1639 hook_cfg.hook_type,
1640 hook_cfg.name,
1641 e,
1642 );
1643 continue;
1644 }
1645 };
1646 let load_result = if hook_backend == rns_hooks::HookBackend::Builtin {
1647 let builtin_id = hook_cfg
1648 .builtin_id
1649 .as_deref()
1650 .filter(|id| !id.is_empty())
1651 .or_else(|| (!hook_cfg.path.is_empty()).then_some(hook_cfg.path.as_str()));
1652 match builtin_id {
1653 Some(id) => mgr.load_builtin(hook_cfg.name.clone(), id, hook_cfg.priority),
1654 None => Err(rns_hooks::HookError::CompileError(
1655 "built-in hook requires builtin/id or path".to_string(),
1656 )),
1657 }
1658 } else {
1659 mgr.load_file_backend(
1660 hook_cfg.name.clone(),
1661 std::path::Path::new(&hook_cfg.path),
1662 hook_cfg.priority,
1663 hook_backend,
1664 )
1665 };
1666 match load_result {
1667 Ok(program) => {
1668 driver.hook_slots[point_idx].attach(program);
1669 log::info!(
1670 "Loaded hook '{}' at point {} (priority {})",
1671 hook_cfg.name,
1672 hook_cfg.attach_point,
1673 hook_cfg.priority,
1674 );
1675 }
1676 Err(e) => {
1677 log::error!(
1678 "Failed to load hook '{}' from '{}': {}",
1679 hook_cfg.name,
1680 hook_cfg.path,
1681 e,
1682 );
1683 }
1684 }
1685 }
1686 }
1687
1688 driver.discover_interfaces = config.discover_interfaces;
1690 driver.autoconnect_interface_mode = config.autoconnect_interface_mode;
1691 driver.autoconnect_interface_gravity = config.autoconnect_interface_gravity;
1692 driver.autoconnect_announces_to_internal =
1693 config.autoconnect_announces_to_internal.then_some(true);
1694 driver.autoconnect_unverified_implementations =
1695 config.autoconnect_unverified_implementations;
1696 if let Some(val) = config.discovery_required_value {
1697 driver.discovery_required_value = val;
1698 }
1699
1700 let next_dynamic_id = Arc::new(AtomicU64::new(10000));
1702 driver.next_dynamic_interface_id = Arc::clone(&next_dynamic_id);
1703
1704 let mut discoverable_interfaces = Vec::new();
1706 #[cfg(feature = "iface-backbone")]
1707 let mut backbone_peer_pool_candidates = Vec::new();
1708
1709 let initial_managed_ids = interface_management
1710 .as_ref()
1711 .map(|management| management.initial_ids.clone())
1712 .unwrap_or_default();
1713 driver.interface_management = interface_management;
1714
1715 let registry = config
1717 .registry
1718 .unwrap_or_else(crate::interface::registry::InterfaceRegistry::with_builtins);
1719 for iface_config in config.interfaces {
1720 let managed_id = initial_managed_ids.get(&iface_config.name).copied();
1721 #[cfg(feature = "iface-backbone")]
1722 if iface_config.type_name == "BackboneInterface" {
1723 if let Some(mode) = iface_config
1724 .config_data
1725 .as_any()
1726 .downcast_ref::<BackboneMode>()
1727 {
1728 if let Some(handle) = runtime_handle_from_mode(mode) {
1729 driver.register_backbone_runtime(handle);
1730 }
1731 if let Some(handle) = peer_state_handle_from_mode(mode) {
1732 driver.register_backbone_peer_state(handle);
1733 }
1734 if let Some(handle) = client_runtime_handle_from_mode(mode) {
1735 driver.register_backbone_client_runtime(handle);
1736 }
1737 if let Some(handle) = backbone_discovery_runtime_from_interface(
1738 &iface_config.name,
1739 mode,
1740 iface_config.discovery.as_ref(),
1741 config.transport_enabled,
1742 iface_config.ifac.as_ref(),
1743 ) {
1744 driver.register_backbone_discovery_runtime(handle);
1745 }
1746 }
1747 }
1748 #[cfg(feature = "iface-tcp")]
1749 if iface_config.type_name == "TCPClientInterface" {
1750 if let Some(tcp_config) = iface_config
1751 .config_data
1752 .as_any()
1753 .downcast_ref::<TcpClientConfig>()
1754 {
1755 driver.register_tcp_client_runtime(tcp_client_runtime_handle_from_config(
1756 tcp_config,
1757 ));
1758 }
1759 }
1760 #[cfg(feature = "iface-tcp")]
1761 if iface_config.type_name == "TCPServerInterface" {
1762 if let Some(tcp_config) = iface_config
1763 .config_data
1764 .as_any()
1765 .downcast_ref::<TcpServerConfig>()
1766 {
1767 driver.register_tcp_server_runtime(tcp_runtime_handle_from_config(tcp_config));
1768 driver.register_tcp_server_discovery_runtime(
1769 tcp_server_discovery_runtime_from_interface(
1770 &iface_config.name,
1771 tcp_config,
1772 iface_config.discovery.as_ref(),
1773 config.transport_enabled,
1774 iface_config.ifac.as_ref(),
1775 ),
1776 );
1777 }
1778 }
1779 #[cfg(feature = "iface-udp")]
1780 if iface_config.type_name == "UDPInterface" {
1781 if let Some(udp_config) = iface_config
1782 .config_data
1783 .as_any()
1784 .downcast_ref::<UdpConfig>()
1785 {
1786 driver.register_udp_runtime(udp_runtime_handle_from_config(udp_config));
1787 }
1788 }
1789 #[cfg(feature = "iface-auto")]
1790 if iface_config.type_name == "AutoInterface" {
1791 if let Some(auto_config) = iface_config
1792 .config_data
1793 .as_any()
1794 .downcast_ref::<AutoConfig>()
1795 {
1796 driver.register_auto_runtime(auto_runtime_handle_from_config(auto_config));
1797 }
1798 }
1799 #[cfg(feature = "iface-i2p")]
1800 if iface_config.type_name == "I2PInterface" {
1801 if let Some(i2p_config) = iface_config
1802 .config_data
1803 .as_any()
1804 .downcast_ref::<I2pConfig>()
1805 {
1806 driver.register_i2p_runtime(i2p_runtime_handle_from_config(i2p_config));
1807 }
1808 }
1809 #[cfg(feature = "iface-pipe")]
1810 if iface_config.type_name == "PipeInterface" {
1811 if let Some(pipe_config) = iface_config
1812 .config_data
1813 .as_any()
1814 .downcast_ref::<PipeConfig>()
1815 {
1816 driver.register_pipe_runtime(pipe_runtime_handle_from_config(pipe_config));
1817 }
1818 }
1819 #[cfg(feature = "iface-rnode")]
1820 if iface_config.type_name == "RNodeInterface" {
1821 if let Some(rnode_config) = iface_config
1822 .config_data
1823 .as_any()
1824 .downcast_ref::<RNodeConfig>()
1825 {
1826 driver.register_rnode_runtime(rnode_runtime_handle_from_config(rnode_config));
1827 }
1828 }
1829
1830 let factory = match registry.get(&iface_config.type_name) {
1831 Some(f) => f,
1832 None => {
1833 log::warn!(
1834 "No factory registered for interface type '{}'",
1835 iface_config.type_name
1836 );
1837 continue;
1838 }
1839 };
1840
1841 let mut ifac_state = derive_ifac_state(iface_config.ifac.as_ref(), &iface_config.name)?;
1842 let ifac_runtime =
1843 ifac_runtime_from_config(iface_config.ifac.as_ref(), factory.default_ifac_size());
1844
1845 #[cfg(feature = "iface-backbone")]
1846 if config.backbone_peer_pool.is_some() && iface_config.type_name == "BackboneInterface"
1847 {
1848 if let Some(mode) = iface_config
1849 .config_data
1850 .as_any()
1851 .downcast_ref::<BackboneMode>()
1852 {
1853 if let Some(client) = client_config_from_mode(mode) {
1854 let priority = client_priority_from_mode(mode).unwrap_or(
1855 crate::driver::BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY,
1856 );
1857 backbone_peer_pool_candidates.push(BackbonePeerPoolCandidateConfig {
1858 client,
1859 mode: iface_config.mode,
1860 gravity: iface_config.gravity,
1861 recursive_prs: iface_config.recursive_prs,
1862 announces_from_internal: iface_config.announces_from_internal,
1863 announces_to_internal: iface_config.announces_to_internal,
1864 ingress_control: iface_config.ingress_control,
1865 ifac_runtime: ifac_runtime.clone(),
1866 ifac_enabled: ifac_state.is_some(),
1867 interface_type_name: iface_config.type_name.clone(),
1868 source: crate::driver::BackbonePeerPoolCandidateSource::Configured,
1869 priority,
1870 discovery: None,
1871 });
1872 if let Some(ref disc) = iface_config.discovery {
1873 discoverable_interfaces.push(discoverable_interface_from_config(
1874 &iface_config.name,
1875 disc,
1876 config.transport_enabled,
1877 iface_config.ifac.as_ref(),
1878 ));
1879 }
1880 continue;
1881 }
1882 }
1883 }
1884
1885 let ctx = crate::interface::StartContext {
1886 tx: tx.clone(),
1887 next_dynamic_id: next_dynamic_id.clone(),
1888 mode: iface_config.mode,
1889 gravity: iface_config.gravity,
1890 recursive_prs: iface_config.recursive_prs,
1891 announces_from_internal: iface_config.announces_from_internal,
1892 announces_to_internal: iface_config.announces_to_internal,
1893 ingress_control: iface_config.ingress_control,
1894 ifac: ifac_state.clone(),
1895 underlay_mark: config.underlay_mark,
1896 };
1897
1898 let result = match factory.start(iface_config.config_data, ctx) {
1899 Ok(r) => r,
1900 Err(e) => {
1901 if config.panic_on_interface_error {
1902 return Err(e);
1903 }
1904 log::error!(
1905 "Interface '{}' ({}) failed to start: {}",
1906 iface_config.name,
1907 iface_config.type_name,
1908 e
1909 );
1910 continue;
1911 }
1912 };
1913
1914 if let Some(ref disc) = iface_config.discovery {
1915 discoverable_interfaces.push(discoverable_interface_from_config(
1916 &iface_config.name,
1917 disc,
1918 config.transport_enabled,
1919 iface_config.ifac.as_ref(),
1920 ));
1921 }
1922
1923 match result {
1924 crate::interface::StartResult::Simple {
1925 id,
1926 info,
1927 writer,
1928 interface_type_name,
1929 control,
1930 } => {
1931 let managed_control = control.clone();
1932 if let Some(control) = control {
1933 driver.register_listener_control(control);
1934 }
1935 if let Some(parent_id) = managed_id {
1936 driver.managed_interfaces.insert(
1937 iface_config.name.clone(),
1938 crate::driver::ManagedInterface {
1939 parent_id,
1940 type_name: iface_config.type_name.clone(),
1941 control: managed_control,
1942 static_ids: vec![id],
1943 },
1944 );
1945 }
1946 register_started_interface(StartedInterface {
1947 driver: &mut driver,
1948 tx: &tx,
1949 queue_capacity: config.interface_writer_queue_capacity,
1950 id,
1951 info,
1952 writer,
1953 interface_type_name,
1954 ifac_state,
1955 ifac_runtime: &ifac_runtime,
1956 });
1957 }
1958 crate::interface::StartResult::Listener { control } => {
1959 let managed_control = control.clone();
1962 if let Some(control) = control {
1963 driver.register_listener_control(control);
1964 }
1965 if let Some(parent_id) = managed_id {
1966 driver.managed_interfaces.insert(
1967 iface_config.name.clone(),
1968 crate::driver::ManagedInterface {
1969 parent_id,
1970 type_name: iface_config.type_name.clone(),
1971 control: managed_control,
1972 static_ids: Vec::new(),
1973 },
1974 );
1975 }
1976 }
1977 crate::interface::StartResult::Multi {
1978 subinterfaces: subs,
1979 control,
1980 } => {
1981 let managed_control = control.clone();
1982 if let Some(control) = control {
1983 driver.register_listener_control(control);
1984 }
1985 let static_ids = subs.iter().map(|sub| sub.id).collect();
1986 if let Some(parent_id) = managed_id {
1987 driver.managed_interfaces.insert(
1988 iface_config.name.clone(),
1989 crate::driver::ManagedInterface {
1990 parent_id,
1991 type_name: iface_config.type_name.clone(),
1992 control: managed_control,
1993 static_ids,
1994 },
1995 );
1996 }
1997 let ifac_cfg = &iface_config.ifac;
1998 let mut first = true;
1999 for sub in subs {
2000 let sub_ifac = if first {
2001 first = false;
2002 ifac_state.take()
2003 } else {
2004 derive_ifac_state(ifac_cfg.as_ref(), &sub.info.name)?
2005 };
2006 register_started_interface(StartedInterface {
2007 driver: &mut driver,
2008 tx: &tx,
2009 queue_capacity: config.interface_writer_queue_capacity,
2010 id: sub.id,
2011 info: sub.info,
2012 writer: sub.writer,
2013 interface_type_name: sub.interface_type_name,
2014 ifac_state: sub_ifac,
2015 ifac_runtime: &ifac_runtime,
2016 });
2017 }
2018 }
2019 }
2020 }
2021
2022 if !discoverable_interfaces.is_empty() {
2024 let transport_id = *identity.hash();
2025 let announcer =
2026 crate::discovery::InterfaceAnnouncer::new(transport_id, discoverable_interfaces);
2027 log::info!("Interface discovery announcer initialized");
2028 driver.interface_announcer = Some(announcer);
2029 }
2030
2031 if let Some(ref cache_dir) = config.cache_dir {
2033 let disc_path = std::path::PathBuf::from(cache_dir)
2034 .parent()
2035 .unwrap_or(std::path::Path::new("."))
2036 .join("storage")
2037 .join("discovery")
2038 .join("interfaces");
2039 let _ = std::fs::create_dir_all(&disc_path);
2040 driver.discovered_interfaces =
2041 crate::discovery::DiscoveredInterfaceStorage::new(disc_path);
2042 }
2043
2044 if let Some(ref storage_dir) = transport_state_dir {
2047 match storage::load_transport_state(storage_dir) {
2048 Ok(snapshot) => {
2049 let announce_cache = driver.announce_cache.as_ref();
2050 let stats = driver.engine.restore_persistence_snapshot(
2051 snapshot,
2052 time::now(),
2053 |packet_hash| {
2054 announce_cache
2055 .and_then(|cache| cache.get(packet_hash).ok().flatten())
2056 .map(|(raw, _)| raw)
2057 },
2058 );
2059 log::debug!(
2060 "restored transport state: {} packet hashes, {} paths, {} tunnels ({} paths skipped)",
2061 stats.packet_hashes,
2062 stats.paths,
2063 stats.tunnels,
2064 stats.skipped_paths,
2065 );
2066 }
2067 Err(err) => log::warn!("failed to restore transport state: {}", err),
2068 }
2069 }
2070
2071 #[cfg(feature = "iface-backbone")]
2072 if let Some(settings) = config.backbone_peer_pool.clone() {
2073 driver.configure_backbone_peer_pool(settings, backbone_peer_pool_candidates);
2074 driver.seed_backbone_peer_pool_from_discovery_cache();
2075 }
2076
2077 if config.management.enable_remote_management {
2079 if let Some(prv_key) = identity.get_private_key() {
2080 let identity_hash = *identity.hash();
2081 let mgmt_dest = crate::management::management_dest_hash(&identity_hash);
2082
2083 let sig_prv = rns_crypto::ed25519::Ed25519PrivateKey::from_bytes(
2085 &prv_key[32..64].try_into().unwrap(),
2086 );
2087 let sig_pub_bytes: [u8; 32] = identity.get_public_key().unwrap()[32..64]
2088 .try_into()
2089 .unwrap();
2090
2091 driver
2093 .engine
2094 .register_destination(mgmt_dest, rns_core::constants::DESTINATION_SINGLE);
2095 driver
2096 .local_destinations
2097 .insert(mgmt_dest, rns_core::constants::DESTINATION_SINGLE);
2098
2099 driver.link_manager.register_link_destination(
2101 mgmt_dest,
2102 sig_prv,
2103 sig_pub_bytes,
2104 crate::link_manager::ResourceStrategy::AcceptNone,
2105 );
2106
2107 driver
2109 .link_manager
2110 .register_management_path(crate::management::status_path_hash());
2111 driver
2112 .link_manager
2113 .register_management_path(crate::management::path_path_hash());
2114
2115 log::info!("Remote management enabled on {:02x?}", &mgmt_dest[..4],);
2116
2117 if !config.management.remote_management_allowed.is_empty() {
2119 log::info!(
2120 "Remote management allowed for {} identities",
2121 config.management.remote_management_allowed.len(),
2122 );
2123 }
2124 }
2125 }
2126
2127 if config.management.publish_blackhole {
2128 if let Some(prv_key) = identity.get_private_key() {
2129 let identity_hash = *identity.hash();
2130 let bh_dest = crate::management::blackhole_dest_hash(&identity_hash);
2131
2132 let sig_prv = rns_crypto::ed25519::Ed25519PrivateKey::from_bytes(
2133 &prv_key[32..64].try_into().unwrap(),
2134 );
2135 let sig_pub_bytes: [u8; 32] = identity.get_public_key().unwrap()[32..64]
2136 .try_into()
2137 .unwrap();
2138
2139 driver
2140 .engine
2141 .register_destination(bh_dest, rns_core::constants::DESTINATION_SINGLE);
2142 driver.link_manager.register_link_destination(
2143 bh_dest,
2144 sig_prv,
2145 sig_pub_bytes,
2146 crate::link_manager::ResourceStrategy::AcceptNone,
2147 );
2148 driver
2149 .link_manager
2150 .register_management_path(crate::management::list_path_hash());
2151
2152 log::info!(
2153 "Blackhole list publishing enabled on {:02x?}",
2154 &bh_dest[..4],
2155 );
2156 }
2157 }
2158
2159 if config.respond_to_probes && config.transport_enabled {
2161 let identity_hash = *identity.hash();
2162 let probe_dest = crate::management::probe_dest_hash(&identity_hash);
2163
2164 driver
2166 .engine
2167 .register_destination(probe_dest, rns_core::constants::DESTINATION_SINGLE);
2168 driver
2169 .local_destinations
2170 .insert(probe_dest, rns_core::constants::DESTINATION_SINGLE);
2171
2172 let probe_identity = rns_crypto::identity::Identity::from_private_key(
2174 &identity.get_private_key().unwrap(),
2175 );
2176 driver.proof_strategies.insert(
2177 probe_dest,
2178 (
2179 rns_core::types::ProofStrategy::ProveAll,
2180 Some(probe_identity),
2181 ),
2182 );
2183
2184 driver.probe_responder_hash = Some(probe_dest);
2185
2186 log::info!("Probe responder enabled on {:02x?}", &probe_dest[..4],);
2187 }
2188
2189 let timer_tx = tx.clone();
2191 let timer_interval = Arc::clone(&tick_interval_ms);
2192 thread::Builder::new()
2193 .name("rns-timer".into())
2194 .spawn(move || {
2195 loop {
2196 let ms = timer_interval.load(Ordering::Relaxed);
2197 thread::sleep(Duration::from_millis(ms));
2198 if timer_tx.send(Event::Tick).is_err() {
2199 break; }
2201 }
2202 })?;
2203
2204 #[cfg(feature = "iface-local")]
2206 if config.share_instance {
2207 let local_server_config = LocalServerConfig {
2208 instance_name: config.instance_name.clone(),
2209 port: config.shared_instance_port,
2210 interface_id: rns_core::transport::types::InterfaceId(0), };
2212 match crate::interface::local::start_server(
2213 local_server_config,
2214 tx.clone(),
2215 next_dynamic_id.clone(),
2216 ) {
2217 Ok(control) => {
2218 driver.register_listener_control(control);
2219 log::info!(
2220 "Local shared instance server started (instance={}, port={})",
2221 config.instance_name,
2222 config.shared_instance_port
2223 );
2224 }
2225 Err(e) => {
2226 log::error!("Failed to start local shared instance server: {}", e);
2227 }
2228 }
2229 }
2230
2231 let rpc_server = if config.share_instance {
2233 let rpc_addr = crate::rpc::RpcAddr::Tcp("127.0.0.1".into(), config.rpc_port);
2234 match crate::rpc::RpcServer::start(&rpc_addr, rpc_auth_key, tx.clone()) {
2235 Ok(server) => {
2236 log::info!("RPC server started on 127.0.0.1:{}", config.rpc_port);
2237 Some(server)
2238 }
2239 Err(e) => {
2240 log::error!("Failed to start RPC server: {}", e);
2241 None
2242 }
2243 }
2244 } else {
2245 None
2246 };
2247 #[cfg(target_os = "linux")]
2248 let rpc_unix_server = if config.share_instance {
2249 match crate::rpc::RpcServer::start_abstract_unix(
2250 &config.instance_name,
2251 rpc_auth_key,
2252 tx.clone(),
2253 ) {
2254 Ok(server) => {
2255 log::info!(
2256 "RPC server started on abstract Unix socket for {}",
2257 config.instance_name
2258 );
2259 Some(server)
2260 }
2261 Err(e) => {
2262 log::error!("Failed to start abstract Unix RPC server: {e}");
2263 None
2264 }
2265 }
2266 } else {
2267 None
2268 };
2269
2270 let announce_verify_queue = Arc::clone(&driver.announce_verify_queue);
2271 let verify_shutdown = Arc::new(AtomicBool::new(false));
2272 let verify_shutdown_thread = Arc::clone(&verify_shutdown);
2273 let verify_tx = tx.clone();
2274 let verify_handle = thread::Builder::new()
2275 .name("rns-verify".into())
2276 .spawn(move || {
2277 #[cfg(target_family = "unix")]
2278 {
2279 unsafe {
2280 libc::nice(5);
2281 }
2282 }
2283
2284 while !verify_shutdown_thread.load(Ordering::Relaxed) {
2285 let batch = {
2286 let mut queue = announce_verify_queue
2287 .lock()
2288 .unwrap_or_else(|poisoned| poisoned.into_inner());
2289 queue.take_pending(time::now())
2290 };
2291
2292 if batch.is_empty() {
2293 thread::sleep(Duration::from_millis(50));
2294 continue;
2295 }
2296
2297 for (key, pending) in batch {
2298 if verify_shutdown_thread.load(Ordering::Relaxed) {
2299 break;
2300 }
2301 let has_ratchet =
2302 pending.packet.flags.context_flag == rns_core::constants::FLAG_SET;
2303 let announce = match rns_core::announce::AnnounceData::unpack(
2304 &pending.packet.data,
2305 has_ratchet,
2306 ) {
2307 Ok(announce) => announce,
2308 Err(_) => {
2309 let signature = [0u8; 64];
2310 let sig_cache_key = {
2311 let mut material = [0u8; 80];
2312 material[..16]
2313 .copy_from_slice(&pending.packet.destination_hash);
2314 material[16..].copy_from_slice(&signature);
2315 rns_core::hash::full_hash(&material)
2316 };
2317 if verify_tx
2318 .send(Event::AnnounceVerifyFailed { key, sig_cache_key })
2319 .is_err()
2320 {
2321 return;
2322 }
2323 continue;
2324 }
2325 };
2326 let mut material = [0u8; 80];
2327 material[..16].copy_from_slice(&pending.packet.destination_hash);
2328 material[16..].copy_from_slice(&announce.signature);
2329 let sig_cache_key = rns_core::hash::full_hash(&material);
2330 match announce.validate(&pending.packet.destination_hash) {
2331 Ok(validated) => {
2332 if verify_tx
2333 .send(Event::AnnounceVerified {
2334 key,
2335 validated,
2336 sig_cache_key,
2337 })
2338 .is_err()
2339 {
2340 return;
2341 }
2342 }
2343 Err(_) => {
2344 if verify_tx
2345 .send(Event::AnnounceVerifyFailed { key, sig_cache_key })
2346 .is_err()
2347 {
2348 return;
2349 }
2350 }
2351 }
2352 }
2353 }
2354 })?;
2355
2356 let driver_handle = thread::Builder::new()
2360 .name("rns-driver".into())
2361 .spawn(move || {
2362 driver.run();
2363 })?;
2364
2365 let (persist_handle, persist_stop) = if let Some(storage_dir) = transport_state_dir.clone()
2369 {
2370 let persist_tx = tx.clone();
2371 let (stop_tx, stop_rx) = std::sync::mpsc::channel();
2372 let handle = thread::Builder::new()
2373 .name("rns-persist".into())
2374 .spawn(move || {
2375 #[cfg(target_family = "unix")]
2376 unsafe {
2377 libc::nice(5);
2378 }
2379
2380 loop {
2381 match stop_rx.recv_timeout(Duration::from_secs(12 * 60 * 60)) {
2382 Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
2383 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
2384 }
2385 let (response_tx, response_rx) = std::sync::mpsc::channel();
2386 if persist_tx
2387 .send(Event::Query(
2388 QueryRequest::TransportStateSnapshot,
2389 response_tx,
2390 ))
2391 .is_err()
2392 {
2393 break;
2394 }
2395 let Ok(QueryResponse::TransportStateSnapshot(snapshot)) =
2396 response_rx.recv()
2397 else {
2398 continue;
2399 };
2400 if let Err(err) =
2401 storage::save_transport_state(&snapshot, &storage_dir, true)
2402 {
2403 log::warn!("failed to persist transport state: {}", err);
2404 }
2405 }
2406 })?;
2407 (Some(handle), Some(stop_tx))
2408 } else {
2409 (None, None)
2410 };
2411
2412 Ok(RnsNode {
2413 tx,
2414 driver_handle: Some(driver_handle),
2415 verify_handle: Some(verify_handle),
2416 verify_shutdown,
2417 persist_handle,
2418 persist_stop,
2419 rpc_server,
2420 #[cfg(target_os = "linux")]
2421 rpc_unix_server,
2422 tick_interval_ms,
2423 probe_server,
2424 known_destinations_path: None,
2425 transport_state_dir,
2426 ratchet_store: config.ratchet_store,
2427 ratchet_expiry: config.ratchet_expiry,
2428 })
2429 }
2430
2431 pub fn query(&self, request: QueryRequest) -> Result<QueryResponse, SendError> {
2433 let (resp_tx, resp_rx) = std::sync::mpsc::channel();
2434 self.tx
2435 .send(Event::Query(request, resp_tx))
2436 .map_err(|_| SendError)?;
2437 resp_rx.recv().map_err(|_| SendError)
2438 }
2439
2440 pub fn attach_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
2442 self.manage_interface(crate::event::InterfaceManagementOperation::Attach, name)
2443 }
2444
2445 pub fn detach_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
2447 self.manage_interface(crate::event::InterfaceManagementOperation::Detach, name)
2448 }
2449
2450 pub fn reload_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
2452 self.manage_interface(crate::event::InterfaceManagementOperation::Reload, name)
2453 }
2454
2455 fn manage_interface(
2456 &self,
2457 operation: crate::event::InterfaceManagementOperation,
2458 name: &str,
2459 ) -> Result<Option<bool>, SendError> {
2460 match self.query(QueryRequest::ManageInterface {
2461 operation,
2462 name: name.to_string(),
2463 })? {
2464 QueryResponse::InterfaceManagement(result) => Ok(result),
2465 _ => Err(SendError),
2466 }
2467 }
2468
2469 pub fn links(&self) -> Result<Vec<crate::event::LinkInfoEntry>, SendError> {
2471 match self.query(QueryRequest::Links)? {
2472 QueryResponse::Links(links) => Ok(links),
2473 _ => Err(SendError),
2474 }
2475 }
2476
2477 pub fn query_link(
2479 &self,
2480 link_id: [u8; 16],
2481 ) -> Result<Option<crate::event::LinkInfoEntry>, SendError> {
2482 Ok(self
2483 .links()?
2484 .into_iter()
2485 .find(|link| link.link_id == link_id))
2486 }
2487
2488 pub fn lowest_interface_bitrate(&self) -> Result<Option<u64>, SendError> {
2493 match self.query(QueryRequest::LowestInterfaceBitrate)? {
2494 QueryResponse::LowestInterfaceBitrate(bitrate) => Ok(bitrate),
2495 _ => Err(SendError),
2496 }
2497 }
2498
2499 pub fn medium_path_timeout(&self) -> Result<f64, SendError> {
2505 match self.query(QueryRequest::MediumPathTimeout)? {
2506 QueryResponse::MediumPathTimeout(timeout) => Ok(timeout),
2507 _ => Err(SendError),
2508 }
2509 }
2510
2511 pub fn begin_drain(&self, timeout: Duration) -> Result<(), SendError> {
2513 self.tx
2514 .send(Event::BeginDrain { timeout })
2515 .map_err(|_| SendError)
2516 }
2517
2518 pub fn drain_status(&self) -> Result<crate::event::DrainStatus, SendError> {
2520 match self.query(QueryRequest::DrainStatus)? {
2521 QueryResponse::DrainStatus(status) => Ok(status),
2522 _ => Err(SendError),
2523 }
2524 }
2525
2526 fn reject_new_work_if_draining(&self) -> Result<(), SendError> {
2527 let status = self.drain_status()?;
2528 if matches!(status.state, crate::event::LifecycleState::Active) {
2529 Ok(())
2530 } else {
2531 Err(SendError)
2532 }
2533 }
2534
2535 pub fn send_raw(
2537 &self,
2538 raw: Vec<u8>,
2539 dest_type: u8,
2540 attached_interface: Option<rns_core::transport::types::InterfaceId>,
2541 ) -> Result<(), SendError> {
2542 self.tx
2543 .send(Event::SendOutbound {
2544 raw,
2545 dest_type,
2546 attached_interface,
2547 })
2548 .map_err(|_| SendError)
2549 }
2550
2551 pub fn register_destination(
2553 &self,
2554 dest_hash: [u8; 16],
2555 dest_type: u8,
2556 ) -> Result<(), SendError> {
2557 self.tx
2558 .send(Event::RegisterDestination {
2559 dest_hash,
2560 dest_type,
2561 })
2562 .map_err(|_| SendError)
2563 }
2564
2565 pub fn deregister_destination(&self, dest_hash: [u8; 16]) -> Result<(), SendError> {
2567 self.tx
2568 .send(Event::DeregisterDestination { dest_hash })
2569 .map_err(|_| SendError)
2570 }
2571
2572 pub fn deregister_link_destination(&self, dest_hash: [u8; 16]) -> Result<(), SendError> {
2574 self.tx
2575 .send(Event::DeregisterLinkDestination { dest_hash })
2576 .map_err(|_| SendError)
2577 }
2578
2579 pub fn register_link_destination(
2585 &self,
2586 dest_hash: [u8; 16],
2587 sig_prv_bytes: [u8; 32],
2588 sig_pub_bytes: [u8; 32],
2589 resource_strategy: u8,
2590 ) -> Result<(), SendError> {
2591 self.register_link_destination_with_max_request_size(
2592 dest_hash,
2593 sig_prv_bytes,
2594 sig_pub_bytes,
2595 resource_strategy,
2596 None,
2597 )
2598 }
2599
2600 pub fn register_link_destination_with_max_request_size(
2602 &self,
2603 dest_hash: [u8; 16],
2604 sig_prv_bytes: [u8; 32],
2605 sig_pub_bytes: [u8; 32],
2606 resource_strategy: u8,
2607 max_request_size: Option<usize>,
2608 ) -> Result<(), SendError> {
2609 self.tx
2610 .send(Event::RegisterLinkDestination {
2611 dest_hash,
2612 sig_prv_bytes,
2613 sig_pub_bytes,
2614 resource_strategy,
2615 max_request_size,
2616 })
2617 .map_err(|_| SendError)
2618 }
2619
2620 pub fn register_request_handler<F>(
2622 &self,
2623 path: &str,
2624 allowed_list: Option<Vec<[u8; 16]>>,
2625 handler: F,
2626 ) -> Result<(), SendError>
2627 where
2628 F: Fn([u8; 16], &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<Vec<u8>>
2629 + Send
2630 + 'static,
2631 {
2632 self.tx
2633 .send(Event::RegisterRequestHandler {
2634 path: path.to_string(),
2635 allowed_list,
2636 handler: Box::new(handler),
2637 })
2638 .map_err(|_| SendError)
2639 }
2640
2641 pub fn register_request_handler_response<F>(
2643 &self,
2644 path: &str,
2645 allowed_list: Option<Vec<[u8; 16]>>,
2646 handler: F,
2647 ) -> Result<(), SendError>
2648 where
2649 F: Fn(
2650 [u8; 16],
2651 &str,
2652 &[u8],
2653 Option<&([u8; 16], [u8; 64])>,
2654 ) -> Option<crate::link_manager::RequestResponse>
2655 + Send
2656 + 'static,
2657 {
2658 self.tx
2659 .send(Event::RegisterRequestHandlerResponse {
2660 path: path.to_string(),
2661 allowed_list,
2662 handler: Box::new(handler),
2663 })
2664 .map_err(|_| SendError)
2665 }
2666
2667 pub fn register_deferred_request_handler<F>(
2670 &self,
2671 path: &str,
2672 allowed_list: Option<Vec<[u8; 16]>>,
2673 handler: F,
2674 ) -> Result<(), SendError>
2675 where
2676 F: Fn([u8; 16], &str, [u8; 16], &[u8], Option<&([u8; 16], [u8; 64])>) + Send + 'static,
2677 {
2678 self.tx
2679 .send(Event::RegisterDeferredRequestHandler {
2680 path: path.to_string(),
2681 allowed_list,
2682 handler: Box::new(handler),
2683 })
2684 .map_err(|_| SendError)
2685 }
2686
2687 pub fn send_deferred_response(
2689 &self,
2690 link_id: [u8; 16],
2691 request_id: [u8; 16],
2692 data: Vec<u8>,
2693 ) -> Result<(), SendError> {
2694 self.tx
2695 .send(Event::SendDeferredResponse {
2696 link_id,
2697 request_id,
2698 data,
2699 })
2700 .map_err(|_| SendError)
2701 }
2702
2703 pub fn create_link(
2707 &self,
2708 dest_hash: [u8; 16],
2709 dest_sig_pub_bytes: [u8; 32],
2710 ) -> Result<[u8; 16], SendError> {
2711 self.reject_new_work_if_draining()?;
2712 let (response_tx, response_rx) = std::sync::mpsc::channel();
2713 self.tx
2714 .send(Event::CreateLink {
2715 dest_hash,
2716 dest_sig_pub_bytes,
2717 response_tx,
2718 })
2719 .map_err(|_| SendError)?;
2720 let link_id = response_rx.recv().map_err(|_| SendError)?;
2721 if link_id == [0u8; 16] {
2722 Err(SendError)
2723 } else {
2724 Ok(link_id)
2725 }
2726 }
2727
2728 pub fn send_request(
2730 &self,
2731 link_id: [u8; 16],
2732 path: &str,
2733 data: &[u8],
2734 ) -> Result<(), SendError> {
2735 self.send_request_with_max_response_size(link_id, path, data, None)
2736 }
2737
2738 pub fn send_request_with_max_response_size(
2740 &self,
2741 link_id: [u8; 16],
2742 path: &str,
2743 data: &[u8],
2744 max_response_size: Option<usize>,
2745 ) -> Result<(), SendError> {
2746 self.reject_new_work_if_draining()?;
2747 self.tx
2748 .send(Event::SendRequest {
2749 link_id,
2750 path: path.to_string(),
2751 data: data.to_vec(),
2752 max_response_size,
2753 })
2754 .map_err(|_| SendError)
2755 }
2756
2757 pub fn identify_on_link(
2759 &self,
2760 link_id: [u8; 16],
2761 identity_prv_key: [u8; 64],
2762 ) -> Result<(), SendError> {
2763 self.reject_new_work_if_draining()?;
2764 self.tx
2765 .send(Event::IdentifyOnLink {
2766 link_id,
2767 identity_prv_key,
2768 })
2769 .map_err(|_| SendError)
2770 }
2771
2772 pub fn teardown_link(&self, link_id: [u8; 16]) -> Result<(), SendError> {
2774 self.tx
2775 .send(Event::TeardownLink { link_id })
2776 .map_err(|_| SendError)
2777 }
2778
2779 pub fn send_resource(
2781 &self,
2782 link_id: [u8; 16],
2783 data: Vec<u8>,
2784 metadata: Option<Vec<u8>>,
2785 ) -> Result<(), SendError> {
2786 self.send_resource_with_auto_compress(link_id, data, metadata, true)
2787 }
2788
2789 pub fn send_resource_with_auto_compress(
2791 &self,
2792 link_id: [u8; 16],
2793 data: Vec<u8>,
2794 metadata: Option<Vec<u8>>,
2795 auto_compress: bool,
2796 ) -> Result<(), SendError> {
2797 self.reject_new_work_if_draining()?;
2798 self.tx
2799 .send(Event::SendResource {
2800 link_id,
2801 data,
2802 metadata,
2803 auto_compress,
2804 })
2805 .map_err(|_| SendError)
2806 }
2807
2808 pub fn send_resource_reader<R: Read + Send + 'static>(
2811 &self,
2812 link_id: [u8; 16],
2813 reader: R,
2814 declared_length: u64,
2815 metadata: Option<Vec<u8>>,
2816 auto_compress: bool,
2817 ) -> Result<crate::resource::ResourceTransferId, SendError> {
2818 self.reject_new_work_if_draining()?;
2819 let transfer_id = crate::resource::ResourceTransferId(
2820 NEXT_RESOURCE_TRANSFER_ID.fetch_add(1, Ordering::Relaxed),
2821 );
2822 self.tx
2823 .send(Event::SendResourceStream {
2824 link_id,
2825 transfer_id,
2826 reader: Box::new(reader),
2827 declared_length,
2828 metadata,
2829 auto_compress,
2830 })
2831 .map_err(|_| SendError)?;
2832 Ok(transfer_id)
2833 }
2834
2835 pub fn send_resource_file(
2837 &self,
2838 link_id: [u8; 16],
2839 path: impl AsRef<Path>,
2840 metadata: Option<Vec<u8>>,
2841 auto_compress: bool,
2842 ) -> io::Result<crate::resource::ResourceTransferId> {
2843 let file = std::fs::File::open(path.as_ref())?;
2844 let length = file.metadata()?.len();
2845 self.send_resource_reader(link_id, file, length, metadata, auto_compress)
2846 .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "RNS driver shut down"))
2847 }
2848
2849 pub fn set_resource_receive_mode(
2851 &self,
2852 link_id: [u8; 16],
2853 mode: crate::resource::ResourceReceiveMode,
2854 ) -> Result<(), SendError> {
2855 self.tx
2856 .send(Event::SetResourceReceiveMode { link_id, mode })
2857 .map_err(|_| SendError)
2858 }
2859
2860 pub fn set_resource_strategy(&self, link_id: [u8; 16], strategy: u8) -> Result<(), SendError> {
2864 self.tx
2865 .send(Event::SetResourceStrategy { link_id, strategy })
2866 .map_err(|_| SendError)
2867 }
2868
2869 pub fn accept_resource(
2871 &self,
2872 link_id: [u8; 16],
2873 resource_hash: Vec<u8>,
2874 accept: bool,
2875 ) -> Result<(), SendError> {
2876 if accept {
2877 self.reject_new_work_if_draining()?;
2878 }
2879 self.tx
2880 .send(Event::AcceptResource {
2881 link_id,
2882 resource_hash,
2883 accept,
2884 })
2885 .map_err(|_| SendError)
2886 }
2887
2888 pub fn send_channel_message(
2890 &self,
2891 link_id: [u8; 16],
2892 msgtype: u16,
2893 payload: Vec<u8>,
2894 ) -> Result<(), SendError> {
2895 self.try_send_channel_message(link_id, msgtype, payload)
2896 .map_err(|_| SendError)
2897 }
2898
2899 pub fn try_send_channel_message(
2903 &self,
2904 link_id: [u8; 16],
2905 msgtype: u16,
2906 payload: Vec<u8>,
2907 ) -> Result<(), ChannelSendError> {
2908 self.reject_new_work_if_draining()
2909 .map_err(|_| ChannelSendError::Unavailable)?;
2910 let (response_tx, response_rx) = std::sync::mpsc::channel();
2911 self.tx
2912 .send(Event::SendChannelMessage {
2913 link_id,
2914 msgtype,
2915 payload,
2916 response_tx,
2917 })
2918 .map_err(|_| ChannelSendError::Unavailable)?;
2919 response_rx
2920 .recv()
2921 .map_err(|_| ChannelSendError::Unavailable)?
2922 .map_err(ChannelSendError::from)
2923 }
2924
2925 pub fn propose_direct_connect(&self, link_id: [u8; 16]) -> Result<(), SendError> {
2930 self.reject_new_work_if_draining()?;
2931 self.tx
2932 .send(Event::ProposeDirectConnect { link_id })
2933 .map_err(|_| SendError)
2934 }
2935
2936 pub fn set_direct_connect_policy(
2938 &self,
2939 policy: crate::holepunch::orchestrator::HolePunchPolicy,
2940 ) -> Result<(), SendError> {
2941 self.tx
2942 .send(Event::SetDirectConnectPolicy { policy })
2943 .map_err(|_| SendError)
2944 }
2945
2946 pub async fn send_on_link(
2958 &self,
2959 link_id: [u8; 16],
2960 data: Vec<u8>,
2961 context: u8,
2962 ) -> Result<(), crate::link_send::LinkSendError> {
2963 use crate::link_send::Completion;
2964 let tracking = self.tx.link_send_pool().track(link_id, true);
2965 let permit = self.tx.link_send_pool().acquire().await?;
2966 let (completion, receipt) = Completion::new_tracked(permit, tracking);
2967 self.tx
2968 .send_async(Event::SendLinkTracked {
2969 link_id,
2970 data,
2971 context,
2972 completion,
2973 })
2974 .await?;
2975 receipt.await
2976 }
2977
2978 pub fn try_send_on_link(
2984 &self,
2985 link_id: [u8; 16],
2986 data: Vec<u8>,
2987 context: u8,
2988 ) -> Result<crate::link_send::LinkSendReceipt, crate::link_send::LinkSendError> {
2989 use crate::link_send::{Completion, LinkSendError};
2990 let permit = self.tx.link_send_pool().try_acquire()?;
2991 let tracking = self.tx.link_send_pool().track(link_id, false);
2992 let (completion, receipt) = Completion::new_tracked(permit, tracking);
2993 self.tx
2994 .try_send(Event::SendLinkTracked {
2995 link_id,
2996 data,
2997 context,
2998 completion,
2999 })
3000 .map_err(|error| match error {
3001 std::sync::mpsc::TrySendError::Full(_) => LinkSendError::QueueFull,
3002 std::sync::mpsc::TrySendError::Disconnected(_) => LinkSendError::DriverStopped,
3003 })?;
3004 Ok(receipt)
3005 }
3006
3007 pub fn try_send_link_datagram(
3012 &self,
3013 link_id: [u8; 16],
3014 data: Vec<u8>,
3015 context: u8,
3016 timeout: Duration,
3017 ) -> Result<(), crate::event::LinkDatagramError> {
3018 use crate::event::LinkDatagramError;
3019 if self.reject_new_work_if_draining().is_err() {
3020 return Err(LinkDatagramError::DriverStopped);
3021 }
3022 let (response_tx, response_rx) = std::sync::mpsc::channel();
3023 self.tx
3024 .try_send(Event::SendOnLink {
3025 link_id,
3026 data,
3027 context,
3028 response_tx: Some(response_tx),
3029 })
3030 .map_err(|error| match error {
3031 std::sync::mpsc::TrySendError::Full(_) => LinkDatagramError::DriverQueueFull,
3032 std::sync::mpsc::TrySendError::Disconnected(_) => LinkDatagramError::DriverStopped,
3033 })?;
3034 response_rx
3035 .recv_timeout(timeout)
3036 .map_err(|error| match error {
3037 std::sync::mpsc::RecvTimeoutError::Timeout => LinkDatagramError::AdmissionTimedOut,
3038 std::sync::mpsc::RecvTimeoutError::Disconnected => LinkDatagramError::DriverStopped,
3039 })?
3040 }
3041
3042 pub async fn announce(
3051 &self,
3052 dest: &crate::destination::Destination,
3053 identity: &Identity,
3054 app_data: Option<&[u8]>,
3055 ) -> Result<(), crate::TransmissionError> {
3056 let permit = self.tx.link_send_pool().acquire().await?;
3057 let (packet, replay) = self
3058 .build_announce(dest, identity, app_data)
3059 .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
3060 let (completion, receipt) = crate::link_send::Completion::new(permit);
3061 self.tx
3062 .send_async(Event::SendOutboundTracked {
3063 raw: packet.raw,
3064 dest_type: dest.dest_type.to_wire_constant(),
3065 replay,
3066 completion,
3067 })
3068 .await?;
3069 receipt.await
3070 }
3071
3072 pub fn try_announce(
3075 &self,
3076 dest: &crate::destination::Destination,
3077 identity: &Identity,
3078 app_data: Option<&[u8]>,
3079 ) -> Result<crate::TransmissionReceipt, crate::TransmissionError> {
3080 let permit = self.tx.link_send_pool().try_acquire()?;
3081 let (packet, replay) = self
3082 .build_announce(dest, identity, app_data)
3083 .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
3084 let (completion, receipt) = crate::link_send::Completion::new(permit);
3085 self.try_confirmed_event(Event::SendOutboundTracked {
3086 raw: packet.raw,
3087 dest_type: dest.dest_type.to_wire_constant(),
3088 replay,
3089 completion,
3090 })?;
3091 Ok(receipt)
3092 }
3093
3094 pub async fn send_packet(
3101 &self,
3102 dest: &crate::destination::Destination,
3103 data: &[u8],
3104 ) -> Result<rns_core::types::PacketHash, crate::TransmissionError> {
3105 let permit = self.tx.link_send_pool().acquire().await?;
3106 let packet = self
3107 .build_packet(dest, data)
3108 .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
3109 let hash = rns_core::types::PacketHash(packet.packet_hash);
3110 let (completion, receipt) = crate::link_send::Completion::new(permit);
3111 self.tx
3112 .send_async(Event::SendOutboundTracked {
3113 raw: packet.raw,
3114 dest_type: dest.dest_type.to_wire_constant(),
3115 replay: None,
3116 completion,
3117 })
3118 .await?;
3119 receipt.await?;
3120 Ok(hash)
3121 }
3122
3123 pub fn try_send_packet(
3127 &self,
3128 dest: &crate::destination::Destination,
3129 data: &[u8],
3130 ) -> Result<crate::PacketSendReceipt, crate::TransmissionError> {
3131 let permit = self.tx.link_send_pool().try_acquire()?;
3132 let packet = self
3133 .build_packet(dest, data)
3134 .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
3135 let hash = rns_core::types::PacketHash(packet.packet_hash);
3136 let (completion, receipt) = crate::link_send::Completion::new(permit);
3137 self.try_confirmed_event(Event::SendOutboundTracked {
3138 raw: packet.raw,
3139 dest_type: dest.dest_type.to_wire_constant(),
3140 replay: None,
3141 completion,
3142 })?;
3143 Ok(crate::PacketSendReceipt {
3144 completion: receipt,
3145 hash,
3146 })
3147 }
3148
3149 fn try_confirmed_event(&self, event: Event) -> Result<(), crate::TransmissionError> {
3150 self.tx.try_send(event).map_err(|error| match error {
3151 std::sync::mpsc::TrySendError::Full(_) => crate::TransmissionError::QueueFull,
3152 std::sync::mpsc::TrySendError::Disconnected(_) => {
3153 crate::TransmissionError::DriverStopped
3154 }
3155 })
3156 }
3157
3158 pub fn announce_queued(
3166 &self,
3167 dest: &crate::destination::Destination,
3168 identity: &Identity,
3169 app_data: Option<&[u8]>,
3170 ) -> Result<(), SendError> {
3171 self.reject_new_work_if_draining()?;
3172 let (packet, replay) = self.build_announce(dest, identity, app_data)?;
3173 if let Some(replay) = replay {
3174 self.tx
3175 .send(Event::StoreSharedAnnounce {
3176 dest_hash: replay.dest_hash,
3177 name_hash: replay.name_hash,
3178 identity_prv_key: replay.identity_prv_key,
3179 app_data: replay.app_data,
3180 })
3181 .map_err(|_| SendError)?;
3182 }
3183 self.send_raw(packet.raw, dest.dest_type.to_wire_constant(), None)
3184 }
3185
3186 fn build_announce(
3187 &self,
3188 dest: &crate::destination::Destination,
3189 identity: &Identity,
3190 app_data: Option<&[u8]>,
3191 ) -> Result<
3192 (
3193 rns_core::packet::RawPacket,
3194 Option<Box<crate::common::event::AnnounceReplay>>,
3195 ),
3196 SendError,
3197 > {
3198 let name_hash = rns_core::destination::name_hash(
3199 &dest.app_name,
3200 &dest.aspects.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
3201 );
3202
3203 let mut random_hash = [0u8; 10];
3204 OsRng.fill_bytes(&mut random_hash[..5]);
3205 let now_secs = std::time::SystemTime::now()
3209 .duration_since(std::time::UNIX_EPOCH)
3210 .unwrap_or_default()
3211 .as_secs();
3212 random_hash[5..10].copy_from_slice(&now_secs.to_be_bytes()[3..8]);
3213
3214 let (announce_data, _has_ratchet) = rns_core::announce::AnnounceData::pack(
3215 identity,
3216 &dest.hash.0,
3217 &name_hash,
3218 &random_hash,
3219 None, app_data,
3221 )
3222 .map_err(|_| SendError)?;
3223
3224 let context_flag = rns_core::constants::FLAG_UNSET;
3225
3226 let flags = rns_core::packet::PacketFlags {
3227 header_type: rns_core::constants::HEADER_1,
3228 context_flag,
3229 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
3230 destination_type: rns_core::constants::DESTINATION_SINGLE,
3231 packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
3232 };
3233
3234 let packet = rns_core::packet::RawPacket::pack(
3235 flags,
3236 0,
3237 &dest.hash.0,
3238 None,
3239 rns_core::constants::CONTEXT_NONE,
3240 &announce_data,
3241 )
3242 .map_err(|_| SendError)?;
3243
3244 let replay = if dest.dest_type == rns_core::types::DestinationType::Single {
3245 identity.get_private_key().map(|identity_prv_key| {
3246 Box::new(crate::common::event::AnnounceReplay {
3247 dest_hash: dest.hash.0,
3248 name_hash,
3249 identity_prv_key,
3250 app_data: app_data.map(|d| d.to_vec()),
3251 })
3252 })
3253 } else {
3254 None
3255 };
3256 Ok((packet, replay))
3257 }
3258
3259 pub fn send_packet_queued(
3267 &self,
3268 dest: &crate::destination::Destination,
3269 data: &[u8],
3270 ) -> Result<rns_core::types::PacketHash, SendError> {
3271 self.reject_new_work_if_draining()?;
3272 let packet = self.build_packet(dest, data)?;
3273 let packet_hash = rns_core::types::PacketHash(packet.packet_hash);
3274 self.tx
3275 .send(Event::SendOutbound {
3276 raw: packet.raw,
3277 dest_type: dest.dest_type.to_wire_constant(),
3278 attached_interface: None,
3279 })
3280 .map_err(|_| SendError)?;
3281 Ok(packet_hash)
3282 }
3283
3284 fn build_packet(
3285 &self,
3286 dest: &crate::destination::Destination,
3287 data: &[u8],
3288 ) -> Result<rns_core::packet::RawPacket, SendError> {
3289 use rns_core::types::DestinationType;
3290
3291 let payload = match dest.dest_type {
3292 DestinationType::Single => self.encrypt_single_payload(dest, data)?,
3293 DestinationType::Plain => data.to_vec(),
3294 DestinationType::Group => dest.encrypt(data).map_err(|_| SendError)?,
3295 };
3296
3297 let flags = rns_core::packet::PacketFlags {
3298 header_type: rns_core::constants::HEADER_1,
3299 context_flag: rns_core::constants::FLAG_UNSET,
3300 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
3301 destination_type: dest.dest_type.to_wire_constant(),
3302 packet_type: rns_core::constants::PACKET_TYPE_DATA,
3303 };
3304
3305 let packet = rns_core::packet::RawPacket::pack(
3306 flags,
3307 0,
3308 &dest.hash.0,
3309 None,
3310 rns_core::constants::CONTEXT_NONE,
3311 &payload,
3312 )
3313 .map_err(|_| SendError)?;
3314
3315 Ok(packet)
3316 }
3317
3318 fn encrypt_single_payload(
3319 &self,
3320 dest: &crate::destination::Destination,
3321 data: &[u8],
3322 ) -> Result<Vec<u8>, SendError> {
3323 let pub_key = dest.public_key.ok_or(SendError)?;
3324 let remote_id = rns_crypto::identity::Identity::from_public_key(&pub_key);
3325 let ratchet = self.ratchet_store.as_ref().and_then(|store| {
3326 match store.current(&dest.hash.0, time::now(), self.ratchet_expiry.as_secs_f64()) {
3327 Ok(entry) => entry.map(|entry| entry.ratchet),
3328 Err(err) => {
3329 log::warn!(
3330 "failed to load ratchet for {:02x}{:02x}{:02x}{:02x}..: {}",
3331 dest.hash.0[0],
3332 dest.hash.0[1],
3333 dest.hash.0[2],
3334 dest.hash.0[3],
3335 err
3336 );
3337 None
3338 }
3339 }
3340 });
3341 remote_id
3342 .encrypt_with_ratchet(data, ratchet.as_ref(), &mut OsRng)
3343 .map_err(|_| SendError)
3344 }
3345
3346 pub fn register_destination_with_proof(
3351 &self,
3352 dest: &crate::destination::Destination,
3353 signing_key: Option<[u8; 64]>,
3354 ) -> Result<(), SendError> {
3355 self.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())?;
3357
3358 if dest.proof_strategy != rns_core::types::ProofStrategy::ProveNone {
3360 self.tx
3361 .send(Event::RegisterProofStrategy {
3362 dest_hash: dest.hash.0,
3363 strategy: dest.proof_strategy,
3364 signing_key,
3365 })
3366 .map_err(|_| SendError)?;
3367 }
3368
3369 Ok(())
3370 }
3371
3372 pub fn request_path(&self, dest_hash: &rns_core::types::DestHash) -> Result<(), SendError> {
3374 self.reject_new_work_if_draining()?;
3375 self.tx
3376 .send(Event::RequestPath {
3377 dest_hash: dest_hash.0,
3378 })
3379 .map_err(|_| SendError)
3380 }
3381
3382 pub fn has_path(&self, dest_hash: &rns_core::types::DestHash) -> Result<bool, SendError> {
3384 match self.query(QueryRequest::HasPath {
3385 dest_hash: dest_hash.0,
3386 })? {
3387 QueryResponse::HasPath(v) => Ok(v),
3388 _ => Ok(false),
3389 }
3390 }
3391
3392 pub fn hops_to(&self, dest_hash: &rns_core::types::DestHash) -> Result<Option<u8>, SendError> {
3394 match self.query(QueryRequest::HopsTo {
3395 dest_hash: dest_hash.0,
3396 })? {
3397 QueryResponse::HopsTo(v) => Ok(v),
3398 _ => Ok(None),
3399 }
3400 }
3401
3402 pub fn recall_identity(
3404 &self,
3405 dest_hash: &rns_core::types::DestHash,
3406 ) -> Result<Option<crate::destination::AnnouncedIdentity>, SendError> {
3407 match self.query(QueryRequest::RecallIdentity {
3408 dest_hash: dest_hash.0,
3409 })? {
3410 QueryResponse::RecallIdentity(v) => Ok(v),
3411 _ => Ok(None),
3412 }
3413 }
3414
3415 pub fn known_destinations(
3417 &self,
3418 ) -> Result<Vec<crate::event::KnownDestinationEntry>, SendError> {
3419 match self.query(QueryRequest::KnownDestinations)? {
3420 QueryResponse::KnownDestinations(entries) => Ok(entries),
3421 _ => Ok(Vec::new()),
3422 }
3423 }
3424
3425 pub fn retain_known_destination(
3427 &self,
3428 dest_hash: &rns_core::types::DestHash,
3429 ) -> Result<bool, SendError> {
3430 match self.query(QueryRequest::RetainKnownDestination {
3431 dest_hash: dest_hash.0,
3432 })? {
3433 QueryResponse::RetainKnownDestination(ok) => Ok(ok),
3434 _ => Ok(false),
3435 }
3436 }
3437
3438 pub fn retain_identity(
3440 &self,
3441 identity_hash: &rns_core::types::IdentityHash,
3442 ) -> Result<bool, SendError> {
3443 match self.query(QueryRequest::RetainIdentity {
3444 identity_hash: identity_hash.0,
3445 })? {
3446 QueryResponse::RetainIdentity(ok) => Ok(ok),
3447 _ => Ok(false),
3448 }
3449 }
3450
3451 pub fn unretain_known_destination(
3453 &self,
3454 dest_hash: &rns_core::types::DestHash,
3455 ) -> Result<bool, SendError> {
3456 match self.query(QueryRequest::UnretainKnownDestination {
3457 dest_hash: dest_hash.0,
3458 })? {
3459 QueryResponse::UnretainKnownDestination(ok) => Ok(ok),
3460 _ => Ok(false),
3461 }
3462 }
3463
3464 pub fn mark_known_destination_used(
3466 &self,
3467 dest_hash: &rns_core::types::DestHash,
3468 ) -> Result<bool, SendError> {
3469 match self.query(QueryRequest::MarkKnownDestinationUsed {
3470 dest_hash: dest_hash.0,
3471 })? {
3472 QueryResponse::MarkKnownDestinationUsed(ok) => Ok(ok),
3473 _ => Ok(false),
3474 }
3475 }
3476
3477 fn persist_known_destinations(&self) {
3478 let Some(path) = self.known_destinations_path.as_ref() else {
3479 return;
3480 };
3481
3482 let Ok(entries) = self.known_destinations() else {
3483 return;
3484 };
3485
3486 let destinations: std::collections::HashMap<[u8; 16], storage::KnownDestination> = entries
3487 .into_iter()
3488 .map(|entry| {
3489 (
3490 entry.dest_hash,
3491 storage::KnownDestination {
3492 identity_hash: entry.identity_hash,
3493 public_key: entry.public_key,
3494 app_data: entry.app_data,
3495 hops: entry.hops,
3496 received_at: entry.received_at,
3497 receiving_interface: entry.receiving_interface.0,
3498 was_used: entry.was_used,
3499 last_used_at: entry.last_used_at,
3500 retained: entry.retained,
3501 },
3502 )
3503 })
3504 .collect();
3505
3506 if let Err(err) = storage::save_known_destinations(&destinations, path) {
3507 log::warn!("failed to persist known destinations: {}", err);
3508 }
3509 }
3510
3511 fn persist_transport_state(&self) {
3512 let Some(storage_dir) = self.transport_state_dir.as_ref() else {
3513 return;
3514 };
3515 let Ok(QueryResponse::TransportStateSnapshot(snapshot)) =
3516 self.query(QueryRequest::TransportStateSnapshot)
3517 else {
3518 return;
3519 };
3520 if let Err(err) = storage::save_transport_state(&snapshot, storage_dir, true) {
3521 log::warn!("failed to persist transport state: {}", err);
3522 }
3523 }
3524
3525 pub fn load_hook(
3527 &self,
3528 name: String,
3529 wasm_bytes: Vec<u8>,
3530 attach_point: String,
3531 priority: i32,
3532 ) -> Result<Result<(), String>, SendError> {
3533 let (response_tx, response_rx) = std::sync::mpsc::channel();
3534 self.tx
3535 .send(Event::LoadHook {
3536 name,
3537 wasm_bytes,
3538 attach_point,
3539 priority,
3540 response_tx,
3541 })
3542 .map_err(|_| SendError)?;
3543 response_rx.recv().map_err(|_| SendError)
3544 }
3545
3546 pub fn load_hook_file(
3548 &self,
3549 name: String,
3550 path: String,
3551 hook_type: String,
3552 attach_point: String,
3553 priority: i32,
3554 ) -> Result<Result<(), String>, SendError> {
3555 let (response_tx, response_rx) = std::sync::mpsc::channel();
3556 self.tx
3557 .send(Event::LoadHookFile {
3558 name,
3559 path,
3560 hook_type,
3561 attach_point,
3562 priority,
3563 response_tx,
3564 })
3565 .map_err(|_| SendError)?;
3566 response_rx.recv().map_err(|_| SendError)
3567 }
3568
3569 pub fn load_builtin_hook(
3571 &self,
3572 name: String,
3573 builtin_id: String,
3574 attach_point: String,
3575 priority: i32,
3576 ) -> Result<Result<(), String>, SendError> {
3577 let (response_tx, response_rx) = std::sync::mpsc::channel();
3578 self.tx
3579 .send(Event::LoadBuiltinHook {
3580 name,
3581 builtin_id,
3582 attach_point,
3583 priority,
3584 response_tx,
3585 })
3586 .map_err(|_| SendError)?;
3587 response_rx.recv().map_err(|_| SendError)
3588 }
3589
3590 pub fn unload_hook(
3592 &self,
3593 name: String,
3594 attach_point: String,
3595 ) -> Result<Result<(), String>, SendError> {
3596 let (response_tx, response_rx) = std::sync::mpsc::channel();
3597 self.tx
3598 .send(Event::UnloadHook {
3599 name,
3600 attach_point,
3601 response_tx,
3602 })
3603 .map_err(|_| SendError)?;
3604 response_rx.recv().map_err(|_| SendError)
3605 }
3606
3607 pub fn reload_hook(
3609 &self,
3610 name: String,
3611 attach_point: String,
3612 wasm_bytes: Vec<u8>,
3613 ) -> Result<Result<(), String>, SendError> {
3614 let (response_tx, response_rx) = std::sync::mpsc::channel();
3615 self.tx
3616 .send(Event::ReloadHook {
3617 name,
3618 attach_point,
3619 wasm_bytes,
3620 response_tx,
3621 })
3622 .map_err(|_| SendError)?;
3623 response_rx.recv().map_err(|_| SendError)
3624 }
3625
3626 pub fn reload_hook_file(
3628 &self,
3629 name: String,
3630 attach_point: String,
3631 path: String,
3632 hook_type: String,
3633 ) -> Result<Result<(), String>, SendError> {
3634 let (response_tx, response_rx) = std::sync::mpsc::channel();
3635 self.tx
3636 .send(Event::ReloadHookFile {
3637 name,
3638 attach_point,
3639 path,
3640 hook_type,
3641 response_tx,
3642 })
3643 .map_err(|_| SendError)?;
3644 response_rx.recv().map_err(|_| SendError)
3645 }
3646
3647 pub fn reload_builtin_hook(
3649 &self,
3650 name: String,
3651 attach_point: String,
3652 builtin_id: String,
3653 ) -> Result<Result<(), String>, SendError> {
3654 let (response_tx, response_rx) = std::sync::mpsc::channel();
3655 self.tx
3656 .send(Event::ReloadBuiltinHook {
3657 name,
3658 attach_point,
3659 builtin_id,
3660 response_tx,
3661 })
3662 .map_err(|_| SendError)?;
3663 response_rx.recv().map_err(|_| SendError)
3664 }
3665
3666 pub fn set_hook_enabled(
3668 &self,
3669 name: String,
3670 attach_point: String,
3671 enabled: bool,
3672 ) -> Result<Result<(), String>, SendError> {
3673 let (response_tx, response_rx) = std::sync::mpsc::channel();
3674 self.tx
3675 .send(Event::SetHookEnabled {
3676 name,
3677 attach_point,
3678 enabled,
3679 response_tx,
3680 })
3681 .map_err(|_| SendError)?;
3682 response_rx.recv().map_err(|_| SendError)
3683 }
3684
3685 pub fn set_hook_priority(
3687 &self,
3688 name: String,
3689 attach_point: String,
3690 priority: i32,
3691 ) -> Result<Result<(), String>, SendError> {
3692 let (response_tx, response_rx) = std::sync::mpsc::channel();
3693 self.tx
3694 .send(Event::SetHookPriority {
3695 name,
3696 attach_point,
3697 priority,
3698 response_tx,
3699 })
3700 .map_err(|_| SendError)?;
3701 response_rx.recv().map_err(|_| SendError)
3702 }
3703
3704 pub fn list_hooks(&self) -> Result<Vec<crate::event::HookInfo>, SendError> {
3706 let (response_tx, response_rx) = std::sync::mpsc::channel();
3707 self.tx
3708 .send(Event::ListHooks { response_tx })
3709 .map_err(|_| SendError)?;
3710 response_rx.recv().map_err(|_| SendError)
3711 }
3712
3713 pub(crate) fn from_parts(
3716 tx: EventSender,
3717 driver_handle: thread::JoinHandle<()>,
3718 rpc_server: Option<crate::rpc::RpcServer>,
3719 tick_interval_ms: Arc<AtomicU64>,
3720 ) -> Self {
3721 RnsNode {
3722 tx,
3723 driver_handle: Some(driver_handle),
3724 verify_handle: None,
3725 verify_shutdown: Arc::new(AtomicBool::new(false)),
3726 persist_handle: None,
3727 persist_stop: None,
3728 rpc_server,
3729 #[cfg(target_os = "linux")]
3730 rpc_unix_server: None,
3731 tick_interval_ms,
3732 probe_server: None,
3733 known_destinations_path: None,
3734 transport_state_dir: None,
3735 ratchet_store: None,
3736 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
3737 }
3738 }
3739
3740 pub fn event_sender(&self) -> &EventSender {
3742 &self.tx
3743 }
3744
3745 pub fn set_tick_interval(&self, ms: u64) -> u64 {
3750 let clamped = ms.clamp(100, 10_000);
3751 if clamped != ms {
3752 log::warn!(
3753 "tick interval {}ms out of range, clamped to {}ms",
3754 ms,
3755 clamped
3756 );
3757 }
3758 self.tick_interval_ms.store(clamped, Ordering::Relaxed);
3759 clamped
3760 }
3761
3762 pub fn tick_interval(&self) -> u64 {
3764 self.tick_interval_ms.load(Ordering::Relaxed)
3765 }
3766
3767 pub fn shutdown(mut self) {
3769 self.shutdown_inner();
3770 }
3771
3772 fn shutdown_inner(&mut self) {
3773 if self.driver_handle.is_none()
3774 && self.verify_handle.is_none()
3775 && self.persist_handle.is_none()
3776 && self.rpc_server.is_none()
3777 && {
3778 #[cfg(target_os = "linux")]
3779 {
3780 self.rpc_unix_server.is_none()
3781 }
3782 #[cfg(not(target_os = "linux"))]
3783 {
3784 true
3785 }
3786 }
3787 {
3788 return;
3789 }
3790 if let Some(mut rpc) = self.rpc_server.take() {
3792 rpc.stop();
3793 }
3794 #[cfg(target_os = "linux")]
3795 if let Some(mut rpc) = self.rpc_unix_server.take() {
3796 rpc.stop();
3797 }
3798 self.persist_known_destinations();
3799 self.persist_transport_state();
3800 if let Some(stop) = self.persist_stop.take() {
3801 let _ = stop.send(());
3802 }
3803 if let Some(handle) = self.persist_handle.take() {
3804 let _ = handle.join();
3805 }
3806 self.verify_shutdown.store(true, Ordering::Relaxed);
3807 let _ = self.tx.send(Event::Shutdown);
3808 if let Some(handle) = self.driver_handle.take() {
3809 let _ = handle.join();
3810 }
3811 if let Some(handle) = self.verify_handle.take() {
3812 let _ = handle.join();
3813 }
3814 }
3815}
3816
3817impl Drop for RnsNode {
3818 fn drop(&mut self) {
3819 self.shutdown_inner();
3820 }
3821}
3822
3823#[cfg(test)]
3824mod tests {
3825 use super::*;
3826 use crate::driver::IfacRuntimeConfig;
3827 use crate::storage::RatchetStore;
3828 use std::fs;
3829 use tempfile::tempdir;
3830
3831 struct NoopCallbacks;
3832
3833 impl Callbacks for NoopCallbacks {
3834 fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
3835 fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
3836 fn on_local_delivery(
3837 &mut self,
3838 _: rns_core::types::DestHash,
3839 _: Vec<u8>,
3840 _: rns_core::types::PacketHash,
3841 ) {
3842 }
3843 }
3844
3845 #[test]
3846 fn managed_interface_attaches_disabled_section_and_reloads_from_disk() {
3847 use crate::event::InterfaceManagementOperation as Operation;
3848
3849 let dir = tempdir().unwrap();
3850 let config_file = dir.path().join("config");
3851 let write_config = |port| {
3852 fs::write(&config_file, format!("[reticulum]\nshare_instance = no\n\n[interfaces]\n [[Managed UDP]]\n type = UDPInterface\n enabled = no\n forward_ip = 127.0.0.1\n forward_port = {port}\n"))
3853 .unwrap();
3854 };
3855 write_config(9);
3856 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
3857 let manage = |operation| {
3858 node.query(QueryRequest::ManageInterface {
3859 operation,
3860 name: "Managed UDP".into(),
3861 })
3862 .unwrap()
3863 };
3864 assert!(matches!(
3865 manage(Operation::Detach),
3866 QueryResponse::InterfaceManagement(None)
3867 ));
3868 assert!(matches!(
3869 manage(Operation::Attach),
3870 QueryResponse::InterfaceManagement(Some(true))
3871 ));
3872 assert!(matches!(
3873 manage(Operation::Attach),
3874 QueryResponse::InterfaceManagement(Some(false))
3875 ));
3876 write_config(10);
3877 assert!(matches!(
3878 manage(Operation::Reload),
3879 QueryResponse::InterfaceManagement(Some(true))
3880 ));
3881 assert!(matches!(
3882 manage(Operation::Detach),
3883 QueryResponse::InterfaceManagement(Some(true))
3884 ));
3885 assert!(matches!(
3886 manage(Operation::Detach),
3887 QueryResponse::InterfaceManagement(None)
3888 ));
3889 node.shutdown();
3890 }
3891
3892 #[test]
3893 fn interface_management_can_be_disabled() {
3894 use crate::event::InterfaceManagementOperation as Operation;
3895
3896 let dir = tempdir().unwrap();
3897 fs::write(
3898 dir.path().join("config"),
3899 "[reticulum]\nshare_instance = no\nenable_interface_management = no\n",
3900 )
3901 .unwrap();
3902 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
3903 assert!(matches!(
3904 node.query(QueryRequest::ManageInterface {
3905 operation: Operation::Attach,
3906 name: "missing".into(),
3907 })
3908 .unwrap(),
3909 QueryResponse::InterfaceManagement(Some(false))
3910 ));
3911 node.shutdown();
3912 }
3913
3914 #[cfg(feature = "iface-tcp")]
3915 #[test]
3916 fn managed_tcp_listener_releases_port_on_detach() {
3917 use crate::event::InterfaceManagementOperation as Operation;
3918 use std::io::Read;
3919
3920 let dir = tempdir().unwrap();
3921 let port = crate::test_support::port();
3922 fs::write(dir.path().join("config"), format!(
3923 "[reticulum]\nshare_instance = no\n\n[interfaces]\n [[Managed TCP]]\n type = TCPServerInterface\n enabled = no\n listen_ip = 127.0.0.1\n listen_port = {port}\n"
3924 )).unwrap();
3925 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
3926 let manage = |operation| {
3927 node.query(QueryRequest::ManageInterface {
3928 operation,
3929 name: "Managed TCP".into(),
3930 })
3931 .unwrap()
3932 };
3933 assert!(matches!(
3934 manage(Operation::Attach),
3935 QueryResponse::InterfaceManagement(Some(true))
3936 ));
3937 let mut client = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
3938 client
3939 .set_read_timeout(Some(Duration::from_secs(2)))
3940 .unwrap();
3941 thread::sleep(Duration::from_millis(100));
3942 assert!(matches!(
3943 manage(Operation::Detach),
3944 QueryResponse::InterfaceManagement(Some(true))
3945 ));
3946 let deadline = std::time::Instant::now() + Duration::from_secs(2);
3947 while std::net::TcpStream::connect(("127.0.0.1", port)).is_ok()
3948 && std::time::Instant::now() < deadline
3949 {
3950 thread::sleep(Duration::from_millis(25));
3951 }
3952 assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_err());
3953 assert_eq!(client.read(&mut [0u8; 1]).unwrap(), 0);
3954 assert!(matches!(
3955 manage(Operation::Attach),
3956 QueryResponse::InterfaceManagement(Some(true))
3957 ));
3958 assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_ok());
3959 node.shutdown();
3960 }
3961
3962 #[test]
3963 fn shared_instance_rpc_manages_interface_by_name() {
3964 use crate::pickle::PickleValue;
3965 use crate::rpc::{derive_auth_key, RpcAddr, RpcClient};
3966
3967 let dir = tempdir().unwrap();
3968 let rpc_port = crate::test_support::port();
3969 let shared_port = crate::test_support::port();
3970 fs::write(dir.path().join("config"), format!(
3971 "[reticulum]\nshare_instance = yes\ninstance_control_port = {rpc_port}\nshared_instance_port = {shared_port}\n\n[interfaces]\n [[Managed UDP]]\n type = UDPInterface\n enabled = no\n forward_ip = 127.0.0.1\n forward_port = 9\n"
3972 )).unwrap();
3973 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
3974 let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
3975 let identity = storage::load_or_create_identity(&paths.identities).unwrap();
3976 let key = derive_auth_key(&identity.get_private_key().unwrap());
3977 let call = |action: &str| {
3978 let mut client =
3979 RpcClient::connect(&RpcAddr::Tcp("127.0.0.1".into(), rpc_port), &key).unwrap();
3980 client
3981 .call(&PickleValue::Dict(vec![
3982 (
3983 PickleValue::String("manage".into()),
3984 PickleValue::String(action.into()),
3985 ),
3986 (
3987 PickleValue::String("name".into()),
3988 PickleValue::String("Managed UDP".into()),
3989 ),
3990 ]))
3991 .unwrap()
3992 };
3993 assert_eq!(call("attach_interface"), PickleValue::Bool(true));
3994 assert_eq!(call("reload_interface"), PickleValue::Bool(true));
3995 assert_eq!(call("detach_interface"), PickleValue::Bool(true));
3996 assert_eq!(call("detach_interface"), PickleValue::None);
3997 node.shutdown();
3998 }
3999
4000 struct TestWriter;
4001
4002 impl crate::interface::Writer for TestWriter {
4003 fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
4004 Ok(())
4005 }
4006 }
4007
4008 fn test_transport_config(transport_enabled: bool) -> TransportConfig {
4009 TransportConfig {
4010 transport_enabled,
4011 identity_hash: None,
4012 local_hops_delta: 0,
4013 prefer_shorter_path: false,
4014 max_paths_per_destination: 1,
4015 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
4016 packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Eager,
4017 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
4018 max_path_destinations: usize::MAX,
4019 max_tunnel_destinations_total: usize::MAX,
4020 destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
4021 announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
4022 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
4023 announce_sig_cache_enabled: true,
4024 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
4025 announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
4026 announce_queue_max_entries: 256,
4027 announce_queue_max_interfaces: 1024,
4028 }
4029 }
4030
4031 fn test_interface_info(id: u64) -> rns_core::transport::types::InterfaceInfo {
4032 rns_core::transport::types::InterfaceInfo {
4033 id: rns_core::transport::types::InterfaceId(id),
4034 name: format!("test-{id}"),
4035 mode: rns_core::constants::MODE_FULL,
4036 gravity: 0,
4037 recursive_prs: false,
4038 announces_from_internal: true,
4039 announces_to_internal: None,
4040 out_capable: true,
4041 in_capable: true,
4042 bitrate: None,
4043 airtime_profile: None,
4044 announce_rate_target: None,
4045 announce_rate_grace: 0,
4046 announce_rate_penalty: 0.0,
4047 announce_cap: rns_core::constants::ANNOUNCE_CAP,
4048 is_local_client: false,
4049 wants_tunnel: false,
4050 tunnel_id: None,
4051 mtu: rns_core::constants::MTU as u32,
4052 ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
4053 ia_freq: 0.0,
4054 ip_freq: 0.0,
4055 op_freq: 0.0,
4056 op_samples: 0,
4057 started: 0.0,
4058 }
4059 }
4060
4061 #[test]
4062 fn static_interface_registration_applies_transport_announce_rate_defaults() {
4063 let (tx, rx) = crate::event::channel();
4064 let mut driver = Driver::new(
4065 test_transport_config(true),
4066 rx,
4067 tx.clone(),
4068 Box::new(NoopCallbacks),
4069 );
4070 driver.set_announce_rate_defaults(AnnounceRateDefaults {
4071 target: Some(7200.0),
4072 penalty: 15.0,
4073 grace: 7,
4074 });
4075
4076 let id = rns_core::transport::types::InterfaceId(7);
4077 register_started_interface(StartedInterface {
4078 driver: &mut driver,
4079 tx: &tx,
4080 queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
4081 id,
4082 info: test_interface_info(id.0),
4083 writer: Box::new(TestWriter),
4084 interface_type_name: "TestInterface".to_string(),
4085 ifac_state: None,
4086 ifac_runtime: &IfacRuntimeConfig {
4087 netname: None,
4088 netkey: None,
4089 size: 16,
4090 },
4091 });
4092
4093 let info = &driver.interfaces[&id].info;
4094 assert_eq!(info.announce_rate_target, Some(7200.0));
4095 assert_eq!(info.announce_rate_penalty, 15.0);
4096 assert_eq!(info.announce_rate_grace, 7);
4097 }
4098
4099 #[test]
4100 fn static_interface_registration_applies_transport_ingress_control_defaults() {
4101 let (tx, rx) = crate::event::channel();
4102 let mut driver = Driver::new(
4103 test_transport_config(true),
4104 rx,
4105 tx.clone(),
4106 Box::new(NoopCallbacks),
4107 );
4108 let mut defaults = rns_core::transport::types::IngressControlConfig::enabled();
4109 defaults.burst_hold = 1.5;
4110 defaults.pr_burst_freq = 5.5;
4111 defaults.egress_enabled = true;
4112 defaults.egress_pr_freq = 9.5;
4113 driver.set_ingress_control_defaults(defaults);
4114
4115 let id = rns_core::transport::types::InterfaceId(8);
4116 register_started_interface(StartedInterface {
4117 driver: &mut driver,
4118 tx: &tx,
4119 queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
4120 id,
4121 info: test_interface_info(id.0),
4122 writer: Box::new(TestWriter),
4123 interface_type_name: "TestInterface".to_string(),
4124 ifac_state: None,
4125 ifac_runtime: &IfacRuntimeConfig {
4126 netname: None,
4127 netkey: None,
4128 size: 16,
4129 },
4130 });
4131
4132 let ic = driver.interfaces[&id].info.ingress_control;
4133 assert_eq!(ic.burst_hold, 1.5);
4134 assert_eq!(ic.pr_burst_freq, 5.5);
4135 assert!(ic.egress_enabled);
4136 assert_eq!(ic.egress_pr_freq, 9.5);
4137 }
4138
4139 struct TestNodeRatchetStore {
4140 entry: storage::RatchetEntry,
4141 current_calls: std::sync::Mutex<Vec<[u8; 16]>>,
4142 }
4143
4144 impl storage::RatchetStore for TestNodeRatchetStore {
4145 fn remember(&self, _dest_hash: [u8; 16], _entry: storage::RatchetEntry) -> io::Result<()> {
4146 Ok(())
4147 }
4148
4149 fn current(
4150 &self,
4151 dest_hash: &[u8; 16],
4152 _now: f64,
4153 _expiry_secs: f64,
4154 ) -> io::Result<Option<storage::RatchetEntry>> {
4155 self.current_calls.lock().unwrap().push(*dest_hash);
4156 Ok(Some(self.entry))
4157 }
4158
4159 fn cleanup(
4160 &self,
4161 _known_destinations: &HashSet<[u8; 16]>,
4162 _now: f64,
4163 _expiry_secs: f64,
4164 ) -> io::Result<storage::RatchetCleanupStats> {
4165 Ok(Default::default())
4166 }
4167 }
4168
4169 #[test]
4170 fn send_packet_checks_ratchet_store_for_single_destinations() {
4171 let store = Arc::new(TestNodeRatchetStore {
4172 entry: storage::RatchetEntry {
4173 ratchet: [0x55; 32],
4174 received_at: time::now(),
4175 },
4176 current_calls: std::sync::Mutex::new(Vec::new()),
4177 });
4178 let ratchet_store: Arc<dyn storage::RatchetStore> = store.clone();
4179 let node = RnsNode::start(
4180 NodeConfig {
4181 ratchet_store: Some(ratchet_store),
4182 ..Default::default()
4183 },
4184 Box::new(NoopCallbacks),
4185 )
4186 .unwrap();
4187
4188 let mut rng = OsRng;
4189 let remote_identity = Identity::new(&mut rng);
4190 let public_key = remote_identity.get_public_key().unwrap();
4191 let announced = crate::destination::AnnouncedIdentity {
4192 dest_hash: rns_core::types::DestHash([0x22; 16]),
4193 identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
4194 public_key,
4195 app_data: None,
4196 hops: 1,
4197 received_at: time::now(),
4198 receiving_interface: rns_core::transport::types::InterfaceId(0),
4199 rssi: Some(-100),
4200 snr: Some(10.5),
4201 };
4202 let dest = crate::destination::Destination::single_out("test", &["ratchet"], &announced);
4203
4204 node.send_packet_queued(&dest, b"hello").unwrap();
4205 assert_eq!(
4206 store.current_calls.lock().unwrap().as_slice(),
4207 &[dest.hash.0]
4208 );
4209
4210 node.shutdown();
4211 }
4212
4213 #[test]
4214 fn single_payload_uses_stored_ratchet_key_material() {
4215 let ratchet_prv = rns_crypto::x25519::X25519PrivateKey::from_bytes(&[0x42; 32]);
4216 let ratchet_pub = ratchet_prv.public_key().public_bytes();
4217 let store = Arc::new(TestNodeRatchetStore {
4218 entry: storage::RatchetEntry {
4219 ratchet: ratchet_pub,
4220 received_at: time::now(),
4221 },
4222 current_calls: std::sync::Mutex::new(Vec::new()),
4223 });
4224 let ratchet_store: Arc<dyn storage::RatchetStore> = store.clone();
4225 let (tx, _rx) = crate::event::channel();
4226 let node = RnsNode {
4227 tx,
4228 driver_handle: None,
4229 verify_handle: None,
4230 verify_shutdown: Arc::new(AtomicBool::new(false)),
4231 persist_handle: None,
4232 persist_stop: None,
4233 rpc_server: None,
4234 #[cfg(target_os = "linux")]
4235 rpc_unix_server: None,
4236 tick_interval_ms: Arc::new(AtomicU64::new(1000)),
4237 probe_server: None,
4238 known_destinations_path: None,
4239 transport_state_dir: None,
4240 ratchet_store: Some(ratchet_store),
4241 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
4242 };
4243
4244 let mut rng = OsRng;
4245 let remote_identity = Identity::new(&mut rng);
4246 let public_key = remote_identity.get_public_key().unwrap();
4247 let announced = crate::destination::AnnouncedIdentity {
4248 dest_hash: rns_core::types::DestHash([0x33; 16]),
4249 identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
4250 public_key,
4251 app_data: None,
4252 hops: 1,
4253 received_at: time::now(),
4254 receiving_interface: rns_core::transport::types::InterfaceId(0),
4255 rssi: Some(-100),
4256 snr: Some(10.5),
4257 };
4258 let dest = crate::destination::Destination::single_out("test", &["ratchet"], &announced);
4259 let payload = node
4260 .encrypt_single_payload(&dest, b"ratchet message")
4261 .unwrap();
4262
4263 assert_eq!(
4264 store.current_calls.lock().unwrap().as_slice(),
4265 &[dest.hash.0]
4266 );
4267 assert!(remote_identity.decrypt(&payload).is_err());
4268
4269 let peer_pub_bytes: [u8; 32] = payload[..32].try_into().unwrap();
4270 let peer_pub = rns_crypto::x25519::X25519PublicKey::from_bytes(&peer_pub_bytes);
4271 let shared_key = ratchet_prv.exchange(&peer_pub);
4272 let derived_key = rns_crypto::hkdf::hkdf(
4273 rns_crypto::identity::DERIVED_KEY_LENGTH,
4274 &shared_key,
4275 Some(remote_identity.hash()),
4276 None,
4277 )
4278 .unwrap();
4279 let token = rns_crypto::token::Token::new(&derived_key).unwrap();
4280 let plaintext = token.decrypt(&payload[32..]).unwrap();
4281 assert_eq!(plaintext, b"ratchet message");
4282 }
4283
4284 #[test]
4285 fn tcp_client_interface_is_not_discoverable_without_kiss_framing() {
4286 let mut params = std::collections::HashMap::new();
4287 params.insert("discoverable".to_string(), "yes".to_string());
4288 params.insert(
4289 "discovery_name".to_string(),
4290 "invalid-tcp-client".to_string(),
4291 );
4292 params.insert("reachable_on".to_string(), "example.com".to_string());
4293 params.insert("target_port".to_string(), "4242".to_string());
4294
4295 let discovery =
4296 super::extract_discovery_config("tcp-client", "TCPClientInterface", ¶ms);
4297
4298 assert!(
4299 discovery.is_none(),
4300 "TCPClientInterface discovery must be rejected unless KISS framing is supported"
4301 );
4302 }
4303
4304 #[test]
4305 fn discovery_config_parses_location_command() {
4306 let mut params = std::collections::HashMap::new();
4307 params.insert("discoverable".into(), "yes".into());
4308 params.insert("location_cmd".into(), "~/bin/reticulum-location".into());
4309
4310 let discovery =
4311 super::extract_discovery_config("public", "BackboneInterface", ¶ms).unwrap();
4312 assert_eq!(
4313 discovery.location_cmd.as_deref(),
4314 Some("~/bin/reticulum-location")
4315 );
4316 }
4317
4318 #[test]
4319 fn discovery_config_parses_valid_operator_lxmf_address() {
4320 let mut params = std::collections::HashMap::new();
4321 params.insert("discoverable".into(), "yes".into());
4322 params.insert("discovery_lxmf_address".into(), "ab".repeat(16));
4323
4324 let discovery =
4325 super::extract_discovery_config("public", "BackboneInterface", ¶ms).unwrap();
4326
4327 assert_eq!(discovery.operator_lxmf_address, Some([0xab; 16]));
4328 }
4329
4330 #[test]
4331 fn discovery_config_ignores_invalid_operator_lxmf_addresses() {
4332 for address in ["ab".repeat(15), "zz".repeat(16)] {
4333 let mut params = std::collections::HashMap::new();
4334 params.insert("discoverable".into(), "yes".into());
4335 params.insert("discovery_lxmf_address".into(), address);
4336
4337 let discovery =
4338 super::extract_discovery_config("public", "BackboneInterface", ¶ms).unwrap();
4339
4340 assert_eq!(discovery.operator_lxmf_address, None);
4341 }
4342 }
4343
4344 #[test]
4345 fn ingress_control_config_defaults_by_interface_type() {
4346 let params = std::collections::HashMap::new();
4347
4348 let tcp = super::parse_ingress_control_config("TCPServerInterface", ¶ms).unwrap();
4349 assert!(tcp.enabled);
4350 assert_eq!(
4351 tcp.max_held_announces,
4352 rns_core::constants::IC_MAX_HELD_ANNOUNCES
4353 );
4354 assert_eq!(tcp.burst_hold, rns_core::constants::IC_BURST_HOLD);
4355
4356 let pipe = super::parse_ingress_control_config("PipeInterface", ¶ms).unwrap();
4357 assert!(!pipe.enabled);
4358 assert_eq!(
4359 pipe.held_release_interval,
4360 rns_core::constants::IC_HELD_RELEASE_INTERVAL
4361 );
4362 }
4363
4364 #[test]
4365 fn ingress_control_config_parses_python_ic_keys() {
4366 let mut params = std::collections::HashMap::new();
4367 params.insert("ingress_control".to_string(), "No".to_string());
4368 params.insert("ic_max_held_announces".to_string(), "17".to_string());
4369 params.insert("ic_burst_hold".to_string(), "1.5".to_string());
4370 params.insert("ic_burst_freq_new".to_string(), "2.5".to_string());
4371 params.insert("ic_burst_freq".to_string(), "3.5".to_string());
4372 params.insert("ic_pr_burst_freq_new".to_string(), "3.25".to_string());
4373 params.insert("ic_pr_burst_freq".to_string(), "8.25".to_string());
4374 params.insert("egress_control".to_string(), "Yes".to_string());
4375 params.insert("ec_pr_freq".to_string(), "5.25".to_string());
4376 params.insert("ic_new_time".to_string(), "4.5".to_string());
4377 params.insert("ic_burst_penalty".to_string(), "5.5".to_string());
4378 params.insert("ic_held_release_interval".to_string(), "6.5".to_string());
4379
4380 let config = super::parse_ingress_control_config("TCPServerInterface", ¶ms).unwrap();
4381
4382 assert!(!config.enabled);
4383 assert_eq!(config.max_held_announces, 17);
4384 assert_eq!(config.burst_hold, 1.5);
4385 assert_eq!(config.burst_freq_new, 2.5);
4386 assert_eq!(config.burst_freq, 3.5);
4387 assert_eq!(config.pr_burst_freq_new, 3.25);
4388 assert_eq!(config.pr_burst_freq, 8.25);
4389 assert!(config.egress_enabled);
4390 assert_eq!(config.egress_pr_freq, 5.25);
4391 assert_eq!(config.new_time, 4.5);
4392 assert_eq!(config.burst_penalty, 5.5);
4393 assert_eq!(config.held_release_interval, 6.5);
4394 }
4395
4396 #[test]
4397 fn ingress_control_config_rejects_invalid_values() {
4398 let mut params = std::collections::HashMap::new();
4399 params.insert("ic_burst_hold".to_string(), "-1".to_string());
4400
4401 let err = super::parse_ingress_control_config("TCPServerInterface", ¶ms).unwrap_err();
4402
4403 assert!(err.contains("ic_burst_hold"));
4404 }
4405
4406 #[test]
4407 fn discovery_stamp_default_is_16_but_explicit_override_is_preserved() {
4408 let mut params = std::collections::HashMap::new();
4409 params.insert("discoverable".to_string(), "yes".to_string());
4410 let defaults =
4411 super::extract_discovery_config("Discovery Test", "BackboneInterface", ¶ms)
4412 .unwrap();
4413 assert_eq!(defaults.stamp_value, 16);
4414
4415 params.insert("discovery_stamp_value".to_string(), "14".to_string());
4416 let overridden =
4417 super::extract_discovery_config("Discovery Test", "BackboneInterface", ¶ms)
4418 .unwrap();
4419 assert_eq!(overridden.stamp_value, 14);
4420 }
4421
4422 #[test]
4423 fn start_and_shutdown() {
4424 let node = RnsNode::start(
4425 NodeConfig {
4426 panic_on_interface_error: false,
4427 transport_enabled: false,
4428 static_transport_identity: false,
4429 local_hops_delta: false,
4430 identity: None,
4431 interfaces: vec![],
4432 share_instance: false,
4433 instance_name: "default".into(),
4434 shared_instance_port: 37428,
4435 rpc_port: 0,
4436 cache_dir: None,
4437 ratchet_store: None,
4438 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
4439 management: Default::default(),
4440 probe_port: None,
4441 probe_addrs: vec![],
4442 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
4443 direct_connect_policy: Default::default(),
4444 device: None,
4445 underlay_mark: None,
4446 hooks: Vec::new(),
4447 discover_interfaces: false,
4448 autoconnect_interface_mode: None,
4449 autoconnect_interface_gravity: 0,
4450 autoconnect_announces_to_internal: false,
4451 autoconnect_unverified_implementations: false,
4452 discovery_required_value: None,
4453 respond_to_probes: false,
4454 prefer_shorter_path: false,
4455 max_paths_per_destination: 1,
4456 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
4457 packet_hashlist_allocation:
4458 rns_core::transport::types::PacketHashlistAllocation::Eager,
4459 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
4460 max_path_destinations: usize::MAX,
4461 max_tunnel_destinations_total: usize::MAX,
4462 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
4463 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
4464 announce_table_ttl: Duration::from_secs(
4465 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
4466 ),
4467 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
4468 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
4469 interface_writer_queue_capacity:
4470 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
4471 announce_rate_defaults: AnnounceRateDefaults::default(),
4472 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
4473 ),
4474 #[cfg(feature = "iface-backbone")]
4475 backbone_peer_pool: None,
4476 announce_sig_cache_enabled: true,
4477 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
4478 announce_sig_cache_ttl: Duration::from_secs(
4479 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
4480 ),
4481 registry: None,
4482 #[cfg(feature = "hooks")]
4483 provider_bridge: None,
4484 },
4485 Box::new(NoopCallbacks),
4486 )
4487 .unwrap();
4488 node.shutdown();
4489 }
4490
4491 #[test]
4492 fn known_destinations_persist_across_restart() {
4493 let dir = tempdir().unwrap();
4494 let dest_hash = [0x91; 16];
4495 let identity = Identity::new(&mut OsRng);
4496 let last_used_at = 77.0;
4497 let receiving_interface = rns_core::transport::types::InterfaceId(42);
4498
4499 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
4500 let (response_tx, response_rx) = std::sync::mpsc::channel();
4501 node.event_sender()
4502 .send(crate::event::Event::Query(
4503 QueryRequest::RestoreKnownDestination(crate::event::KnownDestinationEntry {
4504 dest_hash,
4505 identity_hash: *identity.hash(),
4506 public_key: identity.get_public_key().unwrap(),
4507 app_data: Some(b"persisted".to_vec()),
4508 hops: 2,
4509 received_at: 55.0,
4510 receiving_interface,
4511 was_used: true,
4512 last_used_at: Some(last_used_at),
4513 retained: true,
4514 }),
4515 response_tx,
4516 ))
4517 .unwrap();
4518 assert!(matches!(
4519 response_rx.recv().unwrap(),
4520 QueryResponse::RestoreKnownDestination(true)
4521 ));
4522 node.shutdown();
4523
4524 let restarted = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
4525 let entries = restarted.known_destinations().unwrap();
4526 let entry = entries
4527 .iter()
4528 .find(|entry| entry.dest_hash == dest_hash)
4529 .expect("reloaded destination should appear in lifecycle listing");
4530 assert!(entry.retained);
4531 assert!(entry.was_used);
4532 assert_eq!(entry.hops, 2);
4533 assert_eq!(entry.receiving_interface, receiving_interface);
4534 assert_eq!(entry.last_used_at, Some(last_used_at));
4535
4536 let recalled = restarted
4537 .recall_identity(&rns_core::types::DestHash(dest_hash))
4538 .unwrap()
4539 .expect("known destination should reload from storage");
4540 assert_eq!(recalled.identity_hash.0, *identity.hash());
4541 assert_eq!(recalled.app_data, Some(b"persisted".to_vec()));
4542 restarted.shutdown();
4543 }
4544
4545 #[test]
4546 fn transport_paths_hashes_and_detached_tunnels_restore_from_upstream_files() {
4547 use rns_core::packet::{PacketFlags, RawPacket};
4548 use rns_core::transport::persistence::{
4549 PersistedPath, PersistedTunnel, TransportStateSnapshot,
4550 };
4551
4552 let dir = tempdir().unwrap();
4553 fs::write(
4554 dir.path().join("config"),
4555 r#"[reticulum]
4556enable_transport = True
4557share_instance = False
4558
4559[interfaces]
4560 [[Persistence Test]]
4561 type = UDPInterface
4562 forward_ip = 127.0.0.1
4563 forward_port = 9
4564 interface_mode = full
4565"#,
4566 )
4567 .unwrap();
4568 let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
4569 let destination_hash = [0x31; 16];
4570 let packet = RawPacket::pack(
4571 PacketFlags {
4572 header_type: rns_core::constants::HEADER_1,
4573 context_flag: rns_core::constants::FLAG_UNSET,
4574 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
4575 destination_type: rns_core::constants::DESTINATION_SINGLE,
4576 packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
4577 },
4578 0,
4579 &destination_hash,
4580 None,
4581 rns_core::constants::CONTEXT_NONE,
4582 &[0xA5; 32],
4583 )
4584 .unwrap();
4585 crate::announce_cache::AnnounceCache::new(paths.cache.join("announces"))
4586 .store(&packet.packet_hash, &packet.raw, None)
4587 .unwrap();
4588 let now = time::now();
4589 let route = PersistedPath {
4590 destination_hash,
4591 timestamp: now,
4592 received_from: [0x41; 16],
4593 hops: 2,
4594 expires: now + 600.0,
4595 random_blobs: vec![[0x51; 10]],
4596 interface_hash: Some(rns_core::hash::full_hash(b"Persistence Test")),
4597 packet_hash: packet.packet_hash,
4598 };
4599 storage::save_transport_state(
4600 &TransportStateSnapshot {
4601 packet_hashes: vec![[0x61; 32], [0x62; 32]],
4602 paths: vec![route.clone()],
4603 tunnels: vec![PersistedTunnel {
4604 tunnel_id: [0x71; 32],
4605 interface_hash: None,
4606 paths: vec![PersistedPath {
4607 interface_hash: None,
4608 ..route
4609 }],
4610 expires: now + 1_200.0,
4611 }],
4612 },
4613 &paths.storage,
4614 false,
4615 )
4616 .unwrap();
4617
4618 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
4619 let QueryResponse::TransportStateSnapshot(restored) =
4620 node.query(QueryRequest::TransportStateSnapshot).unwrap()
4621 else {
4622 panic!("unexpected transport snapshot response");
4623 };
4624 assert_eq!(restored.packet_hashes, vec![[0x61; 32], [0x62; 32]]);
4625 assert_eq!(restored.paths.len(), 1);
4626 assert_eq!(restored.paths[0].destination_hash, destination_hash);
4627 assert_eq!(restored.paths[0].received_from, [0x41; 16]);
4628 assert_eq!(restored.tunnels.len(), 1);
4629 assert_eq!(restored.tunnels[0].tunnel_id, [0x71; 32]);
4630 assert_eq!(restored.tunnels[0].interface_hash, None);
4631 node.shutdown();
4632 }
4633
4634 #[test]
4635 fn from_config_cleans_persistent_ratchets_after_loading_known_destinations() {
4636 fn hex(bytes: &[u8]) -> String {
4637 bytes.iter().map(|byte| format!("{:02x}", byte)).collect()
4638 }
4639
4640 let dir = tempdir().unwrap();
4641 fs::write(
4642 dir.path().join("config"),
4643 "[reticulum]\nenable_transport = False\nshare_instance = False\nratchet_expiry = 300\n",
4644 )
4645 .unwrap();
4646 let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
4647 let known_live = [0x01; 16];
4648 let known_expired = [0x02; 16];
4649 let unknown = [0x03; 16];
4650 let known_corrupt = [0x04; 16];
4651 let now = time::now();
4652
4653 let identity = Identity::new(&mut OsRng);
4654 let known = std::collections::HashMap::from([
4655 (
4656 known_live,
4657 storage::KnownDestination {
4658 identity_hash: *identity.hash(),
4659 public_key: identity.get_public_key().unwrap(),
4660 app_data: None,
4661 hops: 1,
4662 received_at: now,
4663 receiving_interface: 0,
4664 was_used: false,
4665 last_used_at: None,
4666 retained: false,
4667 },
4668 ),
4669 (
4670 known_expired,
4671 storage::KnownDestination {
4672 identity_hash: *identity.hash(),
4673 public_key: identity.get_public_key().unwrap(),
4674 app_data: None,
4675 hops: 1,
4676 received_at: now,
4677 receiving_interface: 0,
4678 was_used: false,
4679 last_used_at: None,
4680 retained: false,
4681 },
4682 ),
4683 (
4684 known_corrupt,
4685 storage::KnownDestination {
4686 identity_hash: *identity.hash(),
4687 public_key: identity.get_public_key().unwrap(),
4688 app_data: None,
4689 hops: 1,
4690 received_at: now,
4691 receiving_interface: 0,
4692 was_used: false,
4693 last_used_at: None,
4694 retained: false,
4695 },
4696 ),
4697 ]);
4698 storage::save_known_destinations(&known, &paths.storage.join("known_destinations"))
4699 .unwrap();
4700
4701 let store = storage::FsRatchetStore::new(paths.ratchets.clone());
4702 store
4703 .remember(
4704 known_live,
4705 storage::RatchetEntry {
4706 ratchet: [0x11; 32],
4707 received_at: now,
4708 },
4709 )
4710 .unwrap();
4711 store
4712 .remember(
4713 known_expired,
4714 storage::RatchetEntry {
4715 ratchet: [0x22; 32],
4716 received_at: now - 1000.0,
4717 },
4718 )
4719 .unwrap();
4720 store
4721 .remember(
4722 unknown,
4723 storage::RatchetEntry {
4724 ratchet: [0x33; 32],
4725 received_at: now,
4726 },
4727 )
4728 .unwrap();
4729 fs::write(paths.ratchets.join(hex(&known_corrupt)), b"not msgpack").unwrap();
4730 fs::write(paths.ratchets.join("0102.out"), b"temp").unwrap();
4731
4732 let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
4733
4734 assert!(paths.ratchets.join(hex(&known_live)).exists());
4735 assert!(!paths.ratchets.join(hex(&known_expired)).exists());
4736 assert!(!paths.ratchets.join(hex(&unknown)).exists());
4737 assert!(!paths.ratchets.join(hex(&known_corrupt)).exists());
4738 assert!(!paths.ratchets.join("0102.out").exists());
4739
4740 node.shutdown();
4741 }
4742
4743 #[test]
4744 fn start_with_identity() {
4745 let identity = Identity::new(&mut OsRng);
4746 let hash = *identity.hash();
4747 let node = RnsNode::start(
4748 NodeConfig {
4749 panic_on_interface_error: false,
4750 transport_enabled: true,
4751 static_transport_identity: false,
4752 local_hops_delta: false,
4753 identity: Some(identity),
4754 interfaces: vec![],
4755 share_instance: false,
4756 instance_name: "default".into(),
4757 shared_instance_port: 37428,
4758 rpc_port: 0,
4759 cache_dir: None,
4760 ratchet_store: None,
4761 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
4762 management: Default::default(),
4763 probe_port: None,
4764 probe_addrs: vec![],
4765 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
4766 direct_connect_policy: Default::default(),
4767 device: None,
4768 underlay_mark: None,
4769 hooks: Vec::new(),
4770 discover_interfaces: false,
4771 autoconnect_interface_mode: None,
4772 autoconnect_interface_gravity: 0,
4773 autoconnect_announces_to_internal: false,
4774 autoconnect_unverified_implementations: false,
4775 discovery_required_value: None,
4776 respond_to_probes: false,
4777 prefer_shorter_path: false,
4778 max_paths_per_destination: 1,
4779 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
4780 packet_hashlist_allocation:
4781 rns_core::transport::types::PacketHashlistAllocation::Eager,
4782 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
4783 max_path_destinations: usize::MAX,
4784 max_tunnel_destinations_total: usize::MAX,
4785 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
4786 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
4787 announce_table_ttl: Duration::from_secs(
4788 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
4789 ),
4790 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
4791 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
4792 interface_writer_queue_capacity:
4793 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
4794 announce_rate_defaults: AnnounceRateDefaults::default(),
4795 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
4796 ),
4797 #[cfg(feature = "iface-backbone")]
4798 backbone_peer_pool: None,
4799 announce_sig_cache_enabled: true,
4800 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
4801 announce_sig_cache_ttl: Duration::from_secs(
4802 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
4803 ),
4804 registry: None,
4805 #[cfg(feature = "hooks")]
4806 provider_bridge: None,
4807 },
4808 Box::new(NoopCallbacks),
4809 )
4810 .unwrap();
4811 let _ = hash;
4813 node.shutdown();
4814 }
4815
4816 #[test]
4817 fn start_generates_identity() {
4818 let node = RnsNode::start(
4819 NodeConfig {
4820 panic_on_interface_error: false,
4821 transport_enabled: false,
4822 static_transport_identity: false,
4823 local_hops_delta: false,
4824 identity: None,
4825 interfaces: vec![],
4826 share_instance: false,
4827 instance_name: "default".into(),
4828 shared_instance_port: 37428,
4829 rpc_port: 0,
4830 cache_dir: None,
4831 ratchet_store: None,
4832 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
4833 management: Default::default(),
4834 probe_port: None,
4835 probe_addrs: vec![],
4836 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
4837 direct_connect_policy: Default::default(),
4838 device: None,
4839 underlay_mark: None,
4840 hooks: Vec::new(),
4841 discover_interfaces: false,
4842 autoconnect_interface_mode: None,
4843 autoconnect_interface_gravity: 0,
4844 autoconnect_announces_to_internal: false,
4845 autoconnect_unverified_implementations: false,
4846 discovery_required_value: None,
4847 respond_to_probes: false,
4848 prefer_shorter_path: false,
4849 max_paths_per_destination: 1,
4850 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
4851 packet_hashlist_allocation:
4852 rns_core::transport::types::PacketHashlistAllocation::Eager,
4853 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
4854 max_path_destinations: usize::MAX,
4855 max_tunnel_destinations_total: usize::MAX,
4856 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
4857 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
4858 announce_table_ttl: Duration::from_secs(
4859 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
4860 ),
4861 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
4862 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
4863 interface_writer_queue_capacity:
4864 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
4865 announce_rate_defaults: AnnounceRateDefaults::default(),
4866 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
4867 ),
4868 #[cfg(feature = "iface-backbone")]
4869 backbone_peer_pool: None,
4870 announce_sig_cache_enabled: true,
4871 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
4872 announce_sig_cache_ttl: Duration::from_secs(
4873 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
4874 ),
4875 registry: None,
4876 #[cfg(feature = "hooks")]
4877 provider_bridge: None,
4878 },
4879 Box::new(NoopCallbacks),
4880 )
4881 .unwrap();
4882 node.shutdown();
4884 }
4885
4886 #[test]
4887 fn from_config_creates_identity() {
4888 let dir = std::env::temp_dir().join(format!("rns-test-fc-{}", std::process::id()));
4889 let _ = fs::remove_dir_all(&dir);
4890 fs::create_dir_all(&dir).unwrap();
4891
4892 fs::write(
4894 dir.join("config"),
4895 "[reticulum]\nenable_transport = False\n",
4896 )
4897 .unwrap();
4898
4899 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
4900
4901 assert!(dir.join("storage/identities/identity").exists());
4903
4904 node.shutdown();
4905 let _ = fs::remove_dir_all(&dir);
4906 }
4907
4908 #[test]
4909 fn private_config_keeps_state_out_of_config_directory() {
4910 let root = tempfile::tempdir().unwrap();
4911 let config_dir = root.path().join("config");
4912 let state_dir = root.path().join("state");
4913 fs::create_dir_all(&config_dir).unwrap();
4914 fs::write(
4915 config_dir.join("config"),
4916 "[reticulum]\nenable_transport = False\nshare_instance = Yes\n",
4917 )
4918 .unwrap();
4919 let node =
4920 RnsNode::from_private_config(&config_dir, &state_dir, None, Box::new(NoopCallbacks))
4921 .unwrap();
4922 assert!(state_dir.join("storage/identities/identity").exists());
4923 assert!(!config_dir.join("storage").exists());
4924 node.shutdown();
4925 }
4926
4927 #[test]
4928 fn from_config_loads_identity() {
4929 let dir = std::env::temp_dir().join(format!("rns-test-fl-{}", std::process::id()));
4930 let _ = fs::remove_dir_all(&dir);
4931 fs::create_dir_all(dir.join("storage/identities")).unwrap();
4932
4933 let identity = Identity::new(&mut OsRng);
4935 let hash = *identity.hash();
4936 storage::save_identity(&identity, &dir.join("storage/identities/identity")).unwrap();
4937
4938 fs::write(
4939 dir.join("config"),
4940 "[reticulum]\nenable_transport = False\n",
4941 )
4942 .unwrap();
4943
4944 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
4945
4946 let loaded = storage::load_identity(&dir.join("storage/identities/identity")).unwrap();
4948 assert_eq!(*loaded.hash(), hash);
4949
4950 node.shutdown();
4951 let _ = fs::remove_dir_all(&dir);
4952 }
4953
4954 #[test]
4955 fn from_config_tcp_server() {
4956 let dir = std::env::temp_dir().join(format!("rns-test-fts-{}", std::process::id()));
4957 let _ = fs::remove_dir_all(&dir);
4958 fs::create_dir_all(&dir).unwrap();
4959
4960 let port = crate::test_support::port();
4961
4962 let config = format!(
4963 r#"
4964[reticulum]
4965enable_transport = False
4966
4967[interfaces]
4968 [[Test TCP Server]]
4969 type = TCPServerInterface
4970 listen_ip = 127.0.0.1
4971 listen_port = {}
4972"#,
4973 port
4974 );
4975
4976 fs::write(dir.join("config"), config).unwrap();
4977
4978 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
4979
4980 thread::sleep(Duration::from_millis(100));
4982
4983 let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
4985
4986 node.shutdown();
4987 let _ = fs::remove_dir_all(&dir);
4988 }
4989
4990 #[test]
4991 fn from_config_starts_rpc_when_share_instance_enabled() {
4992 let dir = std::env::temp_dir().join(format!("rns-test-rpc-{}", std::process::id()));
4993 let _ = fs::remove_dir_all(&dir);
4994 fs::create_dir_all(&dir).unwrap();
4995
4996 let rpc_port = crate::test_support::port();
4997
4998 let config = format!(
4999 r#"
5000[reticulum]
5001enable_transport = False
5002share_instance = Yes
5003instance_control_port = {}
5004
5005[interfaces]
5006"#,
5007 rpc_port
5008 );
5009
5010 fs::write(dir.join("config"), config).unwrap();
5011
5012 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
5013
5014 thread::sleep(Duration::from_millis(100));
5015
5016 let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();
5017
5018 node.shutdown();
5019 let _ = fs::remove_dir_all(&dir);
5020 }
5021
5022 #[test]
5023 fn from_config_starts_rpc_when_transport_enabled() {
5024 let dir =
5025 std::env::temp_dir().join(format!("rns-test-rpc-transport-{}", std::process::id()));
5026 let _ = fs::remove_dir_all(&dir);
5027 fs::create_dir_all(&dir).unwrap();
5028
5029 let rpc_port = crate::test_support::port();
5030
5031 let config = format!(
5032 r#"
5033[reticulum]
5034enable_transport = True
5035share_instance = Yes
5036instance_control_port = {}
5037
5038[interfaces]
5039"#,
5040 rpc_port
5041 );
5042
5043 fs::write(dir.join("config"), config).unwrap();
5044
5045 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
5046
5047 thread::sleep(Duration::from_millis(100));
5048
5049 let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();
5050
5051 node.shutdown();
5052 let _ = fs::remove_dir_all(&dir);
5053 }
5054
5055 #[test]
5056 fn from_config_starts_rpc_when_tcp_client_is_unreachable() {
5057 let dir =
5058 std::env::temp_dir().join(format!("rns-test-rpc-unreachable-{}", std::process::id()));
5059 let _ = fs::remove_dir_all(&dir);
5060 fs::create_dir_all(&dir).unwrap();
5061
5062 let rpc_port = crate::test_support::port();
5063 let unreachable_port = crate::test_support::port();
5064
5065 let config = format!(
5066 r#"
5067[reticulum]
5068enable_transport = True
5069share_instance = Yes
5070instance_control_port = {}
5071
5072[interfaces]
5073 [[Unreachable Upstream]]
5074 type = TCPClientInterface
5075 target_host = 127.0.0.1
5076 target_port = {}
5077"#,
5078 rpc_port, unreachable_port
5079 );
5080
5081 fs::write(dir.join("config"), config).unwrap();
5082
5083 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
5084
5085 thread::sleep(Duration::from_millis(100));
5086
5087 let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();
5088
5089 node.shutdown();
5090 let _ = fs::remove_dir_all(&dir);
5091 }
5092
5093 #[test]
5094 fn test_parse_interface_mode() {
5095 use rns_core::constants::*;
5096
5097 assert_eq!(parse_interface_mode("full"), MODE_FULL);
5098 assert_eq!(parse_interface_mode("Full"), MODE_FULL);
5099 assert_eq!(parse_interface_mode("access_point"), MODE_ACCESS_POINT);
5100 assert_eq!(parse_interface_mode("accesspoint"), MODE_ACCESS_POINT);
5101 assert_eq!(parse_interface_mode("ap"), MODE_ACCESS_POINT);
5102 assert_eq!(parse_interface_mode("AP"), MODE_ACCESS_POINT);
5103 assert_eq!(parse_interface_mode("pointtopoint"), MODE_POINT_TO_POINT);
5104 assert_eq!(parse_interface_mode("ptp"), MODE_POINT_TO_POINT);
5105 assert_eq!(parse_interface_mode("roaming"), MODE_ROAMING);
5106 assert_eq!(parse_interface_mode("boundary"), MODE_BOUNDARY);
5107 assert_eq!(parse_interface_mode("gateway"), MODE_GATEWAY);
5108 assert_eq!(parse_interface_mode("gw"), MODE_GATEWAY);
5109 assert_eq!(parse_interface_mode("internal"), MODE_INTERNAL);
5110 assert_eq!(parse_interface_mode("Internal"), MODE_INTERNAL);
5111 assert_eq!(parse_interface_mode("invalid"), MODE_FULL);
5113 }
5114
5115 #[test]
5116 fn discovery_mode_normalization_matrix() {
5117 use rns_core::constants::*;
5118
5119 let interface_types = [
5120 "RNodeInterface",
5121 "RNodeMultiInterface",
5122 "BackboneInterface",
5123 "TCPServerInterface",
5124 "AutoInterface",
5125 ];
5126 let modes = [
5127 MODE_FULL,
5128 MODE_ACCESS_POINT,
5129 MODE_POINT_TO_POINT,
5130 MODE_ROAMING,
5131 MODE_BOUNDARY,
5132 MODE_GATEWAY,
5133 MODE_INTERNAL,
5134 ];
5135
5136 for interface_type in interface_types {
5137 for configured_mode in modes {
5138 for discoverable in [false, true] {
5139 for ignore_warnings in [false, true] {
5140 let expected = if !discoverable
5141 || ignore_warnings
5142 || matches!(
5143 configured_mode,
5144 MODE_ACCESS_POINT | MODE_GATEWAY | MODE_INTERNAL
5145 ) {
5146 configured_mode
5147 } else if matches!(interface_type, "RNodeInterface" | "RNodeMultiInterface")
5148 {
5149 MODE_ACCESS_POINT
5150 } else {
5151 MODE_GATEWAY
5152 };
5153 assert_eq!(
5154 normalize_discovery_mode(
5155 interface_type,
5156 configured_mode,
5157 discoverable,
5158 ignore_warnings,
5159 ),
5160 expected,
5161 "type={interface_type}, mode={configured_mode}, discoverable={discoverable}, ignore={ignore_warnings}",
5162 );
5163 }
5164 }
5165 }
5166 }
5167 }
5168
5169 #[test]
5170 fn to_node_config_serial() {
5171 let dir = std::env::temp_dir().join(format!("rns-test-serial-{}", std::process::id()));
5175 let _ = fs::remove_dir_all(&dir);
5176 fs::create_dir_all(&dir).unwrap();
5177
5178 let config = r#"
5179[reticulum]
5180enable_transport = False
5181
5182[interfaces]
5183 [[Test Serial Port]]
5184 type = SerialInterface
5185 port = /dev/nonexistent_rns_test_serial
5186 speed = 115200
5187 databits = 8
5188 parity = E
5189 stopbits = 1
5190 interface_mode = ptp
5191 networkname = testnet
5192"#;
5193 fs::write(dir.join("config"), config).unwrap();
5194
5195 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
5197 .expect("Config should parse; interface failure is non-fatal");
5198 node.shutdown();
5199
5200 let _ = fs::remove_dir_all(&dir);
5201 }
5202
5203 #[test]
5204 fn to_node_config_kiss() {
5205 let dir = std::env::temp_dir().join(format!("rns-test-kiss-{}", std::process::id()));
5207 let _ = fs::remove_dir_all(&dir);
5208 fs::create_dir_all(&dir).unwrap();
5209
5210 let config = r#"
5211[reticulum]
5212enable_transport = False
5213
5214[interfaces]
5215 [[Test KISS TNC]]
5216 type = KISSInterface
5217 port = /dev/nonexistent_rns_test_kiss
5218 speed = 9600
5219 preamble = 500
5220 txtail = 30
5221 persistence = 128
5222 slottime = 40
5223 flow_control = True
5224 id_interval = 600
5225 id_callsign = TEST0
5226 interface_mode = full
5227 passphrase = secretkey
5228"#;
5229 fs::write(dir.join("config"), config).unwrap();
5230
5231 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
5233 .expect("Config should parse; interface failure is non-fatal");
5234 node.shutdown();
5235
5236 let _ = fs::remove_dir_all(&dir);
5237 }
5238
5239 #[test]
5240 fn config_parser_accepts_recursive_prs_interface_option() {
5241 let config = r#"
5242[reticulum]
5243enable_transport = True
5244
5245[interfaces]
5246 [[Recursive Client]]
5247 type = TCPClientInterface
5248 target_host = 127.0.0.1
5249 target_port = 4242
5250 interface_mode = full
5251 recursive_prs = yes
5252
5253 [[Default Client]]
5254 type = TCPClientInterface
5255 target_host = 127.0.0.1
5256 target_port = 4243
5257 interface_mode = gateway
5258"#;
5259
5260 let parsed = config::parse(config).unwrap();
5261
5262 assert_eq!(parsed.interfaces.len(), 2);
5263 assert_eq!(parsed.interfaces[0].name, "Recursive Client");
5264 assert_eq!(parsed.interfaces[0].mode, "full");
5265 assert_eq!(
5266 parsed.interfaces[0]
5267 .params
5268 .get("recursive_prs")
5269 .map(String::as_str),
5270 Some("yes")
5271 );
5272 assert_eq!(parsed.interfaces[1].name, "Default Client");
5273 assert_eq!(parsed.interfaces[1].mode, "gateway");
5274 assert!(!parsed.interfaces[1].params.contains_key("recursive_prs"));
5275 }
5276
5277 #[test]
5278 fn config_parser_accepts_announces_from_internal_interface_option() {
5279 let config = r#"
5280[reticulum]
5281enable_transport = True
5282
5283[interfaces]
5284 [[Filtered Client]]
5285 type = TCPClientInterface
5286 target_host = 127.0.0.1
5287 target_port = 4242
5288 interface_mode = gateway
5289 announces_from_internal = no
5290
5291 [[Default Client]]
5292 type = TCPClientInterface
5293 target_host = 127.0.0.1
5294 target_port = 4243
5295 interface_mode = full
5296"#;
5297
5298 let parsed = config::parse(config).unwrap();
5299
5300 assert_eq!(parsed.interfaces.len(), 2);
5301 assert_eq!(parsed.interfaces[0].name, "Filtered Client");
5302 assert_eq!(parsed.interfaces[0].mode, "gateway");
5303 assert_eq!(
5304 parsed.interfaces[0]
5305 .params
5306 .get("announces_from_internal")
5307 .map(String::as_str),
5308 Some("no")
5309 );
5310 assert_eq!(parsed.interfaces[1].name, "Default Client");
5311 assert_eq!(parsed.interfaces[1].mode, "full");
5312 assert!(!parsed.interfaces[1]
5313 .params
5314 .contains_key("announces_from_internal"));
5315 }
5316
5317 #[test]
5318 fn config_parser_preserves_announces_to_internal_tristate() {
5319 let config = r#"
5320[interfaces]
5321 [[Allowed Boundary]]
5322 type = TCPClientInterface
5323 target_host = 127.0.0.1
5324 target_port = 4242
5325 announces_to_internal = yes
5326
5327 [[Blocked Boundary]]
5328 type = TCPClientInterface
5329 target_host = 127.0.0.1
5330 target_port = 4243
5331 announces_to_internal = no
5332
5333 [[Default Boundary]]
5334 type = TCPClientInterface
5335 target_host = 127.0.0.1
5336 target_port = 4244
5337"#;
5338
5339 let parsed = config::parse(config).unwrap();
5340 let parsed_value = |index: usize| {
5341 parsed.interfaces[index]
5342 .params
5343 .get("announces_to_internal")
5344 .and_then(|value| config::parse_bool_pub(value))
5345 };
5346
5347 assert_eq!(parsed_value(0), Some(true));
5348 assert_eq!(parsed_value(1), Some(false));
5349 assert_eq!(parsed_value(2), None);
5350 }
5351
5352 #[test]
5353 fn interface_gravity_parser_accepts_signed_values_and_uses_global_default() {
5354 assert_eq!(parse_interface_gravity(Some("7"), -2), 7);
5355 assert_eq!(parse_interface_gravity(Some("-4"), 2), -4);
5356 assert_eq!(parse_interface_gravity(None, -6), -6);
5357 }
5358
5359 #[test]
5360 fn config_parser_accepts_local_hops_delta_option() {
5361 let cfg = r#"
5362[reticulum]
5363local_hops_delta = yes
5364"#;
5365 let parsed = config::parse(cfg).unwrap();
5366 assert!(parsed.reticulum.local_hops_delta);
5367 }
5368
5369 #[test]
5370 fn config_parser_accepts_static_transport_identity_option() {
5371 let config = r#"
5372[reticulum]
5373enable_transport = false
5374static_transport_identity = yes
5375"#;
5376
5377 let parsed = config::parse(config).unwrap();
5378
5379 assert!(!parsed.reticulum.enable_transport);
5380 assert!(parsed.reticulum.static_transport_identity);
5381 }
5382
5383 #[test]
5384 fn non_transport_node_uses_ephemeral_transport_identity_by_default() {
5385 let identity = Identity::new(&mut rns_crypto::FixedRng::new(&[0x41; 32]));
5386 let stored_hash = *identity.hash();
5387 let node = RnsNode::start(
5388 NodeConfig {
5389 transport_enabled: false,
5390 static_transport_identity: false,
5391 local_hops_delta: false,
5392 identity: Some(identity),
5393 share_instance: false,
5394 ..Default::default()
5395 },
5396 Box::new(NoopCallbacks),
5397 )
5398 .unwrap();
5399
5400 let response = node.query(QueryRequest::TransportIdentity).unwrap();
5401 let QueryResponse::TransportIdentity(Some(hash)) = response else {
5402 panic!("transport identity query returned {response:?}");
5403 };
5404 assert_ne!(hash, stored_hash);
5405
5406 node.shutdown();
5407 }
5408
5409 #[test]
5410 fn non_transport_node_can_use_static_transport_identity() {
5411 let identity = Identity::new(&mut rns_crypto::FixedRng::new(&[0x42; 32]));
5412 let stored_hash = *identity.hash();
5413 let node = RnsNode::start(
5414 NodeConfig {
5415 transport_enabled: false,
5416 static_transport_identity: true,
5417 local_hops_delta: false,
5418 identity: Some(identity),
5419 share_instance: false,
5420 ..Default::default()
5421 },
5422 Box::new(NoopCallbacks),
5423 )
5424 .unwrap();
5425
5426 let response = node.query(QueryRequest::TransportIdentity).unwrap();
5427 let QueryResponse::TransportIdentity(Some(hash)) = response else {
5428 panic!("transport identity query returned {response:?}");
5429 };
5430 assert_eq!(hash, stored_hash);
5431
5432 node.shutdown();
5433 }
5434
5435 #[test]
5436 fn discoverable_interface_forces_static_transport_identity_on_non_transport() {
5437 assert!(!effective_static_transport_identity(false, false, false));
5438 assert!(effective_static_transport_identity(true, false, false));
5439 assert!(!effective_static_transport_identity(false, true, false));
5440 assert!(effective_static_transport_identity(false, false, true));
5441 assert!(effective_static_transport_identity(true, false, true));
5442 }
5443
5444 #[test]
5445 fn test_extract_ifac_config() {
5446 use std::collections::HashMap;
5447
5448 let params: HashMap<String, String> = HashMap::new();
5450 assert!(extract_ifac_config(¶ms, 16).is_none());
5451
5452 let mut params = HashMap::new();
5454 params.insert("networkname".into(), "testnet".into());
5455 let ifac = extract_ifac_config(¶ms, 16).unwrap();
5456 assert_eq!(ifac.netname.as_deref(), Some("testnet"));
5457 assert!(ifac.netkey.is_none());
5458 assert_eq!(ifac.size, 16);
5459
5460 let mut params = HashMap::new();
5462 params.insert("passphrase".into(), "secret".into());
5463 params.insert("ifac_size".into(), "64".into()); let ifac = extract_ifac_config(¶ms, 16).unwrap();
5465 assert!(ifac.netname.is_none());
5466 assert_eq!(ifac.netkey.as_deref(), Some("secret"));
5467 assert_eq!(ifac.size, 8);
5468
5469 let mut params = HashMap::new();
5471 params.insert("network_name".into(), "mynet".into());
5472 params.insert("pass_phrase".into(), "mykey".into());
5473 let ifac = extract_ifac_config(¶ms, 8).unwrap();
5474 assert_eq!(ifac.netname.as_deref(), Some("mynet"));
5475 assert_eq!(ifac.netkey.as_deref(), Some("mykey"));
5476 assert_eq!(ifac.size, 8);
5477
5478 let mut params = HashMap::new();
5480 params.insert("networkname".into(), "None".into());
5481 params.insert("passphrase".into(), "None".into());
5482 assert!(extract_ifac_config(¶ms, 16).is_none());
5483
5484 let mut params = HashMap::new();
5486 params.insert("networkname".into(), String::new());
5487 params.insert("passphrase".into(), "secret".into());
5488 let ifac = extract_ifac_config(¶ms, 16).unwrap();
5489 assert!(ifac.netname.is_none());
5490 assert_eq!(ifac.netkey.as_deref(), Some("secret"));
5491 }
5492
5493 #[test]
5494 fn to_node_config_rnode() {
5495 let dir = std::env::temp_dir().join(format!("rns-test-rnode-{}", std::process::id()));
5498 let _ = fs::remove_dir_all(&dir);
5499 fs::create_dir_all(&dir).unwrap();
5500
5501 let config = r#"
5502[reticulum]
5503enable_transport = False
5504
5505[interfaces]
5506 [[Test RNode]]
5507 type = RNodeInterface
5508 port = /dev/nonexistent_rns_test_rnode
5509 frequency = 867200000
5510 bandwidth = 125000
5511 txpower = 7
5512 spreadingfactor = 8
5513 codingrate = 5
5514 flow_control = True
5515 st_alock = 5.0
5516 lt_alock = 2.5
5517 interface_mode = full
5518 networkname = testnet
5519"#;
5520 fs::write(dir.join("config"), config).unwrap();
5521
5522 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
5524 .expect("Config should parse; interface failure is non-fatal");
5525 node.shutdown();
5526
5527 let _ = fs::remove_dir_all(&dir);
5528 }
5529
5530 #[test]
5531 fn to_node_config_pipe() {
5532 let dir = std::env::temp_dir().join(format!("rns-test-pipe-{}", std::process::id()));
5535 let _ = fs::remove_dir_all(&dir);
5536 fs::create_dir_all(&dir).unwrap();
5537
5538 let config = r#"
5539[reticulum]
5540enable_transport = False
5541
5542[interfaces]
5543 [[Test Pipe]]
5544 type = PipeInterface
5545 command = cat
5546 respawn_delay = 5000
5547 interface_mode = full
5548"#;
5549 fs::write(dir.join("config"), config).unwrap();
5550
5551 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
5552 node.shutdown();
5554
5555 let _ = fs::remove_dir_all(&dir);
5556 }
5557
5558 #[test]
5559 fn to_node_config_backbone() {
5560 let dir = std::env::temp_dir().join(format!("rns-test-backbone-{}", std::process::id()));
5562 let _ = fs::remove_dir_all(&dir);
5563 fs::create_dir_all(&dir).unwrap();
5564
5565 let port = crate::test_support::port();
5566
5567 let config = format!(
5568 r#"
5569[reticulum]
5570enable_transport = False
5571
5572[interfaces]
5573 [[Test Backbone]]
5574 type = BackboneInterface
5575 listen_ip = 127.0.0.1
5576 listen_port = {}
5577 interface_mode = full
5578"#,
5579 port
5580 );
5581
5582 fs::write(dir.join("config"), config).unwrap();
5583
5584 let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
5585
5586 thread::sleep(Duration::from_millis(100));
5588
5589 {
5591 let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
5592 }
5594
5595 thread::sleep(Duration::from_millis(50));
5597
5598 node.shutdown();
5599 let _ = fs::remove_dir_all(&dir);
5600 }
5601
5602 #[test]
5603 fn rnode_config_defaults() {
5604 use crate::interface::rnode::{RNodeConfig, RNodeSubConfig};
5605
5606 let config = RNodeConfig::default();
5607 assert_eq!(config.speed, 115200);
5608 assert!(config.subinterfaces.is_empty());
5609 assert!(config.id_interval.is_none());
5610 assert!(config.id_callsign.is_none());
5611
5612 let sub = RNodeSubConfig {
5613 name: "test".into(),
5614 vport: 0,
5615 outgoing: true,
5616 frequency: 868_000_000,
5617 bandwidth: 125_000,
5618 txpower: 7,
5619 spreading_factor: 8,
5620 coding_rate: 5,
5621 flow_control: false,
5622 st_alock: None,
5623 lt_alock: None,
5624 };
5625 assert_eq!(sub.frequency, 868_000_000);
5626 assert_eq!(sub.bandwidth, 125_000);
5627 assert!(!sub.flow_control);
5628 }
5629
5630 #[test]
5635 fn announce_builds_valid_packet() {
5636 let identity = Identity::new(&mut OsRng);
5637 let identity_hash = rns_core::types::IdentityHash(*identity.hash());
5638
5639 let node = RnsNode::start(
5640 NodeConfig {
5641 panic_on_interface_error: false,
5642 transport_enabled: false,
5643 static_transport_identity: false,
5644 local_hops_delta: false,
5645 identity: None,
5646 interfaces: vec![],
5647 share_instance: false,
5648 instance_name: "default".into(),
5649 shared_instance_port: 37428,
5650 rpc_port: 0,
5651 cache_dir: None,
5652 ratchet_store: None,
5653 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
5654 management: Default::default(),
5655 probe_port: None,
5656 probe_addrs: vec![],
5657 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
5658 direct_connect_policy: Default::default(),
5659 device: None,
5660 underlay_mark: None,
5661 hooks: Vec::new(),
5662 discover_interfaces: false,
5663 autoconnect_interface_mode: None,
5664 autoconnect_interface_gravity: 0,
5665 autoconnect_announces_to_internal: false,
5666 autoconnect_unverified_implementations: false,
5667 discovery_required_value: None,
5668 respond_to_probes: false,
5669 prefer_shorter_path: false,
5670 max_paths_per_destination: 1,
5671 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
5672 packet_hashlist_allocation:
5673 rns_core::transport::types::PacketHashlistAllocation::Eager,
5674 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
5675 max_path_destinations: usize::MAX,
5676 max_tunnel_destinations_total: usize::MAX,
5677 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
5678 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
5679 announce_table_ttl: Duration::from_secs(
5680 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
5681 ),
5682 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
5683 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
5684 interface_writer_queue_capacity:
5685 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
5686 announce_rate_defaults: AnnounceRateDefaults::default(),
5687 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
5688 ),
5689 #[cfg(feature = "iface-backbone")]
5690 backbone_peer_pool: None,
5691 announce_sig_cache_enabled: true,
5692 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
5693 announce_sig_cache_ttl: Duration::from_secs(
5694 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
5695 ),
5696 registry: None,
5697 #[cfg(feature = "hooks")]
5698 provider_bridge: None,
5699 },
5700 Box::new(NoopCallbacks),
5701 )
5702 .unwrap();
5703
5704 let dest = crate::destination::Destination::single_in("test", &["echo"], identity_hash);
5705
5706 node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
5708 .unwrap();
5709
5710 let result = node.announce_queued(&dest, &identity, Some(b"hello"));
5712 assert!(result.is_ok());
5713
5714 node.shutdown();
5715 }
5716
5717 #[test]
5718 fn has_path_and_hops_to() {
5719 let node = RnsNode::start(
5720 NodeConfig {
5721 panic_on_interface_error: false,
5722 transport_enabled: false,
5723 static_transport_identity: false,
5724 local_hops_delta: false,
5725 identity: None,
5726 interfaces: vec![],
5727 share_instance: false,
5728 instance_name: "default".into(),
5729 shared_instance_port: 37428,
5730 rpc_port: 0,
5731 cache_dir: None,
5732 ratchet_store: None,
5733 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
5734 management: Default::default(),
5735 probe_port: None,
5736 probe_addrs: vec![],
5737 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
5738 direct_connect_policy: Default::default(),
5739 device: None,
5740 underlay_mark: None,
5741 hooks: Vec::new(),
5742 discover_interfaces: false,
5743 autoconnect_interface_mode: None,
5744 autoconnect_interface_gravity: 0,
5745 autoconnect_announces_to_internal: false,
5746 autoconnect_unverified_implementations: false,
5747 discovery_required_value: None,
5748 respond_to_probes: false,
5749 prefer_shorter_path: false,
5750 max_paths_per_destination: 1,
5751 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
5752 packet_hashlist_allocation:
5753 rns_core::transport::types::PacketHashlistAllocation::Eager,
5754 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
5755 max_path_destinations: usize::MAX,
5756 max_tunnel_destinations_total: usize::MAX,
5757 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
5758 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
5759 announce_table_ttl: Duration::from_secs(
5760 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
5761 ),
5762 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
5763 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
5764 interface_writer_queue_capacity:
5765 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
5766 announce_rate_defaults: AnnounceRateDefaults::default(),
5767 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
5768 ),
5769 #[cfg(feature = "iface-backbone")]
5770 backbone_peer_pool: None,
5771 announce_sig_cache_enabled: true,
5772 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
5773 announce_sig_cache_ttl: Duration::from_secs(
5774 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
5775 ),
5776 registry: None,
5777 #[cfg(feature = "hooks")]
5778 provider_bridge: None,
5779 },
5780 Box::new(NoopCallbacks),
5781 )
5782 .unwrap();
5783
5784 let dh = rns_core::types::DestHash([0xAA; 16]);
5785
5786 assert!(!node.has_path(&dh).unwrap());
5788 assert_eq!(node.hops_to(&dh).unwrap(), None);
5789
5790 node.shutdown();
5791 }
5792
5793 #[test]
5794 fn recall_identity_none_when_unknown() {
5795 let node = RnsNode::start(
5796 NodeConfig {
5797 panic_on_interface_error: false,
5798 transport_enabled: false,
5799 static_transport_identity: false,
5800 local_hops_delta: false,
5801 identity: None,
5802 interfaces: vec![],
5803 share_instance: false,
5804 instance_name: "default".into(),
5805 shared_instance_port: 37428,
5806 rpc_port: 0,
5807 cache_dir: None,
5808 ratchet_store: None,
5809 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
5810 management: Default::default(),
5811 probe_port: None,
5812 probe_addrs: vec![],
5813 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
5814 direct_connect_policy: Default::default(),
5815 device: None,
5816 underlay_mark: None,
5817 hooks: Vec::new(),
5818 discover_interfaces: false,
5819 autoconnect_interface_mode: None,
5820 autoconnect_interface_gravity: 0,
5821 autoconnect_announces_to_internal: false,
5822 autoconnect_unverified_implementations: false,
5823 discovery_required_value: None,
5824 respond_to_probes: false,
5825 prefer_shorter_path: false,
5826 max_paths_per_destination: 1,
5827 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
5828 packet_hashlist_allocation:
5829 rns_core::transport::types::PacketHashlistAllocation::Eager,
5830 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
5831 max_path_destinations: usize::MAX,
5832 max_tunnel_destinations_total: usize::MAX,
5833 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
5834 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
5835 announce_table_ttl: Duration::from_secs(
5836 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
5837 ),
5838 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
5839 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
5840 interface_writer_queue_capacity:
5841 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
5842 announce_rate_defaults: AnnounceRateDefaults::default(),
5843 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
5844 ),
5845 #[cfg(feature = "iface-backbone")]
5846 backbone_peer_pool: None,
5847 announce_sig_cache_enabled: true,
5848 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
5849 announce_sig_cache_ttl: Duration::from_secs(
5850 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
5851 ),
5852 registry: None,
5853 #[cfg(feature = "hooks")]
5854 provider_bridge: None,
5855 },
5856 Box::new(NoopCallbacks),
5857 )
5858 .unwrap();
5859
5860 let dh = rns_core::types::DestHash([0xBB; 16]);
5861 assert!(node.recall_identity(&dh).unwrap().is_none());
5862
5863 node.shutdown();
5864 }
5865
5866 #[test]
5867 fn request_path_does_not_crash() {
5868 let node = RnsNode::start(
5869 NodeConfig {
5870 panic_on_interface_error: false,
5871 transport_enabled: false,
5872 static_transport_identity: false,
5873 local_hops_delta: false,
5874 identity: None,
5875 interfaces: vec![],
5876 share_instance: false,
5877 instance_name: "default".into(),
5878 shared_instance_port: 37428,
5879 rpc_port: 0,
5880 cache_dir: None,
5881 ratchet_store: None,
5882 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
5883 management: Default::default(),
5884 probe_port: None,
5885 probe_addrs: vec![],
5886 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
5887 direct_connect_policy: Default::default(),
5888 device: None,
5889 underlay_mark: None,
5890 hooks: Vec::new(),
5891 discover_interfaces: false,
5892 autoconnect_interface_mode: None,
5893 autoconnect_interface_gravity: 0,
5894 autoconnect_announces_to_internal: false,
5895 autoconnect_unverified_implementations: false,
5896 discovery_required_value: None,
5897 respond_to_probes: false,
5898 prefer_shorter_path: false,
5899 max_paths_per_destination: 1,
5900 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
5901 packet_hashlist_allocation:
5902 rns_core::transport::types::PacketHashlistAllocation::Eager,
5903 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
5904 max_path_destinations: usize::MAX,
5905 max_tunnel_destinations_total: usize::MAX,
5906 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
5907 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
5908 announce_table_ttl: Duration::from_secs(
5909 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
5910 ),
5911 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
5912 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
5913 interface_writer_queue_capacity:
5914 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
5915 announce_rate_defaults: AnnounceRateDefaults::default(),
5916 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
5917 ),
5918 #[cfg(feature = "iface-backbone")]
5919 backbone_peer_pool: None,
5920 announce_sig_cache_enabled: true,
5921 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
5922 announce_sig_cache_ttl: Duration::from_secs(
5923 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
5924 ),
5925 registry: None,
5926 #[cfg(feature = "hooks")]
5927 provider_bridge: None,
5928 },
5929 Box::new(NoopCallbacks),
5930 )
5931 .unwrap();
5932
5933 let dh = rns_core::types::DestHash([0xCC; 16]);
5934 assert!(node.request_path(&dh).is_ok());
5935
5936 thread::sleep(Duration::from_millis(50));
5938
5939 node.shutdown();
5940 }
5941
5942 #[test]
5943 fn create_link_returns_error_while_draining() {
5944 let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
5945
5946 node.begin_drain(Duration::from_secs(1)).unwrap();
5947 assert!(node.create_link([0xAB; 16], [0xCD; 32]).is_err());
5948
5949 node.shutdown();
5950 }
5951
5952 #[test]
5953 fn request_path_returns_error_while_draining() {
5954 let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
5955
5956 node.begin_drain(Duration::from_secs(1)).unwrap();
5957 assert!(node
5958 .request_path(&rns_core::types::DestHash([0xAB; 16]))
5959 .is_err());
5960
5961 node.shutdown();
5962 }
5963
5964 #[test]
5969 fn send_packet_returns_error_while_draining() {
5970 let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
5971 let dest = crate::destination::Destination::plain("drain-test", &["send"]);
5972
5973 node.begin_drain(Duration::from_secs(1)).unwrap();
5974 assert!(node.send_packet_queued(&dest, b"hello").is_err());
5975
5976 node.shutdown();
5977 }
5978
5979 #[test]
5980 fn send_packet_plain() {
5981 let node = RnsNode::start(
5982 NodeConfig {
5983 panic_on_interface_error: false,
5984 transport_enabled: false,
5985 static_transport_identity: false,
5986 local_hops_delta: false,
5987 identity: None,
5988 interfaces: vec![],
5989 share_instance: false,
5990 instance_name: "default".into(),
5991 shared_instance_port: 37428,
5992 rpc_port: 0,
5993 cache_dir: None,
5994 ratchet_store: None,
5995 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
5996 management: Default::default(),
5997 probe_port: None,
5998 probe_addrs: vec![],
5999 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
6000 direct_connect_policy: Default::default(),
6001 device: None,
6002 underlay_mark: None,
6003 hooks: Vec::new(),
6004 discover_interfaces: false,
6005 autoconnect_interface_mode: None,
6006 autoconnect_interface_gravity: 0,
6007 autoconnect_announces_to_internal: false,
6008 autoconnect_unverified_implementations: false,
6009 discovery_required_value: None,
6010 respond_to_probes: false,
6011 prefer_shorter_path: false,
6012 max_paths_per_destination: 1,
6013 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
6014 packet_hashlist_allocation:
6015 rns_core::transport::types::PacketHashlistAllocation::Eager,
6016 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
6017 max_path_destinations: usize::MAX,
6018 max_tunnel_destinations_total: usize::MAX,
6019 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
6020 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
6021 announce_table_ttl: Duration::from_secs(
6022 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
6023 ),
6024 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
6025 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
6026 interface_writer_queue_capacity:
6027 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
6028 announce_rate_defaults: AnnounceRateDefaults::default(),
6029 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
6030 ),
6031 #[cfg(feature = "iface-backbone")]
6032 backbone_peer_pool: None,
6033 announce_sig_cache_enabled: true,
6034 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
6035 announce_sig_cache_ttl: Duration::from_secs(
6036 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
6037 ),
6038 registry: None,
6039 #[cfg(feature = "hooks")]
6040 provider_bridge: None,
6041 },
6042 Box::new(NoopCallbacks),
6043 )
6044 .unwrap();
6045
6046 let dest = crate::destination::Destination::plain("test", &["echo"]);
6047 let result = node.send_packet_queued(&dest, b"hello world");
6048 assert!(result.is_ok());
6049
6050 let packet_hash = result.unwrap();
6051 assert_ne!(packet_hash.0, [0u8; 32]);
6053
6054 thread::sleep(Duration::from_millis(50));
6056
6057 node.shutdown();
6058 }
6059
6060 #[test]
6061 fn send_packet_single_requires_public_key() {
6062 let node = RnsNode::start(
6063 NodeConfig {
6064 panic_on_interface_error: false,
6065 transport_enabled: false,
6066 static_transport_identity: false,
6067 local_hops_delta: false,
6068 identity: None,
6069 interfaces: vec![],
6070 share_instance: false,
6071 instance_name: "default".into(),
6072 shared_instance_port: 37428,
6073 rpc_port: 0,
6074 cache_dir: None,
6075 ratchet_store: None,
6076 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
6077 management: Default::default(),
6078 probe_port: None,
6079 probe_addrs: vec![],
6080 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
6081 direct_connect_policy: Default::default(),
6082 device: None,
6083 underlay_mark: None,
6084 hooks: Vec::new(),
6085 discover_interfaces: false,
6086 autoconnect_interface_mode: None,
6087 autoconnect_interface_gravity: 0,
6088 autoconnect_announces_to_internal: false,
6089 autoconnect_unverified_implementations: false,
6090 discovery_required_value: None,
6091 respond_to_probes: false,
6092 prefer_shorter_path: false,
6093 max_paths_per_destination: 1,
6094 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
6095 packet_hashlist_allocation:
6096 rns_core::transport::types::PacketHashlistAllocation::Eager,
6097 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
6098 max_path_destinations: usize::MAX,
6099 max_tunnel_destinations_total: usize::MAX,
6100 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
6101 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
6102 announce_table_ttl: Duration::from_secs(
6103 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
6104 ),
6105 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
6106 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
6107 interface_writer_queue_capacity:
6108 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
6109 announce_rate_defaults: AnnounceRateDefaults::default(),
6110 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
6111 ),
6112 #[cfg(feature = "iface-backbone")]
6113 backbone_peer_pool: None,
6114 announce_sig_cache_enabled: true,
6115 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
6116 announce_sig_cache_ttl: Duration::from_secs(
6117 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
6118 ),
6119 registry: None,
6120 #[cfg(feature = "hooks")]
6121 provider_bridge: None,
6122 },
6123 Box::new(NoopCallbacks),
6124 )
6125 .unwrap();
6126
6127 let dest = crate::destination::Destination::single_in(
6129 "test",
6130 &["echo"],
6131 rns_core::types::IdentityHash([0x42; 16]),
6132 );
6133 let result = node.send_packet_queued(&dest, b"hello");
6134 assert!(result.is_err(), "single_in has no public_key, should fail");
6135
6136 node.shutdown();
6137 }
6138
6139 #[test]
6140 fn send_packet_single_encrypts() {
6141 let node = RnsNode::start(
6142 NodeConfig {
6143 panic_on_interface_error: false,
6144 transport_enabled: false,
6145 static_transport_identity: false,
6146 local_hops_delta: false,
6147 identity: None,
6148 interfaces: vec![],
6149 share_instance: false,
6150 instance_name: "default".into(),
6151 shared_instance_port: 37428,
6152 rpc_port: 0,
6153 cache_dir: None,
6154 ratchet_store: None,
6155 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
6156 management: Default::default(),
6157 probe_port: None,
6158 probe_addrs: vec![],
6159 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
6160 direct_connect_policy: Default::default(),
6161 device: None,
6162 underlay_mark: None,
6163 hooks: Vec::new(),
6164 discover_interfaces: false,
6165 autoconnect_interface_mode: None,
6166 autoconnect_interface_gravity: 0,
6167 autoconnect_announces_to_internal: false,
6168 autoconnect_unverified_implementations: false,
6169 discovery_required_value: None,
6170 respond_to_probes: false,
6171 prefer_shorter_path: false,
6172 max_paths_per_destination: 1,
6173 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
6174 packet_hashlist_allocation:
6175 rns_core::transport::types::PacketHashlistAllocation::Eager,
6176 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
6177 max_path_destinations: usize::MAX,
6178 max_tunnel_destinations_total: usize::MAX,
6179 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
6180 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
6181 announce_table_ttl: Duration::from_secs(
6182 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
6183 ),
6184 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
6185 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
6186 interface_writer_queue_capacity:
6187 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
6188 announce_rate_defaults: AnnounceRateDefaults::default(),
6189 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
6190 ),
6191 #[cfg(feature = "iface-backbone")]
6192 backbone_peer_pool: None,
6193 announce_sig_cache_enabled: true,
6194 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
6195 announce_sig_cache_ttl: Duration::from_secs(
6196 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
6197 ),
6198 registry: None,
6199 #[cfg(feature = "hooks")]
6200 provider_bridge: None,
6201 },
6202 Box::new(NoopCallbacks),
6203 )
6204 .unwrap();
6205
6206 let remote_identity = Identity::new(&mut OsRng);
6208 let recalled = crate::destination::AnnouncedIdentity {
6209 dest_hash: rns_core::types::DestHash([0xAA; 16]),
6210 identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
6211 public_key: remote_identity.get_public_key().unwrap(),
6212 app_data: None,
6213 hops: 1,
6214 received_at: 0.0,
6215 receiving_interface: rns_core::transport::types::InterfaceId(0),
6216 rssi: Some(-100),
6217 snr: Some(10.5),
6218 };
6219 let dest = crate::destination::Destination::single_out("test", &["echo"], &recalled);
6220
6221 let result = node.send_packet_queued(&dest, b"secret message");
6222 assert!(result.is_ok());
6223
6224 let packet_hash = result.unwrap();
6225 assert_ne!(packet_hash.0, [0u8; 32]);
6226
6227 thread::sleep(Duration::from_millis(50));
6228 node.shutdown();
6229 }
6230
6231 #[test]
6232 fn register_destination_with_proof_prove_all() {
6233 let node = RnsNode::start(
6234 NodeConfig {
6235 panic_on_interface_error: false,
6236 transport_enabled: false,
6237 static_transport_identity: false,
6238 local_hops_delta: false,
6239 identity: None,
6240 interfaces: vec![],
6241 share_instance: false,
6242 instance_name: "default".into(),
6243 shared_instance_port: 37428,
6244 rpc_port: 0,
6245 cache_dir: None,
6246 ratchet_store: None,
6247 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
6248 management: Default::default(),
6249 probe_port: None,
6250 probe_addrs: vec![],
6251 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
6252 direct_connect_policy: Default::default(),
6253 device: None,
6254 underlay_mark: None,
6255 hooks: Vec::new(),
6256 discover_interfaces: false,
6257 autoconnect_interface_mode: None,
6258 autoconnect_interface_gravity: 0,
6259 autoconnect_announces_to_internal: false,
6260 autoconnect_unverified_implementations: false,
6261 discovery_required_value: None,
6262 respond_to_probes: false,
6263 prefer_shorter_path: false,
6264 max_paths_per_destination: 1,
6265 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
6266 packet_hashlist_allocation:
6267 rns_core::transport::types::PacketHashlistAllocation::Eager,
6268 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
6269 max_path_destinations: usize::MAX,
6270 max_tunnel_destinations_total: usize::MAX,
6271 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
6272 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
6273 announce_table_ttl: Duration::from_secs(
6274 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
6275 ),
6276 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
6277 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
6278 interface_writer_queue_capacity:
6279 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
6280 announce_rate_defaults: AnnounceRateDefaults::default(),
6281 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
6282 ),
6283 #[cfg(feature = "iface-backbone")]
6284 backbone_peer_pool: None,
6285 announce_sig_cache_enabled: true,
6286 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
6287 announce_sig_cache_ttl: Duration::from_secs(
6288 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
6289 ),
6290 registry: None,
6291 #[cfg(feature = "hooks")]
6292 provider_bridge: None,
6293 },
6294 Box::new(NoopCallbacks),
6295 )
6296 .unwrap();
6297
6298 let identity = Identity::new(&mut OsRng);
6299 let ih = rns_core::types::IdentityHash(*identity.hash());
6300 let dest = crate::destination::Destination::single_in("echo", &["request"], ih)
6301 .set_proof_strategy(rns_core::types::ProofStrategy::ProveAll);
6302 let prv_key = identity.get_private_key().unwrap();
6303
6304 let result = node.register_destination_with_proof(&dest, Some(prv_key));
6305 assert!(result.is_ok());
6306
6307 thread::sleep(Duration::from_millis(50));
6309
6310 node.shutdown();
6311 }
6312
6313 #[test]
6314 fn register_destination_with_proof_prove_none() {
6315 let node = RnsNode::start(
6316 NodeConfig {
6317 panic_on_interface_error: false,
6318 transport_enabled: false,
6319 static_transport_identity: false,
6320 local_hops_delta: false,
6321 identity: None,
6322 interfaces: vec![],
6323 share_instance: false,
6324 instance_name: "default".into(),
6325 shared_instance_port: 37428,
6326 rpc_port: 0,
6327 cache_dir: None,
6328 ratchet_store: None,
6329 ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
6330 management: Default::default(),
6331 probe_port: None,
6332 probe_addrs: vec![],
6333 probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
6334 direct_connect_policy: Default::default(),
6335 device: None,
6336 underlay_mark: None,
6337 hooks: Vec::new(),
6338 discover_interfaces: false,
6339 autoconnect_interface_mode: None,
6340 autoconnect_interface_gravity: 0,
6341 autoconnect_announces_to_internal: false,
6342 autoconnect_unverified_implementations: false,
6343 discovery_required_value: None,
6344 respond_to_probes: false,
6345 prefer_shorter_path: false,
6346 max_paths_per_destination: 1,
6347 packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
6348 packet_hashlist_allocation:
6349 rns_core::transport::types::PacketHashlistAllocation::Eager,
6350 max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
6351 max_path_destinations: usize::MAX,
6352 max_tunnel_destinations_total: usize::MAX,
6353 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
6354 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
6355 announce_table_ttl: Duration::from_secs(
6356 rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
6357 ),
6358 announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
6359 driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
6360 interface_writer_queue_capacity:
6361 crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
6362 announce_rate_defaults: AnnounceRateDefaults::default(),
6363 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
6364 ),
6365 #[cfg(feature = "iface-backbone")]
6366 backbone_peer_pool: None,
6367 announce_sig_cache_enabled: true,
6368 announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
6369 announce_sig_cache_ttl: Duration::from_secs(
6370 rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
6371 ),
6372 registry: None,
6373 #[cfg(feature = "hooks")]
6374 provider_bridge: None,
6375 },
6376 Box::new(NoopCallbacks),
6377 )
6378 .unwrap();
6379
6380 let dest = crate::destination::Destination::plain("test", &["data"])
6382 .set_proof_strategy(rns_core::types::ProofStrategy::ProveNone);
6383
6384 let result = node.register_destination_with_proof(&dest, None);
6385 assert!(result.is_ok());
6386
6387 thread::sleep(Duration::from_millis(50));
6388 node.shutdown();
6389 }
6390}