fips_core/node/lifecycle/
runtime.rs

1use super::*;
2
3impl Node {
4    // === State Transitions ===
5
6    /// Start the node.
7    ///
8    /// Initializes the TUN interface (if configured), spawns I/O threads,
9    /// and transitions to the Running state.
10    pub async fn start(&mut self) -> Result<(), NodeError> {
11        node_start_debug_log("Node::start begin");
12        if !self.state.can_start() {
13            return Err(NodeError::AlreadyStarted);
14        }
15        self.state = NodeState::Starting;
16        node_start_debug_log("Node::start state set to starting");
17
18        // Create packet channel for transport -> Node communication
19        let packet_buffer_size = self.config.node.buffers.packet_channel;
20        let (mut packet_tx, packet_rx) = packet_channel(packet_buffer_size);
21        self.packet_mover2_fast_ingress_rx = Some(
22            self.packet_mover2
23                .attach_established_fast_ingress(&mut packet_tx),
24        );
25        self.packet_tx = Some(packet_tx.clone());
26        self.packet_rx = Some(packet_rx);
27        node_start_debug_log("Node::start packet channel created");
28
29        // Initialize transports first (before TUN, before Nostr discovery).
30        node_start_debug_log("Node::start create transports begin");
31        let transport_handles = self.create_transports(&packet_tx).await;
32        node_start_debug_log(format!(
33            "Node::start create transports complete count={}",
34            transport_handles.len()
35        ));
36
37        for mut handle in transport_handles {
38            let transport_id = handle.transport_id();
39            let transport_type = handle.transport_type().name;
40            let name = handle.name().map(|s| s.to_string());
41
42            node_start_debug_log(format!(
43                "Node::start transport start begin id={} type={} name={:?}",
44                transport_id, transport_type, name
45            ));
46            match handle.start().await {
47                Ok(()) => {
48                    node_start_debug_log(format!(
49                        "Node::start transport start ok id={} type={}",
50                        transport_id, transport_type
51                    ));
52                    self.udp_transport_resolution_cache.clear();
53                    self.transports.insert(transport_id, handle);
54                }
55                Err(e) => {
56                    node_start_debug_log(format!(
57                        "Node::start transport start error id={} type={} error={}",
58                        transport_id, transport_type, e
59                    ));
60                    if let Some(ref n) = name {
61                        warn!(transport_type, name = %n, error = %e, "Transport failed to start");
62                    } else {
63                        warn!(transport_type, error = %e, "Transport failed to start");
64                    }
65                }
66            }
67        }
68
69        if !self.transports.is_empty() {
70            info!(count = self.transports.len(), "Transports initialized");
71        }
72
73        if self.config.node.discovery.nostr.enabled {
74            node_start_debug_log("Node::start nostr discovery start begin");
75            match NostrDiscovery::start(&self.identity, self.config.node.discovery.nostr.clone())
76                .await
77            {
78                Ok(runtime) => {
79                    node_start_debug_log("Node::start nostr discovery runtime created");
80                    if let Err(err) = self.refresh_overlay_advert(&runtime).await {
81                        warn!(error = %err, "Failed to publish initial Nostr overlay advert");
82                    }
83                    node_start_debug_log("Node::start nostr overlay advert refreshed");
84                    self.nostr_discovery = Some(runtime);
85                    self.nostr_discovery_started_at_ms = Some(Self::now_ms());
86                    info!("Nostr overlay discovery enabled");
87                }
88                Err(err) => {
89                    node_start_debug_log(format!(
90                        "Node::start nostr discovery start error error={}",
91                        err
92                    ));
93                    warn!(error = %err, "Failed to start Nostr overlay discovery");
94                }
95            }
96        }
97
98        // mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even
99        // when Nostr is disabled, since it gives us sub-second pairing
100        // on the same link without any relay or NAT-traversal roundtrip.
101        if self.config.node.discovery.lan.enabled {
102            node_start_debug_log("Node::start lan discovery start begin");
103            let advertised_udp_port = self
104                .transports
105                .values()
106                .filter(|h| h.is_operational())
107                .filter(|h| h.transport_type().name == "udp")
108                .find_map(|h| h.local_addr().map(|addr| addr.port()))
109                .unwrap_or(0);
110            let scope = self.lan_discovery_scope();
111            match crate::discovery::lan::LanDiscovery::start(
112                &self.identity,
113                scope,
114                advertised_udp_port,
115                self.config.node.discovery.lan.clone(),
116            )
117            .await
118            {
119                Ok(runtime) => {
120                    node_start_debug_log("Node::start lan discovery start ok");
121                    self.lan_discovery = Some(runtime);
122                    info!("LAN mDNS discovery enabled");
123                }
124                Err(err) => {
125                    node_start_debug_log(format!(
126                        "Node::start lan discovery start error error={}",
127                        err
128                    ));
129                    debug!(error = %err, "LAN mDNS discovery not started");
130                }
131            }
132        }
133
134        self.start_local_instance_discovery();
135        self.poll_local_instance_discovery().await;
136
137        // Connect to static peers before TUN is active
138        // This allows handshake messages to be sent before we start accepting packets
139        node_start_debug_log("Node::start initiate peer connections begin");
140        self.initiate_peer_connections().await;
141        node_start_debug_log("Node::start initiate peer connections complete");
142
143        // Initialize TUN interface last, after transports and peers are ready
144        if self.config.tun.enabled {
145            node_start_debug_log("Node::start tun init begin");
146            let address = *self.identity.address();
147            match TunDevice::create(&self.config.tun, address).await {
148                Ok(device) => {
149                    let mtu = device.mtu();
150                    let name = device.name().to_string();
151                    let our_addr = *device.address();
152
153                    info!("TUN device active:");
154                    info!("     name: {}", name);
155                    info!("  address: {}", device.address());
156                    info!("      mtu: {}", mtu);
157
158                    // Calculate max MSS for TCP clamping
159                    let effective_mtu = self.effective_ipv6_mtu();
160                    let max_mss = effective_mtu.saturating_sub(40).saturating_sub(20); // IPv6 + TCP headers
161
162                    info!("effective MTU: {} bytes", effective_mtu);
163                    debug!("   max TCP MSS: {} bytes", max_mss);
164
165                    // On macOS, create a shutdown pipe. Writing to it unblocks the
166                    // reader thread's select() loop without closing the TUN fd
167                    // (which would cause a double-close when TunDevice drops).
168                    #[cfg(target_os = "macos")]
169                    let (shutdown_read_fd, shutdown_write_fd) = {
170                        let mut fds = [0i32; 2];
171                        if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 {
172                            return Err(NodeError::Tun(crate::upper::tun::TunError::Configure(
173                                "failed to create shutdown pipe".into(),
174                            )));
175                        }
176                        (fds[0], fds[1])
177                    };
178
179                    // Create writer (dups the fd for independent write access).
180                    // Pass path_mtu_lookup so inbound SYN-ACK clamp can read
181                    // per-destination path MTU learned via discovery.
182                    let (writer, tun_tx) =
183                        device.create_writer(max_mss, self.path_mtu_lookup.clone())?;
184
185                    // Spawn writer thread
186                    let writer_handle = thread::spawn(move || {
187                        writer.run();
188                    });
189
190                    // Clone tun_tx for the reader
191                    let reader_tun_tx = tun_tx.clone();
192
193                    // Create outbound channel for TUN reader → Node
194                    let tun_channel_size = self.config.node.buffers.tun_channel;
195                    let (outbound_tx, outbound_rx) =
196                        crate::upper::tun::tun_outbound_channel(tun_channel_size);
197
198                    // Spawn reader thread
199                    let transport_mtu = self.transport_mtu();
200                    let path_mtu_lookup = self.path_mtu_lookup.clone();
201                    #[cfg(target_os = "macos")]
202                    let reader_handle = thread::spawn(move || {
203                        run_tun_reader(
204                            device,
205                            mtu,
206                            our_addr,
207                            reader_tun_tx,
208                            outbound_tx,
209                            transport_mtu,
210                            path_mtu_lookup,
211                            shutdown_read_fd,
212                        );
213                    });
214                    #[cfg(not(target_os = "macos"))]
215                    let reader_handle = thread::spawn(move || {
216                        run_tun_reader(
217                            device,
218                            mtu,
219                            our_addr,
220                            reader_tun_tx,
221                            outbound_tx,
222                            transport_mtu,
223                            path_mtu_lookup,
224                        );
225                    });
226
227                    self.tun_state = TunState::Active;
228                    self.tun_name = Some(name);
229                    self.tun_tx = Some(tun_tx);
230                    self.tun_outbound_rx = Some(outbound_rx);
231                    self.tun_reader_handle = Some(reader_handle);
232                    self.tun_writer_handle = Some(writer_handle);
233                    #[cfg(target_os = "macos")]
234                    {
235                        self.tun_shutdown_fd = Some(shutdown_write_fd);
236                    }
237                }
238                Err(e) => {
239                    self.tun_state = TunState::Failed;
240                    warn!(error = %e, "Failed to initialize TUN, continuing without it");
241                }
242            }
243            node_start_debug_log("Node::start tun init complete");
244        }
245
246        // Initialize DNS responder (independent of TUN).
247        //
248        // Default bind_addr is "::1" (IPv6 loopback). The shipped
249        // fips-dns-setup configures systemd-resolved via a global
250        // /etc/systemd/resolved.conf.d/fips.conf drop-in pointing at
251        // [::1]:5354, which sidesteps a Linux IPV6_PKTINFO behaviour
252        // where self-destined traffic to fips0's address is attributed
253        // to fips0 in PKTINFO and gets silently dropped by the
254        // mesh-interface filter in src/upper/dns.rs.
255        //
256        // For mesh-reachable resolution (rare), set bind_addr: "::"
257        // in fips.yaml. The mesh-interface filter remains active to
258        // prevent hosts-file alias enumeration in that mode.
259        // `IPV6_V6ONLY=0` is set explicitly so IPv4 clients on
260        // 127.0.0.1 still reach us regardless of kernel sysctl
261        // defaults — but only when bind is on a wildcard / IPv6 path.
262        if self.config.dns.enabled {
263            node_start_debug_log("Node::start dns init begin");
264            let addr_str = self.config.dns.bind_addr();
265            match addr_str.parse::<std::net::IpAddr>() {
266                Ok(ip) => {
267                    let bind = std::net::SocketAddr::new(ip, self.config.dns.port());
268                    match Self::bind_dns_socket(bind) {
269                        Ok(socket) => {
270                            let dns_channel_size = self.config.node.buffers.dns_channel;
271                            let (identity_tx, identity_rx) =
272                                tokio::sync::mpsc::channel(dns_channel_size);
273                            let dns_ttl = self.config.dns.ttl();
274                            let base_hosts = crate::upper::hosts::HostMap::from_peer_configs(
275                                self.config.peers(),
276                            );
277                            let reloader = if self.config.node.system_files_enabled {
278                                let hosts_path = std::path::PathBuf::from(
279                                    crate::upper::hosts::DEFAULT_HOSTS_PATH,
280                                );
281                                crate::upper::hosts::HostMapReloader::new(base_hosts, hosts_path)
282                            } else {
283                                crate::upper::hosts::HostMapReloader::memory_only(base_hosts)
284                            };
285                            // Resolve the TUN ifindex so the responder can
286                            // drop queries arriving on the mesh interface
287                            // (fips0). Without this, the `::` bind exposes
288                            // /etc/fips/hosts alias probing to any mesh peer.
289                            // When TUN isn't enabled or the name can't be
290                            // resolved, `None` disables the filter (there
291                            // is no mesh surface to defend anyway).
292                            let mesh_ifindex = Self::lookup_mesh_ifindex(self.config.tun.name());
293                            info!(
294                                bind = %bind,
295                                hosts = reloader.hosts().len(),
296                                mesh_ifindex = ?mesh_ifindex,
297                                "DNS responder started for .fips domain (auto-reload enabled)"
298                            );
299                            let handle = tokio::spawn(crate::upper::dns::run_dns_responder(
300                                socket,
301                                identity_tx,
302                                dns_ttl,
303                                reloader,
304                                mesh_ifindex,
305                            ));
306                            self.dns_identity_rx = Some(identity_rx);
307                            self.dns_task = Some(handle);
308                        }
309                        Err(e) => {
310                            warn!(bind = %bind, error = %e, "Failed to start DNS responder");
311                        }
312                    }
313                }
314                Err(e) => {
315                    warn!(addr = %addr_str, error = %e, "Invalid dns.bind_addr; DNS responder not started");
316                }
317            }
318            node_start_debug_log("Node::start dns init complete");
319        }
320
321        self.state = NodeState::Running;
322        node_start_debug_log("Node::start running");
323        info!("Node started:");
324        info!("       state: {}", self.state);
325        info!("  transports: {}", self.transports.len());
326        info!(" connections: {}", self.peers.connection_len());
327        Ok(())
328    }
329
330    /// Bind a UDP socket for the DNS responder.
331    ///
332    /// For IPv6 binds (including `::`), sets `IPV6_V6ONLY=0` so the socket
333    /// also accepts IPv4-mapped addresses. This guarantees dual-stack
334    /// delivery regardless of `net.ipv6.bindv6only` sysctl on the host —
335    /// v4 clients on 127.0.0.1 and v6 clients on the fips0 address both
336    /// land on the same socket.
337    ///
338    /// Also enables `IPV6_RECVPKTINFO` on IPv6 sockets so the responder
339    /// can learn the arrival interface per packet. The responder uses that
340    /// to drop queries arriving on the mesh TUN, closing the hosts-file
341    /// probing side-channel created by the `::` bind.
342    pub(super) fn bind_dns_socket(
343        addr: std::net::SocketAddr,
344    ) -> Result<tokio::net::UdpSocket, std::io::Error> {
345        use socket2::{Domain, Protocol, Socket, Type};
346        let domain = if addr.is_ipv4() {
347            Domain::IPV4
348        } else {
349            Domain::IPV6
350        };
351        let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
352        if addr.is_ipv6() {
353            sock.set_only_v6(false)?;
354            #[cfg(unix)]
355            Self::set_recv_pktinfo_v6(&sock)?;
356        }
357        sock.set_nonblocking(true)?;
358        sock.bind(&addr.into())?;
359        tokio::net::UdpSocket::from_std(sock.into())
360    }
361
362    /// Enable `IPV6_RECVPKTINFO` on an IPv6 UDP socket.
363    ///
364    /// After this setsockopt, each `recvmsg()` call on the socket receives
365    /// an `IPV6_PKTINFO` control message containing the arrival interface
366    /// index, which the DNS responder uses for its mesh-interface filter.
367    #[cfg(unix)]
368    pub(super) fn set_recv_pktinfo_v6(sock: &socket2::Socket) -> Result<(), std::io::Error> {
369        use std::os::fd::AsRawFd;
370        let enable: libc::c_int = 1;
371        let ret = unsafe {
372            libc::setsockopt(
373                sock.as_raw_fd(),
374                libc::IPPROTO_IPV6,
375                libc::IPV6_RECVPKTINFO,
376                &enable as *const _ as *const libc::c_void,
377                std::mem::size_of::<libc::c_int>() as libc::socklen_t,
378            )
379        };
380        if ret < 0 {
381            return Err(std::io::Error::last_os_error());
382        }
383        Ok(())
384    }
385
386    /// Resolve the mesh TUN interface index by name.
387    ///
388    /// Returns `None` if the interface does not exist (e.g. TUN disabled
389    /// or not yet created). A `None` result disables the DNS responder's
390    /// mesh-interface filter — safe, because if there is no fips0 there
391    /// is no mesh exposure to defend against.
392    pub(super) fn lookup_mesh_ifindex(name: &str) -> Option<u32> {
393        #[cfg(unix)]
394        {
395            let c_name = std::ffi::CString::new(name).ok()?;
396            let idx = unsafe { libc::if_nametoindex(c_name.as_ptr()) };
397            if idx == 0 { None } else { Some(idx) }
398        }
399        #[cfg(not(unix))]
400        {
401            let _ = name;
402            None
403        }
404    }
405
406    /// Stop the node.
407    ///
408    /// Shuts down TUN interface, stops I/O threads, and transitions to
409    /// the Stopped state.
410    pub async fn stop(&mut self) -> Result<(), NodeError> {
411        if !self.state.can_stop() {
412            return Err(NodeError::NotStarted);
413        }
414        self.state = NodeState::Stopping;
415        info!(state = %self.state, "Node stopping");
416
417        // Stop DNS responder
418        if let Some(handle) = self.dns_task.take() {
419            handle.abort();
420            debug!("DNS responder stopped");
421        }
422
423        // Send disconnect notifications to all active peers before closing transports
424        self.send_disconnect_to_all_peers(DisconnectReason::Shutdown)
425            .await;
426
427        // Stop Nostr overlay discovery background work and withdraw any advert.
428        if let Some(bootstrap) = self.nostr_discovery.take()
429            && let Err(e) = bootstrap.shutdown().await
430        {
431            warn!(error = %e, "Failed to shutdown Nostr overlay discovery");
432        }
433
434        // Tear down LAN mDNS responder + browser. Best-effort: the
435        // OS will eventually time the advert out via its TTL even if
436        // we don't get a clean unregister out before the daemon exits.
437        if let Some(lan) = self.lan_discovery.take() {
438            lan.shutdown().await;
439        }
440
441        if let Some(registry) = self.local_instance_registry.take()
442            && let Err(err) = registry.remove()
443        {
444            debug!(error = %err, "failed to remove same-host FIPS instance record");
445        }
446
447        // Shutdown transports (they're packet producers)
448        let transport_ids: Vec<_> = self.transports.keys().cloned().collect();
449        for transport_id in transport_ids {
450            if let Some(mut handle) = self.transports.remove(&transport_id) {
451                self.udp_transport_resolution_cache.clear();
452                let transport_type = handle.transport_type().name;
453                match handle.stop().await {
454                    Ok(()) => {
455                        info!(transport_id = %transport_id, transport_type, "Transport stopped");
456                    }
457                    Err(e) => {
458                        warn!(
459                            transport_id = %transport_id,
460                            transport_type,
461                            error = %e,
462                            "Transport stop failed"
463                        );
464                    }
465                }
466            }
467        }
468
469        // Drop packet channels
470        self.packet_tx.take();
471        self.packet_rx.take();
472
473        // Shutdown TUN interface
474        if let Some(name) = self.tun_name.take() {
475            info!(name = %name, "Shutting down TUN interface");
476
477            // Drop the tun_tx to signal the writer to stop
478            self.tun_tx.take();
479
480            // Delete the interface (on Linux, causes reader to get EFAULT)
481            if let Err(e) = shutdown_tun_interface(&name).await {
482                warn!(name = %name, error = %e, "Failed to shutdown TUN interface");
483            }
484
485            // On macOS, signal the reader thread to exit by writing to the
486            // shutdown pipe. The reader's select() will wake up and break.
487            #[cfg(target_os = "macos")]
488            if let Some(fd) = self.tun_shutdown_fd.take() {
489                unsafe {
490                    libc::write(fd, b"x".as_ptr() as *const libc::c_void, 1);
491                    libc::close(fd);
492                }
493            }
494
495            // Wait for threads to finish
496            if let Some(handle) = self.tun_reader_handle.take() {
497                let _ = handle.join();
498            }
499            if let Some(handle) = self.tun_writer_handle.take() {
500                let _ = handle.join();
501            }
502
503            self.tun_state = TunState::Disabled;
504        }
505
506        self.state = NodeState::Stopped;
507        info!(state = %self.state, "Node stopped");
508        Ok(())
509    }
510
511    /// Send disconnect notifications to all active peers.
512    ///
513    /// Best-effort: send failures are logged and ignored since the transport
514    /// may already be degraded. This runs before transports are shut down.
515    pub(super) async fn send_disconnect_to_all_peers(&mut self, reason: DisconnectReason) {
516        // Collect node_addrs to avoid borrow conflict with send helper
517        let peer_addrs: Vec<NodeAddr> = self
518            .peers
519            .iter()
520            .filter(|(_, peer)| peer.can_send() && peer.has_session())
521            .map(|(addr, _)| *addr)
522            .collect();
523
524        if peer_addrs.is_empty() {
525            debug!(
526                total_peers = self.peers.len(),
527                "No sendable peers for disconnect notification"
528            );
529            return;
530        }
531
532        let mut sent = 0usize;
533        for node_addr in &peer_addrs {
534            if self.send_disconnect_to_peer(node_addr, reason).await {
535                sent += 1;
536            }
537        }
538
539        info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications");
540    }
541
542    /// Send a Disconnect notification to one peer, swallowing transport failures.
543    pub(super) async fn send_disconnect_to_peer(
544        &mut self,
545        node_addr: &NodeAddr,
546        reason: DisconnectReason,
547    ) -> bool {
548        let plaintext = Disconnect::new(reason).encode();
549        match self
550            .send_packet_mover2_fmp_link_plaintext(node_addr, &plaintext, false)
551            .await
552        {
553            Ok(()) => true,
554            Err(e) => {
555                debug!(
556                    peer = %self.peer_display_name(node_addr),
557                    error = %e,
558                    "Failed to send disconnect (transport may be down)"
559                );
560                false
561            }
562        }
563    }
564}
fips_core/node/lifecycle/runtime.rs

fips_core/node/lifecycle/
runtime.rs
