fips_core/node/
core_impl.rs

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

fips_core/node/
core_impl.rs
