fips_core/node/
core_impl.rs

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    /// Create a new node from configuration.
12    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        // Initialize tree state with signed self-declaration
35        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_peer_send_weights = ConfiguredPeerSendWeights::from_config(&config);
67
68        Ok(Self {
69            identity,
70            startup_epoch,
71            started_at: std::time::Instant::now(),
72            config,
73            state: NodeState::Created,
74            is_leaf_only,
75            tree_state,
76            bloom_state,
77            coord_cache,
78            learned_routes: LearnedRouteTable::default(),
79            session_direct_degradation: SessionDirectDegradation::default(),
80            recent_requests: RecentDiscoveryRequests::default(),
81            transports: HashMap::new(),
82            transport_drops: TransportDropTracker::default(),
83            transport_socket_drops: TransportDropTracker::default(),
84            transport_namespace_drops: TransportDropTracker::default(),
85            links: LinkRegistry::default(),
86            packet_tx: None,
87            packet_rx: None,
88            dataplane: Self::new_dataplane_node(),
89            deferred_dataplane_control_turns: VecDeque::new(),
90            dataplane_fast_ingress_rx: None,
91            dataplane_direct_fsp_sources: Arc::new(HashMap::new()),
92            dataplane_direct_fsp_sources_dirty: true,
93            dataplane_transport_send_batch_packets: DATAPLANE_TRANSPORT_SEND_BATCH_PACKETS,
94            peers: PeerLifecycleRegistry::default(),
95            sessions: SessionRegistry::default(),
96            identity_cache: IdentityCache::default(),
97            pending_session_traffic: PendingSessionTrafficQueues::default(),
98            pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
99            max_connections,
100            max_peers,
101            max_links,
102            next_link_id: 1,
103            next_transport_id: 1,
104            stats: stats::NodeStats::new(),
105            stats_history: stats_history::StatsHistory::new(),
106            tun_state,
107            tun_name: None,
108            tun_tx: None,
109            tun_outbound_rx: None,
110            external_packet_tx: None,
111            endpoint_control_rx: None,
112            endpoint_data_rx: None,
113            endpoint_events: EndpointEventRuntime::default(),
114            endpoint_services: EndpointServiceRuntime::default(),
115            tun_reader_handle: None,
116            tun_writer_handle: None,
117            #[cfg(target_os = "macos")]
118            tun_shutdown_fd: None,
119            dns_identity_rx: None,
120            dns_task: None,
121            index_allocator: IndexAllocator::new(),
122            pending_outbound: PendingOutboundHandshakes::default(),
123            msg1_rate_limiter,
124            icmp_rate_limiter: IcmpRateLimiter::new(),
125            routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
126            coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
127                std::time::Duration::from_millis(coords_response_interval_ms),
128            ),
129            discovery_backoff: DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs),
130            discovery_forward_limiter: DiscoveryForwardRateLimiter::with_interval(
131                std::time::Duration::from_secs(forward_min_interval_secs),
132            ),
133            pending_connects: Vec::new(),
134            retry_pending: retry::PendingRouteRetries::default(),
135            nostr_discovery: None,
136            nostr_discovery_started_at_ms: None,
137            lan_discovery: None,
138            local_instance_registry: None,
139            local_instance_started_at_ms: None,
140            last_local_instance_publish_ms: None,
141            last_local_instance_scan_ms: None,
142            startup_open_discovery_sweep_done: false,
143            bootstrap_transports: BootstrapTransports::default(),
144            discovery_fallback_transit: DiscoveryFallbackTransit::default(),
145            last_parent_reeval: None,
146            last_congestion_log: None,
147            estimated_mesh_size: None,
148            last_mesh_size_log: None,
149            last_self_warn: None,
150            local_send_failures: LocalSendFailures::default(),
151            last_rx_loop_maintenance_timeout_at: None,
152            peer_aliases: HashMap::new(),
153            configured_peer_send_weights,
154            peer_acl,
155            host_map,
156            path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
157        })
158    }
159
160    /// Create a node with a specific identity.
161    ///
162    /// This constructor validates cross-field config invariants before
163    /// constructing the node, same as [`Node::new`].
164    pub fn with_identity(identity: Identity, config: Config) -> Result<Self, NodeError> {
165        config.validate()?;
166        let node_addr = *identity.node_addr();
167
168        let mut startup_epoch = [0u8; 8];
169        rand::rng().fill_bytes(&mut startup_epoch);
170
171        let tun_state = if config.tun.enabled {
172            TunState::Configured
173        } else {
174            TunState::Disabled
175        };
176
177        // Initialize tree state with signed self-declaration
178        let mut tree_state = TreeState::new(node_addr);
179        tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
180        tree_state.set_hold_down(config.node.tree.hold_down_secs);
181        tree_state.set_flap_dampening(
182            config.node.tree.flap_threshold,
183            config.node.tree.flap_window_secs,
184            config.node.tree.flap_dampening_secs,
185        );
186        tree_state
187            .sign_declaration(&identity)
188            .expect("signing own declaration should never fail");
189
190        let mut bloom_state = BloomState::new(node_addr);
191        bloom_state.set_update_debounce_ms(config.node.bloom.update_debounce_ms);
192
193        let coord_cache = CoordCache::new(
194            config.node.cache.coord_size,
195            config.node.cache.coord_ttl_secs * 1000,
196        );
197        let rl = &config.node.rate_limit;
198        let msg1_rate_limiter = HandshakeRateLimiter::with_params(
199            rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
200            config.node.limits.max_pending_inbound,
201        );
202
203        let max_connections = config.node.limits.max_connections;
204        let max_peers = config.node.limits.max_peers;
205        let max_links = config.node.limits.max_links;
206        let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
207
208        let (host_map, peer_acl) = Self::host_map_and_peer_acl(&config);
209        let configured_peer_send_weights = ConfiguredPeerSendWeights::from_config(&config);
210
211        Ok(Self {
212            identity,
213            startup_epoch,
214            started_at: std::time::Instant::now(),
215            config,
216            state: NodeState::Created,
217            is_leaf_only: false,
218            tree_state,
219            bloom_state,
220            coord_cache,
221            learned_routes: LearnedRouteTable::default(),
222            session_direct_degradation: SessionDirectDegradation::default(),
223            recent_requests: RecentDiscoveryRequests::default(),
224            transports: HashMap::new(),
225            transport_drops: TransportDropTracker::default(),
226            transport_socket_drops: TransportDropTracker::default(),
227            transport_namespace_drops: TransportDropTracker::default(),
228            links: LinkRegistry::default(),
229            packet_tx: None,
230            packet_rx: None,
231            dataplane: Self::new_dataplane_node(),
232            deferred_dataplane_control_turns: VecDeque::new(),
233            dataplane_fast_ingress_rx: None,
234            dataplane_direct_fsp_sources: Arc::new(HashMap::new()),
235            dataplane_direct_fsp_sources_dirty: true,
236            dataplane_transport_send_batch_packets: DATAPLANE_TRANSPORT_SEND_BATCH_PACKETS,
237            peers: PeerLifecycleRegistry::default(),
238            sessions: SessionRegistry::default(),
239            identity_cache: IdentityCache::default(),
240            pending_session_traffic: PendingSessionTrafficQueues::default(),
241            pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
242            max_connections,
243            max_peers,
244            max_links,
245            next_link_id: 1,
246            next_transport_id: 1,
247            stats: stats::NodeStats::new(),
248            stats_history: stats_history::StatsHistory::new(),
249            tun_state,
250            tun_name: None,
251            tun_tx: None,
252            tun_outbound_rx: None,
253            external_packet_tx: None,
254            endpoint_control_rx: None,
255            endpoint_data_rx: None,
256            endpoint_events: EndpointEventRuntime::default(),
257            endpoint_services: EndpointServiceRuntime::default(),
258            tun_reader_handle: None,
259            tun_writer_handle: None,
260            #[cfg(target_os = "macos")]
261            tun_shutdown_fd: None,
262            dns_identity_rx: None,
263            dns_task: None,
264            index_allocator: IndexAllocator::new(),
265            pending_outbound: PendingOutboundHandshakes::default(),
266            msg1_rate_limiter,
267            icmp_rate_limiter: IcmpRateLimiter::new(),
268            routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
269            coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
270                std::time::Duration::from_millis(coords_response_interval_ms),
271            ),
272            discovery_backoff: DiscoveryBackoff::new(),
273            discovery_forward_limiter: DiscoveryForwardRateLimiter::new(),
274            pending_connects: Vec::new(),
275            retry_pending: retry::PendingRouteRetries::default(),
276            nostr_discovery: None,
277            nostr_discovery_started_at_ms: None,
278            lan_discovery: None,
279            local_instance_registry: None,
280            local_instance_started_at_ms: None,
281            last_local_instance_publish_ms: None,
282            last_local_instance_scan_ms: None,
283            startup_open_discovery_sweep_done: false,
284            bootstrap_transports: BootstrapTransports::default(),
285            discovery_fallback_transit: DiscoveryFallbackTransit::default(),
286            last_parent_reeval: None,
287            last_congestion_log: None,
288            estimated_mesh_size: None,
289            last_mesh_size_log: None,
290            last_self_warn: None,
291            local_send_failures: LocalSendFailures::default(),
292            last_rx_loop_maintenance_timeout_at: None,
293            peer_aliases: HashMap::new(),
294            configured_peer_send_weights,
295            peer_acl,
296            host_map,
297            path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
298        })
299    }
300
301    /// Create a leaf-only node (simplified state).
302    pub fn leaf_only(config: Config) -> Result<Self, NodeError> {
303        let mut node = Self::new(config)?;
304        node.is_leaf_only = true;
305        node.bloom_state = BloomState::leaf_only(*node.identity.node_addr());
306        Ok(node)
307    }
308
309    pub(super) fn host_map_and_peer_acl(config: &Config) -> (Arc<HostMap>, acl::PeerAclReloader) {
310        let base_host_map = HostMap::from_peer_configs(config.peers());
311        if !config.node.system_files_enabled {
312            return (
313                Arc::new(base_host_map.clone()),
314                acl::PeerAclReloader::memory_only(base_host_map),
315            );
316        }
317
318        let mut host_map = base_host_map.clone();
319        let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH);
320        let hosts_file = HostMap::load_hosts_file(std::path::Path::new(
321            crate::upper::hosts::DEFAULT_HOSTS_PATH,
322        ));
323        host_map.merge(hosts_file);
324        let peer_acl = acl::PeerAclReloader::with_alias_sources(
325            std::path::PathBuf::from(acl::DEFAULT_PEERS_ALLOW_PATH),
326            std::path::PathBuf::from(acl::DEFAULT_PEERS_DENY_PATH),
327            base_host_map,
328            hosts_path,
329        );
330        (Arc::new(host_map), peer_acl)
331    }
332
333    /// Create transport instances from configuration.
334    ///
335    /// Returns a vector of TransportHandles for all configured transports.
336    pub(super) async fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec<TransportHandle> {
337        let mut transports = Vec::new();
338
339        // Collect UDP configs with optional names to avoid borrow conflicts
340        let udp_instances: Vec<_> = self
341            .config
342            .transports
343            .udp
344            .iter()
345            .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
346            .collect();
347
348        // Create UDP transport instances
349        for (name, udp_config) in udp_instances {
350            let transport_id = self.allocate_transport_id();
351            let udp = UdpTransport::new(transport_id, name, udp_config, packet_tx.clone());
352            transports.push(TransportHandle::Udp(udp));
353        }
354
355        #[cfg(feature = "sim-transport")]
356        {
357            let sim_instances: Vec<_> = self
358                .config
359                .transports
360                .sim
361                .iter()
362                .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
363                .collect();
364
365            for (name, sim_config) in sim_instances {
366                let transport_id = self.allocate_transport_id();
367                let sim = crate::transport::sim::SimTransport::new(
368                    transport_id,
369                    name,
370                    sim_config,
371                    packet_tx.clone(),
372                );
373                transports.push(TransportHandle::Sim(sim));
374            }
375        }
376
377        // Create Ethernet transport instances where raw-socket support exists.
378        #[cfg(any(target_os = "linux", target_os = "macos"))]
379        {
380            let eth_instances: Vec<_> = self
381                .config
382                .transports
383                .ethernet
384                .iter()
385                .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
386                .collect();
387            let xonly = self.identity.pubkey();
388            for (name, eth_config) in eth_instances {
389                let mut eth_config = eth_config;
390                if eth_config.discovery_scope.is_none() {
391                    eth_config.discovery_scope = self.lan_discovery_scope();
392                }
393                let transport_id = self.allocate_transport_id();
394                let mut eth =
395                    EthernetTransport::new(transport_id, name, eth_config, packet_tx.clone());
396                eth.set_local_pubkey(xonly);
397                transports.push(TransportHandle::Ethernet(eth));
398            }
399        }
400
401        // Create TCP transport instances
402        let tcp_instances: Vec<_> = self
403            .config
404            .transports
405            .tcp
406            .iter()
407            .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
408            .collect();
409
410        for (name, tcp_config) in tcp_instances {
411            let transport_id = self.allocate_transport_id();
412            let tcp = TcpTransport::new(transport_id, name, tcp_config, packet_tx.clone());
413            transports.push(TransportHandle::Tcp(tcp));
414        }
415
416        // Create Tor transport instances
417        let tor_instances: Vec<_> = self
418            .config
419            .transports
420            .tor
421            .iter()
422            .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
423            .collect();
424
425        for (name, tor_config) in tor_instances {
426            let transport_id = self.allocate_transport_id();
427            let tor = TorTransport::new(transport_id, name, tor_config, packet_tx.clone());
428            transports.push(TransportHandle::Tor(tor));
429        }
430
431        let webrtc_instances: Vec<_> = self
432            .config
433            .transports
434            .webrtc
435            .iter()
436            .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
437            .collect();
438
439        #[cfg(feature = "webrtc-transport")]
440        {
441            for (name, webrtc_config) in webrtc_instances {
442                let transport_id = self.allocate_transport_id();
443                match WebRtcTransport::new(
444                    transport_id,
445                    name,
446                    webrtc_config,
447                    packet_tx.clone(),
448                    &self.identity,
449                    &self.config.node.discovery.nostr,
450                ) {
451                    Ok(webrtc) => transports.push(TransportHandle::WebRtc(Box::new(webrtc))),
452                    Err(err) => {
453                        warn!(
454                            transport_id = %transport_id,
455                            error = %err,
456                            "failed to initialize WebRTC transport"
457                        );
458                    }
459                }
460            }
461        }
462        #[cfg(not(feature = "webrtc-transport"))]
463        if !webrtc_instances.is_empty() {
464            warn!("WebRTC transport configured but this build lacks WebRTC transport support");
465        }
466
467        // Create BLE transport instances
468        #[cfg(bluer_available)]
469        {
470            let ble_instances: Vec<_> = self
471                .config
472                .transports
473                .ble
474                .iter()
475                .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
476                .collect();
477
478            #[cfg(all(bluer_available, not(test)))]
479            for (name, ble_config) in ble_instances {
480                let transport_id = self.allocate_transport_id();
481                let adapter = ble_config.adapter().to_string();
482                let mtu = ble_config.mtu();
483                match crate::transport::ble::io::BluerIo::new(&adapter, mtu).await {
484                    Ok(io) => {
485                        let mut ble = crate::transport::ble::BleTransport::new(
486                            transport_id,
487                            name,
488                            ble_config,
489                            io,
490                            packet_tx.clone(),
491                        );
492                        ble.set_local_pubkey(self.identity.pubkey().serialize());
493                        transports.push(TransportHandle::Ble(ble));
494                    }
495                    Err(e) => {
496                        tracing::warn!(adapter = %adapter, error = %e, "failed to initialize BLE adapter");
497                    }
498                }
499            }
500
501            #[cfg(any(not(bluer_available), test))]
502            if !ble_instances.is_empty() {
503                #[cfg(not(test))]
504                tracing::warn!("BLE transport configured but this build lacks BlueZ support");
505            }
506        }
507
508        transports
509    }
510
511    /// Find an operational transport that matches the given transport type name.
512    ///
513    /// Adopted UDP bootstrap transports are point-to-point sockets handed off
514    /// from Nostr/STUN traversal. They must not be reused for ordinary
515    /// `udp host:port` dials discovered through static config, mDNS, or overlay
516    /// adverts: on macOS a `send_to` through the wrong adopted socket can fail
517    /// with `EINVAL`, and even on platforms that allow it the packet would use
518    /// the wrong 5-tuple/NAT mapping. Prefer configured transports and make the
519    /// choice deterministic by lowest transport id instead of HashMap order.
520    pub(super) fn find_transport_for_type(&self, transport_type: &str) -> Option<TransportId> {
521        self.transports
522            .iter()
523            .filter(|(id, handle)| {
524                handle.transport_type().name == transport_type
525                    && handle.is_operational()
526                    && !self.bootstrap_transports.contains(id)
527            })
528            .min_by_key(|(id, _)| id.as_u32())
529            .map(|(id, _)| *id)
530    }
531
532    /// Resolve an Ethernet peer address ("interface/mac") to a transport ID
533    /// and binary TransportAddr.
534    ///
535    /// Finds the Ethernet transport instance bound to the named interface
536    /// and parses the MAC portion into a 6-byte TransportAddr.
537    #[cfg(any(target_os = "linux", target_os = "macos"))]
538    pub(super) fn resolve_ethernet_addr(
539        &self,
540        addr_str: &str,
541    ) -> Result<(TransportId, TransportAddr), NodeError> {
542        let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| {
543            NodeError::NoTransportForType(format!(
544                "invalid Ethernet address format '{}': expected 'interface/mac'",
545                addr_str
546            ))
547        })?;
548
549        // Find the Ethernet transport bound to this interface
550        let transport_id = self
551            .transports
552            .iter()
553            .find(|(_, handle)| {
554                handle.transport_type().name == "ethernet"
555                    && handle.is_operational()
556                    && handle.interface_name() == Some(iface)
557            })
558            .map(|(id, _)| *id)
559            .ok_or_else(|| {
560                NodeError::NoTransportForType(format!(
561                    "no operational Ethernet transport for interface '{}'",
562                    iface
563                ))
564            })?;
565
566        let mac = crate::transport::ethernet::parse_mac_string(mac_str).map_err(|e| {
567            NodeError::NoTransportForType(format!("invalid MAC in '{}': {}", addr_str, e))
568        })?;
569
570        Ok((transport_id, TransportAddr::from_bytes(&mac)))
571    }
572
573    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
574    pub(super) fn resolve_ethernet_addr(
575        &self,
576        _addr_str: &str,
577    ) -> Result<(TransportId, TransportAddr), NodeError> {
578        Err(NodeError::NoTransportForType(
579            "Ethernet transport is not supported on this platform".to_string(),
580        ))
581    }
582
583    /// Resolve a BLE address string (`"adapter/AA:BB:CC:DD:EE:FF"`) to a
584    /// (TransportId, TransportAddr) pair by finding the BLE transport
585    /// instance matching the adapter name.
586    #[cfg(bluer_available)]
587    pub(super) fn resolve_ble_addr(
588        &self,
589        addr_str: &str,
590    ) -> Result<(TransportId, TransportAddr), NodeError> {
591        let ta = TransportAddr::from_string(addr_str);
592        let adapter = crate::transport::ble::addr::adapter_from_addr(&ta).ok_or_else(|| {
593            NodeError::NoTransportForType(format!(
594                "invalid BLE address format '{}': expected 'adapter/mac'",
595                addr_str
596            ))
597        })?;
598
599        // Find the BLE transport for this adapter
600        let transport_id = self
601            .transports
602            .iter()
603            .find(|(_, handle)| handle.transport_type().name == "ble" && handle.is_operational())
604            .map(|(id, _)| *id)
605            .ok_or_else(|| {
606                NodeError::NoTransportForType(format!(
607                    "no operational BLE transport for adapter '{}'",
608                    adapter
609                ))
610            })?;
611
612        // Validate the address format
613        crate::transport::ble::addr::BleAddr::parse(addr_str).map_err(|e| {
614            NodeError::NoTransportForType(format!("invalid BLE address '{}': {}", addr_str, e))
615        })?;
616
617        Ok((transport_id, TransportAddr::from_string(addr_str)))
618    }
619}
fips_core/node/core_impl.rs

fips_core/node/
core_impl.rs
