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