Spanning tree init and Node lifecycle refactoring

Spanning tree initialization:
- TreeState now uses sequence=1 and current timestamp per protocol spec
- Added ParentDeclaration::sign() and TreeState::sign_declaration() methods
- Node initialization logs identity (node_id, address)

Node lifecycle refactoring:
- Moved TUN init and thread spawning into async Node::start()
- Moved TUN shutdown and thread cleanup into async Node::stop()
- Node owns I/O infrastructure (thread handles, TunTx channel)
- Removed: init_tun(), take_tun_device(), tun_device(), tun_device_mut()
- Added: tun_tx() accessor for packet sending

tun.rs:
- Added run_tun_reader() function (moved from binary)

fips.rs binary reduced from 230 to 117 lines. All 171 tests pass.
This commit is contained in:
Johnathan Corgan
2026-01-30 17:40:33 +00:00
parent ca24ba02b7
commit c421dad525
4 changed files with 280 additions and 218 deletions
+28 -141
View File
@@ -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");
+154 -75
View File
@@ -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<TunDevice>,
/// TUN interface name (for cleanup).
tun_name: Option<String>,
/// TUN packet sender channel.
tun_tx: Option<TunTx>,
/// TUN reader thread handle.
tun_reader_handle: Option<JoinHandle<()>>,
/// TUN writer thread handle.
tun_writer_handle: Option<JoinHandle<()>>,
}
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<TunDevice> {
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<bool, NodeError> {
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)));
}
+32 -2
View File
@@ -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 {
+66
View File
@@ -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 {