diff --git a/src/config/node.rs b/src/config/node.rs index 2b52159..289d77e 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -119,6 +119,9 @@ pub struct CacheConfig { /// Max entries in route cache (`node.cache.route_size`). #[serde(default = "CacheConfig::default_route_size")] pub route_size: usize, + /// Identity cache entry TTL in seconds (`node.cache.identity_ttl_secs`). + #[serde(default = "CacheConfig::default_identity_ttl_secs")] + pub identity_ttl_secs: u64, } impl Default for CacheConfig { @@ -127,6 +130,7 @@ impl Default for CacheConfig { coord_size: 50_000, coord_ttl_secs: 300, route_size: 10_000, + identity_ttl_secs: 60, } } } @@ -135,6 +139,7 @@ impl CacheConfig { fn default_coord_size() -> usize { 50_000 } fn default_coord_ttl_secs() -> u64 { 300 } fn default_route_size() -> usize { 10_000 } + fn default_identity_ttl_secs() -> u64 { 60 } } /// Discovery protocol (`node.discovery.*`). @@ -227,6 +232,10 @@ pub struct SessionConfig { /// Max destinations with pending packets (`node.session.pending_max_destinations`). #[serde(default = "SessionConfig::default_pending_max_destinations")] pub pending_max_destinations: usize, + /// Idle session timeout in seconds (`node.session.idle_timeout_secs`). + /// Established sessions with no activity for this duration are removed. + #[serde(default = "SessionConfig::default_idle_timeout_secs")] + pub idle_timeout_secs: u64, } impl Default for SessionConfig { @@ -235,6 +244,7 @@ impl Default for SessionConfig { default_hop_limit: 64, pending_packets_per_dest: 16, pending_max_destinations: 256, + idle_timeout_secs: 90, } } } @@ -243,6 +253,7 @@ impl SessionConfig { fn default_hop_limit() -> u8 { 64 } fn default_pending_packets_per_dest() -> usize { 16 } fn default_pending_max_destinations() -> usize { 256 } + fn default_idle_timeout_secs() -> u64 { 90 } } /// Internal buffers (`node.buffers.*`). diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 5453bbb..51084e8 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -79,6 +79,7 @@ impl Node { .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); + self.purge_idle_sessions(now_ms); self.process_pending_retries(now_ms).await; self.check_tree_state().await; self.check_bloom_state().await; diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 500a475..a993bf2 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -535,7 +535,7 @@ impl Node { // Look up in identity cache let (dest_addr, dest_pubkey) = match self.lookup_by_fips_prefix(&prefix) { - Some(&(addr, pk)) => (addr, pk), + Some((addr, pk)) => (addr, pk), None => { self.send_icmpv6_dest_unreachable(&ipv6_packet); return; @@ -630,7 +630,7 @@ impl Node { let mut prefix = [0u8; 15]; prefix.copy_from_slice(&dest_addr.as_bytes()[0..15]); let dest_pubkey = match self.lookup_by_fips_prefix(&prefix) { - Some(&(_, pk)) => pk, + Some((_, pk)) => pk, None => { debug!(dest = %dest_addr, "Discovery complete but no identity for session retry"); return; diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index 550db9b..e6cbdd4 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -1,8 +1,8 @@ -//! Timeout management for stale handshake connections. +//! Timeout management for stale handshake connections and idle sessions. use crate::node::Node; use crate::transport::LinkId; -use tracing::info; +use tracing::{debug, info}; impl Node { /// Check for timed-out handshake connections and clean them up. @@ -78,4 +78,33 @@ impl Node { // Remove link and addr_to_link self.remove_link(&link_id); } + + /// Remove established sessions that have been idle too long. + /// + /// Only targets sessions in the Established state. Initiating/Responding + /// sessions are handled by the handshake timeout. + pub(in crate::node) fn purge_idle_sessions(&mut self, now_ms: u64) { + let timeout_ms = self.config.node.session.idle_timeout_secs * 1000; + if timeout_ms == 0 { + return; // disabled + } + + let idle: Vec<_> = self.sessions.iter() + .filter(|(_, entry)| { + entry.is_established() + && now_ms.saturating_sub(entry.last_activity()) > timeout_ms + }) + .map(|(addr, _)| *addr) + .collect(); + + for addr in idle { + self.sessions.remove(&addr); + self.pending_tun_packets.remove(&addr); + debug!( + dest = %addr, + idle_secs = timeout_ms / 1000, + "Idle session removed" + ); + } + } } diff --git a/src/node/mod.rs b/src/node/mod.rs index 919804b..1df41d1 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -254,7 +254,7 @@ pub struct Node { // === Identity Cache === /// Maps FipsAddress prefix bytes (bytes 1-15) to (NodeAddr, PublicKey). /// Enables reverse lookup from IPv6 destination to session/routing identity. - identity_cache: HashMap<[u8; 15], (NodeAddr, secp256k1::PublicKey)>, + identity_cache: HashMap<[u8; 15], (NodeAddr, secp256k1::PublicKey, u64)>, // === Pending TUN Packets === /// Packets queued while waiting for session establishment. @@ -862,12 +862,23 @@ impl Node { pub(crate) fn register_identity(&mut self, node_addr: NodeAddr, pubkey: secp256k1::PublicKey) { let mut prefix = [0u8; 15]; prefix.copy_from_slice(&node_addr.as_bytes()[0..15]); - self.identity_cache.insert(prefix, (node_addr, pubkey)); + self.identity_cache.insert(prefix, (node_addr, pubkey, Self::now_ms())); } /// Look up a destination by FipsAddress prefix (bytes 1-15 of the IPv6 address). - pub(crate) fn lookup_by_fips_prefix(&self, prefix: &[u8; 15]) -> Option<&(NodeAddr, secp256k1::PublicKey)> { - self.identity_cache.get(prefix) + /// Returns None if the entry has expired (lazy expiry). + pub(crate) fn lookup_by_fips_prefix(&mut self, prefix: &[u8; 15]) -> Option<(NodeAddr, secp256k1::PublicKey)> { + let ttl_ms = self.config.node.cache.identity_ttl_secs * 1000; + let now_ms = Self::now_ms(); + if let Some(&(addr, pk, registered_at)) = self.identity_cache.get(prefix) { + if now_ms.saturating_sub(registered_at) > ttl_ms { + self.identity_cache.remove(prefix); + return None; + } + Some((addr, pk)) + } else { + None + } } /// Number of identity cache entries. diff --git a/src/node/session.rs b/src/node/session.rs index 18ad8db..31c9b0c 100644 --- a/src/node/session.rs +++ b/src/node/session.rs @@ -113,7 +113,6 @@ impl SessionEntry { } /// Get last activity time. - #[cfg(test)] pub(crate) fn last_activity(&self) -> u64 { self.last_activity } diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index f6b6705..d9a533d 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -925,8 +925,8 @@ fn test_identity_cache_populated_on_promote() { let cached = node.lookup_by_fips_prefix(&prefix); assert!(cached.is_some(), "Identity cache should contain promoted peer"); let (cached_addr, cached_pk) = cached.unwrap(); - assert_eq!(*cached_addr, peer_addr); - assert_eq!(*cached_pk, peer_identity.pubkey_full()); + assert_eq!(cached_addr, peer_addr); + assert_eq!(cached_pk, peer_identity.pubkey_full()); } #[tokio::test] @@ -1145,3 +1145,212 @@ async fn test_tun_outbound_pending_queue_flush() { cleanup_nodes(&mut nodes).await; } + +// ============================================================================ +// Unit tests: Session idle timeout +// ============================================================================ + +/// Helper: complete a Noise IK handshake and return the initiator's NoiseSession. +fn make_noise_session( + our_identity: &Identity, + remote_identity: &Identity, +) -> crate::noise::NoiseSession { + use crate::noise::HandshakeState; + + let mut initiator = HandshakeState::new_initiator( + our_identity.keypair(), + remote_identity.pubkey_full(), + ); + let mut responder = HandshakeState::new_responder(remote_identity.keypair()); + + let msg1 = initiator.write_message_1().unwrap(); + responder.read_message_1(&msg1).unwrap(); + let msg2 = responder.write_message_2().unwrap(); + initiator.read_message_2(&msg2).unwrap(); + + initiator.into_session().unwrap() +} + +#[test] +fn test_purge_idle_sessions_removes_expired() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let session = make_noise_session(node.identity(), &remote); + let entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Established(session), + 1000, // created at t=1000ms + ); + + node.sessions.insert(remote_addr, entry); + assert_eq!(node.session_count(), 1); + assert!(node.get_session(&remote_addr).unwrap().is_established()); + + // Purge at t=92s — should exceed default 90s idle timeout + let now_ms = 1000 + 92_000; + node.purge_idle_sessions(now_ms); + + assert_eq!(node.session_count(), 0, "Idle session should be purged"); +} + +#[test] +fn test_purge_idle_sessions_keeps_active() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let session = make_noise_session(node.identity(), &remote); + let mut entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Established(session), + 1000, + ); + + // Touch at t=80s — recent activity + entry.touch(81_000); + + node.sessions.insert(remote_addr, entry); + + // Purge at t=92s — only 11s since last activity, well within 90s timeout + let now_ms = 92_000; + node.purge_idle_sessions(now_ms); + + assert_eq!(node.session_count(), 1, "Active session should survive purge"); +} + +#[test] +fn test_purge_idle_sessions_ignores_initiating() { + use crate::noise::HandshakeState; + + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let handshake = HandshakeState::new_initiator( + node.identity().keypair(), + remote.pubkey_full(), + ); + let entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Initiating(handshake), + 1000, + ); + + node.sessions.insert(remote_addr, entry); + + // Purge well past the idle timeout — Initiating sessions should not be touched + let now_ms = 1000 + 200_000; + node.purge_idle_sessions(now_ms); + + assert_eq!(node.session_count(), 1, "Initiating session should not be purged by idle timeout"); +} + +#[test] +fn test_purge_idle_sessions_cleans_pending_packets() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let session = make_noise_session(node.identity(), &remote); + let entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Established(session), + 1000, + ); + + node.sessions.insert(remote_addr, entry); + + // Insert some pending packets for this destination + let mut queue = std::collections::VecDeque::new(); + queue.push_back(vec![1, 2, 3]); + node.pending_tun_packets.insert(remote_addr, queue); + assert!(node.pending_tun_packets.contains_key(&remote_addr)); + + // Purge after idle timeout + let now_ms = 1000 + 92_000; + node.purge_idle_sessions(now_ms); + + assert_eq!(node.session_count(), 0); + assert!(!node.pending_tun_packets.contains_key(&remote_addr), + "Pending packets should be cleaned up with idle session"); +} + +#[test] +fn test_purge_idle_sessions_disabled_when_zero() { + let mut node = make_node(); + node.config.node.session.idle_timeout_secs = 0; + + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let session = make_noise_session(node.identity(), &remote); + let entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Established(session), + 1000, + ); + + node.sessions.insert(remote_addr, entry); + + // Even way past any timeout, sessions should survive when disabled + let now_ms = 1000 + 1_000_000; + node.purge_idle_sessions(now_ms); + + assert_eq!(node.session_count(), 1, "Sessions should not be purged when idle timeout is disabled"); +} + +// ============================================================================ +// Unit tests: Identity cache expiry +// ============================================================================ + +#[test] +fn test_identity_cache_expiry() { + let mut node = make_node(); + // Use a short TTL (1s) and insert with an old timestamp + node.config.node.cache.identity_ttl_secs = 1; + + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&remote_addr.as_bytes()[0..15]); + + // Insert directly with a timestamp far in the past (time 0) + node.identity_cache.insert(prefix, (remote_addr, remote.pubkey_full(), 0)); + + // Lookup should find the entry expired (registered_at=0, TTL=1s, now >> 1s) + let result = node.lookup_by_fips_prefix(&prefix); + assert!(result.is_none(), "Expired identity should return None"); + + // Entry should have been removed from cache + assert!(!node.identity_cache.contains_key(&prefix), + "Expired entry should be removed from cache"); +} + +#[test] +fn test_identity_cache_survives_before_ttl() { + let mut node = make_node(); + // Default 60s TTL — just registered, should be available + + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + + node.register_identity(remote_addr, remote.pubkey_full()); + + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&remote_addr.as_bytes()[0..15]); + + let result = node.lookup_by_fips_prefix(&prefix); + assert!(result.is_some(), "Fresh identity should be available"); + + let (addr, pk) = result.unwrap(); + assert_eq!(addr, remote_addr); + assert_eq!(pk, remote.pubkey_full()); +}