diff --git a/src/bloom.rs b/src/bloom.rs new file mode 100644 index 0000000..6185b2e --- /dev/null +++ b/src/bloom.rs @@ -0,0 +1,810 @@ +//! Bloom Filter Implementation +//! +//! 4KB Bloom filters for K-hop reachability in FIPS routing. Each node +//! maintains filters that summarize which destinations are reachable +//! through each peer, enabling efficient routing decisions without +//! global network knowledge. + +use crate::NodeId; +use std::collections::{HashMap, HashSet}; +use std::fmt; +use thiserror::Error; + +/// Default filter size in bits (4KB = 32,768 bits). +pub const DEFAULT_FILTER_SIZE_BITS: usize = 32768; + +/// Default filter size in bytes. +pub const DEFAULT_FILTER_SIZE_BYTES: usize = DEFAULT_FILTER_SIZE_BITS / 8; + +/// Default number of hash functions. +pub const DEFAULT_HASH_COUNT: u8 = 7; + +/// Errors related to Bloom filter operations. +#[derive(Debug, Error)] +pub enum BloomError { + #[error("invalid filter size: expected {expected} bits, got {got}")] + InvalidSize { expected: usize, got: usize }, + + #[error("filter size must be a multiple of 8, got {0}")] + SizeNotByteAligned(usize), + + #[error("hash count must be positive")] + ZeroHashCount, +} + +/// A Bloom filter for probabilistic set membership. +/// +/// Used in FIPS to track which destinations are reachable through a peer. +/// The filter uses double hashing to generate k hash functions from two +/// base hashes derived from the input. +#[derive(Clone)] +pub struct BloomFilter { + /// Bit array storage (packed as bytes). + bits: Vec, + /// Number of bits in the filter. + num_bits: usize, + /// Number of hash functions to use. + hash_count: u8, +} + +impl BloomFilter { + /// Create a new empty Bloom filter with default parameters. + pub fn new() -> Self { + Self::with_params(DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT) + .expect("default params are valid") + } + + /// Create a Bloom filter with custom parameters. + pub fn with_params(num_bits: usize, hash_count: u8) -> Result { + if num_bits == 0 || !num_bits.is_multiple_of(8) { + return Err(BloomError::SizeNotByteAligned(num_bits)); + } + if hash_count == 0 { + return Err(BloomError::ZeroHashCount); + } + + let num_bytes = num_bits / 8; + Ok(Self { + bits: vec![0u8; num_bytes], + num_bits, + hash_count, + }) + } + + /// Create a Bloom filter from raw bytes. + pub fn from_bytes(bytes: Vec, hash_count: u8) -> Result { + if hash_count == 0 { + return Err(BloomError::ZeroHashCount); + } + if bytes.is_empty() { + return Err(BloomError::SizeNotByteAligned(0)); + } + let num_bits = bytes.len() * 8; + Ok(Self { + bits: bytes, + num_bits, + hash_count, + }) + } + + /// Create a Bloom filter from a byte slice. + pub fn from_slice(bytes: &[u8], hash_count: u8) -> Result { + Self::from_bytes(bytes.to_vec(), hash_count) + } + + /// Insert a NodeId into the filter. + pub fn insert(&mut self, node_id: &NodeId) { + for i in 0..self.hash_count { + let bit_index = self.hash(node_id.as_bytes(), i); + self.set_bit(bit_index); + } + } + + /// Insert raw bytes into the filter. + pub fn insert_bytes(&mut self, data: &[u8]) { + for i in 0..self.hash_count { + let bit_index = self.hash(data, i); + self.set_bit(bit_index); + } + } + + /// Check if the filter might contain a NodeId. + /// + /// Returns `true` if the item might be in the set (possible false positive). + /// Returns `false` if the item is definitely not in the set. + pub fn contains(&self, node_id: &NodeId) -> bool { + self.contains_bytes(node_id.as_bytes()) + } + + /// Check if the filter might contain raw bytes. + pub fn contains_bytes(&self, data: &[u8]) -> bool { + for i in 0..self.hash_count { + let bit_index = self.hash(data, i); + if !self.get_bit(bit_index) { + return false; + } + } + true + } + + /// Merge another filter into this one (OR operation). + /// + /// After merge, this filter contains all elements from both filters. + pub fn merge(&mut self, other: &BloomFilter) -> Result<(), BloomError> { + if self.num_bits != other.num_bits { + return Err(BloomError::InvalidSize { + expected: self.num_bits, + got: other.num_bits, + }); + } + + for (a, b) in self.bits.iter_mut().zip(other.bits.iter()) { + *a |= b; + } + Ok(()) + } + + /// Create a new filter that is the union of this and another. + pub fn union(&self, other: &BloomFilter) -> Result { + let mut result = self.clone(); + result.merge(other)?; + Ok(result) + } + + /// Clear all bits in the filter. + pub fn clear(&mut self) { + self.bits.fill(0); + } + + /// Count the number of set bits (population count). + pub fn count_ones(&self) -> usize { + self.bits.iter().map(|b| b.count_ones() as usize).sum() + } + + /// Estimate the fill ratio (set bits / total bits). + pub fn fill_ratio(&self) -> f64 { + self.count_ones() as f64 / self.num_bits as f64 + } + + /// Estimate the number of elements in the filter. + /// + /// Uses the formula: n = -(m/k) * ln(1 - X/m) + /// where m = num_bits, k = hash_count, X = count_ones + pub fn estimated_count(&self) -> f64 { + let m = self.num_bits as f64; + let k = self.hash_count as f64; + let x = self.count_ones() as f64; + + if x >= m { + return f64::INFINITY; + } + + -(m / k) * (1.0 - x / m).ln() + } + + /// Check if the filter is empty. + pub fn is_empty(&self) -> bool { + self.bits.iter().all(|&b| b == 0) + } + + /// Get the raw bytes. + pub fn as_bytes(&self) -> &[u8] { + &self.bits + } + + /// Get the filter size in bits. + pub fn num_bits(&self) -> usize { + self.num_bits + } + + /// Get the filter size in bytes. + pub fn num_bytes(&self) -> usize { + self.bits.len() + } + + /// Get the number of hash functions. + pub fn hash_count(&self) -> u8 { + self.hash_count + } + + /// Compute a hash index for the given data and hash function number. + /// + /// Uses double hashing: h(x,i) = (h1(x) + i*h2(x)) mod m + fn hash(&self, data: &[u8], k: u8) -> usize { + // Use first 16 bytes of SHA-256 for h1 and h2 + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(data); + let hash = hasher.finalize(); + + // h1 from first 8 bytes + let h1 = u64::from_le_bytes(hash[0..8].try_into().unwrap()); + // h2 from next 8 bytes + let h2 = u64::from_le_bytes(hash[8..16].try_into().unwrap()); + + let combined = h1.wrapping_add((k as u64).wrapping_mul(h2)); + (combined as usize) % self.num_bits + } + + fn set_bit(&mut self, index: usize) { + let byte_index = index / 8; + let bit_offset = index % 8; + self.bits[byte_index] |= 1 << bit_offset; + } + + fn get_bit(&self, index: usize) -> bool { + let byte_index = index / 8; + let bit_offset = index % 8; + (self.bits[byte_index] >> bit_offset) & 1 == 1 + } +} + +impl Default for BloomFilter { + fn default() -> Self { + Self::new() + } +} + +impl PartialEq for BloomFilter { + fn eq(&self, other: &Self) -> bool { + self.num_bits == other.num_bits + && self.hash_count == other.hash_count + && self.bits == other.bits + } +} + +impl Eq for BloomFilter {} + +impl fmt::Debug for BloomFilter { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BloomFilter") + .field("bits", &self.num_bits) + .field("hash_count", &self.hash_count) + .field("fill_ratio", &format!("{:.2}%", self.fill_ratio() * 100.0)) + .field("est_count", &format!("{:.0}", self.estimated_count())) + .finish() + } +} + +/// State for managing Bloom filter announcements. +/// +/// Tracks local filter state and what needs to be sent to peers. +#[derive(Clone, Debug)] +pub struct BloomState { + /// This node's NodeId (always included in outgoing filters). + own_node_id: NodeId, + /// Leaf-only nodes we speak for (included in our filter). + leaf_dependents: HashSet, + /// Whether this node operates in leaf-only mode. + is_leaf_only: bool, + /// Rate limiting: minimum interval between outgoing updates (milliseconds). + update_debounce_ms: u64, + /// Timestamp of last update sent (per peer, in milliseconds). + last_update_sent: HashMap, + /// Peers that need a filter update. + pending_updates: HashSet, + /// Current sequence number for outgoing filters. + sequence: u64, +} + +impl BloomState { + /// Create new Bloom state for a node. + pub fn new(own_node_id: NodeId) -> Self { + Self { + own_node_id, + leaf_dependents: HashSet::new(), + is_leaf_only: false, + update_debounce_ms: 500, + last_update_sent: HashMap::new(), + pending_updates: HashSet::new(), + sequence: 0, + } + } + + /// Create state for a leaf-only node. + pub fn leaf_only(own_node_id: NodeId) -> Self { + let mut state = Self::new(own_node_id); + state.is_leaf_only = true; + state + } + + /// Get the node's own ID. + pub fn own_node_id(&self) -> &NodeId { + &self.own_node_id + } + + /// Check if this is a leaf-only node. + pub fn is_leaf_only(&self) -> bool { + self.is_leaf_only + } + + /// Get the current sequence number. + pub fn sequence(&self) -> u64 { + self.sequence + } + + /// Increment and return the next sequence number. + pub fn next_sequence(&mut self) -> u64 { + self.sequence += 1; + self.sequence + } + + /// Get the update debounce interval in milliseconds. + pub fn update_debounce_ms(&self) -> u64 { + self.update_debounce_ms + } + + /// Set the update debounce interval. + pub fn set_update_debounce_ms(&mut self, ms: u64) { + self.update_debounce_ms = ms; + } + + /// Add a leaf dependent that we'll include in our filter. + pub fn add_leaf_dependent(&mut self, node_id: NodeId) { + self.leaf_dependents.insert(node_id); + } + + /// Remove a leaf dependent. + pub fn remove_leaf_dependent(&mut self, node_id: &NodeId) -> bool { + self.leaf_dependents.remove(node_id) + } + + /// Get the set of leaf dependents. + pub fn leaf_dependents(&self) -> &HashSet { + &self.leaf_dependents + } + + /// Number of leaf dependents. + pub fn leaf_dependent_count(&self) -> usize { + self.leaf_dependents.len() + } + + /// Mark that a peer needs an update. + pub fn mark_update_needed(&mut self, peer_id: NodeId) { + self.pending_updates.insert(peer_id); + } + + /// Mark all peers as needing updates. + pub fn mark_all_updates_needed(&mut self, peer_ids: impl IntoIterator) { + self.pending_updates.extend(peer_ids); + } + + /// Check if a peer needs an update. + pub fn needs_update(&self, peer_id: &NodeId) -> bool { + self.pending_updates.contains(peer_id) + } + + /// Check if we should send an update to a peer (respecting debounce). + pub fn should_send_update(&self, peer_id: &NodeId, current_time_ms: u64) -> bool { + if !self.pending_updates.contains(peer_id) { + return false; + } + + match self.last_update_sent.get(peer_id) { + Some(&last_time) => current_time_ms >= last_time + self.update_debounce_ms, + None => true, + } + } + + /// Record that we sent an update to a peer. + pub fn record_update_sent(&mut self, peer_id: NodeId, current_time_ms: u64) { + self.last_update_sent.insert(peer_id, current_time_ms); + self.pending_updates.remove(&peer_id); + } + + /// Clear all pending updates. + pub fn clear_pending_updates(&mut self) { + self.pending_updates.clear(); + } + + /// Compute the outgoing filter for a specific peer. + /// + /// The filter includes: + /// - This node's own ID + /// - All leaf dependents + /// - Entries from other peers' inbound filters (excluding the destination peer) + /// + /// The `peer_filters` map contains inbound filters from each peer. + /// The filter for `exclude_peer` is excluded to prevent routing loops. + pub fn compute_outgoing_filter( + &self, + exclude_peer: &NodeId, + peer_filters: &HashMap, // (filter, ttl) + ) -> BloomFilter { + let mut filter = BloomFilter::new(); + + // Always include ourselves + filter.insert(&self.own_node_id); + + // Include leaf dependents + for dep in &self.leaf_dependents { + filter.insert(dep); + } + + // Merge filters from other peers (with TTL > 0) + for (peer_id, (peer_filter, ttl)) in peer_filters { + if peer_id != exclude_peer && *ttl > 0 { + // Ignore merge errors (size mismatches) - just skip that filter + let _ = filter.merge(peer_filter); + } + } + + filter + } + + /// Create a base filter containing just this node and its dependents. + pub fn base_filter(&self) -> BloomFilter { + let mut filter = BloomFilter::new(); + filter.insert(&self.own_node_id); + for dep in &self.leaf_dependents { + filter.insert(dep); + } + filter + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + // ===== BloomFilter Tests ===== + + #[test] + fn test_bloom_filter_new() { + let filter = BloomFilter::new(); + assert_eq!(filter.num_bits(), DEFAULT_FILTER_SIZE_BITS); + assert_eq!(filter.hash_count(), DEFAULT_HASH_COUNT); + assert_eq!(filter.count_ones(), 0); + assert!(filter.is_empty()); + } + + #[test] + fn test_bloom_filter_insert_contains() { + let mut filter = BloomFilter::new(); + let node1 = make_node_id(1); + let node2 = make_node_id(2); + + assert!(!filter.contains(&node1)); + assert!(!filter.contains(&node2)); + + filter.insert(&node1); + + assert!(filter.contains(&node1)); + // node2 might have false positive, but very unlikely with single insert + assert!(!filter.is_empty()); + } + + #[test] + fn test_bloom_filter_multiple_inserts() { + let mut filter = BloomFilter::new(); + + for i in 0..100 { + let node = make_node_id(i); + filter.insert(&node); + } + + // All inserted items should be found + for i in 0..100 { + let node = make_node_id(i); + assert!(filter.contains(&node), "Node {} not found", i); + } + + // Fill ratio should be reasonable + let fill = filter.fill_ratio(); + assert!(fill > 0.0 && fill < 0.5, "Unexpected fill ratio: {}", fill); + } + + #[test] + fn test_bloom_filter_merge() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + let node1 = make_node_id(1); + let node2 = make_node_id(2); + + filter1.insert(&node1); + filter2.insert(&node2); + + filter1.merge(&filter2).unwrap(); + + assert!(filter1.contains(&node1)); + assert!(filter1.contains(&node2)); + } + + #[test] + fn test_bloom_filter_union() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + let node1 = make_node_id(1); + let node2 = make_node_id(2); + + filter1.insert(&node1); + filter2.insert(&node2); + + let union = filter1.union(&filter2).unwrap(); + + assert!(union.contains(&node1)); + assert!(union.contains(&node2)); + // Original filters unchanged + assert!(!filter1.contains(&node2)); + assert!(!filter2.contains(&node1)); + } + + #[test] + fn test_bloom_filter_clear() { + let mut filter = BloomFilter::new(); + let node = make_node_id(1); + + filter.insert(&node); + assert!(!filter.is_empty()); + + filter.clear(); + assert!(filter.is_empty()); + assert_eq!(filter.count_ones(), 0); + assert!(!filter.contains(&node)); + } + + #[test] + fn test_bloom_filter_merge_size_mismatch() { + let mut filter1 = BloomFilter::with_params(1024, 7).unwrap(); + let filter2 = BloomFilter::with_params(2048, 7).unwrap(); + + let result = filter1.merge(&filter2); + assert!(matches!(result, Err(BloomError::InvalidSize { .. }))); + } + + #[test] + fn test_bloom_filter_custom_params() { + let filter = BloomFilter::with_params(1024, 5).unwrap(); + assert_eq!(filter.num_bits(), 1024); + assert_eq!(filter.num_bytes(), 128); + assert_eq!(filter.hash_count(), 5); + } + + #[test] + fn test_bloom_filter_invalid_params() { + // Not byte-aligned (1001 is not divisible by 8) + assert!(matches!( + BloomFilter::with_params(1001, 7), + Err(BloomError::SizeNotByteAligned(1001)) + )); + + // Zero size + assert!(matches!( + BloomFilter::with_params(0, 7), + Err(BloomError::SizeNotByteAligned(0)) + )); + + // Zero hash count + assert!(matches!( + BloomFilter::with_params(1024, 0), + Err(BloomError::ZeroHashCount) + )); + } + + #[test] + fn test_bloom_filter_from_bytes() { + let original = BloomFilter::new(); + let bytes = original.as_bytes().to_vec(); + + let restored = + BloomFilter::from_bytes(bytes, original.hash_count()).unwrap(); + + assert_eq!(original, restored); + } + + #[test] + fn test_bloom_filter_estimated_count() { + let mut filter = BloomFilter::new(); + + // Empty filter + assert_eq!(filter.estimated_count(), 0.0); + + // Insert some items + for i in 0..50 { + filter.insert(&make_node_id(i)); + } + + // Estimate should be reasonably close to 50 + let estimate = filter.estimated_count(); + assert!( + estimate > 30.0 && estimate < 100.0, + "Unexpected estimate: {}", + estimate + ); + } + + #[test] + fn test_bloom_filter_equality() { + let mut filter1 = BloomFilter::new(); + let mut filter2 = BloomFilter::new(); + + assert_eq!(filter1, filter2); + + filter1.insert(&make_node_id(1)); + assert_ne!(filter1, filter2); + + filter2.insert(&make_node_id(1)); + assert_eq!(filter1, filter2); + } + + // ===== BloomState Tests ===== + + #[test] + fn test_bloom_state_new() { + let node = make_node_id(0); + let state = BloomState::new(node); + + assert_eq!(state.own_node_id(), &node); + assert!(!state.is_leaf_only()); + assert_eq!(state.sequence(), 0); + assert_eq!(state.leaf_dependent_count(), 0); + } + + #[test] + fn test_bloom_state_leaf_only() { + let node = make_node_id(0); + let state = BloomState::leaf_only(node); + + assert!(state.is_leaf_only()); + } + + #[test] + fn test_bloom_state_leaf_dependents() { + let node = make_node_id(0); + let mut state = BloomState::new(node); + + let leaf1 = make_node_id(1); + let leaf2 = make_node_id(2); + + state.add_leaf_dependent(leaf1); + state.add_leaf_dependent(leaf2); + assert_eq!(state.leaf_dependent_count(), 2); + + assert!(state.remove_leaf_dependent(&leaf1)); + assert_eq!(state.leaf_dependent_count(), 1); + + assert!(!state.remove_leaf_dependent(&leaf1)); // already removed + } + + #[test] + fn test_bloom_state_debounce() { + let node = make_node_id(0); + let peer = make_node_id(1); + let mut state = BloomState::new(node); + state.set_update_debounce_ms(500); + + state.mark_update_needed(peer); + + // Should send initially + assert!(state.should_send_update(&peer, 1000)); + + // Record send + state.record_update_sent(peer, 1000); + state.mark_update_needed(peer); + + // Should not send immediately (within debounce) + assert!(!state.should_send_update(&peer, 1200)); + + // Should send after debounce period + assert!(state.should_send_update(&peer, 1600)); + } + + #[test] + fn test_bloom_state_sequence() { + let node = make_node_id(0); + let mut state = BloomState::new(node); + + assert_eq!(state.sequence(), 0); + assert_eq!(state.next_sequence(), 1); + assert_eq!(state.next_sequence(), 2); + assert_eq!(state.sequence(), 2); + } + + #[test] + fn test_bloom_state_pending_updates() { + let node = make_node_id(0); + let mut state = BloomState::new(node); + + let peer1 = make_node_id(1); + let peer2 = make_node_id(2); + + assert!(!state.needs_update(&peer1)); + + state.mark_update_needed(peer1); + assert!(state.needs_update(&peer1)); + assert!(!state.needs_update(&peer2)); + + state.mark_all_updates_needed(vec![peer1, peer2]); + assert!(state.needs_update(&peer1)); + assert!(state.needs_update(&peer2)); + + state.clear_pending_updates(); + assert!(!state.needs_update(&peer1)); + assert!(!state.needs_update(&peer2)); + } + + #[test] + fn test_bloom_state_base_filter() { + let node = make_node_id(0); + let mut state = BloomState::new(node); + + let leaf = make_node_id(1); + state.add_leaf_dependent(leaf); + + let filter = state.base_filter(); + + assert!(filter.contains(&node)); + assert!(filter.contains(&leaf)); + assert!(!filter.contains(&make_node_id(99))); + } + + #[test] + fn test_bloom_state_compute_outgoing_filter() { + let my_node = make_node_id(0); + let mut state = BloomState::new(my_node); + + let leaf = make_node_id(1); + state.add_leaf_dependent(leaf); + + let peer1 = make_node_id(10); + let peer2 = make_node_id(20); + + // Create peer filters + let mut filter1 = BloomFilter::new(); + filter1.insert(&make_node_id(100)); + filter1.insert(&make_node_id(101)); + + let mut filter2 = BloomFilter::new(); + filter2.insert(&make_node_id(200)); + + let mut peer_filters = HashMap::new(); + peer_filters.insert(peer1, (filter1, 2)); // TTL 2 + peer_filters.insert(peer2, (filter2, 1)); // TTL 1 + + // Filter for peer1 should exclude peer1's contributions + let outgoing1 = state.compute_outgoing_filter(&peer1, &peer_filters); + assert!(outgoing1.contains(&my_node)); // self + assert!(outgoing1.contains(&leaf)); // leaf dependent + assert!(outgoing1.contains(&make_node_id(200))); // from peer2 + // peer1's nodes may or may not be present (depends on split brain) + + // Filter for peer2 should exclude peer2's contributions + let outgoing2 = state.compute_outgoing_filter(&peer2, &peer_filters); + assert!(outgoing2.contains(&my_node)); + assert!(outgoing2.contains(&leaf)); + assert!(outgoing2.contains(&make_node_id(100))); // from peer1 + assert!(outgoing2.contains(&make_node_id(101))); // from peer1 + } + + #[test] + fn test_bloom_state_ttl_filtering() { + let my_node = make_node_id(0); + let state = BloomState::new(my_node); + + let peer1 = make_node_id(10); + let peer2 = make_node_id(20); + + let mut filter1 = BloomFilter::new(); + filter1.insert(&make_node_id(100)); + + let mut filter2 = BloomFilter::new(); + filter2.insert(&make_node_id(200)); + + let mut peer_filters = HashMap::new(); + peer_filters.insert(peer1, (filter1, 1)); // TTL 1 - included + peer_filters.insert(peer2, (filter2, 0)); // TTL 0 - excluded + + let outgoing = state.compute_outgoing_filter(&make_node_id(99), &peer_filters); + + assert!(outgoing.contains(&make_node_id(100))); // TTL 1 + assert!(!outgoing.contains(&make_node_id(200))); // TTL 0 excluded + } +} diff --git a/src/cache.rs b/src/cache.rs new file mode 100644 index 0000000..921d0a1 --- /dev/null +++ b/src/cache.rs @@ -0,0 +1,798 @@ +//! Caching Entities +//! +//! Coordinate and route caching for FIPS routing. The CoordCache stores +//! address-to-coordinate mappings populated by session setup, while +//! RouteCache stores coordinates learned from discovery queries. + +use crate::tree::TreeCoordinate; +use crate::{FipsAddress, NodeId}; +use std::collections::HashMap; +use thiserror::Error; + +/// Default maximum entries in coordinate cache. +pub const DEFAULT_COORD_CACHE_SIZE: usize = 50_000; + +/// Default TTL for coordinate cache entries (5 minutes in milliseconds). +pub const DEFAULT_COORD_CACHE_TTL_MS: u64 = 300_000; + +/// Default maximum entries in route cache. +pub const DEFAULT_ROUTE_CACHE_SIZE: usize = 10_000; + +/// Errors related to cache operations. +#[derive(Debug, Error)] +pub enum CacheError { + #[error("cache full: max {max} entries")] + CacheFull { max: usize }, + + #[error("entry not found")] + NotFound, + + #[error("entry expired")] + Expired, +} + +/// A cached coordinate entry. +#[derive(Clone, Debug)] +pub struct CacheEntry { + /// The cached coordinates. + coords: TreeCoordinate, + /// When this entry was created (Unix milliseconds). + created_at: u64, + /// When this entry was last used (Unix milliseconds). + last_used: u64, + /// When this entry expires (Unix milliseconds). + expires_at: u64, +} + +impl CacheEntry { + /// Create a new cache entry. + pub fn new(coords: TreeCoordinate, current_time_ms: u64, ttl_ms: u64) -> Self { + Self { + coords, + created_at: current_time_ms, + last_used: current_time_ms, + expires_at: current_time_ms.saturating_add(ttl_ms), + } + } + + /// Get the cached coordinates. + pub fn coords(&self) -> &TreeCoordinate { + &self.coords + } + + /// Get the creation timestamp. + pub fn created_at(&self) -> u64 { + self.created_at + } + + /// Get the last used timestamp. + pub fn last_used(&self) -> u64 { + self.last_used + } + + /// Get the expiry timestamp. + pub fn expires_at(&self) -> u64 { + self.expires_at + } + + /// Check if this entry has expired. + pub fn is_expired(&self, current_time_ms: u64) -> bool { + current_time_ms > self.expires_at + } + + /// Touch the entry to update last_used time. + pub fn touch(&mut self, current_time_ms: u64) { + self.last_used = current_time_ms; + } + + /// Refresh the expiry time. + pub fn refresh(&mut self, current_time_ms: u64, ttl_ms: u64) { + self.expires_at = current_time_ms.saturating_add(ttl_ms); + self.last_used = current_time_ms; + } + + /// Update the coordinates and refresh timestamps. + pub fn update(&mut self, coords: TreeCoordinate, current_time_ms: u64, ttl_ms: u64) { + self.coords = coords; + self.last_used = current_time_ms; + self.expires_at = current_time_ms.saturating_add(ttl_ms); + } + + /// Time since last use (for LRU eviction). + pub fn idle_time(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.last_used) + } + + /// Age of the entry. + pub fn age(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.created_at) + } + + /// Time until expiry (0 if already expired). + pub fn time_to_expiry(&self, current_time_ms: u64) -> u64 { + self.expires_at.saturating_sub(current_time_ms) + } +} + +/// Coordinate cache for routing decisions. +/// +/// Maps FIPS addresses to their tree coordinates, enabling data packets +/// to be routed without carrying coordinates in every packet. Populated +/// by SessionSetup packets. +#[derive(Clone, Debug)] +pub struct CoordCache { + /// Address -> coordinates mapping. + entries: HashMap, + /// Maximum number of entries. + max_entries: usize, + /// Default TTL for entries (milliseconds). + default_ttl_ms: u64, +} + +impl CoordCache { + /// Create a new coordinate cache. + pub fn new(max_entries: usize, default_ttl_ms: u64) -> Self { + Self { + entries: HashMap::with_capacity(max_entries.min(1000)), + max_entries, + default_ttl_ms, + } + } + + /// Create a cache with default parameters. + pub fn with_defaults() -> Self { + Self::new(DEFAULT_COORD_CACHE_SIZE, DEFAULT_COORD_CACHE_TTL_MS) + } + + /// Get the maximum capacity. + pub fn max_entries(&self) -> usize { + self.max_entries + } + + /// Get the default TTL. + pub fn default_ttl_ms(&self) -> u64 { + self.default_ttl_ms + } + + /// Set the default TTL. + pub fn set_default_ttl_ms(&mut self, ttl_ms: u64) { + self.default_ttl_ms = ttl_ms; + } + + /// Insert or update a cache entry. + pub fn insert(&mut self, addr: FipsAddress, coords: TreeCoordinate, current_time_ms: u64) { + // Update existing entry if present + if let Some(entry) = self.entries.get_mut(&addr) { + entry.update(coords, current_time_ms, self.default_ttl_ms); + return; + } + + // Evict if at capacity + if self.entries.len() >= self.max_entries { + self.evict_one(current_time_ms); + } + + let entry = CacheEntry::new(coords, current_time_ms, self.default_ttl_ms); + self.entries.insert(addr, entry); + } + + /// Insert with a custom TTL. + pub fn insert_with_ttl( + &mut self, + addr: FipsAddress, + coords: TreeCoordinate, + current_time_ms: u64, + ttl_ms: u64, + ) { + if let Some(entry) = self.entries.get_mut(&addr) { + entry.update(coords, current_time_ms, ttl_ms); + return; + } + + if self.entries.len() >= self.max_entries { + self.evict_one(current_time_ms); + } + + let entry = CacheEntry::new(coords, current_time_ms, ttl_ms); + self.entries.insert(addr, entry); + } + + /// Look up coordinates for an address (without touching). + pub fn get(&self, addr: &FipsAddress, current_time_ms: u64) -> Option<&TreeCoordinate> { + self.entries.get(addr).and_then(|entry| { + if entry.is_expired(current_time_ms) { + None + } else { + Some(entry.coords()) + } + }) + } + + /// Look up coordinates and touch (update last_used). + pub fn get_and_touch( + &mut self, + addr: &FipsAddress, + current_time_ms: u64, + ) -> Option<&TreeCoordinate> { + // Check and remove if expired + if let Some(entry) = self.entries.get(addr) + && entry.is_expired(current_time_ms) + { + self.entries.remove(addr); + return None; + } + + // Touch and return + if let Some(entry) = self.entries.get_mut(addr) { + entry.touch(current_time_ms); + Some(entry.coords()) + } else { + None + } + } + + /// Get the full cache entry. + pub fn get_entry(&self, addr: &FipsAddress) -> Option<&CacheEntry> { + self.entries.get(addr) + } + + /// Remove an entry. + pub fn remove(&mut self, addr: &FipsAddress) -> Option { + self.entries.remove(addr) + } + + /// Check if an address is cached (and not expired). + pub fn contains(&self, addr: &FipsAddress, current_time_ms: u64) -> bool { + self.get(addr, current_time_ms).is_some() + } + + /// Number of entries (including expired). + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Check if empty. + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Remove all expired entries. + pub fn purge_expired(&mut self, current_time_ms: u64) -> usize { + let before = self.entries.len(); + self.entries + .retain(|_, entry| !entry.is_expired(current_time_ms)); + before - self.entries.len() + } + + /// Clear all entries. + pub fn clear(&mut self) { + self.entries.clear(); + } + + /// Evict one entry (expired first, then LRU). + fn evict_one(&mut self, current_time_ms: u64) { + // First try to evict an expired entry + let expired_key = self + .entries + .iter() + .find(|(_, e)| e.is_expired(current_time_ms)) + .map(|(k, _)| *k); + + if let Some(key) = expired_key { + self.entries.remove(&key); + return; + } + + // Otherwise evict LRU (oldest last_used) + let lru_key = self + .entries + .iter() + .max_by_key(|(_, e)| e.idle_time(current_time_ms)) + .map(|(k, _)| *k); + + if let Some(key) = lru_key { + self.entries.remove(&key); + } + } + + /// Get cache statistics. + pub fn stats(&self, current_time_ms: u64) -> CacheStats { + let mut expired = 0; + let mut total_age = 0u64; + + for entry in self.entries.values() { + if entry.is_expired(current_time_ms) { + expired += 1; + } + total_age += entry.age(current_time_ms); + } + + CacheStats { + entries: self.entries.len(), + max_entries: self.max_entries, + expired, + avg_age_ms: if self.entries.is_empty() { + 0 + } else { + total_age / self.entries.len() as u64 + }, + } + } +} + +impl Default for CoordCache { + fn default() -> Self { + Self::with_defaults() + } +} + +/// Cache statistics. +#[derive(Clone, Debug)] +pub struct CacheStats { + /// Current number of entries. + pub entries: usize, + /// Maximum capacity. + pub max_entries: usize, + /// Number of expired entries. + pub expired: usize, + /// Average entry age in milliseconds. + pub avg_age_ms: u64, +} + +impl CacheStats { + /// Fill ratio (entries / max_entries). + pub fn fill_ratio(&self) -> f64 { + if self.max_entries == 0 { + 0.0 + } else { + self.entries as f64 / self.max_entries as f64 + } + } +} + +/// A cached route from discovery. +#[derive(Clone, Debug)] +pub struct CachedCoords { + /// The coordinates discovered. + coords: TreeCoordinate, + /// When this was discovered (Unix milliseconds). + discovered_at: u64, + /// Last time we used this route (Unix milliseconds). + last_used: u64, +} + +impl CachedCoords { + /// Create a new cached route. + pub fn new(coords: TreeCoordinate, discovered_at: u64) -> Self { + Self { + coords, + discovered_at, + last_used: discovered_at, + } + } + + /// Get the coordinates. + pub fn coords(&self) -> &TreeCoordinate { + &self.coords + } + + /// Get the discovery timestamp. + pub fn discovered_at(&self) -> u64 { + self.discovered_at + } + + /// Get the last used timestamp. + pub fn last_used(&self) -> u64 { + self.last_used + } + + /// Touch (update last_used). + pub fn touch(&mut self, current_time_ms: u64) { + self.last_used = current_time_ms; + } + + /// Age since discovery. + pub fn age(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.discovered_at) + } + + /// Idle time since last use. + pub fn idle_time(&self, current_time_ms: u64) -> u64 { + current_time_ms.saturating_sub(self.last_used) + } + + /// Update coordinates (re-discovered). + pub fn update(&mut self, coords: TreeCoordinate, current_time_ms: u64) { + self.coords = coords; + self.discovered_at = current_time_ms; + self.last_used = current_time_ms; + } +} + +/// Route cache for discovered destinations. +/// +/// Separate from CoordCache, this stores routes learned from the discovery +/// protocol (LookupRequest/LookupResponse) rather than session establishment. +/// Keyed by NodeId rather than FipsAddress. +#[derive(Clone, Debug)] +pub struct RouteCache { + /// NodeId -> discovered coordinates. + entries: HashMap, + /// Maximum entries. + max_entries: usize, +} + +impl RouteCache { + /// Create a new route cache. + pub fn new(max_entries: usize) -> Self { + Self { + entries: HashMap::with_capacity(max_entries.min(1000)), + max_entries, + } + } + + /// Create with default capacity. + pub fn with_defaults() -> Self { + Self::new(DEFAULT_ROUTE_CACHE_SIZE) + } + + /// Get the maximum capacity. + pub fn max_entries(&self) -> usize { + self.max_entries + } + + /// Insert a discovered route. + pub fn insert(&mut self, node_id: NodeId, coords: TreeCoordinate, current_time_ms: u64) { + // Update existing + if let Some(entry) = self.entries.get_mut(&node_id) { + entry.update(coords, current_time_ms); + return; + } + + // Evict if full + if self.entries.len() >= self.max_entries { + self.evict_lru(current_time_ms); + } + + self.entries + .insert(node_id, CachedCoords::new(coords, current_time_ms)); + } + + /// Look up a route (without touching). + pub fn get(&self, node_id: &NodeId) -> Option<&CachedCoords> { + self.entries.get(node_id) + } + + /// Look up and touch. + pub fn get_and_touch( + &mut self, + node_id: &NodeId, + current_time_ms: u64, + ) -> Option<&TreeCoordinate> { + if let Some(entry) = self.entries.get_mut(node_id) { + entry.touch(current_time_ms); + Some(entry.coords()) + } else { + None + } + } + + /// Remove a route (e.g., after route failure). + pub fn invalidate(&mut self, node_id: &NodeId) -> Option { + self.entries.remove(node_id) + } + + /// Check if a node is cached. + pub fn contains(&self, node_id: &NodeId) -> bool { + self.entries.contains_key(node_id) + } + + /// Number of cached routes. + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Check if empty. + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Clear all routes. + pub fn clear(&mut self) { + self.entries.clear(); + } + + /// Evict routes older than a threshold. + pub fn evict_older_than(&mut self, max_age_ms: u64, current_time_ms: u64) -> usize { + let before = self.entries.len(); + self.entries + .retain(|_, entry| entry.age(current_time_ms) < max_age_ms); + before - self.entries.len() + } + + fn evict_lru(&mut self, current_time_ms: u64) { + let lru_id = self + .entries + .iter() + .max_by_key(|(_, e)| e.idle_time(current_time_ms)) + .map(|(k, _)| *k); + + if let Some(id) = lru_id { + self.entries.remove(&id); + } + } +} + +impl Default for RouteCache { + fn default() -> Self { + Self::with_defaults() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + fn make_address(val: u8) -> FipsAddress { + let mut bytes = [0xfdu8; 16]; + bytes[1] = val; + FipsAddress::from_bytes(bytes).unwrap() + } + + fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::new(ids.iter().map(|&v| make_node_id(v)).collect()).unwrap() + } + + // ===== CacheEntry Tests ===== + + #[test] + fn test_cache_entry_expiry() { + let coords = make_coords(&[1, 0]); + let entry = CacheEntry::new(coords, 1000, 500); + + assert!(!entry.is_expired(1000)); + assert!(!entry.is_expired(1500)); // expires_at = 1500, not yet expired + assert!(entry.is_expired(1501)); // one ms after expiry + assert!(entry.is_expired(2000)); + } + + #[test] + fn test_cache_entry_refresh() { + let coords = make_coords(&[1, 0]); + let mut entry = CacheEntry::new(coords, 1000, 500); + + assert!(entry.is_expired(1501)); // expires_at = 1500 + + entry.refresh(1400, 500); // new expires_at = 1900 + + assert!(!entry.is_expired(1600)); + assert!(!entry.is_expired(1900)); // at exactly expiry, not expired + assert!(entry.is_expired(1901)); // one ms after expiry + } + + #[test] + fn test_cache_entry_times() { + let coords = make_coords(&[1, 0]); + let entry = CacheEntry::new(coords, 1000, 500); + + assert_eq!(entry.created_at(), 1000); + assert_eq!(entry.last_used(), 1000); + assert_eq!(entry.expires_at(), 1500); + assert_eq!(entry.age(1200), 200); + assert_eq!(entry.idle_time(1200), 200); + assert_eq!(entry.time_to_expiry(1200), 300); + assert_eq!(entry.time_to_expiry(1600), 0); + } + + // ===== CoordCache Tests ===== + + #[test] + fn test_coord_cache_basic() { + let mut cache = CoordCache::new(100, 1000); + let addr = make_address(1); + let coords = make_coords(&[1, 0]); + + cache.insert(addr, coords.clone(), 0); + + assert!(cache.contains(&addr, 0)); + assert_eq!(cache.get(&addr, 0), Some(&coords)); + assert_eq!(cache.len(), 1); + } + + #[test] + fn test_coord_cache_expiry() { + let mut cache = CoordCache::new(100, 1000); + let addr = make_address(1); + let coords = make_coords(&[1, 0]); + + cache.insert(addr, coords, 0); + + assert!(cache.contains(&addr, 500)); + assert!(!cache.contains(&addr, 1500)); + } + + #[test] + fn test_coord_cache_update() { + let mut cache = CoordCache::new(100, 1000); + let addr = make_address(1); + + cache.insert(addr, make_coords(&[1, 0]), 0); + cache.insert(addr, make_coords(&[1, 2, 0]), 500); + + assert_eq!(cache.len(), 1); + let coords = cache.get(&addr, 500).unwrap(); + assert_eq!(coords.depth(), 2); + } + + #[test] + fn test_coord_cache_eviction() { + let mut cache = CoordCache::new(2, 10000); + + let addr1 = make_address(1); + let addr2 = make_address(2); + let addr3 = make_address(3); + + cache.insert(addr1, make_coords(&[1, 0]), 0); + cache.insert(addr2, make_coords(&[2, 0]), 100); + + // Touch addr2 to make it more recent + let _ = cache.get_and_touch(&addr2, 200); + + // Insert addr3, should evict addr1 (LRU) + cache.insert(addr3, make_coords(&[3, 0]), 300); + + assert!(!cache.contains(&addr1, 300)); + assert!(cache.contains(&addr2, 300)); + assert!(cache.contains(&addr3, 300)); + } + + #[test] + fn test_coord_cache_evict_expired_first() { + let mut cache = CoordCache::new(2, 100); + + cache.insert(make_address(1), make_coords(&[1, 0]), 0); + cache.insert(make_address(2), make_coords(&[2, 0]), 50); + + // At time 150, addr1 is expired, addr2 is not + cache.insert(make_address(3), make_coords(&[3, 0]), 150); + + // addr1 should be evicted (expired), not addr2 (LRU but not expired) + assert!(!cache.contains(&make_address(1), 150)); + assert!(cache.contains(&make_address(2), 150)); + assert!(cache.contains(&make_address(3), 150)); + } + + #[test] + fn test_coord_cache_purge_expired() { + let mut cache = CoordCache::new(100, 100); + + cache.insert(make_address(1), make_coords(&[1, 0]), 0); // expires at 100 + cache.insert(make_address(2), make_coords(&[2, 0]), 50); // expires at 150 + cache.insert(make_address(3), make_coords(&[3, 0]), 200); // expires at 300 + + assert_eq!(cache.len(), 3); + + let purged = cache.purge_expired(151); // both addr1 and addr2 expired + + // Entry 1 and 2 expired, entry 3 still valid + assert_eq!(purged, 2); + assert_eq!(cache.len(), 1); + assert!(cache.contains(&make_address(3), 151)); + } + + #[test] + fn test_coord_cache_stats() { + let mut cache = CoordCache::new(100, 100); + + cache.insert(make_address(1), make_coords(&[1, 0]), 0); + cache.insert(make_address(2), make_coords(&[2, 0]), 50); + + let stats = cache.stats(150); + + assert_eq!(stats.entries, 2); + assert_eq!(stats.max_entries, 100); + assert_eq!(stats.expired, 1); // addr1 expired + assert!(stats.avg_age_ms > 0); + } + + // ===== CachedCoords Tests ===== + + #[test] + fn test_cached_coords() { + let coords = make_coords(&[1, 0]); + let mut cached = CachedCoords::new(coords.clone(), 1000); + + assert_eq!(cached.coords(), &coords); + assert_eq!(cached.discovered_at(), 1000); + assert_eq!(cached.last_used(), 1000); + + cached.touch(1500); + assert_eq!(cached.last_used(), 1500); + assert_eq!(cached.idle_time(1600), 100); + assert_eq!(cached.age(1600), 600); + } + + // ===== RouteCache Tests ===== + + #[test] + fn test_route_cache_basic() { + let mut cache = RouteCache::new(100); + let node = make_node_id(1); + let coords = make_coords(&[1, 0]); + + cache.insert(node, coords.clone(), 0); + + assert!(cache.contains(&node)); + assert_eq!(cache.get(&node).unwrap().coords(), &coords); + } + + #[test] + fn test_route_cache_invalidate() { + let mut cache = RouteCache::new(100); + let node = make_node_id(1); + let coords = make_coords(&[1, 0]); + + cache.insert(node, coords, 0); + assert!(cache.contains(&node)); + + cache.invalidate(&node); + assert!(!cache.contains(&node)); + } + + #[test] + fn test_route_cache_lru_eviction() { + let mut cache = RouteCache::new(2); + + let node1 = make_node_id(1); + let node2 = make_node_id(2); + let node3 = make_node_id(3); + + cache.insert(node1, make_coords(&[1, 0]), 0); + cache.insert(node2, make_coords(&[2, 0]), 100); + + // Touch node2 + let _ = cache.get_and_touch(&node2, 200); + + // Insert node3 + cache.insert(node3, make_coords(&[3, 0]), 300); + + // node1 should be evicted + assert!(!cache.contains(&node1)); + assert!(cache.contains(&node2)); + assert!(cache.contains(&node3)); + } + + #[test] + fn test_route_cache_evict_older_than() { + let mut cache = RouteCache::new(100); + + cache.insert(make_node_id(1), make_coords(&[1, 0]), 0); + cache.insert(make_node_id(2), make_coords(&[2, 0]), 500); + cache.insert(make_node_id(3), make_coords(&[3, 0]), 1000); + + let evicted = cache.evict_older_than(600, 1000); + + assert_eq!(evicted, 1); // node1 is > 600ms old + assert_eq!(cache.len(), 2); + } + + #[test] + fn test_route_cache_update() { + let mut cache = RouteCache::new(100); + let node = make_node_id(1); + + cache.insert(node, make_coords(&[1, 0]), 0); + cache.insert(node, make_coords(&[1, 2, 0]), 500); + + assert_eq!(cache.len(), 1); + let cached = cache.get(&node).unwrap(); + assert_eq!(cached.coords().depth(), 2); + assert_eq!(cached.discovered_at(), 500); + } +} diff --git a/src/lib.rs b/src/lib.rs index 328ff93..d043966 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,11 +3,49 @@ //! A distributed, decentralized network routing protocol for mesh nodes //! connecting over arbitrary transports. +pub mod bloom; +pub mod cache; pub mod config; pub mod identity; +pub mod node; +pub mod peer; +pub mod protocol; +pub mod transport; +pub mod tree; -pub use config::{Config, ConfigError, IdentityConfig}; +// Re-export identity types pub use identity::{ decode_npub, decode_nsec, decode_secret, encode_npub, encode_nsec, AuthChallenge, AuthResponse, FipsAddress, Identity, IdentityError, NodeId, PeerIdentity, }; + +// Re-export config types +pub use config::{Config, ConfigError, IdentityConfig}; + +// Re-export tree types +pub use tree::{ParentDeclaration, TreeCoordinate, TreeError, TreeState}; + +// Re-export bloom filter types +pub use bloom::{BloomError, BloomFilter, BloomState}; + +// Re-export transport types +pub use transport::{ + DiscoveredPeer, Link, LinkDirection, LinkId, LinkState, LinkStats, Transport, TransportAddr, + TransportError, TransportId, TransportState, TransportType, +}; + +// Re-export protocol types +pub use protocol::{ + Auth, AuthAck, Challenge, CoordsRequired, DataFlags, DataPacket, FilterAnnounce, Hello, + LookupRequest, LookupResponse, MessageType, PathBroken, ProtocolError, SessionAck, + SessionFlags, SessionSetup, TreeAnnounce, +}; + +// Re-export cache types +pub use cache::{CacheEntry, CacheError, CacheStats, CachedCoords, CoordCache, RouteCache}; + +// Re-export peer types +pub use peer::{Peer, PeerError, PeerState, UpstreamPeer}; + +// Re-export node types +pub use node::{Node, NodeError, NodeState}; diff --git a/src/node.rs b/src/node.rs new file mode 100644 index 0000000..566ec9d --- /dev/null +++ b/src/node.rs @@ -0,0 +1,737 @@ +//! FIPS Node Entity +//! +//! Top-level structure representing a running FIPS instance. The Node +//! holds all state required for mesh routing: identity, tree state, +//! Bloom filters, coordinate caches, transports, links, and peers. + +use crate::bloom::BloomState; +use crate::cache::CoordCache; +use crate::peer::Peer; +use crate::transport::{Link, LinkId, TransportId}; +use crate::tree::TreeState; +use crate::{Config, ConfigError, Identity, IdentityError, NodeId}; +use std::collections::HashMap; +use std::fmt; +use thiserror::Error; + +/// Errors related to node operations. +#[derive(Debug, Error)] +pub enum NodeError { + #[error("node not started")] + NotStarted, + + #[error("node already started")] + AlreadyStarted, + + #[error("node already stopped")] + AlreadyStopped, + + #[error("transport not found: {0}")] + TransportNotFound(TransportId), + + #[error("link not found: {0}")] + LinkNotFound(LinkId), + + #[error("peer not found: {0:?}")] + PeerNotFound(NodeId), + + #[error("peer already exists: {0:?}")] + PeerAlreadyExists(NodeId), + + #[error("max peers exceeded: {max}")] + MaxPeersExceeded { max: usize }, + + #[error("max links exceeded: {max}")] + MaxLinksExceeded { max: usize }, + + #[error("config error: {0}")] + Config(#[from] ConfigError), + + #[error("identity error: {0}")] + Identity(#[from] IdentityError), +} + +/// Node operational state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum NodeState { + /// Created but not started. + Created, + /// Starting up (initializing transports). + Starting, + /// Fully operational. + Running, + /// Shutting down. + Stopping, + /// Stopped. + Stopped, +} + +impl NodeState { + /// Check if node is operational. + pub fn is_operational(&self) -> bool { + matches!(self, NodeState::Running) + } + + /// Check if node can be started. + pub fn can_start(&self) -> bool { + matches!(self, NodeState::Created | NodeState::Stopped) + } + + /// Check if node can be stopped. + pub fn can_stop(&self) -> bool { + matches!(self, NodeState::Running) + } +} + +impl fmt::Display for NodeState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + NodeState::Created => "created", + NodeState::Starting => "starting", + NodeState::Running => "running", + NodeState::Stopping => "stopping", + NodeState::Stopped => "stopped", + }; + write!(f, "{}", s) + } +} + +/// A running FIPS node instance. +/// +/// This is the top-level container holding all node state. +pub struct Node { + // === Identity === + /// This node's cryptographic identity. + identity: Identity, + + // === Configuration === + /// Loaded configuration. + config: Config, + + // === State === + /// Node operational state. + state: NodeState, + + /// Whether this is a leaf-only node. + is_leaf_only: bool, + + // === Spanning Tree === + /// Local spanning tree state. + tree_state: TreeState, + + // === Bloom Filter === + /// Local Bloom filter state. + bloom_state: BloomState, + + // === Routing === + /// Address -> coordinates cache. + coord_cache: CoordCache, + + // === Transports & Links === + /// Active transport IDs. + transport_ids: Vec, + /// Active links. + links: HashMap, + + // === Peers === + /// Authenticated peers. + peers: HashMap, + + // === Resource Limits === + /// Maximum peers (0 = unlimited). + max_peers: usize, + /// Maximum links (0 = unlimited). + max_links: usize, + + // === Counters === + /// Next link ID to allocate. + next_link_id: u64, + /// Next transport ID to allocate. + next_transport_id: u32, +} + +impl Node { + /// Create a new node from configuration. + pub fn new(config: Config) -> Result { + let identity = config.create_identity()?; + let node_id = *identity.node_id(); + + Ok(Self { + identity, + config, + state: NodeState::Created, + is_leaf_only: false, + tree_state: TreeState::new(node_id), + bloom_state: BloomState::new(node_id), + coord_cache: CoordCache::with_defaults(), + transport_ids: Vec::new(), + links: HashMap::new(), + peers: HashMap::new(), + max_peers: 128, + max_links: 256, + next_link_id: 1, + next_transport_id: 1, + }) + } + + /// Create a node with a specific identity. + pub fn with_identity(identity: Identity, config: Config) -> Self { + let node_id = *identity.node_id(); + Self { + identity, + config, + state: NodeState::Created, + is_leaf_only: false, + tree_state: TreeState::new(node_id), + bloom_state: BloomState::new(node_id), + coord_cache: CoordCache::with_defaults(), + transport_ids: Vec::new(), + links: HashMap::new(), + peers: HashMap::new(), + max_peers: 128, + max_links: 256, + next_link_id: 1, + next_transport_id: 1, + } + } + + /// Create a leaf-only node (simplified state). + pub fn leaf_only(config: Config) -> Result { + let mut node = Self::new(config)?; + node.is_leaf_only = true; + node.bloom_state = BloomState::leaf_only(*node.identity.node_id()); + Ok(node) + } + + // === Identity Accessors === + + /// Get this node's identity. + pub fn identity(&self) -> &Identity { + &self.identity + } + + /// Get this node's NodeId. + pub fn node_id(&self) -> &NodeId { + self.identity.node_id() + } + + /// Get this node's npub. + pub fn npub(&self) -> String { + self.identity.npub() + } + + // === Configuration === + + /// Get the configuration. + pub fn config(&self) -> &Config { + &self.config + } + + // === State === + + /// Get the node state. + pub fn state(&self) -> NodeState { + self.state + } + + /// Check if node is operational. + pub fn is_running(&self) -> bool { + self.state.is_operational() + } + + /// Check if this is a leaf-only node. + pub fn is_leaf_only(&self) -> bool { + self.is_leaf_only + } + + // === Tree State === + + /// Get the tree state. + pub fn tree_state(&self) -> &TreeState { + &self.tree_state + } + + /// Get mutable tree state. + pub fn tree_state_mut(&mut self) -> &mut TreeState { + &mut self.tree_state + } + + // === Bloom State === + + /// Get the Bloom filter state. + pub fn bloom_state(&self) -> &BloomState { + &self.bloom_state + } + + /// Get mutable Bloom filter state. + pub fn bloom_state_mut(&mut self) -> &mut BloomState { + &mut self.bloom_state + } + + // === Coord Cache === + + /// Get the coordinate cache. + pub fn coord_cache(&self) -> &CoordCache { + &self.coord_cache + } + + /// Get mutable coordinate cache. + pub fn coord_cache_mut(&mut self) -> &mut CoordCache { + &mut self.coord_cache + } + + // === Resource Limits === + + /// Set the maximum number of peers. + pub fn set_max_peers(&mut self, max: usize) { + self.max_peers = max; + } + + /// Set the maximum number of links. + pub fn set_max_links(&mut self, max: usize) { + self.max_links = max; + } + + // === Counts === + + /// Number of authenticated peers. + pub fn peer_count(&self) -> usize { + self.peers.len() + } + + /// Number of active links. + pub fn link_count(&self) -> usize { + self.links.len() + } + + /// Number of transports. + pub fn transport_count(&self) -> usize { + self.transport_ids.len() + } + + // === Transport Management === + + /// Allocate a new transport ID. + pub fn allocate_transport_id(&mut self) -> TransportId { + let id = TransportId::new(self.next_transport_id); + self.next_transport_id += 1; + id + } + + /// Register a transport. + pub fn add_transport(&mut self, transport_id: TransportId) { + if !self.transport_ids.contains(&transport_id) { + self.transport_ids.push(transport_id); + } + } + + /// Unregister a transport. + pub fn remove_transport(&mut self, transport_id: &TransportId) { + self.transport_ids.retain(|id| id != transport_id); + } + + /// Get all transport IDs. + pub fn transport_ids(&self) -> &[TransportId] { + &self.transport_ids + } + + // === Link Management === + + /// Allocate a new link ID. + pub fn allocate_link_id(&mut self) -> LinkId { + let id = LinkId::new(self.next_link_id); + self.next_link_id += 1; + id + } + + /// Add a link. + pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> { + if self.max_links > 0 && self.links.len() >= self.max_links { + return Err(NodeError::MaxLinksExceeded { max: self.max_links }); + } + self.links.insert(link.link_id(), link); + Ok(()) + } + + /// Get a link by ID. + pub fn get_link(&self, link_id: &LinkId) -> Option<&Link> { + self.links.get(link_id) + } + + /// Get a mutable link by ID. + pub fn get_link_mut(&mut self, link_id: &LinkId) -> Option<&mut Link> { + self.links.get_mut(link_id) + } + + /// Remove a link. + pub fn remove_link(&mut self, link_id: &LinkId) -> Option { + self.links.remove(link_id) + } + + /// Iterate over all links. + pub fn links(&self) -> impl Iterator { + self.links.values() + } + + // === Peer Management === + + /// Add an authenticated peer. + pub fn add_peer(&mut self, peer: Peer) -> Result<(), NodeError> { + let node_id = *peer.node_id(); + + if self.peers.contains_key(&node_id) { + return Err(NodeError::PeerAlreadyExists(node_id)); + } + + if self.max_peers > 0 && self.peers.len() >= self.max_peers { + return Err(NodeError::MaxPeersExceeded { max: self.max_peers }); + } + + self.peers.insert(node_id, peer); + Ok(()) + } + + /// Get a peer by NodeId. + pub fn get_peer(&self, node_id: &NodeId) -> Option<&Peer> { + self.peers.get(node_id) + } + + /// Get a mutable peer by NodeId. + pub fn get_peer_mut(&mut self, node_id: &NodeId) -> Option<&mut Peer> { + self.peers.get_mut(node_id) + } + + /// Remove a peer. + pub fn remove_peer(&mut self, node_id: &NodeId) -> Option { + self.peers.remove(node_id) + } + + /// Iterate over all peers. + pub fn peers(&self) -> impl Iterator { + self.peers.values() + } + + /// Iterate over all peer node IDs. + pub fn peer_ids(&self) -> impl Iterator { + self.peers.keys() + } + + /// Iterate over all active peers. + pub fn active_peers(&self) -> impl Iterator { + self.peers.values().filter(|p| p.state().is_active()) + } + + /// Number of active peers. + pub fn active_peer_count(&self) -> usize { + self.peers.values().filter(|p| p.state().is_active()).count() + } + + // === Routing (stubs) === + + /// Find next hop for a destination (stub). + /// + /// Returns the peer that minimizes tree distance to the destination. + pub fn find_next_hop(&self, _dest_node_id: &NodeId) -> Option<&Peer> { + // Stub: would implement greedy tree routing + None + } + + /// Check if a destination is in any peer's bloom filter. + pub fn destination_in_filters(&self, dest: &NodeId) -> Vec<&Peer> { + self.peers.values().filter(|p| p.may_reach(dest)).collect() + } + + // === State Transitions === + + /// Start the node (stub). + /// + /// In a full implementation, this would: + /// - Initialize transports + /// - Bind TUN interface + /// - Start event loop + pub 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. + self.state = NodeState::Running; + Ok(()) + } + + /// Stop the node (stub). + /// + /// 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> { + if !self.state.can_stop() { + return Err(NodeError::NotStarted); + } + self.state = NodeState::Stopping; + // Actual shutdown would close transports, links, etc. + self.state = NodeState::Stopped; + Ok(()) + } +} + +impl fmt::Debug for Node { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Node") + .field("node_id", self.node_id()) + .field("state", &self.state) + .field("is_leaf_only", &self.is_leaf_only) + .field("peers", &self.peer_count()) + .field("links", &self.link_count()) + .field("transports", &self.transport_count()) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::{LinkDirection, TransportAddr}; + use std::time::Duration; + + fn make_node() -> Node { + let config = Config::new(); + Node::new(config).unwrap() + } + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + #[test] + fn test_node_creation() { + let node = make_node(); + + assert_eq!(node.state(), NodeState::Created); + assert_eq!(node.peer_count(), 0); + assert_eq!(node.link_count(), 0); + assert!(!node.is_leaf_only()); + } + + #[test] + fn test_node_with_identity() { + let identity = Identity::generate(); + let expected_node_id = *identity.node_id(); + let config = Config::new(); + + let node = Node::with_identity(identity, config); + + assert_eq!(node.node_id(), &expected_node_id); + } + + #[test] + fn test_node_leaf_only() { + let config = Config::new(); + let node = Node::leaf_only(config).unwrap(); + + assert!(node.is_leaf_only()); + assert!(node.bloom_state().is_leaf_only()); + } + + #[test] + fn test_node_state_transitions() { + let mut node = make_node(); + + assert!(!node.is_running()); + assert!(node.state().can_start()); + + node.start().unwrap(); + assert!(node.is_running()); + assert!(!node.state().can_start()); + + node.stop().unwrap(); + assert!(!node.is_running()); + assert_eq!(node.state(), NodeState::Stopped); + } + + #[test] + fn test_node_double_start() { + let mut node = make_node(); + node.start().unwrap(); + + let result = node.start(); + assert!(matches!(result, Err(NodeError::AlreadyStarted))); + } + + #[test] + fn test_node_stop_not_started() { + let mut node = make_node(); + + let result = node.stop(); + assert!(matches!(result, Err(NodeError::NotStarted))); + } + + #[test] + fn test_node_link_management() { + let mut node = make_node(); + + let link_id = node.allocate_link_id(); + let link = Link::connectionless( + link_id, + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + + node.add_link(link).unwrap(); + assert_eq!(node.link_count(), 1); + + assert!(node.get_link(&link_id).is_some()); + + node.remove_link(&link_id); + assert_eq!(node.link_count(), 0); + } + + #[test] + fn test_node_link_limit() { + let mut node = make_node(); + node.set_max_links(2); + + for _ in 0..2 { + let link_id = node.allocate_link_id(); + let link = Link::connectionless( + link_id, + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + node.add_link(link).unwrap(); + } + + let link_id = node.allocate_link_id(); + let link = Link::connectionless( + link_id, + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + + let result = node.add_link(link); + assert!(matches!(result, Err(NodeError::MaxLinksExceeded { .. }))); + } + + #[test] + fn test_node_peer_management() { + let mut node = make_node(); + + let peer_identity = Identity::generate(); + let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); + let peer = Peer::discovered(peer_pub, LinkId::new(1)); + let peer_node_id = *peer.node_id(); + + node.add_peer(peer).unwrap(); + assert_eq!(node.peer_count(), 1); + + assert!(node.get_peer(&peer_node_id).is_some()); + + node.remove_peer(&peer_node_id); + assert_eq!(node.peer_count(), 0); + } + + #[test] + fn test_node_peer_duplicate() { + let mut node = make_node(); + + let peer_identity = Identity::generate(); + let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); + let peer1 = Peer::discovered(peer_pub.clone(), LinkId::new(1)); + let peer2 = Peer::discovered(peer_pub, LinkId::new(2)); + + node.add_peer(peer1).unwrap(); + let result = node.add_peer(peer2); + + assert!(matches!(result, Err(NodeError::PeerAlreadyExists(_)))); + } + + #[test] + fn test_node_peer_limit() { + let mut node = make_node(); + node.set_max_peers(2); + + for _ in 0..2 { + let peer_identity = Identity::generate(); + let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); + let peer = Peer::discovered(peer_pub, LinkId::new(1)); + node.add_peer(peer).unwrap(); + } + + let peer_identity = Identity::generate(); + let peer_pub = crate::PeerIdentity::from_pubkey(peer_identity.pubkey()); + let peer = Peer::discovered(peer_pub, LinkId::new(1)); + + let result = node.add_peer(peer); + assert!(matches!(result, Err(NodeError::MaxPeersExceeded { .. }))); + } + + #[test] + fn test_node_link_id_allocation() { + let mut node = make_node(); + + let id1 = node.allocate_link_id(); + let id2 = node.allocate_link_id(); + let id3 = node.allocate_link_id(); + + assert_ne!(id1, id2); + assert_ne!(id2, id3); + assert_eq!(id1.as_u64(), 1); + assert_eq!(id2.as_u64(), 2); + assert_eq!(id3.as_u64(), 3); + } + + #[test] + fn test_node_transport_management() { + let mut node = make_node(); + + let id1 = node.allocate_transport_id(); + let id2 = node.allocate_transport_id(); + + node.add_transport(id1); + node.add_transport(id2); + assert_eq!(node.transport_count(), 2); + + // Adding same ID again doesn't duplicate + node.add_transport(id1); + assert_eq!(node.transport_count(), 2); + + node.remove_transport(&id1); + assert_eq!(node.transport_count(), 1); + } + + #[test] + fn test_node_active_peers() { + let mut node = make_node(); + + // Add a discovered peer + let peer_identity1 = Identity::generate(); + let peer_pub1 = crate::PeerIdentity::from_pubkey(peer_identity1.pubkey()); + let peer1 = Peer::discovered(peer_pub1, LinkId::new(1)); + node.add_peer(peer1).unwrap(); + + // Add an active peer + let peer_identity2 = Identity::generate(); + let peer_pub2 = crate::PeerIdentity::from_pubkey(peer_identity2.pubkey()); + let mut peer2 = Peer::discovered(peer_pub2, LinkId::new(2)); + peer2.set_active(1000); + let peer2_id = *peer2.node_id(); + node.add_peer(peer2).unwrap(); + + assert_eq!(node.peer_count(), 2); + assert_eq!(node.active_peer_count(), 1); + + let active: Vec<_> = node.active_peers().collect(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].node_id(), &peer2_id); + } +} diff --git a/src/peer.rs b/src/peer.rs new file mode 100644 index 0000000..7e46355 --- /dev/null +++ b/src/peer.rs @@ -0,0 +1,691 @@ +//! Peer Management Entities +//! +//! Structures for tracking authenticated remote FIPS nodes. A Peer +//! represents an authenticated connection to another node in the mesh. + +use crate::bloom::BloomFilter; +use crate::transport::{LinkId, LinkStats}; +use crate::tree::{ParentDeclaration, TreeCoordinate}; +use crate::{FipsAddress, NodeId, PeerIdentity}; +use secp256k1::XOnlyPublicKey; +use std::fmt; +use thiserror::Error; + +/// Errors related to peer operations. +#[derive(Debug, Error)] +pub enum PeerError { + #[error("peer not authenticated")] + NotAuthenticated, + + #[error("peer not found: {0:?}")] + NotFound(NodeId), + + #[error("peer already exists: {0:?}")] + AlreadyExists(NodeId), + + #[error("peer state invalid for operation: expected {expected}, got {actual}")] + InvalidState { expected: &'static str, actual: PeerState }, + + #[error("peer disconnected")] + Disconnected, +} + +/// Peer lifecycle state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PeerState { + /// Known via discovery or config, no link yet. + Discovered, + /// Link establishment in progress (connection-oriented transports). + Connecting, + /// FIPS authentication handshake in progress. + Authenticating, + /// Fully integrated peer. + Active, + /// Was active, now disconnected. + Disconnected, +} + +impl PeerState { + /// Check if the peer is fully operational. + pub fn is_active(&self) -> bool { + matches!(self, PeerState::Active) + } + + /// Check if peer can receive data. + pub fn can_send(&self) -> bool { + matches!(self, PeerState::Active) + } + + /// Check if this is a terminal state. + pub fn is_terminal(&self) -> bool { + matches!(self, PeerState::Disconnected) + } + + /// Check if the peer is in the process of connecting. + pub fn is_connecting(&self) -> bool { + matches!(self, PeerState::Connecting | PeerState::Authenticating) + } +} + +impl fmt::Display for PeerState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + PeerState::Discovered => "discovered", + PeerState::Connecting => "connecting", + PeerState::Authenticating => "authenticating", + PeerState::Active => "active", + PeerState::Disconnected => "disconnected", + }; + write!(f, "{}", s) + } +} + +/// An authenticated remote FIPS node. +#[derive(Clone, Debug)] +pub struct Peer { + // === Identity === + /// Cryptographic identity (includes pubkey, node_id, address). + identity: PeerIdentity, + + // === Connection === + /// Link used to reach this peer. + link_id: LinkId, + /// Current lifecycle state. + state: PeerState, + + // === Spanning Tree === + /// Their latest parent declaration. + declaration: Option, + /// Their path to root. + ancestry: Option, + + // === Bloom Filter === + /// What's reachable through them (inbound filter). + inbound_filter: Option, + /// Their filter's sequence number. + filter_sequence: u64, + /// Remaining propagation hops on their filter. + filter_ttl: u8, + /// When we received their last filter (Unix milliseconds). + filter_received_at: u64, + /// Whether we owe them a filter update. + pending_filter_update: bool, + + // === Statistics === + /// Link statistics. + link_stats: LinkStats, + /// When this peer was first connected (Unix milliseconds). + connected_at: Option, + /// When this peer was last seen (any activity, Unix milliseconds). + last_seen: u64, +} + +impl Peer { + /// Create a new peer in Discovered state. + pub fn discovered(identity: PeerIdentity, link_id: LinkId) -> Self { + Self { + identity, + link_id, + state: PeerState::Discovered, + declaration: None, + ancestry: None, + inbound_filter: None, + filter_sequence: 0, + filter_ttl: 0, + filter_received_at: 0, + pending_filter_update: false, + link_stats: LinkStats::new(), + connected_at: None, + last_seen: 0, + } + } + + /// Create a new peer from a public key. + pub fn from_pubkey(pubkey: XOnlyPublicKey, link_id: LinkId) -> Self { + Self::discovered(PeerIdentity::from_pubkey(pubkey), link_id) + } + + // === Identity Accessors === + + /// Get the peer's identity. + pub fn identity(&self) -> &PeerIdentity { + &self.identity + } + + /// Get the peer's NodeId. + pub fn node_id(&self) -> &NodeId { + self.identity.node_id() + } + + /// Get the peer's FIPS address. + pub fn address(&self) -> &FipsAddress { + self.identity.address() + } + + /// Get the peer's public key. + pub fn pubkey(&self) -> XOnlyPublicKey { + self.identity.pubkey() + } + + /// Get the peer's npub string. + pub fn npub(&self) -> String { + self.identity.npub() + } + + // === Connection Accessors === + + /// Get the link ID. + pub fn link_id(&self) -> LinkId { + self.link_id + } + + /// Get the current state. + pub fn state(&self) -> PeerState { + self.state + } + + /// Check if the peer is active. + pub fn is_active(&self) -> bool { + self.state.is_active() + } + + /// Check if the peer can receive data. + pub fn can_send(&self) -> bool { + self.state.can_send() + } + + // === Tree Accessors === + + /// Get the peer's tree coordinates, if known. + pub fn coords(&self) -> Option<&TreeCoordinate> { + self.ancestry.as_ref() + } + + /// Get the peer's parent declaration, if known. + pub fn declaration(&self) -> Option<&ParentDeclaration> { + self.declaration.as_ref() + } + + /// Check if this peer has a known tree position. + pub fn has_tree_position(&self) -> bool { + self.declaration.is_some() && self.ancestry.is_some() + } + + // === Filter Accessors === + + /// Get the peer's inbound filter, if known. + pub fn inbound_filter(&self) -> Option<&BloomFilter> { + self.inbound_filter.as_ref() + } + + /// Get the filter sequence number. + pub fn filter_sequence(&self) -> u64 { + self.filter_sequence + } + + /// Get the filter TTL. + pub fn filter_ttl(&self) -> u8 { + self.filter_ttl + } + + /// Check if this peer's filter is stale. + pub fn filter_is_stale(&self, current_time_ms: u64, stale_threshold_ms: u64) -> bool { + if self.filter_received_at == 0 { + return true; + } + current_time_ms.saturating_sub(self.filter_received_at) > stale_threshold_ms + } + + /// Check if a destination might be reachable through this peer. + pub fn may_reach(&self, node_id: &NodeId) -> bool { + match &self.inbound_filter { + Some(filter) => filter.contains(node_id), + None => false, + } + } + + /// Check if we need to send this peer a filter update. + pub fn needs_filter_update(&self) -> bool { + self.pending_filter_update + } + + // === Statistics Accessors === + + /// Get link statistics. + pub fn link_stats(&self) -> &LinkStats { + &self.link_stats + } + + /// Get mutable link statistics. + pub fn link_stats_mut(&mut self) -> &mut LinkStats { + &mut self.link_stats + } + + /// Get when this peer was connected. + pub fn connected_at(&self) -> Option { + self.connected_at + } + + /// Get when this peer was last seen. + pub fn last_seen(&self) -> u64 { + self.last_seen + } + + /// Time since last activity. + pub fn idle_time(&self, current_time_ms: u64) -> u64 { + if self.last_seen == 0 { + return u64::MAX; + } + current_time_ms.saturating_sub(self.last_seen) + } + + /// Connection duration. + pub fn connection_duration(&self, current_time_ms: u64) -> Option { + self.connected_at + .map(|t| current_time_ms.saturating_sub(t)) + } + + // === State Transitions === + + /// Transition to Connecting state. + pub fn set_connecting(&mut self) { + self.state = PeerState::Connecting; + } + + /// Transition to Authenticating state. + pub fn set_authenticating(&mut self) { + self.state = PeerState::Authenticating; + } + + /// Transition to Active state. + pub fn set_active(&mut self, current_time_ms: u64) { + self.state = PeerState::Active; + self.connected_at = Some(current_time_ms); + self.last_seen = current_time_ms; + } + + /// Transition to Disconnected state. + pub fn set_disconnected(&mut self) { + self.state = PeerState::Disconnected; + } + + /// Update last seen timestamp. + pub fn touch(&mut self, current_time_ms: u64) { + self.last_seen = current_time_ms; + } + + // === Tree Updates === + + /// Update peer's tree position. + pub fn update_tree_position( + &mut self, + declaration: ParentDeclaration, + ancestry: TreeCoordinate, + current_time_ms: u64, + ) { + self.declaration = Some(declaration); + self.ancestry = Some(ancestry); + self.last_seen = current_time_ms; + } + + /// Clear peer's tree position. + pub fn clear_tree_position(&mut self) { + self.declaration = None; + self.ancestry = None; + } + + // === Filter Updates === + + /// Update peer's inbound filter. + pub fn update_filter( + &mut self, + filter: BloomFilter, + sequence: u64, + ttl: u8, + current_time_ms: u64, + ) { + self.inbound_filter = Some(filter); + self.filter_sequence = sequence; + self.filter_ttl = ttl; + self.filter_received_at = current_time_ms; + self.last_seen = current_time_ms; + } + + /// Clear peer's inbound filter. + pub fn clear_filter(&mut self) { + self.inbound_filter = None; + self.filter_sequence = 0; + self.filter_ttl = 0; + self.filter_received_at = 0; + } + + /// Mark that we need to send this peer a filter update. + pub fn mark_filter_update_needed(&mut self) { + self.pending_filter_update = true; + } + + /// Clear the pending filter update flag. + pub fn clear_filter_update_needed(&mut self) { + self.pending_filter_update = false; + } + + // === Link Updates === + + /// Update the link ID (e.g., on reconnect). + pub fn set_link_id(&mut self, link_id: LinkId) { + self.link_id = link_id; + } +} + +/// Simplified peer for leaf-only nodes. +/// +/// Leaf-only nodes maintain a single upstream peer without tree state +/// or Bloom filter management. +#[derive(Clone, Debug)] +pub struct UpstreamPeer { + /// Peer identity. + identity: PeerIdentity, + /// Link to upstream. + link_id: LinkId, + /// Lifecycle state (auth lifecycle only). + state: PeerState, + /// Link statistics. + link_stats: LinkStats, + /// When connected. + connected_at: Option, + /// Last activity. + last_seen: u64, +} + +impl UpstreamPeer { + /// Create a new upstream peer. + pub fn new(identity: PeerIdentity, link_id: LinkId) -> Self { + Self { + identity, + link_id, + state: PeerState::Discovered, + link_stats: LinkStats::new(), + connected_at: None, + last_seen: 0, + } + } + + /// Create from public key. + pub fn from_pubkey(pubkey: XOnlyPublicKey, link_id: LinkId) -> Self { + Self::new(PeerIdentity::from_pubkey(pubkey), link_id) + } + + /// Get the identity. + pub fn identity(&self) -> &PeerIdentity { + &self.identity + } + + /// Get the node ID. + pub fn node_id(&self) -> &NodeId { + self.identity.node_id() + } + + /// Get the FIPS address. + pub fn address(&self) -> &FipsAddress { + self.identity.address() + } + + /// Get the link ID. + pub fn link_id(&self) -> LinkId { + self.link_id + } + + /// Get the state. + pub fn state(&self) -> PeerState { + self.state + } + + /// Check if active. + pub fn is_active(&self) -> bool { + self.state.is_active() + } + + /// Get link statistics. + pub fn link_stats(&self) -> &LinkStats { + &self.link_stats + } + + /// Get mutable link statistics. + pub fn link_stats_mut(&mut self) -> &mut LinkStats { + &mut self.link_stats + } + + /// Set connecting state. + pub fn set_connecting(&mut self) { + self.state = PeerState::Connecting; + } + + /// Set authenticating state. + pub fn set_authenticating(&mut self) { + self.state = PeerState::Authenticating; + } + + /// Set active state. + pub fn set_active(&mut self, current_time_ms: u64) { + self.state = PeerState::Active; + self.connected_at = Some(current_time_ms); + self.last_seen = current_time_ms; + } + + /// Set disconnected state. + pub fn set_disconnected(&mut self) { + self.state = PeerState::Disconnected; + } + + /// Update last seen. + pub fn touch(&mut self, current_time_ms: u64) { + self.last_seen = current_time_ms; + } + + /// Get connected timestamp. + pub fn connected_at(&self) -> Option { + self.connected_at + } + + /// Get last seen timestamp. + pub fn last_seen(&self) -> u64 { + self.last_seen + } + + /// Set link ID. + pub fn set_link_id(&mut self, link_id: LinkId) { + self.link_id = link_id; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Identity; + + fn make_peer() -> Peer { + let identity = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); + Peer::discovered(peer_identity, LinkId::new(1)) + } + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::new(ids.iter().map(|&v| make_node_id(v)).collect()).unwrap() + } + + // ===== PeerState Tests ===== + + #[test] + fn test_peer_state_properties() { + assert!(!PeerState::Discovered.is_active()); + assert!(!PeerState::Connecting.is_active()); + assert!(!PeerState::Authenticating.is_active()); + assert!(PeerState::Active.is_active()); + assert!(!PeerState::Disconnected.is_active()); + + assert!(PeerState::Connecting.is_connecting()); + assert!(PeerState::Authenticating.is_connecting()); + assert!(!PeerState::Active.is_connecting()); + + assert!(PeerState::Disconnected.is_terminal()); + assert!(!PeerState::Active.is_terminal()); + } + + // ===== Peer Tests ===== + + #[test] + fn test_peer_state_transitions() { + let mut peer = make_peer(); + + assert_eq!(peer.state(), PeerState::Discovered); + assert!(!peer.is_active()); + + peer.set_connecting(); + assert_eq!(peer.state(), PeerState::Connecting); + + peer.set_authenticating(); + assert_eq!(peer.state(), PeerState::Authenticating); + + peer.set_active(1000); + assert_eq!(peer.state(), PeerState::Active); + assert!(peer.is_active()); + assert_eq!(peer.connected_at(), Some(1000)); + + peer.set_disconnected(); + assert_eq!(peer.state(), PeerState::Disconnected); + assert!(peer.state().is_terminal()); + } + + #[test] + fn test_peer_filter_stale() { + let mut peer = make_peer(); + + // No filter received yet + assert!(peer.filter_is_stale(1000, 500)); + + // Update filter + peer.update_filter(BloomFilter::new(), 1, 2, 1000); + + // Not stale yet + assert!(!peer.filter_is_stale(1200, 500)); + + // Stale after threshold + assert!(peer.filter_is_stale(1600, 500)); + } + + #[test] + fn test_peer_may_reach() { + let mut peer = make_peer(); + let target = make_node_id(42); + + // No filter yet + assert!(!peer.may_reach(&target)); + + // Add filter with target + let mut filter = BloomFilter::new(); + filter.insert(&target); + peer.update_filter(filter, 1, 2, 0); + + assert!(peer.may_reach(&target)); + } + + #[test] + fn test_peer_tree_position() { + let mut peer = make_peer(); + + assert!(!peer.has_tree_position()); + assert!(peer.coords().is_none()); + assert!(peer.declaration().is_none()); + + let node = make_node_id(1); + let parent = make_node_id(2); + let decl = ParentDeclaration::new(node, parent, 1, 1000); + let coords = make_coords(&[1, 2, 0]); + + peer.update_tree_position(decl, coords, 2000); + + assert!(peer.has_tree_position()); + assert!(peer.coords().is_some()); + assert!(peer.declaration().is_some()); + assert_eq!(peer.last_seen(), 2000); + } + + #[test] + fn test_peer_filter_update_flag() { + let mut peer = make_peer(); + + assert!(!peer.needs_filter_update()); + + peer.mark_filter_update_needed(); + assert!(peer.needs_filter_update()); + + peer.clear_filter_update_needed(); + assert!(!peer.needs_filter_update()); + } + + #[test] + fn test_peer_idle_time() { + let mut peer = make_peer(); + + // No activity yet + assert_eq!(peer.idle_time(1000), u64::MAX); + + peer.touch(500); + assert_eq!(peer.idle_time(1000), 500); + assert_eq!(peer.idle_time(500), 0); + } + + #[test] + fn test_peer_connection_duration() { + let mut peer = make_peer(); + + // Not connected + assert!(peer.connection_duration(1000).is_none()); + + peer.set_active(500); + assert_eq!(peer.connection_duration(1000), Some(500)); + } + + // ===== UpstreamPeer Tests ===== + + #[test] + fn test_upstream_peer_state_transitions() { + let identity = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); + let mut upstream = UpstreamPeer::new(peer_identity, LinkId::new(1)); + + assert!(!upstream.is_active()); + assert_eq!(upstream.state(), PeerState::Discovered); + + upstream.set_connecting(); + assert_eq!(upstream.state(), PeerState::Connecting); + + upstream.set_authenticating(); + assert_eq!(upstream.state(), PeerState::Authenticating); + + upstream.set_active(1000); + assert!(upstream.is_active()); + assert_eq!(upstream.connected_at(), Some(1000)); + + upstream.set_disconnected(); + assert!(!upstream.is_active()); + } + + #[test] + fn test_upstream_peer_touch() { + let identity = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey(identity.pubkey()); + let mut upstream = UpstreamPeer::new(peer_identity, LinkId::new(1)); + + assert_eq!(upstream.last_seen(), 0); + + upstream.touch(1000); + assert_eq!(upstream.last_seen(), 1000); + } +} diff --git a/src/protocol.rs b/src/protocol.rs new file mode 100644 index 0000000..dc26e0c --- /dev/null +++ b/src/protocol.rs @@ -0,0 +1,966 @@ +//! FIPS Protocol Messages +//! +//! Wire format message types for FIPS protocol communication, including +//! authentication handshake, spanning tree announcements, Bloom filter +//! propagation, discovery protocol, and data packets. + +use crate::bloom::BloomFilter; +use crate::tree::{ParentDeclaration, TreeCoordinate}; +use crate::{FipsAddress, NodeId}; +use rand::Rng; +use secp256k1::schnorr::Signature; +use secp256k1::XOnlyPublicKey; +use std::fmt; +use thiserror::Error; + +/// Protocol version for message compatibility. +pub const PROTOCOL_VERSION: u8 = 1; + +/// Data packet header size in bytes (excluding payload). +pub const DATA_HEADER_SIZE: usize = 36; + +/// Message type identifiers. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub enum MessageType { + // Authentication (0x00-0x0F) + Hello = 0x00, + Challenge = 0x01, + Auth = 0x02, + AuthAck = 0x03, + + // Tree protocol (0x10-0x1F) + TreeAnnounce = 0x10, + + // Bloom filter (0x20-0x2F) + FilterAnnounce = 0x20, + + // Discovery (0x30-0x3F) + LookupRequest = 0x30, + LookupResponse = 0x31, + + // Session (0x40-0x4F) + SessionSetup = 0x40, + SessionAck = 0x41, + + // Data (0x50-0x5F) + DataPacket = 0x50, + + // Errors (0x60-0x6F) + CoordsRequired = 0x60, + PathBroken = 0x61, +} + +impl MessageType { + /// Try to convert from a byte. + pub fn from_byte(b: u8) -> Option { + match b { + 0x00 => Some(MessageType::Hello), + 0x01 => Some(MessageType::Challenge), + 0x02 => Some(MessageType::Auth), + 0x03 => Some(MessageType::AuthAck), + 0x10 => Some(MessageType::TreeAnnounce), + 0x20 => Some(MessageType::FilterAnnounce), + 0x30 => Some(MessageType::LookupRequest), + 0x31 => Some(MessageType::LookupResponse), + 0x40 => Some(MessageType::SessionSetup), + 0x41 => Some(MessageType::SessionAck), + 0x50 => Some(MessageType::DataPacket), + 0x60 => Some(MessageType::CoordsRequired), + 0x61 => Some(MessageType::PathBroken), + _ => None, + } + } + + /// Convert to a byte. + pub fn to_byte(self) -> u8 { + self as u8 + } +} + +impl fmt::Display for MessageType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let name = match self { + MessageType::Hello => "Hello", + MessageType::Challenge => "Challenge", + MessageType::Auth => "Auth", + MessageType::AuthAck => "AuthAck", + MessageType::TreeAnnounce => "TreeAnnounce", + MessageType::FilterAnnounce => "FilterAnnounce", + MessageType::LookupRequest => "LookupRequest", + MessageType::LookupResponse => "LookupResponse", + MessageType::SessionSetup => "SessionSetup", + MessageType::SessionAck => "SessionAck", + MessageType::DataPacket => "DataPacket", + MessageType::CoordsRequired => "CoordsRequired", + MessageType::PathBroken => "PathBroken", + }; + write!(f, "{}", name) + } +} + +/// Errors related to protocol message handling. +#[derive(Debug, Error)] +pub enum ProtocolError { + #[error("invalid message type: 0x{0:02x}")] + InvalidMessageType(u8), + + #[error("message too short: expected at least {expected}, got {got}")] + MessageTooShort { expected: usize, got: usize }, + + #[error("message too long: max {max}, got {got}")] + MessageTooLong { max: usize, got: usize }, + + #[error("invalid signature")] + InvalidSignature, + + #[error("unsupported protocol version: {0}")] + UnsupportedVersion(u8), + + #[error("malformed message: {0}")] + Malformed(String), + + #[error("hop limit exceeded")] + HopLimitExceeded, + + #[error("ttl expired")] + TtlExpired, +} + +// ============ Authentication Messages ============ + +/// Initial hello message from initiator. +/// +/// The first message in the 4-step authentication handshake. +#[derive(Clone, Debug)] +pub struct Hello { + /// Initiator's public key. + pub pubkey: XOnlyPublicKey, +} + +impl Hello { + /// Create a new Hello message. + pub fn new(pubkey: XOnlyPublicKey) -> Self { + Self { pubkey } + } +} + +/// Challenge from responder, also contains responder's identity. +/// +/// The second message in the authentication handshake. +#[derive(Clone, Debug)] +pub struct Challenge { + /// Responder's public key. + pub pubkey: XOnlyPublicKey, + /// Random challenge for initiator to sign. + pub challenge: [u8; 32], +} + +impl Challenge { + /// Create a new Challenge with a specific challenge value. + pub fn new(pubkey: XOnlyPublicKey, challenge: [u8; 32]) -> Self { + Self { pubkey, challenge } + } + + /// Generate a Challenge with a random challenge value. + pub fn generate(pubkey: XOnlyPublicKey) -> Self { + let mut challenge = [0u8; 32]; + rand::thread_rng().fill(&mut challenge); + Self { pubkey, challenge } + } +} + +/// Authentication response from initiator. +/// +/// The third message in the authentication handshake. Contains the +/// initiator's challenge and their response to the responder's challenge. +#[derive(Clone, Debug)] +pub struct Auth { + /// Challenge for responder to sign. + pub challenge: [u8; 32], + /// Initiator's response to responder's challenge. + pub response: Signature, + /// Timestamp included in signed response (Unix seconds). + pub timestamp: u64, +} + +impl Auth { + /// Create a new Auth message. + pub fn new(challenge: [u8; 32], response: Signature, timestamp: u64) -> Self { + Self { + challenge, + response, + timestamp, + } + } + + /// Generate a new Auth with a random challenge. + pub fn generate(response: Signature, timestamp: u64) -> Self { + let mut challenge = [0u8; 32]; + rand::thread_rng().fill(&mut challenge); + Self { + challenge, + response, + timestamp, + } + } +} + +/// Final acknowledgement from responder. +/// +/// The fourth and final message in the authentication handshake. +#[derive(Clone, Debug)] +pub struct AuthAck { + /// Responder's response to initiator's challenge. + pub response: Signature, + /// Timestamp included in signed response (Unix seconds). + pub timestamp: u64, +} + +impl AuthAck { + /// Create a new AuthAck message. + pub fn new(response: Signature, timestamp: u64) -> Self { + Self { response, timestamp } + } +} + +// ============ Tree Protocol Messages ============ + +/// Spanning tree announcement carrying parent declaration and ancestry. +/// +/// Sent to peers to propagate tree state. The declaration proves the +/// sender's parent selection; the ancestry provides path to root for +/// routing decisions. +#[derive(Clone, Debug)] +pub struct TreeAnnounce { + /// The sender's parent declaration. + pub declaration: ParentDeclaration, + /// Full ancestry from sender to root. + pub ancestry: TreeCoordinate, +} + +impl TreeAnnounce { + /// Create a new TreeAnnounce message. + pub fn new(declaration: ParentDeclaration, ancestry: TreeCoordinate) -> Self { + Self { + declaration, + ancestry, + } + } +} + +// ============ Bloom Filter Messages ============ + +/// Bloom filter announcement for reachability propagation. +/// +/// Sent to peers to advertise which destinations are reachable. +/// The TTL controls propagation depth (decremented at each hop). +#[derive(Clone, Debug)] +pub struct FilterAnnounce { + /// The bloom filter contents. + pub filter: BloomFilter, + /// Remaining propagation hops (decremented at each forward). + pub ttl: u8, + /// Sequence number for freshness/dedup. + pub sequence: u64, +} + +impl FilterAnnounce { + /// Create a new FilterAnnounce message. + pub fn new(filter: BloomFilter, ttl: u8, sequence: u64) -> Self { + Self { + filter, + ttl, + sequence, + } + } + + /// Check if this filter can be forwarded (TTL > 0). + pub fn can_forward(&self) -> bool { + self.ttl > 0 + } + + /// Create a forwarded version with decremented TTL. + pub fn forwarded(&self) -> Option { + if self.ttl == 0 { + return None; + } + Some(Self { + filter: self.filter.clone(), + ttl: self.ttl - 1, + sequence: self.sequence, + }) + } +} + +// ============ Discovery Messages ============ + +/// Request to discover a node's coordinates. +/// +/// Flooded through the network with TTL limiting scope. The visited +/// filter prevents routing loops. +#[derive(Clone, Debug)] +pub struct LookupRequest { + /// Unique request identifier. + pub request_id: u64, + /// Node we're looking for. + pub target: NodeId, + /// Who's asking (for response routing). + pub origin: NodeId, + /// Origin's coordinates (for return path). + pub origin_coords: TreeCoordinate, + /// Remaining propagation hops. + pub ttl: u8, + /// Visited nodes filter (loop prevention). + pub visited: BloomFilter, +} + +impl LookupRequest { + /// Create a new lookup request. + pub fn new( + request_id: u64, + target: NodeId, + origin: NodeId, + origin_coords: TreeCoordinate, + ttl: u8, + ) -> Self { + // Small filter for visited tracking + let visited = BloomFilter::with_params(256 * 8, 5).expect("valid params"); + Self { + request_id, + target, + origin, + origin_coords, + ttl, + visited, + } + } + + /// Generate a new request with a random ID. + pub fn generate( + target: NodeId, + origin: NodeId, + origin_coords: TreeCoordinate, + ttl: u8, + ) -> Self { + use rand::Rng; + let request_id = rand::thread_rng().r#gen(); + Self::new(request_id, target, origin, origin_coords, ttl) + } + + /// Decrement TTL and add self to visited. + /// + /// Returns false if TTL was already 0. + pub fn forward(&mut self, my_node_id: &NodeId) -> bool { + if self.ttl == 0 { + return false; + } + self.ttl -= 1; + self.visited.insert(my_node_id); + true + } + + /// Check if this request can still be forwarded. + pub fn can_forward(&self) -> bool { + self.ttl > 0 + } + + /// Check if a node was already visited. + pub fn was_visited(&self, node_id: &NodeId) -> bool { + self.visited.contains(node_id) + } +} + +/// Response to a lookup request with target's coordinates. +/// +/// Routed back to the origin using the origin_coords from the request. +#[derive(Clone, Debug)] +pub struct LookupResponse { + /// Echoed request identifier. + pub request_id: u64, + /// The target node. + pub target: NodeId, + /// Target's coordinates in the tree. + pub target_coords: TreeCoordinate, + /// Proof that target authorized this response (signature over request). + pub proof: Signature, +} + +impl LookupResponse { + /// Create a new lookup response. + pub fn new( + request_id: u64, + target: NodeId, + target_coords: TreeCoordinate, + proof: Signature, + ) -> Self { + Self { + request_id, + target, + target_coords, + proof, + } + } + + /// Get the bytes that should be signed as proof. + /// + /// Format: request_id (8) || target (32) + pub fn proof_bytes(request_id: u64, target: &NodeId) -> Vec { + let mut bytes = Vec::with_capacity(40); + bytes.extend_from_slice(&request_id.to_le_bytes()); + bytes.extend_from_slice(target.as_bytes()); + bytes + } +} + +// ============ Session Messages ============ + +/// Session flags for setup options. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct SessionFlags { + /// Request acknowledgement from destination. + pub request_ack: bool, + /// Set up bidirectional session. + pub bidirectional: bool, +} + +impl SessionFlags { + /// Create default flags. + pub fn new() -> Self { + Self::default() + } + + /// Set request_ack flag. + pub fn with_ack(mut self) -> Self { + self.request_ack = true; + self + } + + /// Set bidirectional flag. + pub fn bidirectional(mut self) -> Self { + self.bidirectional = true; + self + } + + /// Convert to a byte. + pub fn to_byte(&self) -> u8 { + let mut flags = 0u8; + if self.request_ack { + flags |= 0x01; + } + if self.bidirectional { + flags |= 0x02; + } + flags + } + + /// Convert from a byte. + pub fn from_byte(byte: u8) -> Self { + Self { + request_ack: byte & 0x01 != 0, + bidirectional: byte & 0x02 != 0, + } + } +} + +/// Session setup to establish cached coordinate state. +/// +/// Sent before data packets to warm router caches with coordinate +/// information. Routers along the path cache the mappings. +#[derive(Clone, Debug)] +pub struct SessionSetup { + /// Source FIPS address. + pub src_addr: FipsAddress, + /// Destination FIPS address. + pub dest_addr: FipsAddress, + /// Source coordinates (for return path caching). + pub src_coords: TreeCoordinate, + /// Destination coordinates (for forward routing). + pub dest_coords: TreeCoordinate, + /// Session options. + pub flags: SessionFlags, +} + +impl SessionSetup { + /// Create a new session setup message. + pub fn new( + src_addr: FipsAddress, + dest_addr: FipsAddress, + src_coords: TreeCoordinate, + dest_coords: TreeCoordinate, + ) -> Self { + Self { + src_addr, + dest_addr, + src_coords, + dest_coords, + flags: SessionFlags::new(), + } + } + + /// Set session flags. + pub fn with_flags(mut self, flags: SessionFlags) -> Self { + self.flags = flags; + self + } +} + +/// Session acknowledgement. +/// +/// Sent in response to SessionSetup when request_ack is set. +#[derive(Clone, Debug)] +pub struct SessionAck { + /// Source address (the acknowledger). + pub src_addr: FipsAddress, + /// Destination address (original session initiator). + pub dest_addr: FipsAddress, + /// Acknowledger's coordinates. + pub src_coords: TreeCoordinate, +} + +impl SessionAck { + /// Create a new session acknowledgement. + pub fn new(src_addr: FipsAddress, dest_addr: FipsAddress, src_coords: TreeCoordinate) -> Self { + Self { + src_addr, + dest_addr, + src_coords, + } + } +} + +// ============ Data Messages ============ + +/// Data packet flags. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct DataFlags { + /// Reserved bits for future use. + reserved: u8, +} + +impl DataFlags { + /// Create default flags. + pub fn new() -> Self { + Self::default() + } + + /// Convert to a byte. + pub fn to_byte(&self) -> u8 { + self.reserved + } + + /// Convert from a byte. + pub fn from_byte(byte: u8) -> Self { + Self { reserved: byte } + } +} + +/// Minimal data packet with addresses only (no coordinates). +/// +/// The 36-byte header contains: +/// - flags (1 byte) +/// - hop_limit (1 byte) +/// - payload_length (2 bytes) +/// - src_addr (16 bytes) +/// - dest_addr (16 bytes) +/// +/// Routers use cached coordinates for routing decisions. +#[derive(Clone, Debug)] +pub struct DataPacket { + /// Packet flags. + pub flags: DataFlags, + /// Hop limit (TTL). + pub hop_limit: u8, + /// Source FIPS address. + pub src_addr: FipsAddress, + /// Destination FIPS address. + pub dest_addr: FipsAddress, + /// Payload data. + pub payload: Vec, +} + +impl DataPacket { + /// Create a new data packet. + pub fn new(src_addr: FipsAddress, dest_addr: FipsAddress, payload: Vec) -> Self { + Self { + flags: DataFlags::new(), + hop_limit: 64, + src_addr, + dest_addr, + payload, + } + } + + /// Set the hop limit. + pub fn with_hop_limit(mut self, hop_limit: u8) -> Self { + self.hop_limit = hop_limit; + self + } + + /// Set the flags. + pub fn with_flags(mut self, flags: DataFlags) -> Self { + self.flags = flags; + self + } + + /// Decrement hop limit, returning false if exhausted. + pub fn decrement_hop_limit(&mut self) -> bool { + if self.hop_limit > 0 { + self.hop_limit -= 1; + true + } else { + false + } + } + + /// Check if the packet can be forwarded. + pub fn can_forward(&self) -> bool { + self.hop_limit > 0 + } + + /// Get the payload length. + pub fn payload_len(&self) -> usize { + self.payload.len() + } + + /// Total packet size (header + payload). + pub fn total_size(&self) -> usize { + DATA_HEADER_SIZE + self.payload.len() + } + + /// Header size in bytes. + pub fn header_size(&self) -> usize { + DATA_HEADER_SIZE + } +} + +// ============ Error Messages ============ + +/// Error indicating router cache miss - needs coordinates. +/// +/// Sent back to the source when a router doesn't have cached +/// coordinates for the destination. +#[derive(Clone, Debug)] +pub struct CoordsRequired { + /// Destination that couldn't be routed. + pub dest_addr: FipsAddress, + /// Router reporting the miss. + pub reporter: NodeId, +} + +impl CoordsRequired { + /// Create a new CoordsRequired error. + pub fn new(dest_addr: FipsAddress, reporter: NodeId) -> Self { + Self { dest_addr, reporter } + } +} + +/// Error indicating routing failure (local minimum or unreachable). +/// +/// Sent back to the source when greedy routing fails. +#[derive(Clone, Debug)] +pub struct PathBroken { + /// Original source of the failed packet. + pub original_src: FipsAddress, + /// Destination that couldn't be reached. + pub dest_addr: FipsAddress, + /// Node that detected the failure. + pub reporter: NodeId, + /// Optional: last known coordinates of destination. + pub last_known_coords: Option, +} + +impl PathBroken { + /// Create a new PathBroken error. + pub fn new(original_src: FipsAddress, dest_addr: FipsAddress, reporter: NodeId) -> Self { + Self { + original_src, + dest_addr, + reporter, + last_known_coords: None, + } + } + + /// Add last known coordinates. + pub fn with_last_coords(mut self, coords: TreeCoordinate) -> Self { + self.last_known_coords = Some(coords); + self + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + fn make_address(val: u8) -> FipsAddress { + let mut bytes = [0xfdu8; 16]; + bytes[1] = val; + FipsAddress::from_bytes(bytes).unwrap() + } + + fn make_coords(ids: &[u8]) -> TreeCoordinate { + TreeCoordinate::new(ids.iter().map(|&v| make_node_id(v)).collect()).unwrap() + } + + // ===== MessageType Tests ===== + + #[test] + fn test_message_type_roundtrip() { + let types = [ + MessageType::Hello, + MessageType::Challenge, + MessageType::Auth, + MessageType::AuthAck, + MessageType::TreeAnnounce, + MessageType::FilterAnnounce, + MessageType::LookupRequest, + MessageType::LookupResponse, + MessageType::SessionSetup, + MessageType::SessionAck, + MessageType::DataPacket, + MessageType::CoordsRequired, + MessageType::PathBroken, + ]; + + for ty in types { + let byte = ty.to_byte(); + let restored = MessageType::from_byte(byte); + assert_eq!(restored, Some(ty)); + } + } + + #[test] + fn test_message_type_invalid() { + assert!(MessageType::from_byte(0xFF).is_none()); + assert!(MessageType::from_byte(0x99).is_none()); + } + + // ===== SessionFlags Tests ===== + + #[test] + fn test_session_flags() { + let flags = SessionFlags::new().with_ack().bidirectional(); + + assert!(flags.request_ack); + assert!(flags.bidirectional); + + let byte = flags.to_byte(); + let restored = SessionFlags::from_byte(byte); + + assert_eq!(flags, restored); + } + + #[test] + fn test_session_flags_default() { + let flags = SessionFlags::new(); + assert!(!flags.request_ack); + assert!(!flags.bidirectional); + assert_eq!(flags.to_byte(), 0); + } + + // ===== DataPacket Tests ===== + + #[test] + fn test_data_packet_size() { + let packet = DataPacket::new(make_address(1), make_address(2), vec![0u8; 100]); + + // 36 byte header + 100 byte payload + assert_eq!(packet.total_size(), 136); + assert_eq!(packet.header_size(), 36); + assert_eq!(packet.payload_len(), 100); + } + + #[test] + fn test_data_packet_hop_limit() { + let mut packet = DataPacket::new(make_address(1), make_address(2), vec![]); + + packet.hop_limit = 2; + assert!(packet.can_forward()); + + assert!(packet.decrement_hop_limit()); + assert_eq!(packet.hop_limit, 1); + + assert!(packet.decrement_hop_limit()); + assert_eq!(packet.hop_limit, 0); + assert!(!packet.can_forward()); + + assert!(!packet.decrement_hop_limit()); + assert_eq!(packet.hop_limit, 0); + } + + #[test] + fn test_data_packet_builder() { + let packet = DataPacket::new(make_address(1), make_address(2), vec![1, 2, 3]) + .with_hop_limit(32) + .with_flags(DataFlags::from_byte(0x80)); + + assert_eq!(packet.hop_limit, 32); + assert_eq!(packet.flags.to_byte(), 0x80); + } + + // ===== LookupRequest Tests ===== + + #[test] + fn test_lookup_request_forward() { + let target = make_node_id(1); + let origin = make_node_id(2); + let coords = make_coords(&[2, 0]); + let forwarder = make_node_id(3); + + let mut request = LookupRequest::new(123, target, origin, coords, 5); + + assert!(request.can_forward()); + assert!(!request.was_visited(&forwarder)); + + assert!(request.forward(&forwarder)); + + assert_eq!(request.ttl, 4); + assert!(request.was_visited(&forwarder)); + } + + #[test] + fn test_lookup_request_ttl_exhausted() { + let target = make_node_id(1); + let origin = make_node_id(2); + let coords = make_coords(&[2, 0]); + + let mut request = LookupRequest::new(123, target, origin, coords, 1); + + assert!(request.forward(&make_node_id(3))); + assert!(!request.can_forward()); + assert!(!request.forward(&make_node_id(4))); + } + + #[test] + fn test_lookup_request_generate() { + let target = make_node_id(1); + let origin = make_node_id(2); + let coords = make_coords(&[2, 0]); + + let req1 = LookupRequest::generate(target, origin, coords.clone(), 5); + let req2 = LookupRequest::generate(target, origin, coords, 5); + + // Random IDs should differ + assert_ne!(req1.request_id, req2.request_id); + } + + // ===== LookupResponse Tests ===== + + #[test] + fn test_lookup_response_proof_bytes() { + let target = make_node_id(42); + let bytes = LookupResponse::proof_bytes(12345, &target); + + assert_eq!(bytes.len(), 40); // 8 + 32 + assert_eq!(&bytes[0..8], &12345u64.to_le_bytes()); + assert_eq!(&bytes[8..40], target.as_bytes()); + } + + // ===== Challenge Tests ===== + + #[test] + fn test_challenge_generate() { + let secp = secp256k1::Secp256k1::new(); + let keypair = secp256k1::Keypair::new(&secp, &mut rand::thread_rng()); + let pubkey = keypair.x_only_public_key().0; + + let challenge1 = Challenge::generate(pubkey); + let challenge2 = Challenge::generate(pubkey); + + // Challenges should be different (random) + assert_ne!(challenge1.challenge, challenge2.challenge); + } + + // ===== FilterAnnounce Tests ===== + + #[test] + fn test_filter_announce_forward() { + let filter = BloomFilter::new(); + let announce = FilterAnnounce::new(filter, 2, 100); + + assert!(announce.can_forward()); + + let forwarded = announce.forwarded().unwrap(); + assert_eq!(forwarded.ttl, 1); + assert_eq!(forwarded.sequence, 100); + + let forwarded2 = forwarded.forwarded().unwrap(); + assert_eq!(forwarded2.ttl, 0); + assert!(!forwarded2.can_forward()); + + assert!(forwarded2.forwarded().is_none()); + } + + // ===== SessionSetup Tests ===== + + #[test] + fn test_session_setup() { + let setup = SessionSetup::new( + make_address(1), + make_address(2), + make_coords(&[1, 0]), + make_coords(&[2, 0]), + ) + .with_flags(SessionFlags::new().with_ack()); + + assert!(setup.flags.request_ack); + assert!(!setup.flags.bidirectional); + } + + // ===== CoordsRequired Tests ===== + + #[test] + fn test_coords_required() { + let err = CoordsRequired::new(make_address(1), make_node_id(2)); + + assert_eq!(err.dest_addr, make_address(1)); + assert_eq!(err.reporter, make_node_id(2)); + } + + // ===== PathBroken Tests ===== + + #[test] + fn test_path_broken() { + let err = PathBroken::new(make_address(1), make_address(2), make_node_id(3)) + .with_last_coords(make_coords(&[2, 0])); + + assert!(err.last_known_coords.is_some()); + } + + // ===== Auth Tests ===== + + #[test] + fn test_auth_generate() { + let secp = secp256k1::Secp256k1::new(); + let keypair = secp256k1::Keypair::new(&secp, &mut rand::thread_rng()); + let digest = [0u8; 32]; + let sig = secp.sign_schnorr(&digest, &keypair); + + let auth1 = Auth::generate(sig, 1000); + let auth2 = Auth::generate(sig, 1000); + + // Random challenges should differ + assert_ne!(auth1.challenge, auth2.challenge); + } + + // ===== TreeAnnounce Tests ===== + + #[test] + fn test_tree_announce() { + let node = make_node_id(1); + let parent = make_node_id(2); + let decl = ParentDeclaration::new(node, parent, 1, 1000); + let ancestry = make_coords(&[1, 2, 0]); + + let announce = TreeAnnounce::new(decl, ancestry); + + assert_eq!(announce.declaration.node_id(), &node); + assert_eq!(announce.ancestry.depth(), 2); + } +} diff --git a/src/transport.rs b/src/transport.rs new file mode 100644 index 0000000..d31766d --- /dev/null +++ b/src/transport.rs @@ -0,0 +1,886 @@ +//! Transport Layer Abstractions +//! +//! Traits and types for FIPS transport drivers. Transports provide the +//! underlying communication mechanisms (UDP, Ethernet, Tor, etc.) over +//! which FIPS links are established. + +use secp256k1::XOnlyPublicKey; +use std::fmt; +use std::time::Duration; +use thiserror::Error; + +/// Unique identifier for a transport instance. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct TransportId(u32); + +impl TransportId { + /// Create a new transport ID. + pub fn new(id: u32) -> Self { + Self(id) + } + + /// Get the raw ID value. + pub fn as_u32(&self) -> u32 { + self.0 + } +} + +impl fmt::Display for TransportId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "transport:{}", self.0) + } +} + +/// Unique identifier for a link instance. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct LinkId(u64); + +impl LinkId { + /// Create a new link ID. + pub fn new(id: u64) -> Self { + Self(id) + } + + /// Get the raw ID value. + pub fn as_u64(&self) -> u64 { + self.0 + } +} + +impl fmt::Display for LinkId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "link:{}", self.0) + } +} + +/// Errors related to transport operations. +#[derive(Debug, Error)] +pub enum TransportError { + #[error("transport not started")] + NotStarted, + + #[error("transport already started")] + AlreadyStarted, + + #[error("transport failed to start: {0}")] + StartFailed(String), + + #[error("transport shutdown failed: {0}")] + ShutdownFailed(String), + + #[error("link failed: {0}")] + LinkFailed(String), + + #[error("send failed: {0}")] + SendFailed(String), + + #[error("receive failed: {0}")] + RecvFailed(String), + + #[error("invalid transport address: {0}")] + InvalidAddress(String), + + #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")] + MtuExceeded { packet_size: usize, mtu: u16 }, + + #[error("transport timeout")] + Timeout, + + #[error("connection refused")] + ConnectionRefused, + + #[error("transport not supported: {0}")] + NotSupported(String), + + #[error("io error: {0}")] + Io(#[from] std::io::Error), +} + +/// Static metadata about a transport type. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TransportType { + /// Human-readable name (e.g., "udp", "ethernet", "tor"). + pub name: &'static str, + /// Whether this transport requires connection establishment. + pub connection_oriented: bool, + /// Whether the transport guarantees delivery. + pub reliable: bool, +} + +impl TransportType { + /// UDP/IP transport. + pub const UDP: TransportType = TransportType { + name: "udp", + connection_oriented: false, + reliable: false, + }; + + /// TCP/IP transport. + pub const TCP: TransportType = TransportType { + name: "tcp", + connection_oriented: true, + reliable: true, + }; + + /// Raw Ethernet transport. + pub const ETHERNET: TransportType = TransportType { + name: "ethernet", + connection_oriented: false, + reliable: false, + }; + + /// WiFi (same characteristics as Ethernet). + pub const WIFI: TransportType = TransportType { + name: "wifi", + connection_oriented: false, + reliable: false, + }; + + /// Tor onion transport. + pub const TOR: TransportType = TransportType { + name: "tor", + connection_oriented: true, + reliable: true, + }; + + /// Serial/UART transport. + pub const SERIAL: TransportType = TransportType { + name: "serial", + connection_oriented: false, + reliable: true, // typically uses framing with checksums + }; + + /// Check if the transport is connectionless. + pub fn is_connectionless(&self) -> bool { + !self.connection_oriented + } +} + +impl fmt::Display for TransportType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.name) + } +} + +/// Transport lifecycle state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TransportState { + /// Configured but not started. + Configured, + /// Initialization in progress. + Starting, + /// Ready for links. + Up, + /// Was up, now unavailable. + Down, + /// Failed to start. + Failed, +} + +impl TransportState { + /// Check if the transport is operational. + pub fn is_operational(&self) -> bool { + matches!(self, TransportState::Up) + } + + /// Check if the transport can be started. + pub fn can_start(&self) -> bool { + matches!( + self, + TransportState::Configured | TransportState::Down | TransportState::Failed + ) + } + + /// Check if the transport is in a terminal state. + pub fn is_terminal(&self) -> bool { + matches!(self, TransportState::Failed) + } +} + +impl fmt::Display for TransportState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + TransportState::Configured => "configured", + TransportState::Starting => "starting", + TransportState::Up => "up", + TransportState::Down => "down", + TransportState::Failed => "failed", + }; + write!(f, "{}", s) + } +} + +/// Link lifecycle state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LinkState { + /// Connection in progress (connection-oriented only). + Connecting, + /// Ready for traffic. + Connected, + /// Was connected, now gone. + Disconnected, + /// Connection attempt failed. + Failed, +} + +impl LinkState { + /// Check if the link is operational. + pub fn is_operational(&self) -> bool { + matches!(self, LinkState::Connected) + } + + /// Check if the link is in a terminal state. + pub fn is_terminal(&self) -> bool { + matches!(self, LinkState::Disconnected | LinkState::Failed) + } +} + +impl fmt::Display for LinkState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + LinkState::Connecting => "connecting", + LinkState::Connected => "connected", + LinkState::Disconnected => "disconnected", + LinkState::Failed => "failed", + }; + write!(f, "{}", s) + } +} + +/// Direction of link establishment. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LinkDirection { + /// We initiated the connection. + Outbound, + /// They initiated the connection. + Inbound, +} + +impl fmt::Display for LinkDirection { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + LinkDirection::Outbound => "outbound", + LinkDirection::Inbound => "inbound", + }; + write!(f, "{}", s) + } +} + +/// Opaque transport-specific address. +/// +/// Each transport type interprets this differently: +/// - UDP: "ip:port" +/// - Ethernet: MAC address (6 bytes) +/// - Tor: ".onion:port" +#[derive(Clone, PartialEq, Eq, Hash)] +pub struct TransportAddr(Vec); + +impl TransportAddr { + /// Create a transport address from raw bytes. + pub fn new(bytes: Vec) -> Self { + Self(bytes) + } + + /// Create a transport address from a byte slice. + pub fn from_bytes(bytes: &[u8]) -> Self { + Self(bytes.to_vec()) + } + + /// Create a transport address from a string. + pub fn from_string(s: &str) -> Self { + Self(s.as_bytes().to_vec()) + } + + /// Get the raw bytes. + pub fn as_bytes(&self) -> &[u8] { + &self.0 + } + + /// Try to interpret as a UTF-8 string. + pub fn as_str(&self) -> Option<&str> { + std::str::from_utf8(&self.0).ok() + } + + /// Get the length in bytes. + pub fn len(&self) -> usize { + self.0.len() + } + + /// Check if empty. + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + +impl fmt::Debug for TransportAddr { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.as_str() { + Some(s) => write!(f, "TransportAddr(\"{}\")", s), + None => write!(f, "TransportAddr({:?})", self.0), + } + } +} + +impl fmt::Display for TransportAddr { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Best-effort display as string if valid UTF-8, else hex + match self.as_str() { + Some(s) => write!(f, "{}", s), + None => { + for byte in &self.0 { + write!(f, "{:02x}", byte)?; + } + Ok(()) + } + } + } +} + +impl From<&str> for TransportAddr { + fn from(s: &str) -> Self { + Self::from_string(s) + } +} + +impl From for TransportAddr { + fn from(s: String) -> Self { + Self(s.into_bytes()) + } +} + +/// Statistics for a link. +#[derive(Clone, Debug, Default)] +pub struct LinkStats { + /// Total packets sent. + pub packets_sent: u64, + /// Total packets received. + pub packets_recv: u64, + /// Total bytes sent. + pub bytes_sent: u64, + /// Total bytes received. + pub bytes_recv: u64, + /// Timestamp of last received packet (Unix milliseconds). + pub last_recv_ms: u64, + /// Estimated round-trip time. + rtt_estimate: Option, + /// Observed packet loss rate (0.0-1.0). + pub loss_rate: f32, + /// Estimated throughput in bytes/second. + pub throughput_estimate: u64, +} + +impl LinkStats { + /// Create new link statistics. + pub fn new() -> Self { + Self::default() + } + + /// Record a sent packet. + pub fn record_sent(&mut self, bytes: usize) { + self.packets_sent += 1; + self.bytes_sent += bytes as u64; + } + + /// Record a received packet. + pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) { + self.packets_recv += 1; + self.bytes_recv += bytes as u64; + self.last_recv_ms = timestamp_ms; + } + + /// Get the RTT estimate, if available. + pub fn rtt_estimate(&self) -> Option { + self.rtt_estimate + } + + /// Update RTT estimate from a probe response. + /// + /// Uses exponential moving average with alpha=0.2. + pub fn update_rtt(&mut self, rtt: Duration) { + match self.rtt_estimate { + Some(old_rtt) => { + let alpha = 0.2; + let new_rtt_nanos = (alpha * rtt.as_nanos() as f64 + + (1.0 - alpha) * old_rtt.as_nanos() as f64) + as u64; + self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos)); + } + None => { + self.rtt_estimate = Some(rtt); + } + } + } + + /// Time since last receive (for keepalive/timeout). + pub fn time_since_recv(&self, current_time_ms: u64) -> u64 { + if self.last_recv_ms == 0 { + return u64::MAX; + } + current_time_ms.saturating_sub(self.last_recv_ms) + } + + /// Reset all statistics. + pub fn reset(&mut self) { + *self = Self::default(); + } +} + +/// A link to a remote endpoint over a transport. +#[derive(Clone, Debug)] +pub struct Link { + /// Unique link identifier. + link_id: LinkId, + /// Which transport this link uses. + transport_id: TransportId, + /// Transport-specific remote address. + remote_addr: TransportAddr, + /// Whether we initiated or they initiated. + direction: LinkDirection, + /// Current link state. + state: LinkState, + /// Base RTT hint from transport type. + base_rtt: Duration, + /// Measured statistics. + stats: LinkStats, + /// When this link was created (Unix milliseconds). + created_at: u64, +} + +impl Link { + /// Create a new link in Connecting state. + pub fn new( + link_id: LinkId, + transport_id: TransportId, + remote_addr: TransportAddr, + direction: LinkDirection, + base_rtt: Duration, + ) -> Self { + Self { + link_id, + transport_id, + remote_addr, + direction, + state: LinkState::Connecting, + base_rtt, + stats: LinkStats::new(), + created_at: 0, + } + } + + /// Create a link with a creation timestamp. + pub fn new_with_timestamp( + link_id: LinkId, + transport_id: TransportId, + remote_addr: TransportAddr, + direction: LinkDirection, + base_rtt: Duration, + created_at: u64, + ) -> Self { + let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt); + link.created_at = created_at; + link + } + + /// Create a connectionless link (immediately connected). + /// + /// For connectionless transports (UDP, Ethernet), links are immediately + /// in the Connected state. + pub fn connectionless( + link_id: LinkId, + transport_id: TransportId, + remote_addr: TransportAddr, + direction: LinkDirection, + base_rtt: Duration, + ) -> Self { + let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt); + link.state = LinkState::Connected; + link + } + + /// Get the link ID. + pub fn link_id(&self) -> LinkId { + self.link_id + } + + /// Get the transport ID. + pub fn transport_id(&self) -> TransportId { + self.transport_id + } + + /// Get the remote address. + pub fn remote_addr(&self) -> &TransportAddr { + &self.remote_addr + } + + /// Get the link direction. + pub fn direction(&self) -> LinkDirection { + self.direction + } + + /// Get the current state. + pub fn state(&self) -> LinkState { + self.state + } + + /// Get the base RTT hint. + pub fn base_rtt(&self) -> Duration { + self.base_rtt + } + + /// Get the link statistics. + pub fn stats(&self) -> &LinkStats { + &self.stats + } + + /// Get mutable access to link statistics. + pub fn stats_mut(&mut self) -> &mut LinkStats { + &mut self.stats + } + + /// Get the creation timestamp. + pub fn created_at(&self) -> u64 { + self.created_at + } + + /// Set the creation timestamp. + pub fn set_created_at(&mut self, timestamp: u64) { + self.created_at = timestamp; + } + + /// Mark the link as connected. + pub fn set_connected(&mut self) { + self.state = LinkState::Connected; + } + + /// Mark the link as disconnected. + pub fn set_disconnected(&mut self) { + self.state = LinkState::Disconnected; + } + + /// Mark the link as failed. + pub fn set_failed(&mut self) { + self.state = LinkState::Failed; + } + + /// Check if this link is operational. + pub fn is_operational(&self) -> bool { + self.state.is_operational() + } + + /// Check if this link is in a terminal state. + pub fn is_terminal(&self) -> bool { + self.state.is_terminal() + } + + /// Get effective RTT (measured if available, else base hint). + pub fn effective_rtt(&self) -> Duration { + self.stats.rtt_estimate().unwrap_or(self.base_rtt) + } + + /// Age of the link in milliseconds. + pub fn age(&self, current_time_ms: u64) -> u64 { + if self.created_at == 0 { + return 0; + } + current_time_ms.saturating_sub(self.created_at) + } +} + +/// A peer discovered via transport-layer discovery. +#[derive(Clone, Debug)] +pub struct DiscoveredPeer { + /// Transport that discovered this peer. + pub transport_id: TransportId, + /// Transport address where the peer was found. + pub addr: TransportAddr, + /// Optional hint about the peer's identity (if known from discovery). + pub pubkey_hint: Option, +} + +impl DiscoveredPeer { + /// Create a discovered peer without identity hint. + pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self { + Self { + transport_id, + addr, + pubkey_hint: None, + } + } + + /// Create a discovered peer with identity hint. + pub fn with_hint( + transport_id: TransportId, + addr: TransportAddr, + pubkey: XOnlyPublicKey, + ) -> Self { + Self { + transport_id, + addr, + pubkey_hint: Some(pubkey), + } + } +} + +/// Transport trait defining the interface for transport drivers. +/// +/// This is a simplified synchronous trait. Actual implementations would +/// be async and use channels for event delivery. +pub trait Transport { + /// Get the transport identifier. + fn transport_id(&self) -> TransportId; + + /// Get the transport type metadata. + fn transport_type(&self) -> &TransportType; + + /// Get the current state. + fn state(&self) -> TransportState; + + /// Get the MTU for this transport. + fn mtu(&self) -> u16; + + /// Start the transport. + fn start(&mut self) -> Result<(), TransportError>; + + /// Stop the transport. + fn stop(&mut self) -> Result<(), TransportError>; + + /// Send data to a transport address. + fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>; + + /// Discover potential peers (if supported). + fn discover(&self) -> Result, TransportError>; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_transport_id() { + let id = TransportId::new(42); + assert_eq!(id.as_u32(), 42); + assert_eq!(format!("{}", id), "transport:42"); + } + + #[test] + fn test_link_id() { + let id = LinkId::new(12345); + assert_eq!(id.as_u64(), 12345); + assert_eq!(format!("{}", id), "link:12345"); + } + + #[test] + fn test_transport_state_transitions() { + assert!(TransportState::Configured.can_start()); + assert!(TransportState::Down.can_start()); + assert!(TransportState::Failed.can_start()); + assert!(!TransportState::Starting.can_start()); + assert!(!TransportState::Up.can_start()); + + assert!(TransportState::Up.is_operational()); + assert!(!TransportState::Starting.is_operational()); + assert!(!TransportState::Failed.is_operational()); + } + + #[test] + fn test_link_state() { + assert!(LinkState::Connected.is_operational()); + assert!(!LinkState::Connecting.is_operational()); + assert!(!LinkState::Disconnected.is_operational()); + assert!(!LinkState::Failed.is_operational()); + + assert!(LinkState::Disconnected.is_terminal()); + assert!(LinkState::Failed.is_terminal()); + assert!(!LinkState::Connected.is_terminal()); + } + + #[test] + fn test_transport_type_constants() { + assert!(!TransportType::UDP.connection_oriented); + assert!(!TransportType::UDP.reliable); + assert!(TransportType::UDP.is_connectionless()); + + assert!(TransportType::TOR.connection_oriented); + assert!(TransportType::TOR.reliable); + assert!(!TransportType::TOR.is_connectionless()); + + assert_eq!(TransportType::UDP.name, "udp"); + assert_eq!(TransportType::ETHERNET.name, "ethernet"); + } + + #[test] + fn test_transport_addr_string() { + let addr = TransportAddr::from_string("192.168.1.1:4000"); + assert_eq!(format!("{}", addr), "192.168.1.1:4000"); + assert_eq!(addr.as_str(), Some("192.168.1.1:4000")); + } + + #[test] + fn test_transport_addr_binary() { + // Binary address with invalid UTF-8 bytes (0xff, 0x80 are invalid UTF-8) + let binary = TransportAddr::new(vec![0xff, 0x80, 0x2b, 0x3c, 0x4d, 0x5e]); + assert_eq!(format!("{}", binary), "ff802b3c4d5e"); + assert!(binary.as_str().is_none()); + assert_eq!(binary.len(), 6); + } + + #[test] + fn test_transport_addr_from_string() { + let addr: TransportAddr = "test:1234".into(); + assert_eq!(addr.as_str(), Some("test:1234")); + + let addr2: TransportAddr = String::from("hello").into(); + assert_eq!(addr2.as_str(), Some("hello")); + } + + #[test] + fn test_link_stats_basic() { + let mut stats = LinkStats::new(); + + stats.record_sent(100); + stats.record_recv(200, 1000); + + assert_eq!(stats.packets_sent, 1); + assert_eq!(stats.bytes_sent, 100); + assert_eq!(stats.packets_recv, 1); + assert_eq!(stats.bytes_recv, 200); + assert_eq!(stats.last_recv_ms, 1000); + } + + #[test] + fn test_link_stats_rtt() { + let mut stats = LinkStats::new(); + + assert!(stats.rtt_estimate().is_none()); + + stats.update_rtt(Duration::from_millis(100)); + assert_eq!(stats.rtt_estimate(), Some(Duration::from_millis(100))); + + // Second update uses EMA + stats.update_rtt(Duration::from_millis(200)); + // EMA: 0.2 * 200 + 0.8 * 100 = 120ms + let rtt = stats.rtt_estimate().unwrap(); + assert!(rtt.as_millis() >= 110 && rtt.as_millis() <= 130); + } + + #[test] + fn test_link_stats_time_since_recv() { + let mut stats = LinkStats::new(); + + // No receive yet + assert_eq!(stats.time_since_recv(1000), u64::MAX); + + stats.record_recv(100, 500); + assert_eq!(stats.time_since_recv(1000), 500); + assert_eq!(stats.time_since_recv(500), 0); + } + + #[test] + fn test_link_creation() { + let link = Link::new( + LinkId::new(1), + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + + assert_eq!(link.state(), LinkState::Connecting); + assert!(!link.is_operational()); + assert_eq!(link.direction(), LinkDirection::Outbound); + } + + #[test] + fn test_link_connectionless() { + let link = Link::connectionless( + LinkId::new(1), + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Inbound, + Duration::from_millis(5), + ); + + assert_eq!(link.state(), LinkState::Connected); + assert!(link.is_operational()); + } + + #[test] + fn test_link_state_changes() { + let mut link = Link::new( + LinkId::new(1), + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + + assert!(!link.is_operational()); + + link.set_connected(); + assert!(link.is_operational()); + assert!(!link.is_terminal()); + + link.set_disconnected(); + assert!(!link.is_operational()); + assert!(link.is_terminal()); + } + + #[test] + fn test_link_effective_rtt() { + let mut link = Link::connectionless( + LinkId::new(1), + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Inbound, + Duration::from_millis(50), + ); + + // Before measurement, uses base RTT + assert_eq!(link.effective_rtt(), Duration::from_millis(50)); + + // After measurement, uses measured RTT + link.stats_mut().update_rtt(Duration::from_millis(100)); + assert_eq!(link.effective_rtt(), Duration::from_millis(100)); + } + + #[test] + fn test_link_age() { + let mut link = Link::new( + LinkId::new(1), + TransportId::new(1), + TransportAddr::from_string("test"), + LinkDirection::Outbound, + Duration::from_millis(50), + ); + + // No timestamp set + assert_eq!(link.age(1000), 0); + + link.set_created_at(500); + assert_eq!(link.age(1000), 500); + assert_eq!(link.age(500), 0); + } + + #[test] + fn test_discovered_peer() { + let peer = DiscoveredPeer::new( + TransportId::new(1), + TransportAddr::from_string("192.168.1.1:4000"), + ); + + assert_eq!(peer.transport_id, TransportId::new(1)); + assert!(peer.pubkey_hint.is_none()); + } + + #[test] + fn test_link_direction_display() { + assert_eq!(format!("{}", LinkDirection::Outbound), "outbound"); + assert_eq!(format!("{}", LinkDirection::Inbound), "inbound"); + } + + #[test] + fn test_transport_state_display() { + assert_eq!(format!("{}", TransportState::Up), "up"); + assert_eq!(format!("{}", TransportState::Failed), "failed"); + } +} diff --git a/src/tree.rs b/src/tree.rs new file mode 100644 index 0000000..1a12bf8 --- /dev/null +++ b/src/tree.rs @@ -0,0 +1,876 @@ +//! Spanning Tree Protocol Entities +//! +//! Tree coordinates and parent declarations for the FIPS spanning tree. +//! 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 secp256k1::schnorr::Signature; +use secp256k1::XOnlyPublicKey; +use std::collections::HashMap; +use std::fmt; +use thiserror::Error; + +/// Errors related to spanning tree operations. +#[derive(Debug, Error)] +pub enum TreeError { + #[error("invalid tree coordinate: empty path")] + EmptyCoordinate, + + #[error("invalid ancestry: does not reach claimed root")] + AncestryNotToRoot, + + #[error("signature verification failed for node {0:?}")] + InvalidSignature(NodeId), + + #[error("sequence number regression: got {got}, expected > {expected}")] + SequenceRegression { got: u64, expected: u64 }, + + #[error("parent not in peers: {0:?}")] + ParentNotPeer(NodeId), + + #[error("identity error: {0}")] + Identity(#[from] IdentityError), +} + +/// A node's declaration of its parent in the spanning tree. +/// +/// Each node periodically announces its parent selection. The declaration +/// includes a monotonic sequence number for freshness and a signature +/// for authenticity. When `parent_id == node_id`, the node declares itself +/// as a root candidate. +#[derive(Clone)] +pub struct ParentDeclaration { + /// The node making this declaration. + node_id: NodeId, + /// The selected parent (equals node_id if self-declaring as root). + parent_id: NodeId, + /// Monotonically increasing sequence number. + sequence: u64, + /// Timestamp when this declaration was created (Unix seconds). + timestamp: u64, + /// Schnorr signature over the declaration fields. + signature: Option, +} + +impl ParentDeclaration { + /// Create a new unsigned parent declaration. + /// + /// The declaration must be signed before transmission using `set_signature()`. + pub fn new(node_id: NodeId, parent_id: NodeId, sequence: u64, timestamp: u64) -> Self { + Self { + node_id, + parent_id, + sequence, + timestamp, + signature: None, + } + } + + /// Create a self-declaration (node is root candidate). + pub fn self_root(node_id: NodeId, sequence: u64, timestamp: u64) -> Self { + Self::new(node_id, node_id, sequence, timestamp) + } + + /// Create a declaration with a pre-computed signature. + pub fn with_signature( + node_id: NodeId, + parent_id: NodeId, + sequence: u64, + timestamp: u64, + signature: Signature, + ) -> Self { + Self { + node_id, + parent_id, + sequence, + timestamp, + signature: Some(signature), + } + } + + /// Get the declaring node's ID. + pub fn node_id(&self) -> &NodeId { + &self.node_id + } + + /// Get the parent node's ID. + pub fn parent_id(&self) -> &NodeId { + &self.parent_id + } + + /// Get the sequence number. + pub fn sequence(&self) -> u64 { + self.sequence + } + + /// Get the timestamp. + pub fn timestamp(&self) -> u64 { + self.timestamp + } + + /// Get the signature, if set. + pub fn signature(&self) -> Option<&Signature> { + self.signature.as_ref() + } + + /// Set the signature after signing. + pub fn set_signature(&mut self, signature: Signature) { + self.signature = Some(signature); + } + + /// Check if this is a root declaration (parent == self). + pub fn is_root(&self) -> bool { + self.node_id == self.parent_id + } + + /// Check if this declaration is signed. + pub fn is_signed(&self) -> bool { + self.signature.is_some() + } + + /// Get the bytes that should be signed. + /// + /// Format: node_id (32) || parent_id (32) || sequence (8) || timestamp (8) + pub fn signing_bytes(&self) -> Vec { + let mut bytes = Vec::with_capacity(80); + bytes.extend_from_slice(self.node_id.as_bytes()); + bytes.extend_from_slice(self.parent_id.as_bytes()); + bytes.extend_from_slice(&self.sequence.to_le_bytes()); + bytes.extend_from_slice(&self.timestamp.to_le_bytes()); + bytes + } + + /// Verify the signature on this declaration. + /// + /// Returns Ok(()) if the signature is valid, or an error otherwise. + pub fn verify(&self, pubkey: &XOnlyPublicKey) -> Result<(), TreeError> { + let signature = self + .signature + .as_ref() + .ok_or(TreeError::InvalidSignature(self.node_id))?; + + let secp = secp256k1::Secp256k1::verification_only(); + let hash = self.signing_hash(); + + secp.verify_schnorr(signature, &hash, pubkey) + .map_err(|_| TreeError::InvalidSignature(self.node_id)) + } + + /// Compute the SHA-256 hash of the signing bytes. + fn signing_hash(&self) -> [u8; 32] { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(self.signing_bytes()); + hasher.finalize().into() + } + + /// Check if this declaration is fresher than another. + pub fn is_fresher_than(&self, other: &ParentDeclaration) -> bool { + self.sequence > other.sequence + } +} + +impl fmt::Debug for ParentDeclaration { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ParentDeclaration") + .field("node_id", &self.node_id) + .field("parent_id", &self.parent_id) + .field("sequence", &self.sequence) + .field("is_root", &self.is_root()) + .field("signed", &self.is_signed()) + .finish() + } +} + +impl PartialEq for ParentDeclaration { + fn eq(&self, other: &Self) -> bool { + self.node_id == other.node_id + && self.parent_id == other.parent_id + && self.sequence == other.sequence + && self.timestamp == other.timestamp + } +} + +impl Eq for ParentDeclaration {} + +/// A node's coordinates in the spanning tree. +/// +/// Coordinates are the path from the node to the root: +/// `[self, parent, grandparent, ..., root]` +/// +/// The coordinate enables greedy routing via tree distance calculation. +/// Two nodes can compute the hops between them by finding their lowest +/// common ancestor (LCA) in the tree. +#[derive(Clone, PartialEq, Eq)] +pub struct TreeCoordinate(Vec); + +impl TreeCoordinate { + /// Create a coordinate from a path (self to root). + /// + /// The path must be non-empty and ordered from the node to the root. + pub fn new(path: Vec) -> Result { + if path.is_empty() { + return Err(TreeError::EmptyCoordinate); + } + Ok(Self(path)) + } + + /// Create a coordinate for a root node. + pub fn root(node_id: NodeId) -> Self { + Self(vec![node_id]) + } + + /// The node this coordinate belongs to (first element). + pub fn node_id(&self) -> &NodeId { + &self.0[0] + } + + /// The root of the tree (last element). + pub fn root_id(&self) -> &NodeId { + self.0.last().expect("coordinate never empty") + } + + /// The immediate parent (second element, or self if root). + pub fn parent_id(&self) -> &NodeId { + self.0.get(1).unwrap_or(&self.0[0]) + } + + /// Depth in the tree (0 = root). + pub fn depth(&self) -> usize { + self.0.len() - 1 + } + + /// The full ancestry path. + pub fn path(&self) -> &[NodeId] { + &self.0 + } + + /// Check if this coordinate is a root (length 1). + pub fn is_root(&self) -> bool { + self.0.len() == 1 + } + + /// Calculate tree distance to another coordinate. + /// + /// Distance is hops through the lowest common ancestor (LCA). + /// If the coordinates have different roots, returns usize::MAX. + pub fn distance_to(&self, other: &TreeCoordinate) -> usize { + // Different trees have infinite distance + if self.root_id() != other.root_id() { + return usize::MAX; + } + + let lca_depth = self.lca_depth(other); + let self_to_lca = self.depth() - lca_depth; + let other_to_lca = other.depth() - lca_depth; + self_to_lca + other_to_lca + } + + /// Find the depth of the lowest common ancestor. + /// + /// Since coordinates are self-to-root, common ancestry is a suffix match. + /// Returns the depth (from root) of the LCA. + pub fn lca_depth(&self, other: &TreeCoordinate) -> usize { + let mut common: usize = 0; + let self_rev = self.0.iter().rev(); + let other_rev = other.0.iter().rev(); + + for (a, b) in self_rev.zip(other_rev) { + if a == b { + common += 1; + } else { + break; + } + } + + // LCA depth is counted from root (depth 0) + common.saturating_sub(1) + } + + /// Get the lowest common ancestor node ID. + pub fn lca(&self, other: &TreeCoordinate) -> Option<&NodeId> { + let self_rev: Vec<_> = self.0.iter().rev().collect(); + let other_rev: Vec<_> = other.0.iter().rev().collect(); + + let mut lca = None; + for (a, b) in self_rev.iter().zip(other_rev.iter()) { + if a == b { + lca = Some(*a); + } else { + break; + } + } + lca + } + + /// Check if `other` is an ancestor (appears in our path after self). + pub fn has_ancestor(&self, other: &NodeId) -> bool { + self.0.iter().skip(1).any(|id| id == other) + } + + /// Check if `other` is in our ancestry (including self). + pub fn contains(&self, other: &NodeId) -> bool { + self.0.iter().any(|id| id == other) + } + + /// Get the ancestor at a specific depth from self. + /// + /// `ancestor_at(0)` returns self, `ancestor_at(1)` returns parent, etc. + pub fn ancestor_at(&self, depth: usize) -> Option<&NodeId> { + self.0.get(depth) + } +} + +impl fmt::Debug for TreeCoordinate { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "TreeCoordinate(depth={}, path=[", self.depth())?; + for (i, id) in self.0.iter().enumerate() { + if i > 0 { + write!(f, " → ")?; + } + // Show first 4 bytes of each node ID + write!(f, "{:02x}{:02x}", id.as_bytes()[0], id.as_bytes()[1])?; + } + write!(f, "])") + } +} + +impl AsRef<[NodeId]> for TreeCoordinate { + fn as_ref(&self) -> &[NodeId] { + &self.0 + } +} + +/// Local spanning tree state for a node. +/// +/// Contains this node's declaration, coordinates, and view of peers' +/// tree positions. State is bounded by O(P × D) where P is peer count +/// and D is tree depth. +pub struct TreeState { + /// This node's NodeId. + my_node_id: NodeId, + /// This node's current parent declaration. + my_declaration: ParentDeclaration, + /// This node's current coordinates (computed from declaration chain). + my_coords: TreeCoordinate, + /// The current elected root (smallest reachable node_id). + root: NodeId, + /// Each peer's most recent parent declaration. + peer_declarations: HashMap, + /// Each peer's full ancestry to root. + peer_ancestry: HashMap, +} + +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. + pub fn new(my_node_id: NodeId) -> Self { + let my_declaration = ParentDeclaration::self_root(my_node_id, 0, 0); + let my_coords = TreeCoordinate::root(my_node_id); + + Self { + my_node_id, + my_declaration, + my_coords, + root: my_node_id, + peer_declarations: HashMap::new(), + peer_ancestry: HashMap::new(), + } + } + + /// Get this node's NodeId. + pub fn my_node_id(&self) -> &NodeId { + &self.my_node_id + } + + /// Get this node's current declaration. + pub fn my_declaration(&self) -> &ParentDeclaration { + &self.my_declaration + } + + /// Get this node's current coordinates. + pub fn my_coords(&self) -> &TreeCoordinate { + &self.my_coords + } + + /// Get the current root. + pub fn root(&self) -> &NodeId { + &self.root + } + + /// Check if this node is currently the root. + pub fn is_root(&self) -> bool { + self.root == self.my_node_id + } + + /// Get coordinates for a peer, if known. + pub fn peer_coords(&self, peer_id: &NodeId) -> Option<&TreeCoordinate> { + self.peer_ancestry.get(peer_id) + } + + /// Get declaration for a peer, if known. + pub fn peer_declaration(&self, peer_id: &NodeId) -> Option<&ParentDeclaration> { + self.peer_declarations.get(peer_id) + } + + /// Number of known peers. + pub fn peer_count(&self) -> usize { + self.peer_declarations.len() + } + + /// Iterate over all peer node IDs. + pub fn peer_ids(&self) -> impl Iterator { + self.peer_declarations.keys() + } + + /// Add or update a peer's tree state. + /// + /// Returns true if the state was updated (new or fresher declaration). + pub fn update_peer( + &mut self, + declaration: ParentDeclaration, + ancestry: TreeCoordinate, + ) -> bool { + let peer_id = *declaration.node_id(); + + // Check if this is a fresh update + if let Some(existing) = self.peer_declarations.get(&peer_id) + && !declaration.is_fresher_than(existing) + { + return false; + } + + self.peer_declarations.insert(peer_id, declaration); + self.peer_ancestry.insert(peer_id, ancestry); + true + } + + /// Remove a peer from the tree state. + pub fn remove_peer(&mut self, peer_id: &NodeId) { + self.peer_declarations.remove(peer_id); + self.peer_ancestry.remove(peer_id); + } + + /// Update this node's parent selection. + /// + /// Call this when switching parents. Updates the declaration and coordinates. + pub fn set_parent(&mut self, parent_id: NodeId, sequence: u64, timestamp: u64) { + self.my_declaration = ParentDeclaration::new(self.my_node_id, parent_id, sequence, timestamp); + // Coordinates will be recomputed when ancestry is available + } + + /// Update this node's coordinates based on current parent's ancestry. + pub fn recompute_coords(&mut self) { + if self.my_declaration.is_root() { + self.my_coords = TreeCoordinate::root(self.my_node_id); + self.root = self.my_node_id; + return; + } + + let parent_id = self.my_declaration.parent_id(); + if let Some(parent_coords) = self.peer_ancestry.get(parent_id) { + // Our coords = [self] ++ parent_coords + let mut path = vec![self.my_node_id]; + path.extend_from_slice(parent_coords.path()); + self.my_coords = TreeCoordinate::new(path).expect("non-empty path"); + self.root = *self.my_coords.root_id(); + } + } + + /// Calculate tree distance to a peer. + pub fn distance_to_peer(&self, peer_id: &NodeId) -> Option { + self.peer_ancestry + .get(peer_id) + .map(|coords| self.my_coords.distance_to(coords)) + } + + /// Find the best next hop toward a destination. + /// + /// Returns the peer that minimizes tree distance to the destination. + /// This is a stub - full implementation requires greedy routing logic. + pub fn find_next_hop(&self, _dest_coords: &TreeCoordinate) -> Option { + // Stub: would implement greedy tree routing + None + } + + /// Check if a parent switch to `candidate` would be beneficial. + /// + /// This is a stub - full implementation requires policy decisions. + pub fn should_switch_parent(&self, _candidate: &NodeId) -> bool { + // Stub: would evaluate parent switch criteria + false + } +} + +impl fmt::Debug for TreeState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TreeState") + .field("my_node_id", &self.my_node_id) + .field("root", &self.root) + .field("is_root", &self.is_root()) + .field("depth", &self.my_coords.depth()) + .field("peers", &self.peer_count()) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_node_id(val: u8) -> NodeId { + let mut bytes = [0u8; 32]; + bytes[0] = val; + NodeId::from_bytes(bytes) + } + + // ===== TreeCoordinate Tests ===== + + #[test] + fn test_tree_coordinate_root() { + let root_id = make_node_id(1); + let coord = TreeCoordinate::root(root_id); + + assert!(coord.is_root()); + assert_eq!(coord.depth(), 0); + assert_eq!(coord.node_id(), &root_id); + assert_eq!(coord.root_id(), &root_id); + assert_eq!(coord.parent_id(), &root_id); + } + + #[test] + fn test_tree_coordinate_path() { + let node = make_node_id(1); + let parent = make_node_id(2); + let root = make_node_id(3); + + let coord = TreeCoordinate::new(vec![node, parent, root]).unwrap(); + + assert!(!coord.is_root()); + assert_eq!(coord.depth(), 2); + assert_eq!(coord.node_id(), &node); + assert_eq!(coord.parent_id(), &parent); + assert_eq!(coord.root_id(), &root); + } + + #[test] + fn test_tree_coordinate_empty_fails() { + let result = TreeCoordinate::new(vec![]); + assert!(matches!(result, Err(TreeError::EmptyCoordinate))); + } + + #[test] + fn test_tree_distance_same_node() { + let node = make_node_id(1); + let coord = TreeCoordinate::root(node); + + assert_eq!(coord.distance_to(&coord), 0); + } + + #[test] + fn test_tree_distance_siblings() { + let root = make_node_id(0); + let a = make_node_id(1); + let b = make_node_id(2); + + let coord_a = TreeCoordinate::new(vec![a, root]).unwrap(); + let coord_b = TreeCoordinate::new(vec![b, root]).unwrap(); + + // a -> root -> b = 2 hops + assert_eq!(coord_a.distance_to(&coord_b), 2); + } + + #[test] + fn test_tree_distance_ancestor() { + let root = make_node_id(0); + let parent = make_node_id(1); + let child = make_node_id(2); + + let coord_parent = TreeCoordinate::new(vec![parent, root]).unwrap(); + let coord_child = TreeCoordinate::new(vec![child, parent, root]).unwrap(); + + // child -> parent = 1 hop + assert_eq!(coord_child.distance_to(&coord_parent), 1); + } + + #[test] + fn test_tree_distance_cousins() { + // Tree structure: + // root + // / \ + // a b + // / \ + // c d + let root = make_node_id(0); + let a = make_node_id(1); + let b = make_node_id(2); + let c = make_node_id(3); + let d = make_node_id(4); + + let coord_c = TreeCoordinate::new(vec![c, a, root]).unwrap(); + let coord_d = TreeCoordinate::new(vec![d, b, root]).unwrap(); + + // c -> a -> root -> b -> d = 4 hops + assert_eq!(coord_c.distance_to(&coord_d), 4); + } + + #[test] + fn test_tree_distance_different_roots() { + let root1 = make_node_id(1); + let root2 = make_node_id(2); + + let coord1 = TreeCoordinate::root(root1); + let coord2 = TreeCoordinate::root(root2); + + assert_eq!(coord1.distance_to(&coord2), usize::MAX); + } + + #[test] + fn test_has_ancestor() { + let root = make_node_id(0); + let parent = make_node_id(1); + let child = make_node_id(2); + + let coord = TreeCoordinate::new(vec![child, parent, root]).unwrap(); + + assert!(coord.has_ancestor(&parent)); + assert!(coord.has_ancestor(&root)); + assert!(!coord.has_ancestor(&child)); // self is not an ancestor + } + + #[test] + fn test_contains() { + let root = make_node_id(0); + let parent = make_node_id(1); + let child = make_node_id(2); + let other = make_node_id(99); + + let coord = TreeCoordinate::new(vec![child, parent, root]).unwrap(); + + assert!(coord.contains(&child)); + assert!(coord.contains(&parent)); + assert!(coord.contains(&root)); + assert!(!coord.contains(&other)); + } + + #[test] + fn test_ancestor_at() { + let root = make_node_id(0); + let parent = make_node_id(1); + let child = make_node_id(2); + + let coord = TreeCoordinate::new(vec![child, parent, root]).unwrap(); + + assert_eq!(coord.ancestor_at(0), Some(&child)); + assert_eq!(coord.ancestor_at(1), Some(&parent)); + assert_eq!(coord.ancestor_at(2), Some(&root)); + assert_eq!(coord.ancestor_at(3), None); + } + + #[test] + fn test_lca() { + let root = make_node_id(0); + let a = make_node_id(1); + let b = make_node_id(2); + let c = make_node_id(3); + let d = make_node_id(4); + + // c under a, d under b, both under root + let coord_c = TreeCoordinate::new(vec![c, a, root]).unwrap(); + let coord_d = TreeCoordinate::new(vec![d, b, root]).unwrap(); + + assert_eq!(coord_c.lca(&coord_d), Some(&root)); + + // c and a share ancestry through a and root + let coord_a = TreeCoordinate::new(vec![a, root]).unwrap(); + assert_eq!(coord_c.lca(&coord_a), Some(&a)); + } + + // ===== ParentDeclaration Tests ===== + + #[test] + fn test_parent_declaration_new() { + let node = make_node_id(1); + let parent = make_node_id(2); + + let decl = ParentDeclaration::new(node, parent, 1, 1000); + + assert_eq!(decl.node_id(), &node); + assert_eq!(decl.parent_id(), &parent); + assert_eq!(decl.sequence(), 1); + assert_eq!(decl.timestamp(), 1000); + assert!(!decl.is_root()); + assert!(!decl.is_signed()); + } + + #[test] + fn test_parent_declaration_self_root() { + let node = make_node_id(1); + + let decl = ParentDeclaration::self_root(node, 5, 2000); + + assert!(decl.is_root()); + assert_eq!(decl.node_id(), decl.parent_id()); + } + + #[test] + fn test_parent_declaration_freshness() { + let node = make_node_id(1); + let parent = make_node_id(2); + + let old_decl = ParentDeclaration::new(node, parent, 1, 1000); + let new_decl = ParentDeclaration::new(node, parent, 2, 2000); + + assert!(new_decl.is_fresher_than(&old_decl)); + assert!(!old_decl.is_fresher_than(&new_decl)); + assert!(!old_decl.is_fresher_than(&old_decl)); + } + + #[test] + fn test_parent_declaration_signing_bytes() { + let node = make_node_id(1); + let parent = make_node_id(2); + + let decl = ParentDeclaration::new(node, parent, 100, 1234567890); + let bytes = decl.signing_bytes(); + + // Should be 80 bytes: 32 + 32 + 8 + 8 + assert_eq!(bytes.len(), 80); + + // Verify structure + assert_eq!(&bytes[0..32], node.as_bytes()); + assert_eq!(&bytes[32..64], parent.as_bytes()); + assert_eq!(&bytes[64..72], &100u64.to_le_bytes()); + assert_eq!(&bytes[72..80], &1234567890u64.to_le_bytes()); + } + + #[test] + fn test_parent_declaration_equality() { + let node = make_node_id(1); + let parent = make_node_id(2); + + let decl1 = ParentDeclaration::new(node, parent, 1, 1000); + let decl2 = ParentDeclaration::new(node, parent, 1, 1000); + let decl3 = ParentDeclaration::new(node, parent, 2, 1000); + + assert_eq!(decl1, decl2); + assert_ne!(decl1, decl3); + } + + // ===== TreeState Tests ===== + + #[test] + fn test_tree_state_new() { + let node = make_node_id(1); + let state = TreeState::new(node); + + assert_eq!(state.my_node_id(), &node); + assert!(state.is_root()); + assert_eq!(state.root(), &node); + assert_eq!(state.my_coords().depth(), 0); + assert_eq!(state.peer_count(), 0); + } + + #[test] + fn test_tree_state_update_peer() { + let my_node = make_node_id(0); + let mut state = TreeState::new(my_node); + + let peer = make_node_id(1); + let root = make_node_id(2); + + let decl = ParentDeclaration::new(peer, root, 1, 1000); + let coords = TreeCoordinate::new(vec![peer, root]).unwrap(); + + assert!(state.update_peer(decl.clone(), coords.clone())); + assert_eq!(state.peer_count(), 1); + assert!(state.peer_coords(&peer).is_some()); + assert!(state.peer_declaration(&peer).is_some()); + + // Same sequence should not update + let decl2 = ParentDeclaration::new(peer, root, 1, 1000); + assert!(!state.update_peer(decl2, coords.clone())); + + // Higher sequence should update + let decl3 = ParentDeclaration::new(peer, root, 2, 2000); + assert!(state.update_peer(decl3, coords)); + } + + #[test] + fn test_tree_state_remove_peer() { + let my_node = make_node_id(0); + let mut state = TreeState::new(my_node); + + let peer = make_node_id(1); + let root = make_node_id(2); + + let decl = ParentDeclaration::new(peer, root, 1, 1000); + let coords = TreeCoordinate::new(vec![peer, root]).unwrap(); + + state.update_peer(decl, coords); + assert_eq!(state.peer_count(), 1); + + state.remove_peer(&peer); + assert_eq!(state.peer_count(), 0); + assert!(state.peer_coords(&peer).is_none()); + } + + #[test] + fn test_tree_state_distance_to_peer() { + let my_node = make_node_id(0); + let mut state = TreeState::new(my_node); + + let peer = make_node_id(1); + + // Both are roots in their own trees initially - different roots + let peer_coords = TreeCoordinate::root(peer); + let decl = ParentDeclaration::self_root(peer, 1, 1000); + state.update_peer(decl, peer_coords); + + // Different roots = MAX distance + assert_eq!(state.distance_to_peer(&peer), Some(usize::MAX)); + + // If they share a root, distance should be finite + let shared_root = make_node_id(99); + + // Update my state to have shared root + state.set_parent(shared_root, 1, 1000); + let my_new_coords = TreeCoordinate::new(vec![my_node, shared_root]).unwrap(); + // Manually set coords for test (normally done by recompute_coords) + state.my_coords = my_new_coords; + state.root = shared_root; + + // Update peer to have same root + let peer_coords = TreeCoordinate::new(vec![peer, shared_root]).unwrap(); + let decl = ParentDeclaration::new(peer, shared_root, 2, 2000); + state.update_peer(decl, peer_coords); + + // Now distance should be 2 (me -> root -> peer) + assert_eq!(state.distance_to_peer(&peer), Some(2)); + } + + #[test] + fn test_tree_state_peer_ids() { + let my_node = make_node_id(0); + let mut state = TreeState::new(my_node); + + let peer1 = make_node_id(1); + let peer2 = make_node_id(2); + + state.update_peer( + ParentDeclaration::self_root(peer1, 1, 1000), + TreeCoordinate::root(peer1), + ); + state.update_peer( + ParentDeclaration::self_root(peer2, 1, 1000), + TreeCoordinate::root(peer2), + ); + + let ids: Vec<_> = state.peer_ids().collect(); + assert_eq!(ids.len(), 2); + assert!(ids.contains(&&peer1)); + assert!(ids.contains(&&peer2)); + } +}