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

fips_core/node/
core_impl.rs
