fips_core/node/
core_impl.rs

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

fips_core/node/
core_impl.rs
