fips_core/node/
core_impl.rs

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

fips_core/node/
core_impl.rs
