1use std::collections::HashMap;
4use std::io;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant};
8
9use rns_core::packet::RawPacket;
10use rns_core::transport::announce_verify_queue::{AnnounceVerifyQueue, OverflowPolicy};
11use rns_core::transport::tables::PathEntry;
12use rns_core::transport::types::{InterfaceId, TransportAction, TransportConfig};
13use rns_core::transport::TransportEngine;
14use rns_crypto::{OsRng, Rng};
15
16#[cfg(feature = "hooks")]
17use crate::provider_bridge::ProviderBridge;
18#[cfg(feature = "hooks")]
19use rns_hooks::{create_hook_slots, EngineAccess, HookContext, HookManager, HookPoint, HookSlot};
20
21#[cfg(feature = "hooks")]
22use crate::event::BackbonePeerHookEvent;
23use crate::event::{
24 BackbonePeerPoolMemberStatus, BackbonePeerPoolStatus, BackbonePeerStateEntry, BlackholeInfo,
25 DrainStatus, Event, EventReceiver, InterfaceStatsResponse, KnownDestinationEntry,
26 LifecycleState, LocalDestinationEntry, NextHopResponse, PathTableEntry, QueryRequest,
27 QueryResponse, RateTableEntry, RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError,
28 RuntimeConfigErrorCode, RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat,
29};
30use crate::holepunch::orchestrator::{HolePunchManager, HolePunchManagerAction};
31use crate::ifac;
32#[cfg(all(feature = "iface-auto", test))]
33use crate::interface::auto::AutoRuntime;
34#[cfg(feature = "iface-auto")]
35use crate::interface::auto::AutoRuntimeConfigHandle;
36#[cfg(feature = "iface-backbone")]
37use crate::interface::backbone::{
38 start_client, BackboneClientConfig, BackboneClientRuntime, BackboneClientRuntimeConfigHandle,
39 BackbonePeerStateHandle, BackboneRuntimeConfigHandle,
40};
41#[cfg(all(feature = "iface-backbone", target_os = "linux", test))]
42use crate::interface::backbone::{BackboneAbuseConfig, BackboneServerRuntime};
43#[cfg(all(feature = "iface-i2p", test))]
44use crate::interface::i2p::I2pRuntime;
45#[cfg(feature = "iface-i2p")]
46use crate::interface::i2p::I2pRuntimeConfigHandle;
47#[cfg(all(feature = "iface-pipe", test))]
48use crate::interface::pipe::PipeRuntime;
49#[cfg(feature = "iface-pipe")]
50use crate::interface::pipe::PipeRuntimeConfigHandle;
51#[cfg(feature = "iface-rnode")]
52use crate::interface::rnode::{
53 validate_sub_config, RNodeRuntime, RNodeRuntimeConfigHandle, RNodeSubConfig,
54};
55#[cfg(feature = "iface-tcp")]
56use crate::interface::tcp::TcpClientRuntimeConfigHandle;
57#[cfg(all(feature = "iface-tcp", test))]
58use crate::interface::tcp_server::TcpServerRuntime;
59#[cfg(feature = "iface-tcp")]
60use crate::interface::tcp_server::TcpServerRuntimeConfigHandle;
61#[cfg(all(feature = "iface-udp", test))]
62use crate::interface::udp::UdpRuntime;
63#[cfg(feature = "iface-udp")]
64use crate::interface::udp::UdpRuntimeConfigHandle;
65use crate::interface::{InterfaceEntry, InterfaceStats};
66use crate::link_manager::{LinkManager, LinkManagerAction};
67use crate::time;
68
69const DEFAULT_KNOWN_DESTINATIONS_TTL: f64 = 48.0 * 60.0 * 60.0;
70const DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES: usize = 8192;
71const DEFAULT_RATE_LIMITER_TTL_SECS: f64 = 48.0 * 60.0 * 60.0;
72const DEFAULT_TICK_INTERVAL_MS: u64 = 1000;
73const DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS: u32 = 3600;
74const DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS: u32 = 3600;
75const DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE: usize = 10_000;
76const DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS: u32 = 3600;
77const DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS: f64 = 300.0;
78const DEFAULT_LINK_TEARDOWN_FLUSH: Duration = Duration::from_millis(150);
79const SEND_RETRY_BACKOFF_MIN: Duration = Duration::from_millis(25);
80const SEND_RETRY_BACKOFF_MAX: Duration = Duration::from_millis(1000);
81
82#[derive(Debug, Clone, Copy, PartialEq)]
84pub struct AnnounceRateDefaults {
85 pub target: Option<f64>,
86 pub penalty: f64,
87 pub grace: u32,
88}
89
90impl Default for AnnounceRateDefaults {
91 fn default() -> Self {
92 Self {
93 target: Some(3600.0),
94 penalty: 0.0,
95 grace: 5,
96 }
97 }
98}
99
100mod dispatch;
101mod events;
102mod lifecycle;
103mod queries;
104mod runtime_config;
105
106#[cfg(test)]
107mod tests;
108
109fn inject_transport_header(raw: &[u8], next_hop: &[u8; 16]) -> Vec<u8> {
110 if raw.len() < 18 {
111 return raw.to_vec();
112 }
113
114 let new_flags = (rns_core::constants::HEADER_2 << 6)
115 | (rns_core::constants::TRANSPORT_TRANSPORT << 4)
116 | (raw[0] & 0x0F);
117
118 let mut new_raw = Vec::with_capacity(raw.len() + 16);
119 new_raw.push(new_flags);
120 new_raw.push(raw[1]);
121 new_raw.extend_from_slice(next_hop);
122 new_raw.extend_from_slice(&raw[2..]);
123 new_raw
124}
125
126fn recover_mutex_guard<'a, T>(mutex: &'a Mutex<T>, label: &str) -> std::sync::MutexGuard<'a, T> {
127 match mutex.lock() {
128 Ok(guard) => guard,
129 Err(poisoned) => {
130 log::error!("recovering from poisoned mutex: {}", label);
131 poisoned.into_inner()
132 }
133 }
134}
135
136#[derive(Debug, Clone, Copy)]
137pub(crate) struct RuntimeConfigDefaults {
138 pub(crate) tick_interval_ms: u64,
139 pub(crate) known_destinations_ttl: f64,
140 pub(crate) rate_limiter_ttl_secs: f64,
141 pub(crate) known_destinations_cleanup_interval_ticks: u32,
142 pub(crate) announce_cache_cleanup_interval_ticks: u32,
143 pub(crate) announce_cache_cleanup_batch_size: usize,
144 pub(crate) discovery_cleanup_interval_ticks: u32,
145 pub(crate) management_announce_interval_secs: f64,
146 pub(crate) direct_connect_policy: crate::event::HolePunchPolicy,
147 #[cfg(feature = "hooks")]
148 pub(crate) provider_queue_max_events: usize,
149 #[cfg(feature = "hooks")]
150 pub(crate) provider_queue_max_bytes: usize,
151}
152
153#[cfg(feature = "iface-backbone")]
154#[derive(Debug, Clone)]
155pub(crate) struct BackboneDiscoveryRuntime {
156 pub(crate) discoverable: bool,
157 pub(crate) config: crate::discovery::DiscoveryConfig,
158 pub(crate) transport_enabled: bool,
159 pub(crate) ifac_netname: Option<String>,
160 pub(crate) ifac_netkey: Option<String>,
161}
162
163#[cfg(feature = "iface-backbone")]
164#[derive(Debug, Clone)]
165pub(crate) struct BackboneDiscoveryRuntimeHandle {
166 pub(crate) interface_name: String,
167 pub(crate) current: BackboneDiscoveryRuntime,
168 pub(crate) startup: BackboneDiscoveryRuntime,
169}
170
171#[cfg(feature = "iface-tcp")]
172#[derive(Debug, Clone)]
173pub(crate) struct TcpServerDiscoveryRuntime {
174 pub(crate) discoverable: bool,
175 pub(crate) config: crate::discovery::DiscoveryConfig,
176 pub(crate) transport_enabled: bool,
177 pub(crate) ifac_netname: Option<String>,
178 pub(crate) ifac_netkey: Option<String>,
179}
180
181#[cfg(feature = "iface-tcp")]
182#[derive(Debug, Clone)]
183pub(crate) struct TcpServerDiscoveryRuntimeHandle {
184 pub(crate) interface_name: String,
185 pub(crate) current: TcpServerDiscoveryRuntime,
186 pub(crate) startup: TcpServerDiscoveryRuntime,
187}
188
189#[derive(Debug, Clone, PartialEq, Eq)]
190pub(crate) struct IfacRuntimeConfig {
191 pub(crate) netname: Option<String>,
192 pub(crate) netkey: Option<String>,
193 pub(crate) size: usize,
194}
195
196#[cfg(feature = "iface-backbone")]
197impl BackboneDiscoveryRuntimeHandle {
198 pub(crate) fn from_parts(
199 interface_name: String,
200 startup_config: crate::discovery::DiscoveryConfig,
201 transport_enabled: bool,
202 ifac_netname: Option<String>,
203 ifac_netkey: Option<String>,
204 discoverable: bool,
205 ) -> Self {
206 let startup = BackboneDiscoveryRuntime {
207 discoverable,
208 config: startup_config,
209 transport_enabled,
210 ifac_netname,
211 ifac_netkey,
212 };
213 Self {
214 interface_name,
215 current: startup.clone(),
216 startup,
217 }
218 }
219}
220
221#[cfg(feature = "iface-tcp")]
222impl TcpServerDiscoveryRuntimeHandle {
223 pub(crate) fn from_parts(
224 interface_name: String,
225 startup_config: crate::discovery::DiscoveryConfig,
226 transport_enabled: bool,
227 ifac_netname: Option<String>,
228 ifac_netkey: Option<String>,
229 discoverable: bool,
230 ) -> Self {
231 let startup = TcpServerDiscoveryRuntime {
232 discoverable,
233 config: startup_config,
234 transport_enabled,
235 ifac_netname,
236 ifac_netkey,
237 };
238 Self {
239 interface_name,
240 current: startup.clone(),
241 startup,
242 }
243 }
244}
245
246impl IfacRuntimeConfig {
247 pub(crate) fn from_parts(netname: Option<String>, netkey: Option<String>, size: usize) -> Self {
248 Self {
249 netname,
250 netkey,
251 size,
252 }
253 }
254}
255
256#[cfg(feature = "iface-backbone")]
257#[derive(Debug, Clone)]
258pub struct BackbonePeerPoolSettings {
259 pub max_connected: usize,
260 pub failure_threshold: usize,
261 pub failure_window: Duration,
262 pub cooldown: Duration,
263}
264
265#[cfg(feature = "iface-backbone")]
266pub(crate) const BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY: u8 = 60;
267#[cfg(feature = "iface-backbone")]
268pub(crate) const BACKBONE_PEER_POOL_DISCOVERED_PRIORITY: u8 = 40;
269
270#[cfg(feature = "iface-backbone")]
271pub(crate) struct BackbonePeerPoolCandidateConfig {
272 pub(crate) client: BackboneClientConfig,
273 pub(crate) mode: u8,
274 pub(crate) recursive_prs: bool,
275 pub(crate) announces_from_internal: bool,
276 pub(crate) ingress_control: rns_core::transport::types::IngressControlConfig,
277 pub(crate) ifac_runtime: IfacRuntimeConfig,
278 pub(crate) ifac_enabled: bool,
279 pub(crate) interface_type_name: String,
280 pub(crate) source: BackbonePeerPoolCandidateSource,
281 pub(crate) priority: u8,
282 pub(crate) discovery: Option<BackbonePeerPoolDiscoveryCandidate>,
283}
284
285#[cfg(feature = "iface-backbone")]
286#[derive(Debug, Clone, Copy, PartialEq, Eq)]
287pub(crate) enum BackbonePeerPoolCandidateSource {
288 Configured,
289 Discovered,
290}
291
292#[cfg(feature = "iface-backbone")]
293impl BackbonePeerPoolCandidateSource {
294 fn as_str(self) -> &'static str {
295 match self {
296 Self::Configured => "configured",
297 Self::Discovered => "discovered",
298 }
299 }
300}
301
302#[cfg(feature = "iface-backbone")]
303#[derive(Debug, Clone)]
304pub(crate) struct BackbonePeerPoolDiscoveryCandidate {
305 pub(crate) discovery_hash: [u8; 32],
306 pub(crate) status: crate::discovery::DiscoveredStatus,
307 pub(crate) hops: u8,
308 pub(crate) stamp_value: u32,
309 pub(crate) last_heard: f64,
310}
311
312#[cfg(feature = "iface-backbone")]
313struct BackbonePeerPool {
314 settings: BackbonePeerPoolSettings,
315 candidates: Vec<BackbonePeerPoolCandidate>,
316}
317
318#[cfg(feature = "iface-backbone")]
319struct BackbonePeerPoolCandidate {
320 config: BackbonePeerPoolCandidateConfig,
321 active_id: Option<InterfaceId>,
322 failures: Vec<f64>,
323 retry_after: Option<f64>,
324 cooldown_until: Option<f64>,
325 last_error: Option<String>,
326}
327
328#[cfg(feature = "hooks")]
330struct EngineRef<'a> {
331 engine: &'a TransportEngine,
332 interfaces: &'a HashMap<InterfaceId, InterfaceEntry>,
333 link_manager: &'a LinkManager,
334 now: f64,
335}
336
337#[cfg(feature = "hooks")]
338impl<'a> EngineAccess for EngineRef<'a> {
339 fn has_path(&self, dest: &[u8; 16]) -> bool {
340 self.engine.has_path(dest)
341 }
342 fn hops_to(&self, dest: &[u8; 16]) -> Option<u8> {
343 self.engine.hops_to(dest)
344 }
345 fn next_hop(&self, dest: &[u8; 16]) -> Option<[u8; 16]> {
346 self.engine.next_hop(dest)
347 }
348 fn is_blackholed(&self, identity: &[u8; 16]) -> bool {
349 self.engine.is_blackholed(identity, self.now)
350 }
351 fn interface_name(&self, id: u64) -> Option<String> {
352 self.interfaces
353 .get(&InterfaceId(id))
354 .map(|e| e.info.name.clone())
355 }
356 fn interface_mode(&self, id: u64) -> Option<u8> {
357 self.interfaces.get(&InterfaceId(id)).map(|e| e.info.mode)
358 }
359 fn identity_hash(&self) -> Option<[u8; 16]> {
360 self.engine.identity_hash().copied()
361 }
362 fn announce_rate(&self, id: u64) -> Option<i32> {
363 self.interfaces
364 .get(&InterfaceId(id))
365 .map(|e| (e.stats.outgoing_announce_freq() * 1000.0) as i32)
366 }
367 fn link_state(&self, link_hash: &[u8; 16]) -> Option<u8> {
368 use rns_core::link::types::LinkState;
369 self.link_manager.link_state(link_hash).map(|s| match s {
370 LinkState::Pending => 0,
371 LinkState::Handshake => 1,
372 LinkState::Active => 2,
373 LinkState::Stale => 3,
374 LinkState::Closed => 4,
375 })
376 }
377}
378
379#[cfg(any(test, feature = "hooks"))]
384fn extract_dest_hash(raw: &[u8]) -> [u8; 16] {
385 let mut dest = [0u8; 16];
386 if raw.is_empty() {
387 return dest;
388 }
389 let is_header2 = raw[0] & 0x40 != 0;
390 let start = if is_header2 { 18 } else { 2 };
391 let end = start + 16;
392 if raw.len() >= end {
393 dest.copy_from_slice(&raw[start..end]);
394 }
395 dest
396}
397
398#[cfg(feature = "hooks")]
400fn run_hook_inner(
401 programs: &mut [rns_hooks::LoadedProgram],
402 hook_manager: &Option<HookManager>,
403 engine_access: &dyn EngineAccess,
404 ctx: &HookContext,
405 now: f64,
406 provider_events_enabled: bool,
407) -> Option<rns_hooks::ExecuteResult> {
408 if programs.is_empty() {
409 return None;
410 }
411 let mgr = hook_manager.as_ref()?;
412 mgr.run_chain_with_provider_events(programs, ctx, engine_access, now, provider_events_enabled)
413}
414
415#[cfg(feature = "hooks")]
416fn backbone_peer_hook_context(event: &BackbonePeerHookEvent) -> HookContext<'_> {
417 HookContext::BackbonePeer {
418 server_interface_id: event.server_interface_id.0,
419 peer_interface_id: event.peer_interface_id.map(|id| id.0),
420 peer_ip: event.peer_ip,
421 peer_port: event.peer_port,
422 connected_for: event.connected_for,
423 had_received_data: event.had_received_data,
424 penalty_level: event.penalty_level,
425 blacklist_for: event.blacklist_for,
426 }
427}
428
429#[cfg(feature = "hooks")]
431fn convert_injected_actions(actions: Vec<rns_hooks::ActionWire>) -> Vec<TransportAction> {
432 actions
433 .into_iter()
434 .map(|a| {
435 use rns_hooks::ActionWire;
436 match a {
437 ActionWire::SendOnInterface { interface, raw } => {
438 TransportAction::SendOnInterface {
439 interface: InterfaceId(interface),
440 raw: raw.into(),
441 }
442 }
443 ActionWire::BroadcastOnAllInterfaces {
444 raw,
445 exclude,
446 has_exclude,
447 } => TransportAction::BroadcastOnAllInterfaces {
448 raw: raw.into(),
449 exclude: if has_exclude != 0 {
450 Some(InterfaceId(exclude))
451 } else {
452 None
453 },
454 },
455 ActionWire::DeliverLocal {
456 destination_hash,
457 raw,
458 packet_hash,
459 receiving_interface,
460 } => TransportAction::DeliverLocal {
461 destination_hash,
462 raw: raw.into(),
463 packet_hash,
464 receiving_interface: InterfaceId(receiving_interface),
465 },
466 ActionWire::PathUpdated {
467 destination_hash,
468 hops,
469 next_hop,
470 interface,
471 } => TransportAction::PathUpdated {
472 destination_hash,
473 hops,
474 next_hop,
475 interface: InterfaceId(interface),
476 },
477 ActionWire::CacheAnnounce { packet_hash, raw } => TransportAction::CacheAnnounce {
478 packet_hash,
479 raw: raw.into(),
480 },
481 ActionWire::TunnelEstablished {
482 tunnel_id,
483 interface,
484 } => TransportAction::TunnelEstablished {
485 tunnel_id,
486 interface: InterfaceId(interface),
487 },
488 ActionWire::TunnelSynthesize {
489 interface,
490 data,
491 dest_hash,
492 } => TransportAction::TunnelSynthesize {
493 interface: InterfaceId(interface),
494 data,
495 dest_hash,
496 },
497 ActionWire::ForwardToLocalClients {
498 raw,
499 exclude,
500 has_exclude,
501 } => TransportAction::ForwardToLocalClients {
502 raw: raw.into(),
503 exclude: if has_exclude != 0 {
504 Some(InterfaceId(exclude))
505 } else {
506 None
507 },
508 },
509 ActionWire::ForwardPlainBroadcast {
510 raw,
511 to_local,
512 exclude,
513 has_exclude,
514 } => TransportAction::ForwardPlainBroadcast {
515 raw: raw.into(),
516 to_local: to_local != 0,
517 exclude: if has_exclude != 0 {
518 Some(InterfaceId(exclude))
519 } else {
520 None
521 },
522 },
523 ActionWire::AnnounceReceived {
524 destination_hash,
525 identity_hash,
526 public_key,
527 name_hash,
528 random_hash,
529 app_data,
530 hops,
531 receiving_interface,
532 } => TransportAction::AnnounceReceived {
533 destination_hash,
534 identity_hash,
535 public_key,
536 name_hash,
537 random_hash,
538 ratchet: None,
539 app_data,
540 hops,
541 receiving_interface: InterfaceId(receiving_interface),
542 rx: rns_core::transport::RxMetadata::default(),
543 },
544 }
545 })
546 .collect()
547}
548
549fn infer_interface_type(name: &str) -> String {
553 if name.starts_with("TCPServerInterface") {
554 "TCPServerClientInterface".to_string()
555 } else if name.starts_with("BackboneInterface") {
556 "BackboneInterface".to_string()
557 } else if name.starts_with("LocalInterface") {
558 "LocalServerClientInterface".to_string()
559 } else {
560 "AutoInterface".to_string()
563 }
564}
565
566pub use crate::common::callbacks::Callbacks;
567
568#[derive(Clone)]
569struct SharedAnnounceRecord {
570 name_hash: [u8; 10],
571 identity_prv_key: [u8; 64],
572 app_data: Option<Vec<u8>>,
573}
574
575#[derive(Debug, Clone)]
576pub(crate) struct KnownDestinationState {
577 announced: crate::destination::AnnouncedIdentity,
578 was_used: bool,
579 last_used_at: Option<f64>,
580 retained: bool,
581}
582
583pub struct Driver {
585 pub(crate) engine: TransportEngine,
586 pub(crate) interfaces: HashMap<InterfaceId, InterfaceEntry>,
587 pub(crate) rng: OsRng,
588 pub(crate) rx: EventReceiver,
589 pub(crate) callbacks: Box<dyn Callbacks>,
590 pub(crate) started: f64,
591 pub(crate) lifecycle_state: LifecycleState,
592 pub(crate) drain_started_at: Option<Instant>,
593 pub(crate) drain_deadline: Option<Instant>,
594 pub(crate) listener_controls: Vec<crate::interface::ListenerControl>,
595 pub(crate) announce_cache: Option<crate::announce_cache::AnnounceCache>,
596 pub(crate) tunnel_synth_dest: [u8; 16],
598 pub(crate) transport_identity: Option<rns_crypto::identity::Identity>,
600 pub(crate) link_manager: LinkManager,
602 pub(crate) management_config: crate::management::ManagementConfig,
604 pub(crate) last_management_announce: f64,
606 pub(crate) initial_announce_sent: bool,
608 pub(crate) known_destinations: HashMap<[u8; 16], KnownDestinationState>,
610 pub(crate) ratchet_store: Option<Arc<dyn crate::storage::RatchetStore>>,
612 pub(crate) ratchet_expiry_secs: f64,
614 pub(crate) known_destinations_ttl: f64,
616 pub(crate) known_destinations_max_entries: usize,
618 pub(crate) rate_limiter_ttl_secs: f64,
620 pub(crate) path_request_dest: [u8; 16],
622 pub(crate) proof_strategies: HashMap<
625 [u8; 16],
626 (
627 rns_core::types::ProofStrategy,
628 Option<rns_crypto::identity::Identity>,
629 ),
630 >,
631 pub(crate) sent_packets: HashMap<[u8; 32], ([u8; 16], f64)>,
633 pub(crate) completed_proofs: HashMap<[u8; 32], (f64, f64)>,
635 pub(crate) local_destinations: HashMap<[u8; 16], u8>,
637 shared_announces: HashMap<[u8; 16], SharedAnnounceRecord>,
639 shared_reconnect_pending: HashMap<InterfaceId, bool>,
641 pub(crate) holepunch_manager: HolePunchManager,
643 pub(crate) event_tx: crate::event::EventSender,
645 pub(crate) interface_writer_queue_capacity: usize,
647 pub(crate) announce_rate_defaults: AnnounceRateDefaults,
649 pub(crate) ingress_control_defaults: rns_core::transport::types::IngressControlConfig,
651 pub(crate) tick_interval_ms: Arc<AtomicU64>,
653 #[cfg(feature = "iface-backbone")]
655 pub(crate) backbone_runtime: HashMap<String, BackboneRuntimeConfigHandle>,
656 #[cfg(feature = "iface-backbone")]
658 pub(crate) backbone_peer_state: HashMap<String, BackbonePeerStateHandle>,
659 #[cfg(feature = "iface-backbone")]
661 pub(crate) backbone_client_runtime: HashMap<String, BackboneClientRuntimeConfigHandle>,
662 #[cfg(feature = "iface-backbone")]
664 pub(crate) backbone_discovery_runtime: HashMap<String, BackboneDiscoveryRuntimeHandle>,
665 #[cfg(feature = "iface-backbone")]
667 backbone_peer_pool: Option<BackbonePeerPool>,
668 #[cfg(feature = "iface-backbone")]
670 pub(crate) next_dynamic_interface_id: Arc<AtomicU64>,
671 #[cfg(feature = "iface-tcp")]
673 pub(crate) tcp_server_runtime: HashMap<String, TcpServerRuntimeConfigHandle>,
674 #[cfg(feature = "iface-tcp")]
676 pub(crate) tcp_client_runtime: HashMap<String, TcpClientRuntimeConfigHandle>,
677 #[cfg(feature = "iface-tcp")]
679 pub(crate) tcp_server_discovery_runtime: HashMap<String, TcpServerDiscoveryRuntimeHandle>,
680 #[cfg(feature = "iface-udp")]
682 pub(crate) udp_runtime: HashMap<String, UdpRuntimeConfigHandle>,
683 #[cfg(feature = "iface-auto")]
685 pub(crate) auto_runtime: HashMap<String, AutoRuntimeConfigHandle>,
686 #[cfg(feature = "iface-i2p")]
688 pub(crate) i2p_runtime: HashMap<String, I2pRuntimeConfigHandle>,
689 #[cfg(feature = "iface-pipe")]
691 pub(crate) pipe_runtime: HashMap<String, PipeRuntimeConfigHandle>,
692 #[cfg(feature = "iface-rnode")]
694 pub(crate) rnode_runtime: HashMap<String, RNodeRuntimeConfigHandle>,
695 pub(crate) interface_runtime_defaults:
697 HashMap<String, rns_core::transport::types::InterfaceInfo>,
698 pub(crate) interface_ifac_runtime: HashMap<String, IfacRuntimeConfig>,
700 pub(crate) interface_ifac_runtime_defaults: HashMap<String, IfacRuntimeConfig>,
702 pub(crate) discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage,
704 pub(crate) discovery_required_value: u8,
706 pub(crate) discovery_name_hash: [u8; 10],
708 pub(crate) probe_responder_hash: Option<[u8; 16]>,
710 pub(crate) discover_interfaces: bool,
712 pub(crate) interface_announcer: Option<crate::discovery::InterfaceAnnouncer>,
714 pub(crate) announce_verify_queue: Arc<Mutex<AnnounceVerifyQueue>>,
716 pub(crate) async_announce_verification: bool,
718 pub(crate) discovery_cleanup_counter: u32,
720 pub(crate) discovery_cleanup_interval_ticks: u32,
722 pub(crate) memory_stats_counter: u32,
724 pub(crate) cache_cleanup_counter: u32,
726 pub(crate) announce_cache_cleanup_counter: u32,
728 pub(crate) known_destinations_cleanup_interval_ticks: u32,
730 pub(crate) known_destinations_cap_evict_count: usize,
732 pub(crate) announce_cache_cleanup_interval_ticks: u32,
734 pub(crate) cache_cleanup_active_hashes: Option<Vec<[u8; 32]>>,
736 pub(crate) cache_cleanup_entries: Option<std::fs::ReadDir>,
738 pub(crate) cache_cleanup_removed: usize,
740 pub(crate) announce_cache_cleanup_batch_size: usize,
742 pub(crate) management_announce_interval_secs: f64,
744 pub(crate) runtime_config_defaults: RuntimeConfigDefaults,
746 #[cfg(feature = "hooks")]
748 pub(crate) hook_slots: [HookSlot; HookPoint::COUNT],
749 #[cfg(feature = "hooks")]
751 pub(crate) hook_manager: Option<HookManager>,
752 #[cfg(feature = "hooks")]
753 pub(crate) provider_bridge: Option<ProviderBridge>,
754}
755
756impl Driver {
757 pub fn new(
759 config: TransportConfig,
760 rx: EventReceiver,
761 tx: crate::event::EventSender,
762 callbacks: Box<dyn Callbacks>,
763 ) -> Self {
764 let announce_queue_max_entries = config.announce_queue_max_entries;
765 let tunnel_synth_dest = rns_core::destination::destination_hash(
766 "rnstransport",
767 &["tunnel", "synthesize"],
768 None,
769 );
770 let path_request_dest =
771 rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
772 let discovery_name_hash = crate::discovery::discovery_name_hash();
773 let mut engine = TransportEngine::new(config);
774 engine.register_destination(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
775 engine.register_destination(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
777 let mut local_destinations = HashMap::new();
780 local_destinations.insert(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
781 local_destinations.insert(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
782 let runtime_config_defaults = RuntimeConfigDefaults {
783 tick_interval_ms: DEFAULT_TICK_INTERVAL_MS,
784 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
785 rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
786 known_destinations_cleanup_interval_ticks:
787 DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS,
788 announce_cache_cleanup_interval_ticks: DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS,
789 announce_cache_cleanup_batch_size: DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE,
790 discovery_cleanup_interval_ticks: DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS,
791 management_announce_interval_secs: DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS,
792 direct_connect_policy: crate::event::HolePunchPolicy::default(),
793 #[cfg(feature = "hooks")]
794 provider_queue_max_events: crate::provider_bridge::ProviderBridgeConfig::default()
795 .queue_max_events,
796 #[cfg(feature = "hooks")]
797 provider_queue_max_bytes: crate::provider_bridge::ProviderBridgeConfig::default()
798 .queue_max_bytes,
799 };
800 Driver {
801 engine,
802 interfaces: HashMap::new(),
803 rng: OsRng,
804 rx,
805 callbacks,
806 started: time::now(),
807 lifecycle_state: LifecycleState::Active,
808 drain_started_at: None,
809 drain_deadline: None,
810 listener_controls: Vec::new(),
811 announce_cache: None,
812 tunnel_synth_dest,
813 transport_identity: None,
814 link_manager: LinkManager::new(),
815 management_config: Default::default(),
816 last_management_announce: 0.0,
817 initial_announce_sent: false,
818 known_destinations: HashMap::new(),
819 ratchet_store: None,
820 ratchet_expiry_secs: rns_core::constants::RATCHET_EXPIRY as f64,
821 known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
822 known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
823 rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
824 path_request_dest,
825 proof_strategies: HashMap::new(),
826 sent_packets: HashMap::new(),
827 completed_proofs: HashMap::new(),
828 local_destinations,
829 shared_announces: HashMap::new(),
830 shared_reconnect_pending: HashMap::new(),
831 holepunch_manager: HolePunchManager::new(
832 vec![],
833 rns_core::holepunch::ProbeProtocol::Rnsp,
834 None,
835 ),
836 event_tx: tx,
837 interface_writer_queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
838 announce_rate_defaults: AnnounceRateDefaults::default(),
839 ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(),
840 tick_interval_ms: Arc::new(AtomicU64::new(DEFAULT_TICK_INTERVAL_MS)),
841 #[cfg(feature = "iface-backbone")]
842 backbone_runtime: HashMap::new(),
843 #[cfg(feature = "iface-backbone")]
844 backbone_peer_state: HashMap::new(),
845 #[cfg(feature = "iface-backbone")]
846 backbone_client_runtime: HashMap::new(),
847 #[cfg(feature = "iface-backbone")]
848 backbone_discovery_runtime: HashMap::new(),
849 #[cfg(feature = "iface-backbone")]
850 backbone_peer_pool: None,
851 #[cfg(feature = "iface-backbone")]
852 next_dynamic_interface_id: Arc::new(AtomicU64::new(10000)),
853 #[cfg(feature = "iface-tcp")]
854 tcp_server_runtime: HashMap::new(),
855 #[cfg(feature = "iface-tcp")]
856 tcp_client_runtime: HashMap::new(),
857 #[cfg(feature = "iface-tcp")]
858 tcp_server_discovery_runtime: HashMap::new(),
859 #[cfg(feature = "iface-udp")]
860 udp_runtime: HashMap::new(),
861 #[cfg(feature = "iface-auto")]
862 auto_runtime: HashMap::new(),
863 #[cfg(feature = "iface-i2p")]
864 i2p_runtime: HashMap::new(),
865 #[cfg(feature = "iface-pipe")]
866 pipe_runtime: HashMap::new(),
867 #[cfg(feature = "iface-rnode")]
868 rnode_runtime: HashMap::new(),
869 interface_runtime_defaults: HashMap::new(),
870 interface_ifac_runtime: HashMap::new(),
871 interface_ifac_runtime_defaults: HashMap::new(),
872 discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage::new(
873 std::env::temp_dir().join("rns-discovered-interfaces"),
874 ),
875 discovery_required_value: crate::discovery::DEFAULT_STAMP_VALUE,
876 discovery_name_hash,
877 probe_responder_hash: None,
878 discover_interfaces: false,
879 interface_announcer: None,
880 announce_verify_queue: Arc::new(Mutex::new(AnnounceVerifyQueue::new(
881 announce_queue_max_entries,
882 ))),
883 async_announce_verification: false,
884 discovery_cleanup_counter: 0,
885 discovery_cleanup_interval_ticks: runtime_config_defaults
886 .discovery_cleanup_interval_ticks,
887 memory_stats_counter: 0,
888 cache_cleanup_counter: 0,
889 announce_cache_cleanup_counter: 0,
890 known_destinations_cleanup_interval_ticks: runtime_config_defaults
891 .known_destinations_cleanup_interval_ticks,
892 known_destinations_cap_evict_count: 0,
893 announce_cache_cleanup_interval_ticks: runtime_config_defaults
894 .announce_cache_cleanup_interval_ticks,
895 cache_cleanup_active_hashes: None,
896 cache_cleanup_entries: None,
897 cache_cleanup_removed: 0,
898 announce_cache_cleanup_batch_size: runtime_config_defaults
899 .announce_cache_cleanup_batch_size,
900 management_announce_interval_secs: runtime_config_defaults
901 .management_announce_interval_secs,
902 runtime_config_defaults,
903 #[cfg(feature = "hooks")]
904 hook_slots: create_hook_slots(),
905 #[cfg(feature = "hooks")]
906 hook_manager: HookManager::new().ok(),
907 #[cfg(feature = "hooks")]
908 provider_bridge: None,
909 }
910 }
911
912 pub fn set_announce_verify_queue_config(
913 &mut self,
914 max_entries: usize,
915 max_bytes: usize,
916 max_stale_secs: f64,
917 overflow_policy: OverflowPolicy,
918 ) {
919 self.announce_verify_queue = Arc::new(Mutex::new(AnnounceVerifyQueue::with_limits(
920 max_entries,
921 max_bytes,
922 max_stale_secs,
923 overflow_policy,
924 )));
925 }
926
927 pub fn set_announce_rate_defaults(&mut self, defaults: AnnounceRateDefaults) {
928 self.announce_rate_defaults = defaults;
929 }
930
931 pub fn set_ingress_control_defaults(
932 &mut self,
933 defaults: rns_core::transport::types::IngressControlConfig,
934 ) {
935 self.ingress_control_defaults = defaults;
936 }
937
938 pub(crate) fn apply_announce_rate_defaults(
939 &self,
940 info: &mut rns_core::transport::types::InterfaceInfo,
941 ) {
942 if !self.engine.transport_enabled() || info.announce_rate_target.is_some() {
943 return;
944 }
945
946 let Some(target) = self.announce_rate_defaults.target else {
947 return;
948 };
949
950 info.announce_rate_target = Some(target);
951 if info.announce_rate_grace == 0 {
952 info.announce_rate_grace = self.announce_rate_defaults.grace;
953 }
954 if info.announce_rate_penalty == 0.0 {
955 info.announce_rate_penalty = self.announce_rate_defaults.penalty;
956 }
957 }
958
959 pub(crate) fn apply_ingress_control_defaults(
960 &self,
961 info: &mut rns_core::transport::types::InterfaceInfo,
962 ) {
963 if !self.engine.transport_enabled() {
964 return;
965 }
966
967 let baseline = if info.ingress_control.enabled {
968 rns_core::transport::types::IngressControlConfig::enabled()
969 } else {
970 rns_core::transport::types::IngressControlConfig::disabled()
971 };
972 let defaults = self.ingress_control_defaults;
973 let ic = &mut info.ingress_control;
974
975 if ic.max_held_announces == baseline.max_held_announces {
976 ic.max_held_announces = defaults.max_held_announces;
977 }
978 if ic.burst_hold == baseline.burst_hold {
979 ic.burst_hold = defaults.burst_hold;
980 }
981 if ic.burst_freq_new == baseline.burst_freq_new {
982 ic.burst_freq_new = defaults.burst_freq_new;
983 }
984 if ic.burst_freq == baseline.burst_freq {
985 ic.burst_freq = defaults.burst_freq;
986 }
987 if ic.pr_burst_freq_new == baseline.pr_burst_freq_new {
988 ic.pr_burst_freq_new = defaults.pr_burst_freq_new;
989 }
990 if ic.pr_burst_freq == baseline.pr_burst_freq {
991 ic.pr_burst_freq = defaults.pr_burst_freq;
992 }
993 if ic.new_time == baseline.new_time {
994 ic.new_time = defaults.new_time;
995 }
996 if ic.burst_penalty == baseline.burst_penalty {
997 ic.burst_penalty = defaults.burst_penalty;
998 }
999 if ic.held_release_interval == baseline.held_release_interval {
1000 ic.held_release_interval = defaults.held_release_interval;
1001 }
1002 if ic.egress_enabled == baseline.egress_enabled {
1003 ic.egress_enabled = defaults.egress_enabled;
1004 }
1005 if ic.egress_pr_freq == baseline.egress_pr_freq {
1006 ic.egress_pr_freq = defaults.egress_pr_freq;
1007 }
1008 }
1009
1010 fn wrap_interface_writer(
1011 &self,
1012 interface_id: InterfaceId,
1013 interface_name: &str,
1014 writer: Box<dyn crate::interface::Writer>,
1015 ) -> (
1016 Box<dyn crate::interface::Writer>,
1017 crate::interface::AsyncWriterMetrics,
1018 ) {
1019 crate::interface::wrap_async_writer(
1020 writer,
1021 interface_id,
1022 interface_name,
1023 self.event_tx.clone(),
1024 self.interface_writer_queue_capacity,
1025 )
1026 }
1027
1028 #[cfg(feature = "hooks")]
1029 fn provider_events_enabled(&self) -> bool {
1030 self.provider_bridge.is_some()
1031 }
1032
1033 #[cfg(feature = "hooks")]
1034 fn run_backbone_peer_hook(
1035 &mut self,
1036 attach_point: &str,
1037 point: HookPoint,
1038 event: &BackbonePeerHookEvent,
1039 ) {
1040 let ctx = backbone_peer_hook_context(event);
1041 let now = time::now();
1042 let engine_ref = EngineRef {
1043 engine: &self.engine,
1044 interfaces: &self.interfaces,
1045 link_manager: &self.link_manager,
1046 now,
1047 };
1048 let provider_events_enabled = self.provider_events_enabled();
1049 if let Some(ref e) = run_hook_inner(
1050 &mut self.hook_slots[point as usize].programs,
1051 &self.hook_manager,
1052 &engine_ref,
1053 &ctx,
1054 now,
1055 provider_events_enabled,
1056 ) {
1057 self.forward_hook_side_effects(attach_point, e);
1058 }
1059 }
1060
1061 #[cfg(feature = "iface-backbone")]
1062 fn make_discoverable_interface(
1063 runtime: &BackboneDiscoveryRuntimeHandle,
1064 ) -> crate::discovery::DiscoverableInterface {
1065 crate::discovery::DiscoverableInterface {
1066 interface_name: runtime.interface_name.clone(),
1067 config: runtime.current.config.clone(),
1068 transport_enabled: runtime.current.transport_enabled,
1069 ifac_netname: runtime.current.ifac_netname.clone(),
1070 ifac_netkey: runtime.current.ifac_netkey.clone(),
1071 }
1072 }
1073
1074 #[cfg(feature = "iface-backbone")]
1075 fn sync_backbone_discovery_runtime(
1076 &mut self,
1077 interface_name: &str,
1078 ) -> Result<(), RuntimeConfigError> {
1079 let handle = self
1080 .backbone_discovery_runtime
1081 .get(interface_name)
1082 .ok_or(RuntimeConfigError {
1083 code: RuntimeConfigErrorCode::NotFound,
1084 message: format!("backbone interface '{}' not found", interface_name),
1085 })?
1086 .clone();
1087
1088 if handle.current.discoverable {
1089 let iface = Self::make_discoverable_interface(&handle);
1090 if let Some(announcer) = self.interface_announcer.as_mut() {
1091 announcer.upsert_interface(iface);
1092 } else if let Some(identity) = self.transport_identity.as_ref() {
1093 self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
1094 *identity.hash(),
1095 vec![iface],
1096 ));
1097 }
1098 } else if let Some(announcer) = self.interface_announcer.as_mut() {
1099 announcer.remove_interface(interface_name);
1100 if announcer.is_empty() {
1101 self.interface_announcer = None;
1102 }
1103 }
1104
1105 Ok(())
1106 }
1107
1108 #[cfg(feature = "iface-tcp")]
1109 fn make_tcp_server_discoverable_interface(
1110 runtime: &TcpServerDiscoveryRuntimeHandle,
1111 ) -> crate::discovery::DiscoverableInterface {
1112 crate::discovery::DiscoverableInterface {
1113 interface_name: runtime.interface_name.clone(),
1114 config: runtime.current.config.clone(),
1115 transport_enabled: runtime.current.transport_enabled,
1116 ifac_netname: runtime.current.ifac_netname.clone(),
1117 ifac_netkey: runtime.current.ifac_netkey.clone(),
1118 }
1119 }
1120
1121 #[cfg(feature = "iface-tcp")]
1122 fn sync_tcp_server_discovery_runtime(
1123 &mut self,
1124 interface_name: &str,
1125 ) -> Result<(), RuntimeConfigError> {
1126 let handle = self
1127 .tcp_server_discovery_runtime
1128 .get(interface_name)
1129 .ok_or(RuntimeConfigError {
1130 code: RuntimeConfigErrorCode::NotFound,
1131 message: format!("tcp server interface '{}' not found", interface_name),
1132 })?
1133 .clone();
1134
1135 if handle.current.discoverable {
1136 let iface = Self::make_tcp_server_discoverable_interface(&handle);
1137 if let Some(announcer) = self.interface_announcer.as_mut() {
1138 announcer.upsert_interface(iface);
1139 } else if let Some(identity) = self.transport_identity.as_ref() {
1140 self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
1141 *identity.hash(),
1142 vec![iface],
1143 ));
1144 }
1145 } else if let Some(announcer) = self.interface_announcer.as_mut() {
1146 announcer.remove_interface(interface_name);
1147 if announcer.is_empty() {
1148 self.interface_announcer = None;
1149 }
1150 }
1151
1152 Ok(())
1153 }
1154
1155 #[cfg(feature = "hooks")]
1156 fn update_hook_program<F>(
1157 &mut self,
1158 name: &str,
1159 attach_point: &str,
1160 mut update: F,
1161 ) -> Result<(), String>
1162 where
1163 F: FnMut(&mut rns_hooks::LoadedProgram),
1164 {
1165 let point_idx = crate::config::parse_hook_point(attach_point)
1166 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1167 let program = self.hook_slots[point_idx]
1168 .programs
1169 .iter_mut()
1170 .find(|program| program.name == name)
1171 .ok_or_else(|| format!("hook '{}' not found at point '{}'", name, attach_point))?;
1172 update(program);
1173 Ok(())
1174 }
1175
1176 pub(crate) fn set_tick_interval_handle(&mut self, tick_interval_ms: Arc<AtomicU64>) {
1177 self.tick_interval_ms = tick_interval_ms;
1178 }
1179
1180 pub(crate) fn set_packet_hashlist_max_entries(&mut self, max_entries: usize) {
1181 self.engine.set_packet_hashlist_max_entries(max_entries);
1182 }
1183
1184 #[cfg(feature = "hooks")]
1185 fn forward_hook_side_effects(&mut self, attach_point: &str, exec: &rns_hooks::ExecuteResult) {
1186 if !exec.injected_actions.is_empty() {
1187 self.dispatch_all(convert_injected_actions(exec.injected_actions.clone()));
1188 }
1189 if let Some(ref bridge) = self.provider_bridge {
1190 for event in &exec.provider_events {
1191 bridge.emit_event(
1192 attach_point,
1193 event.hook_name.clone(),
1194 event.payload_type.clone(),
1195 event.payload.clone(),
1196 );
1197 }
1198 }
1199 }
1200
1201 #[cfg(feature = "hooks")]
1202 fn collect_hook_side_effects(
1203 &mut self,
1204 attach_point: &str,
1205 exec: &rns_hooks::ExecuteResult,
1206 out: &mut Vec<TransportAction>,
1207 ) {
1208 if !exec.injected_actions.is_empty() {
1209 out.extend(convert_injected_actions(exec.injected_actions.clone()));
1210 }
1211 if let Some(ref bridge) = self.provider_bridge {
1212 for event in &exec.provider_events {
1213 bridge.emit_event(
1214 attach_point,
1215 event.hook_name.clone(),
1216 event.payload_type.clone(),
1217 event.payload.clone(),
1218 );
1219 }
1220 }
1221 }
1222
1223 pub fn set_probe_config(
1225 &mut self,
1226 addrs: Vec<std::net::SocketAddr>,
1227 protocol: rns_core::holepunch::ProbeProtocol,
1228 device: Option<String>,
1229 ) {
1230 self.holepunch_manager = HolePunchManager::new(addrs, protocol, device);
1231 }
1232}