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

fips_core/node/
core_impl.rs
