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