1use super::*;
2
3impl Node {
4 fn new_dataplane_node() -> DataplaneNode {
5 DataplaneLiveNode::new(AdmissionConfig::new(
6 1024,
7 FIPS_ENDPOINT_DIRECT_PACKET_QUEUE_MAX_PACKETS,
8 ))
9 }
10
11 pub fn new(config: Config) -> Result<Self, NodeError> {
13 config.validate()?;
14 let identity = config.create_identity()?;
15 let node_addr = *identity.node_addr();
16 let is_leaf_only = config.is_leaf_only();
17
18 let mut startup_epoch = [0u8; 8];
19 rand::rng().fill_bytes(&mut startup_epoch);
20
21 let mut bloom_state = if is_leaf_only {
22 BloomState::leaf_only(node_addr)
23 } else {
24 BloomState::new(node_addr)
25 };
26 bloom_state.set_update_debounce_ms(config.node.bloom.update_debounce_ms);
27
28 let tun_state = if config.tun.enabled {
29 TunState::Configured
30 } else {
31 TunState::Disabled
32 };
33
34 let mut tree_state = TreeState::new(node_addr);
36 tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
37 tree_state.set_hold_down(config.node.tree.hold_down_secs);
38 tree_state.set_flap_dampening(
39 config.node.tree.flap_threshold,
40 config.node.tree.flap_window_secs,
41 config.node.tree.flap_dampening_secs,
42 );
43 tree_state
44 .sign_declaration(&identity)
45 .expect("signing own declaration should never fail");
46
47 let coord_cache = CoordCache::new(
48 config.node.cache.coord_size,
49 config.node.cache.coord_ttl_secs * 1000,
50 );
51 let rl = &config.node.rate_limit;
52 let msg1_rate_limiter = HandshakeRateLimiter::with_params(
53 rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
54 config.node.limits.max_pending_inbound,
55 );
56
57 let max_connections = config.node.limits.max_connections;
58 let max_peers = config.node.limits.max_peers;
59 let max_links = config.node.limits.max_links;
60 let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
61 let backoff_base_secs = config.node.discovery.backoff_base_secs;
62 let backoff_max_secs = config.node.discovery.backoff_max_secs;
63 let forward_min_interval_secs = config.node.discovery.forward_min_interval_secs;
64
65 let (host_map, peer_acl) = Self::host_map_and_peer_acl(&config);
66 let configured_peers = ConfiguredPeerLookup::from_config(&config);
67 let local_rendezvous =
68 local_rendezvous::LocalRendezvous::new(&config, &identity, startup_epoch);
69
70 Ok(Self {
71 identity,
72 startup_epoch,
73 started_at: std::time::Instant::now(),
74 config,
75 state: NodeState::Created,
76 is_leaf_only,
77 tree_state,
78 bloom_state,
79 coord_cache,
80 learned_routes: LearnedRouteTable::default(),
81 session_direct_degradation: SessionDirectDegradation::default(),
82 recent_requests: RecentDiscoveryRequests::default(),
83 transports: HashMap::new(),
84 transport_drops: TransportDropTracker::default(),
85 transport_socket_drops: TransportDropTracker::default(),
86 transport_namespace_drops: TransportDropTracker::default(),
87 links: LinkRegistry::default(),
88 packet_tx: None,
89 packet_rx: None,
90 dataplane: Self::new_dataplane_node(),
91 deferred_dataplane_control_turns: VecDeque::new(),
92 deferred_session_forwards: Default::default(),
93 dataplane_fast_ingress_rx: None,
94 dataplane_direct_fsp_sources: Arc::new(HashMap::new()),
95 dataplane_direct_fsp_sources_dirty: true,
96 dataplane_transport_send_batch_packets: DATAPLANE_TRANSPORT_SEND_BATCH_PACKETS,
97 peers: PeerLifecycleRegistry::default(),
98 sessions: SessionRegistry::default(),
99 identity_cache: IdentityCache::default(),
100 pending_session_traffic: PendingSessionTrafficQueues::default(),
101 pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
102 max_connections,
103 max_peers,
104 max_links,
105 next_link_id: 1,
106 next_transport_id: 1,
107 stats: stats::NodeStats::new(),
108 stats_history: stats_history::StatsHistory::new(),
109 tun_state,
110 tun_name: None,
111 tun_tx: None,
112 tun_outbound_rx: None,
113 external_packet_tx: None,
114 endpoint_control_rx: None,
115 endpoint_data_rx: None,
116 endpoint_events: EndpointEventRuntime::default(),
117 endpoint_services: EndpointServiceRuntime::default(),
118 tun_reader_handle: None,
119 tun_writer_handle: None,
120 #[cfg(target_os = "macos")]
121 tun_shutdown_fd: None,
122 dns_identity_rx: None,
123 dns_task: None,
124 index_allocator: IndexAllocator::new(),
125 pending_outbound: PendingOutboundHandshakes::default(),
126 msg1_rate_limiter,
127 icmp_rate_limiter: IcmpRateLimiter::new(),
128 routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
129 coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
130 std::time::Duration::from_millis(coords_response_interval_ms),
131 ),
132 discovery_backoff: DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs),
133 discovery_forward_limiter: DiscoveryForwardRateLimiter::with_interval(
134 std::time::Duration::from_secs(forward_min_interval_secs),
135 ),
136 pending_connects: Vec::new(),
137 retry_pending: retry::PendingRouteRetries::default(),
138 nostr_discovery: None,
139 pending_mesh_signals: HashMap::new(),
140 nostr_discovery_started_at_ms: None,
141 lan_discovery: None,
142 local_rendezvous,
143 startup_open_discovery_sweep_done: false,
144 bootstrap_transports: BootstrapTransports::default(),
145 discovery_fallback_transit: DiscoveryFallbackTransit::default(),
146 last_parent_reeval: None,
147 last_congestion_log: None,
148 estimated_mesh_size: None,
149 last_mesh_size_log: None,
150 last_self_warn: None,
151 local_send_failures: LocalSendFailures::default(),
152 last_rx_loop_maintenance_timeout_at: None,
153 peer_aliases: HashMap::new(),
154 configured_peers,
155 peer_acl,
156 host_map,
157 path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
158 })
159 }
160
161 pub fn with_identity(identity: Identity, config: Config) -> Result<Self, NodeError> {
166 config.validate()?;
167 let node_addr = *identity.node_addr();
168
169 let mut startup_epoch = [0u8; 8];
170 rand::rng().fill_bytes(&mut startup_epoch);
171
172 let tun_state = if config.tun.enabled {
173 TunState::Configured
174 } else {
175 TunState::Disabled
176 };
177
178 let mut tree_state = TreeState::new(node_addr);
180 tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
181 tree_state.set_hold_down(config.node.tree.hold_down_secs);
182 tree_state.set_flap_dampening(
183 config.node.tree.flap_threshold,
184 config.node.tree.flap_window_secs,
185 config.node.tree.flap_dampening_secs,
186 );
187 tree_state
188 .sign_declaration(&identity)
189 .expect("signing own declaration should never fail");
190
191 let mut bloom_state = BloomState::new(node_addr);
192 bloom_state.set_update_debounce_ms(config.node.bloom.update_debounce_ms);
193
194 let coord_cache = CoordCache::new(
195 config.node.cache.coord_size,
196 config.node.cache.coord_ttl_secs * 1000,
197 );
198 let rl = &config.node.rate_limit;
199 let msg1_rate_limiter = HandshakeRateLimiter::with_params(
200 rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
201 config.node.limits.max_pending_inbound,
202 );
203
204 let max_connections = config.node.limits.max_connections;
205 let max_peers = config.node.limits.max_peers;
206 let max_links = config.node.limits.max_links;
207 let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
208
209 let (host_map, peer_acl) = Self::host_map_and_peer_acl(&config);
210 let configured_peers = ConfiguredPeerLookup::from_config(&config);
211 let local_rendezvous =
212 local_rendezvous::LocalRendezvous::new(&config, &identity, startup_epoch);
213
214 Ok(Self {
215 identity,
216 startup_epoch,
217 started_at: std::time::Instant::now(),
218 config,
219 state: NodeState::Created,
220 is_leaf_only: false,
221 tree_state,
222 bloom_state,
223 coord_cache,
224 learned_routes: LearnedRouteTable::default(),
225 session_direct_degradation: SessionDirectDegradation::default(),
226 recent_requests: RecentDiscoveryRequests::default(),
227 transports: HashMap::new(),
228 transport_drops: TransportDropTracker::default(),
229 transport_socket_drops: TransportDropTracker::default(),
230 transport_namespace_drops: TransportDropTracker::default(),
231 links: LinkRegistry::default(),
232 packet_tx: None,
233 packet_rx: None,
234 dataplane: Self::new_dataplane_node(),
235 deferred_dataplane_control_turns: VecDeque::new(),
236 deferred_session_forwards: Default::default(),
237 dataplane_fast_ingress_rx: None,
238 dataplane_direct_fsp_sources: Arc::new(HashMap::new()),
239 dataplane_direct_fsp_sources_dirty: true,
240 dataplane_transport_send_batch_packets: DATAPLANE_TRANSPORT_SEND_BATCH_PACKETS,
241 peers: PeerLifecycleRegistry::default(),
242 sessions: SessionRegistry::default(),
243 identity_cache: IdentityCache::default(),
244 pending_session_traffic: PendingSessionTrafficQueues::default(),
245 pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
246 max_connections,
247 max_peers,
248 max_links,
249 next_link_id: 1,
250 next_transport_id: 1,
251 stats: stats::NodeStats::new(),
252 stats_history: stats_history::StatsHistory::new(),
253 tun_state,
254 tun_name: None,
255 tun_tx: None,
256 tun_outbound_rx: None,
257 external_packet_tx: None,
258 endpoint_control_rx: None,
259 endpoint_data_rx: None,
260 endpoint_events: EndpointEventRuntime::default(),
261 endpoint_services: EndpointServiceRuntime::default(),
262 tun_reader_handle: None,
263 tun_writer_handle: None,
264 #[cfg(target_os = "macos")]
265 tun_shutdown_fd: None,
266 dns_identity_rx: None,
267 dns_task: None,
268 index_allocator: IndexAllocator::new(),
269 pending_outbound: PendingOutboundHandshakes::default(),
270 msg1_rate_limiter,
271 icmp_rate_limiter: IcmpRateLimiter::new(),
272 routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
273 coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
274 std::time::Duration::from_millis(coords_response_interval_ms),
275 ),
276 discovery_backoff: DiscoveryBackoff::new(),
277 discovery_forward_limiter: DiscoveryForwardRateLimiter::new(),
278 pending_connects: Vec::new(),
279 retry_pending: retry::PendingRouteRetries::default(),
280 nostr_discovery: None,
281 pending_mesh_signals: HashMap::new(),
282 nostr_discovery_started_at_ms: None,
283 lan_discovery: None,
284 local_rendezvous,
285 startup_open_discovery_sweep_done: false,
286 bootstrap_transports: BootstrapTransports::default(),
287 discovery_fallback_transit: DiscoveryFallbackTransit::default(),
288 last_parent_reeval: None,
289 last_congestion_log: None,
290 estimated_mesh_size: None,
291 last_mesh_size_log: None,
292 last_self_warn: None,
293 local_send_failures: LocalSendFailures::default(),
294 last_rx_loop_maintenance_timeout_at: None,
295 peer_aliases: HashMap::new(),
296 configured_peers,
297 peer_acl,
298 host_map,
299 path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
300 })
301 }
302
303 pub fn leaf_only(config: Config) -> Result<Self, NodeError> {
305 let mut node = Self::new(config)?;
306 node.is_leaf_only = true;
307 node.bloom_state = BloomState::leaf_only(*node.identity.node_addr());
308 Ok(node)
309 }
310
311 pub(super) fn host_map_and_peer_acl(config: &Config) -> (Arc<HostMap>, acl::PeerAclReloader) {
312 let base_host_map = HostMap::from_peer_configs(config.peers());
313 if !config.node.system_files_enabled {
314 return (
315 Arc::new(base_host_map.clone()),
316 acl::PeerAclReloader::memory_only(base_host_map),
317 );
318 }
319
320 let mut host_map = base_host_map.clone();
321 let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH);
322 let hosts_file = HostMap::load_hosts_file(std::path::Path::new(
323 crate::upper::hosts::DEFAULT_HOSTS_PATH,
324 ));
325 host_map.merge(hosts_file);
326 let peer_acl = acl::PeerAclReloader::with_alias_sources(
327 std::path::PathBuf::from(acl::DEFAULT_PEERS_ALLOW_PATH),
328 std::path::PathBuf::from(acl::DEFAULT_PEERS_DENY_PATH),
329 base_host_map,
330 hosts_path,
331 );
332 (Arc::new(host_map), peer_acl)
333 }
334
335 pub(super) async fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec<TransportHandle> {
339 let mut transports = Vec::new();
340
341 let udp_instances: Vec<_> = self
343 .config
344 .transports
345 .udp
346 .iter()
347 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
348 .collect();
349
350 for (name, udp_config) in udp_instances {
352 let transport_id = self.allocate_transport_id();
353 let udp = UdpTransport::new(transport_id, name, udp_config, packet_tx.clone());
354 transports.push(TransportHandle::Udp(udp));
355 }
356
357 let nostr_relay_instances: Vec<_> = self
358 .config
359 .transports
360 .nostr_relay
361 .iter()
362 .map(|(name, config)| (name.map(str::to_string), config.clone()))
363 .collect();
364 for (name, relay_config) in nostr_relay_instances {
365 let transport_id = self.allocate_transport_id();
366 match NostrRelayTransport::new(
367 transport_id,
368 name,
369 relay_config,
370 packet_tx.clone(),
371 &self.identity,
372 ) {
373 Ok(transport) => transports.push(TransportHandle::NostrRelay(Box::new(transport))),
374 Err(error) => warn!(
375 %transport_id,
376 %error,
377 "failed to initialize Nostr relay fallback transport"
378 ),
379 }
380 }
381
382 #[cfg(feature = "sim-transport")]
383 {
384 let sim_instances: Vec<_> = self
385 .config
386 .transports
387 .sim
388 .iter()
389 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
390 .collect();
391
392 for (name, sim_config) in sim_instances {
393 let transport_id = self.allocate_transport_id();
394 let sim = crate::transport::sim::SimTransport::new(
395 transport_id,
396 name,
397 sim_config,
398 packet_tx.clone(),
399 );
400 transports.push(TransportHandle::Sim(sim));
401 }
402 }
403
404 #[cfg(any(target_os = "linux", target_os = "macos"))]
406 {
407 let eth_instances: Vec<_> = self
408 .config
409 .transports
410 .ethernet
411 .iter()
412 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
413 .collect();
414 let xonly = self.identity.pubkey();
415 for (name, eth_config) in eth_instances {
416 let mut eth_config = eth_config;
417 if eth_config.discovery_scope.is_none() {
418 eth_config.discovery_scope = self.lan_discovery_scope();
419 }
420 let transport_id = self.allocate_transport_id();
421 let mut eth =
422 EthernetTransport::new(transport_id, name, eth_config, packet_tx.clone());
423 eth.set_local_pubkey(xonly);
424 transports.push(TransportHandle::Ethernet(eth));
425 }
426 }
427
428 let tcp_instances: Vec<_> = self
430 .config
431 .transports
432 .tcp
433 .iter()
434 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
435 .collect();
436
437 for (name, tcp_config) in tcp_instances {
438 let transport_id = self.allocate_transport_id();
439 let tcp = TcpTransport::new(transport_id, name, tcp_config, packet_tx.clone());
440 transports.push(TransportHandle::Tcp(tcp));
441 }
442
443 let tor_instances: Vec<_> = self
445 .config
446 .transports
447 .tor
448 .iter()
449 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
450 .collect();
451
452 for (name, tor_config) in tor_instances {
453 let transport_id = self.allocate_transport_id();
454 let tor = TorTransport::new(transport_id, name, tor_config, packet_tx.clone());
455 transports.push(TransportHandle::Tor(tor));
456 }
457
458 let webrtc_instances: Vec<_> = self
459 .config
460 .transports
461 .webrtc
462 .iter()
463 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
464 .collect();
465
466 #[cfg(feature = "webrtc-transport")]
467 {
468 for (name, webrtc_config) in webrtc_instances {
469 let transport_id = self.allocate_transport_id();
470 match WebRtcTransport::new(
471 transport_id,
472 name,
473 webrtc_config,
474 packet_tx.clone(),
475 &self.identity,
476 &self.config.node.discovery.nostr,
477 ) {
478 Ok(webrtc) => transports.push(TransportHandle::WebRtc(Box::new(webrtc))),
479 Err(err) => {
480 warn!(
481 transport_id = %transport_id,
482 error = %err,
483 "failed to initialize WebRTC transport"
484 );
485 }
486 }
487 }
488 }
489 #[cfg(not(feature = "webrtc-transport"))]
490 if !webrtc_instances.is_empty() {
491 warn!("WebRTC transport configured but this build lacks WebRTC transport support");
492 }
493
494 #[cfg(bluer_available)]
496 {
497 let ble_instances: Vec<_> = self
498 .config
499 .transports
500 .ble
501 .iter()
502 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
503 .collect();
504
505 #[cfg(all(bluer_available, not(test)))]
506 for (name, ble_config) in ble_instances {
507 let transport_id = self.allocate_transport_id();
508 let adapter = ble_config.adapter().to_string();
509 let mtu = ble_config.mtu();
510 match crate::transport::ble::io::BluerIo::new(&adapter, mtu).await {
511 Ok(io) => {
512 let mut ble = crate::transport::ble::BleTransport::new(
513 transport_id,
514 name,
515 ble_config,
516 io,
517 packet_tx.clone(),
518 );
519 ble.set_local_pubkey(self.identity.pubkey().serialize());
520 transports.push(TransportHandle::Ble(ble));
521 }
522 Err(e) => {
523 tracing::warn!(adapter = %adapter, error = %e, "failed to initialize BLE adapter");
524 }
525 }
526 }
527
528 #[cfg(any(not(bluer_available), test))]
529 if !ble_instances.is_empty() {
530 #[cfg(not(test))]
531 tracing::warn!("BLE transport configured but this build lacks BlueZ support");
532 }
533 }
534
535 transports
536 }
537
538 pub(super) fn find_transport_for_type(&self, transport_type: &str) -> Option<TransportId> {
548 self.transports
549 .iter()
550 .filter(|(id, handle)| {
551 handle.transport_type().name == transport_type
552 && handle.is_operational()
553 && !self.bootstrap_transports.contains(id)
554 && !self.is_local_rendezvous_transport(id)
555 })
556 .min_by_key(|(id, _)| id.as_u32())
557 .map(|(id, _)| *id)
558 }
559
560 #[cfg(any(target_os = "linux", target_os = "macos"))]
566 pub(super) fn resolve_ethernet_addr(
567 &self,
568 addr_str: &str,
569 ) -> Result<(TransportId, TransportAddr), NodeError> {
570 let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| {
571 NodeError::NoTransportForType(format!(
572 "invalid Ethernet address format '{}': expected 'interface/mac'",
573 addr_str
574 ))
575 })?;
576
577 let transport_id = self
579 .transports
580 .iter()
581 .find(|(_, handle)| {
582 handle.transport_type().name == "ethernet"
583 && handle.is_operational()
584 && handle.interface_name() == Some(iface)
585 })
586 .map(|(id, _)| *id)
587 .ok_or_else(|| {
588 NodeError::NoTransportForType(format!(
589 "no operational Ethernet transport for interface '{}'",
590 iface
591 ))
592 })?;
593
594 let mac = crate::transport::ethernet::parse_mac_string(mac_str).map_err(|e| {
595 NodeError::NoTransportForType(format!("invalid MAC in '{}': {}", addr_str, e))
596 })?;
597
598 Ok((transport_id, TransportAddr::from_bytes(&mac)))
599 }
600
601 #[cfg(not(any(target_os = "linux", target_os = "macos")))]
602 pub(super) fn resolve_ethernet_addr(
603 &self,
604 _addr_str: &str,
605 ) -> Result<(TransportId, TransportAddr), NodeError> {
606 Err(NodeError::NoTransportForType(
607 "Ethernet transport is not supported on this platform".to_string(),
608 ))
609 }
610
611 #[cfg(bluer_available)]
615 pub(super) fn resolve_ble_addr(
616 &self,
617 addr_str: &str,
618 ) -> Result<(TransportId, TransportAddr), NodeError> {
619 let ta = TransportAddr::from_string(addr_str);
620 let adapter = crate::transport::ble::addr::adapter_from_addr(&ta).ok_or_else(|| {
621 NodeError::NoTransportForType(format!(
622 "invalid BLE address format '{}': expected 'adapter/mac'",
623 addr_str
624 ))
625 })?;
626
627 let transport_id = self
629 .transports
630 .iter()
631 .find(|(_, handle)| handle.transport_type().name == "ble" && handle.is_operational())
632 .map(|(id, _)| *id)
633 .ok_or_else(|| {
634 NodeError::NoTransportForType(format!(
635 "no operational BLE transport for adapter '{}'",
636 adapter
637 ))
638 })?;
639
640 crate::transport::ble::addr::BleAddr::parse(addr_str).map_err(|e| {
642 NodeError::NoTransportForType(format!("invalid BLE address '{}': {}", addr_str, e))
643 })?;
644
645 Ok((transport_id, TransportAddr::from_string(addr_str)))
646 }
647}