diff --git a/src/bin/fips.rs b/src/bin/fips.rs index 0a34c48..3f15e94 100644 --- a/src/bin/fips.rs +++ b/src/bin/fips.rs @@ -2,64 +2,10 @@ //! //! Loads configuration and creates the top-level node instance. -use fips::{ - build_dest_unreachable, log_ipv6_packet, should_send_icmp_error, shutdown_tun_interface, - Config, DestUnreachableCode, FipsAddress, Node, TunDevice, TunTx, -}; -use tracing::{debug, error, info, warn, Level}; +use fips::{Config, Node, TunState}; +use tracing::{error, info, warn, Level}; use tracing_subscriber::{fmt, EnvFilter}; -/// TUN packet reader loop. -/// -/// Reads packets from the TUN device, logs them, and sends ICMPv6 -/// Destination Unreachable responses for packets we can't route. -/// -/// This runs in a separate thread since TUN reads are blocking. -fn run_tun_reader(mut device: TunDevice, mtu: u16, our_addr: FipsAddress, tun_tx: TunTx) { - let mut buf = vec![0u8; mtu as usize + 100]; // Extra space for headers - - loop { - match device.read_packet(&mut buf) { - Ok(n) if n > 0 => { - let packet = &buf[..n]; - log_ipv6_packet(packet); - - // Currently no routing capability - send ICMPv6 Destination Unreachable - // for all packets that qualify for an error response - if should_send_icmp_error(packet) { - if let Some(response) = build_dest_unreachable( - packet, - DestUnreachableCode::NoRoute, - our_addr.to_ipv6(), - ) { - debug!( - len = response.len(), - "Sending ICMPv6 Destination Unreachable" - ); - if tun_tx.send(response).is_err() { - info!("TUN writer channel closed, reader stopping"); - break; - } - } - } - } - Ok(_) => { - // Zero-length read, continue - } - Err(e) => { - // "Bad address" (EFAULT) is expected during shutdown when interface is deleted - let err_str = e.to_string(); - if err_str.contains("Bad address") { - info!("TUN interface deleted, reader stopping"); - } else { - error!("TUN read error: {}", e); - } - break; - } - } - } -} - #[tokio::main(flavor = "current_thread")] async fn main() { // Initialize logging @@ -119,96 +65,40 @@ async fn main() { info!(" node_id: {}", hex::encode(node.node_id().as_bytes())); info!(" address: {}", node.identity().address()); - // Initialize TUN interface + // Start the node (initializes TUN, spawns I/O threads) info!( tun_state = %node.tun_state(), - "TUN interface" + "Starting node" ); - if node.tun_state() != fips::TunState::Disabled { - info!( - name = node.config().tun.name(), - mtu = node.config().tun.mtu(), - "Initializing TUN device" - ); + if let Err(e) = node.start().await { + error!("Failed to start node: {}", e); + std::process::exit(1); + } - match node.init_tun().await { - Ok(true) => { - let device = node.tun_device().unwrap(); - info!( - name = device.name(), - mtu = device.mtu(), - address = %device.address(), - "TUN device active" - ); - - // Show interface details for debugging - let output = std::process::Command::new("ip") - .args(["link", "show", device.name()]) - .output(); - match output { - Ok(out) => { - if out.status.success() { - info!( - "ip link show {}:\n{}", - device.name(), - String::from_utf8_lossy(&out.stdout) - ); - } - } - Err(e) => { - warn!("Failed to run ip link: {}", e); + // Show TUN interface details if active + if node.tun_state() == TunState::Active { + if let Some(tun_name) = node.config().tun.name.as_deref() { + let output = std::process::Command::new("ip") + .args(["link", "show", tun_name]) + .output(); + match output { + Ok(out) => { + if out.status.success() { + info!( + "ip link show {}:\n{}", + tun_name, + String::from_utf8_lossy(&out.stdout) + ); } } - } - Ok(false) => { - info!("TUN disabled"); - } - Err(e) => { - error!("Failed to initialize TUN: {}", e); - warn!("Continuing without TUN interface"); + Err(e) => { + warn!("Failed to run ip link: {}", e); + } } } } - info!("FIPS initialized successfully"); - - // Spawn TUN reader and writer threads if TUN is active - let tun_name = if let Some(tun_device) = node.take_tun_device() { - let mtu = tun_device.mtu(); - let name = tun_device.name().to_string(); - let our_addr = *tun_device.address(); - - // Create writer (dups the fd for independent write access) - let (writer, tun_tx) = match tun_device.create_writer() { - Ok(w) => w, - Err(e) => { - error!("Failed to create TUN writer: {}", e); - std::process::exit(1); - } - }; - - info!(mtu, name = %name, "Starting TUN reader and writer"); - - // Spawn writer thread - std::thread::spawn(move || { - writer.run(); - }); - - // Spawn reader thread - std::thread::spawn(move || { - run_tun_reader(tun_device, mtu, our_addr, tun_tx); - }); - - Some(name) - } else { - None - }; - - // TODO: Spawn additional event-driven tasks here: - // - Transport listeners/senders - // - Periodic timers (tree announcements, keepalives, etc.) - info!("FIPS running, press Ctrl+C to exit"); match tokio::signal::ctrl_c().await { @@ -218,12 +108,9 @@ async fn main() { info!("FIPS shutting down"); - // Shutdown TUN interface if active - if let Some(name) = tun_name { - info!(name = %name, "Shutting down TUN interface"); - if let Err(e) = shutdown_tun_interface(&name).await { - warn!("Failed to shutdown TUN interface: {}", e); - } + // Stop the node (shuts down TUN, stops I/O threads) + if let Err(e) = node.stop().await { + warn!("Error during shutdown: {}", e); } info!("FIPS shutdown complete"); diff --git a/src/node.rs b/src/node.rs index 4d4899d..6f58402 100644 --- a/src/node.rs +++ b/src/node.rs @@ -9,11 +9,13 @@ use crate::cache::CoordCache; use crate::peer::Peer; use crate::transport::{Link, LinkId, TransportId}; use crate::tree::TreeState; -use crate::tun::{TunDevice, TunError, TunState}; +use crate::tun::{run_tun_reader, shutdown_tun_interface, TunDevice, TunError, TunState, TunTx}; use crate::{Config, ConfigError, Identity, IdentityError, NodeId}; use std::collections::HashMap; use std::fmt; +use std::thread::{self, JoinHandle}; use thiserror::Error; +use tracing::{info, warn}; /// Errors related to node operations. #[derive(Debug, Error)] @@ -156,8 +158,14 @@ pub struct Node { // === TUN Interface === /// TUN device state. tun_state: TunState, - /// TUN device (if active). - tun_device: Option, + /// TUN interface name (for cleanup). + tun_name: Option, + /// TUN packet sender channel. + tun_tx: Option, + /// TUN reader thread handle. + tun_reader_handle: Option>, + /// TUN writer thread handle. + tun_writer_handle: Option>, } impl Node { @@ -179,12 +187,24 @@ impl Node { TunState::Disabled }; + // Initialize tree state with signed self-declaration + let mut tree_state = TreeState::new(node_id); + tree_state + .sign_declaration(&identity) + .expect("signing own declaration should never fail"); + + info!( + node_id = %node_id, + address = %identity.address(), + "Node initialized as root" + ); + Ok(Self { identity, config, state: NodeState::Created, is_leaf_only, - tree_state: TreeState::new(node_id), + tree_state, bloom_state, coord_cache: CoordCache::with_defaults(), transport_ids: Vec::new(), @@ -195,7 +215,10 @@ impl Node { next_link_id: 1, next_transport_id: 1, tun_state, - tun_device: None, + tun_name: None, + tun_tx: None, + tun_reader_handle: None, + tun_writer_handle: None, }) } @@ -207,12 +230,25 @@ impl Node { } else { TunState::Disabled }; + + // Initialize tree state with signed self-declaration + let mut tree_state = TreeState::new(node_id); + tree_state + .sign_declaration(&identity) + .expect("signing own declaration should never fail"); + + info!( + node_id = %node_id, + address = %identity.address(), + "Node initialized as root" + ); + Self { identity, config, state: NodeState::Created, is_leaf_only: false, - tree_state: TreeState::new(node_id), + tree_state, bloom_state: BloomState::new(node_id), coord_cache: CoordCache::with_defaults(), transport_ids: Vec::new(), @@ -223,7 +259,10 @@ impl Node { next_link_id: 1, next_transport_id: 1, tun_state, - tun_device: None, + tun_name: None, + tun_tx: None, + tun_reader_handle: None, + tun_writer_handle: None, } } @@ -319,48 +358,6 @@ impl Node { self.tun_state } - /// Get the TUN device if active. - pub fn tun_device(&self) -> Option<&TunDevice> { - self.tun_device.as_ref() - } - - /// Get mutable TUN device if active. - pub fn tun_device_mut(&mut self) -> Option<&mut TunDevice> { - self.tun_device.as_mut() - } - - /// Take ownership of the TUN device. - /// - /// This removes the TUN device from the node, transferring ownership - /// to the caller. Useful for moving the device into a reader task. - pub fn take_tun_device(&mut self) -> Option { - self.tun_device.take() - } - - /// Initialize the TUN interface. - /// - /// Creates and configures the TUN device based on the node's configuration. - /// Requires CAP_NET_ADMIN capability (run with sudo or setcap). - /// - /// Returns Ok(true) if TUN was initialized, Ok(false) if TUN is disabled. - pub async fn init_tun(&mut self) -> Result { - if !self.config.tun.enabled { - return Ok(false); - } - - let address = *self.identity.address(); - match TunDevice::create(&self.config.tun, address).await { - Ok(device) => { - self.tun_device = Some(device); - self.tun_state = TunState::Active; - Ok(true) - } - Err(e) => { - self.tun_state = TunState::Failed; - Err(e.into()) - } - } - } // === Resource Limits === @@ -525,38 +522,117 @@ impl Node { // === State Transitions === - /// Start the node (stub). + /// Start the node. /// - /// In a full implementation, this would: - /// - Initialize transports - /// - Bind TUN interface - /// - Start event loop - pub fn start(&mut self) -> Result<(), NodeError> { + /// Initializes the TUN interface (if configured), spawns I/O threads, + /// and transitions to the Running state. + pub async fn start(&mut self) -> Result<(), NodeError> { if !self.state.can_start() { return Err(NodeError::AlreadyStarted); } self.state = NodeState::Starting; - // Actual startup would initialize transports, TUN, etc. + + // Initialize TUN interface if configured + if self.config.tun.enabled { + let address = *self.identity.address(); + match TunDevice::create(&self.config.tun, address).await { + Ok(device) => { + let mtu = device.mtu(); + let name = device.name().to_string(); + let our_addr = *device.address(); + + info!( + name = %name, + mtu, + address = %device.address(), + "TUN device active" + ); + + // Create writer (dups the fd for independent write access) + let (writer, tun_tx) = device.create_writer()?; + + info!(mtu, name = %name, "Starting TUN reader and writer"); + + // Spawn writer thread + let writer_handle = thread::spawn(move || { + writer.run(); + }); + + // Clone tun_tx for the reader + let reader_tun_tx = tun_tx.clone(); + + // Spawn reader thread + let reader_handle = thread::spawn(move || { + run_tun_reader(device, mtu, our_addr, reader_tun_tx); + }); + + self.tun_state = TunState::Active; + self.tun_name = Some(name); + self.tun_tx = Some(tun_tx); + self.tun_reader_handle = Some(reader_handle); + self.tun_writer_handle = Some(writer_handle); + } + Err(e) => { + self.tun_state = TunState::Failed; + warn!(error = %e, "Failed to initialize TUN, continuing without it"); + } + } + } + + // TODO: Initialize transports here + self.state = NodeState::Running; + info!(state = %self.state, "Node started"); Ok(()) } - /// Stop the node (stub). + /// Stop the node. /// - /// In a full implementation, this would: - /// - Close all peers - /// - Close all links - /// - Stop all transports - /// - Unbind TUN interface - pub fn stop(&mut self) -> Result<(), NodeError> { + /// Shuts down TUN interface, stops I/O threads, and transitions to + /// the Stopped state. + pub async fn stop(&mut self) -> Result<(), NodeError> { if !self.state.can_stop() { return Err(NodeError::NotStarted); } self.state = NodeState::Stopping; - // Actual shutdown would close transports, links, etc. + info!(state = %self.state, "Node stopping"); + + // Shutdown TUN interface + if let Some(name) = self.tun_name.take() { + info!(name = %name, "Shutting down TUN interface"); + + // Drop the tun_tx to signal the writer to stop + self.tun_tx.take(); + + // Delete the interface (causes reader to get EFAULT) + if let Err(e) = shutdown_tun_interface(&name).await { + warn!(name = %name, error = %e, "Failed to shutdown TUN interface"); + } + + // Wait for threads to finish + if let Some(handle) = self.tun_reader_handle.take() { + let _ = handle.join(); + } + if let Some(handle) = self.tun_writer_handle.take() { + let _ = handle.join(); + } + + self.tun_state = TunState::Disabled; + } + + // TODO: Shutdown transports here + self.state = NodeState::Stopped; + info!(state = %self.state, "Node stopped"); Ok(()) } + + /// Get the TUN packet sender channel. + /// + /// Returns None if TUN is not active or the node hasn't been started. + pub fn tun_tx(&self) -> Option<&TunTx> { + self.tun_tx.as_ref() + } } impl fmt::Debug for Node { @@ -620,36 +696,39 @@ mod tests { assert!(node.bloom_state().is_leaf_only()); } - #[test] - fn test_node_state_transitions() { + #[tokio::test] + async fn test_node_state_transitions() { let mut node = make_node(); assert!(!node.is_running()); assert!(node.state().can_start()); - node.start().unwrap(); + node.start().await.unwrap(); assert!(node.is_running()); assert!(!node.state().can_start()); - node.stop().unwrap(); + node.stop().await.unwrap(); assert!(!node.is_running()); assert_eq!(node.state(), NodeState::Stopped); } - #[test] - fn test_node_double_start() { + #[tokio::test] + async fn test_node_double_start() { let mut node = make_node(); - node.start().unwrap(); + node.start().await.unwrap(); - let result = node.start(); + let result = node.start().await; assert!(matches!(result, Err(NodeError::AlreadyStarted))); + + // Clean up + node.stop().await.unwrap(); } - #[test] - fn test_node_stop_not_started() { + #[tokio::test] + async fn test_node_stop_not_started() { let mut node = make_node(); - let result = node.stop(); + let result = node.stop().await; assert!(matches!(result, Err(NodeError::NotStarted))); } diff --git a/src/tree.rs b/src/tree.rs index 1a12bf8..a3d5362 100644 --- a/src/tree.rs +++ b/src/tree.rs @@ -4,7 +4,7 @@ //! The spanning tree provides a routing topology where each node maintains //! a path to a common root, enabling greedy distance-based routing. -use crate::{IdentityError, NodeId}; +use crate::{Identity, IdentityError, NodeId}; use secp256k1::schnorr::Signature; use secp256k1::XOnlyPublicKey; use std::collections::HashMap; @@ -119,6 +119,19 @@ impl ParentDeclaration { self.signature = Some(signature); } + /// Sign this declaration with the given identity. + /// + /// The identity's node_id must match this declaration's node_id. + /// Returns an error if the node_ids don't match. + pub fn sign(&mut self, identity: &Identity) -> Result<(), TreeError> { + if identity.node_id() != &self.node_id { + return Err(TreeError::InvalidSignature(self.node_id)); + } + let signature = identity.sign(&self.signing_bytes()); + self.signature = Some(signature); + Ok(()) + } + /// Check if this is a root declaration (parent == self). pub fn is_root(&self) -> bool { self.node_id == self.parent_id @@ -366,8 +379,13 @@ impl TreeState { /// Create initial tree state for a node (as root candidate). /// /// The node starts as its own root until it learns of a smaller node_id. + /// Initial sequence is 1 per protocol spec; timestamp is current Unix time. pub fn new(my_node_id: NodeId) -> Self { - let my_declaration = ParentDeclaration::self_root(my_node_id, 0, 0); + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let my_declaration = ParentDeclaration::self_root(my_node_id, 1, timestamp); let my_coords = TreeCoordinate::root(my_node_id); Self { @@ -502,6 +520,18 @@ impl TreeState { // Stub: would evaluate parent switch criteria false } + + /// Sign this node's declaration with the given identity. + /// + /// The identity's node_id must match this TreeState's node_id. + pub fn sign_declaration(&mut self, identity: &Identity) -> Result<(), TreeError> { + self.my_declaration.sign(identity) + } + + /// Check if this node's declaration is signed. + pub fn is_declaration_signed(&self) -> bool { + self.my_declaration.is_signed() + } } impl fmt::Debug for TreeState { diff --git a/src/tun.rs b/src/tun.rs index bae7b4a..83dd98f 100644 --- a/src/tun.rs +++ b/src/tun.rs @@ -224,6 +224,72 @@ impl TunWriter { } } +/// TUN packet reader loop. +/// +/// Reads packets from the TUN device, logs them, and sends ICMPv6 +/// Destination Unreachable responses for packets we can't route. +/// +/// This is designed to run in a dedicated thread since TUN reads are blocking. +/// The loop exits when the TUN interface is deleted (EFAULT) or an unrecoverable +/// error occurs. +pub fn run_tun_reader( + mut device: TunDevice, + mtu: u16, + our_addr: FipsAddress, + tun_tx: TunTx, +) { + use crate::icmp::{build_dest_unreachable, should_send_icmp_error, DestUnreachableCode}; + + let name = device.name().to_string(); + let mut buf = vec![0u8; mtu as usize + 100]; // Extra space for headers + + info!(name = %name, "TUN reader starting"); + + loop { + match device.read_packet(&mut buf) { + Ok(n) if n > 0 => { + let packet = &buf[..n]; + log_ipv6_packet(packet); + + // Currently no routing capability - send ICMPv6 Destination Unreachable + // for all packets that qualify for an error response + if should_send_icmp_error(packet) { + if let Some(response) = build_dest_unreachable( + packet, + DestUnreachableCode::NoRoute, + our_addr.to_ipv6(), + ) { + debug!( + name = %name, + len = response.len(), + "Sending ICMPv6 Destination Unreachable" + ); + if tun_tx.send(response).is_err() { + info!(name = %name, "TUN writer channel closed, reader stopping"); + break; + } + } + } + } + Ok(_) => { + // Zero-length read, continue + } + Err(e) => { + // "Bad address" (EFAULT) is expected during shutdown when interface is deleted + let err_str = format!("{}", e); + if err_str.contains("Bad address") { + info!(name = %name, "TUN interface deleted, reader stopping"); + } else { + error!(name = %name, error = %e, "TUN read error"); + } + break; + } + } + } + + info!(name = %name, "TUN reader stopped"); +} + /// Log basic information about an IPv6 packet at DEBUG level. pub fn log_ipv6_packet(packet: &[u8]) { if packet.len() < 40 {