diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 81cf0ee..7b0a9bf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -145,6 +145,10 @@ jobs: - suite: static-chain type: static topology: chain + # ── Rekey integration test ────────────────────────────────────────── + - suite: rekey + type: rekey + topology: rekey # ── Chaos / stochastic scenarios ─────────────────────────────────── - suite: chaos-smoke-10 type: chaos @@ -233,6 +237,48 @@ jobs: docker compose -f testing/static/docker-compose.yml \ --profile ${{ matrix.topology }} down --volumes --remove-orphans + # ── Rekey integration test ────────────────────────────────────────────── + - name: Install binary (rekey) + if: matrix.type == 'rekey' + run: | + chmod +x _bin/fips _bin/fipsctl + cp _bin/fips testing/static/fips + cp _bin/fipsctl testing/static/fipsctl + + - name: Generate and inject configs (rekey) + if: matrix.type == 'rekey' + run: | + bash testing/static/scripts/generate-configs.sh rekey + bash testing/static/scripts/rekey-test.sh inject-config + + - name: Build Docker images (rekey) + if: matrix.type == 'rekey' + run: | + docker compose -f testing/static/docker-compose.yml \ + --profile rekey build + + - name: Start containers (rekey) + if: matrix.type == 'rekey' + run: | + docker compose -f testing/static/docker-compose.yml \ + --profile rekey up -d + + - name: Run rekey test + if: matrix.type == 'rekey' + run: bash testing/static/scripts/rekey-test.sh + + - name: Collect logs on failure (rekey) + if: matrix.type == 'rekey' && failure() + run: | + docker compose -f testing/static/docker-compose.yml \ + --profile rekey logs --no-color + + - name: Stop containers (rekey) + if: matrix.type == 'rekey' && always() + run: | + docker compose -f testing/static/docker-compose.yml \ + --profile rekey down --volumes --remove-orphans + # ── Chaos simulation ─────────────────────────────────────────────────── - name: Install Python deps (chaos) if: matrix.type == 'chaos' diff --git a/src/config/mod.rs b/src/config/mod.rs index 9abd200..d9d5feb 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -30,7 +30,8 @@ use thiserror::Error; pub use node::{ BloomConfig, BuffersConfig, CacheConfig, ControlConfig, DiscoveryConfig, LimitsConfig, - NodeConfig, RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig, + NodeConfig, RateLimitConfig, RekeyConfig, RetryConfig, SessionConfig, SessionMmpConfig, + TreeConfig, }; pub use peer::{ConnectPolicy, PeerAddress, PeerConfig}; pub use transport::{EthernetConfig, TcpConfig, TransportInstances, TransportsConfig, UdpConfig}; diff --git a/src/config/node.rs b/src/config/node.rs index b101298..02820fd 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -418,6 +418,42 @@ impl BuffersConfig { // ECN Congestion Signaling // ============================================================================ +/// Rekey / session rekeying configuration (`node.rekey.*`). +/// +/// Controls periodic full rekey for both FMP (link layer) and FSP +/// (session layer) Noise sessions. Rekeying provides true forward secrecy +/// with fresh DH randomness, nonce reset, and session index rotation. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RekeyConfig { + /// Enable periodic rekey (`node.rekey.enabled`). + #[serde(default = "RekeyConfig::default_enabled")] + pub enabled: bool, + + /// Initiate rekey after this many seconds (`node.rekey.after_secs`). + #[serde(default = "RekeyConfig::default_after_secs")] + pub after_secs: u64, + + /// Initiate rekey after this many messages sent (`node.rekey.after_messages`). + #[serde(default = "RekeyConfig::default_after_messages")] + pub after_messages: u64, +} + +impl Default for RekeyConfig { + fn default() -> Self { + Self { + enabled: true, + after_secs: 120, + after_messages: 1 << 16, // 65536 + } + } +} + +impl RekeyConfig { + fn default_enabled() -> bool { true } + fn default_after_secs() -> u64 { 120 } + fn default_after_messages() -> u64 { 1 << 16 } +} + /// ECN congestion signaling configuration (`node.ecn.*`). /// /// Controls the FMP CE relay chain: transit nodes detect congestion on outgoing @@ -541,6 +577,10 @@ pub struct NodeConfig { /// ECN congestion signaling (`node.ecn.*`). #[serde(default)] pub ecn: EcnConfig, + + /// Rekey / session rekeying (`node.rekey.*`). + #[serde(default)] + pub rekey: RekeyConfig, } impl Default for NodeConfig { @@ -565,6 +605,7 @@ impl Default for NodeConfig { mmp: MmpConfig::default(), session_mmp: SessionMmpConfig::default(), ecn: EcnConfig::default(), + rekey: RekeyConfig::default(), } } } diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 9209710..284c774 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -117,11 +117,27 @@ impl Node { } let link_id = peer.link_id(); + let transport_id = peer.transport_id(); - // Free session index - if let (Some(tid), Some(idx)) = (peer.transport_id(), peer.our_index()) { - self.peers_by_index.remove(&(tid, idx.as_u32())); - let _ = self.index_allocator.free(idx); + // Free session indices (current, rekey, pending, previous) + if let Some(tid) = transport_id { + if let Some(idx) = peer.our_index() { + self.peers_by_index.remove(&(tid, idx.as_u32())); + let _ = self.index_allocator.free(idx); + } + if let Some(idx) = peer.rekey_our_index() { + self.pending_outbound.remove(&(tid, idx.as_u32())); + self.peers_by_index.remove(&(tid, idx.as_u32())); + let _ = self.index_allocator.free(idx); + } + if let Some(idx) = peer.pending_our_index() { + self.peers_by_index.remove(&(tid, idx.as_u32())); + let _ = self.index_allocator.free(idx); + } + if let Some(idx) = peer.previous_our_index() { + self.peers_by_index.remove(&(tid, idx.as_u32())); + let _ = self.index_allocator.free(idx); + } } // Remove link and address mapping diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index b0c537c..7ba4617 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -2,16 +2,24 @@ use crate::noise::NoiseError; use crate::node::Node; -use crate::node::wire::{EncryptedHeader, strip_inner_header, FLAG_CE, FLAG_SP}; +use crate::node::wire::{EncryptedHeader, strip_inner_header, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP}; use crate::transport::ReceivedPacket; use std::time::Instant; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; + +/// Force-remove a peer after this many consecutive decryption failures. +const DECRYPT_FAILURE_THRESHOLD: u32 = 20; impl Node { /// Handle an encrypted frame (phase 0x0). /// /// This is the hot path for established sessions. We use O(1) /// index-based lookup to find the session, then decrypt. + /// + /// K-bit handling: when the peer flips the K-bit after a rekey, + /// we promote the pending new session to current and demote the old + /// session to previous for a drain window. During drain, we try the + /// current session first, then fall back to the previous session. pub(in crate::node) async fn handle_encrypted_frame(&mut self, packet: ReceivedPacket) { // Parse header (fail fast) let header = match EncryptedHeader::parse(&packet.data) { @@ -24,7 +32,6 @@ impl Node { let node_addr = match self.peers_by_index.get(&key) { Some(id) => *id, None => { - // Unknown index - could be stale session or attack debug!( receiver_idx = %header.receiver_idx, transport_id = %packet.transport_id, @@ -34,72 +41,89 @@ impl Node { } }; - let peer = match self.peers.get_mut(&node_addr) { - Some(p) => p, - None => { - // Peer removed but index not cleaned up - fix it - self.peers_by_index.remove(&key); - return; - } - }; + if !self.peers.contains_key(&node_addr) { + self.peers_by_index.remove(&key); + return; + } - // Get the session (peer must have one for index-based lookup) - let session = match peer.noise_session_mut() { - Some(s) => s, - None => { - warn!( - peer = %self.peer_display_name(&node_addr), - "Peer in index map has no session" + // Extract K-bit from flags + let received_k_bit = header.flags & FLAG_KEY_EPOCH != 0; + + // K-bit flip detection: peer has cut over to the new session. + // Check and perform cutover in a scoped borrow. + { + let peer = self.peers.get(&node_addr).unwrap(); + let k_bit_flipped = received_k_bit != peer.current_k_bit() + && peer.pending_new_session().is_some(); + + if k_bit_flipped { + let display_name = self.peer_display_name(&node_addr); + info!( + peer = %display_name, + "Peer K-bit flip detected, promoting new session" ); - return; - } - }; - // Decrypt with replay check and AAD (this is the expensive part) - let ciphertext = &packet.data[header.ciphertext_offset()..]; - let plaintext = match session.decrypt_with_replay_check_and_aad( - ciphertext, - header.counter, - &header.header_bytes, - ) { - Ok(p) => p, - Err(e) => { - if matches!(e, NoiseError::ReplayDetected(_)) { - // Suppress repeated replay detections during link transitions. - // Re-borrow peer mutably for suppression counter update. - if let Some(peer) = self.peers.get_mut(&node_addr) { - let count = peer.increment_replay_suppressed(); - if count <= 3 { - debug!( - peer = %self.peer_display_name(&node_addr), - counter = header.counter, - error = %e, - "Decryption failed" - ); - } else if count == 4 { - debug!( - peer = %self.peer_display_name(&node_addr), - "Suppressing further replay detection messages" - ); - } - // count > 4: silently suppress - } else { - debug!( - peer = %self.peer_display_name(&node_addr), - counter = header.counter, - error = %e, - "Decryption failed" - ); - } - } else { - debug!( - peer = %self.peer_display_name(&node_addr), - counter = header.counter, - error = %e, - "Decryption failed" + let peer = self.peers.get_mut(&node_addr).unwrap(); + if let Some(_old_our_index) = peer.handle_peer_kbit_flip() + && let (Some(transport_id), Some(new_our_index)) = + (peer.transport_id(), peer.our_index()) + { + self.peers_by_index.insert( + (transport_id, new_our_index.as_u32()), + node_addr, ); } - return; + } + } + + // Decrypt: try current session first, then previous (drain fallback) + let ciphertext = &packet.data[header.ciphertext_offset()..]; + let plaintext = { + let peer = self.peers.get_mut(&node_addr).unwrap(); + let session = match peer.noise_session_mut() { + Some(s) => s, + None => { + warn!( + peer = %self.peer_display_name(&node_addr), + "Peer in index map has no session" + ); + return; + } + }; + + match session.decrypt_with_replay_check_and_aad( + ciphertext, + header.counter, + &header.header_bytes, + ) { + Ok(p) => { + peer.reset_decrypt_failures(); + p + } + Err(e) => { + // Current session failed — try previous session (drain window) + if let Some(prev_session) = peer.previous_session_mut() { + match prev_session.decrypt_with_replay_check_and_aad( + ciphertext, + header.counter, + &header.header_bytes, + ) { + Ok(p) => { + peer.reset_decrypt_failures(); + p + } + Err(_) => { + self.log_decrypt_failure(&node_addr, &header, &e); + self.handle_decrypt_failure(&node_addr); + return; + } + } + } else { + self.log_decrypt_failure(&node_addr, &header, &e); + self.handle_decrypt_failure(&node_addr); + return; + } + } } }; @@ -118,34 +142,84 @@ impl Node { } }; - // MMP per-frame processing: feed counter, timestamp, flags to receiver state + // MMP per-frame processing and statistics let now = Instant::now(); let ce_flag = header.flags & FLAG_CE != 0; let sp_flag = header.flags & FLAG_SP != 0; - if let Some(mmp) = peer.mmp_mut() { - mmp.receiver.record_recv( - header.counter, - timestamp, - packet.data.len(), - ce_flag, - now, - ); - // Spin bit: advance state machine for correct TX reflection. - // RTT samples from spin bit are not used for SRTT because - // inter-frame timing in the mesh is irregular, inflating - // spin-bit RTT by variable processing delays on both sides. - // Timestamp-echo in ReceiverReport provides accurate RTT. - let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now); + + if let Some(peer) = self.peers.get_mut(&node_addr) { + if let Some(mmp) = peer.mmp_mut() { + mmp.receiver.record_recv( + header.counter, + timestamp, + packet.data.len(), + ce_flag, + now, + ); + let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now); + } + peer.set_current_addr(packet.transport_id, packet.remote_addr.clone()); + peer.link_stats_mut().record_recv(packet.data.len(), packet.timestamp_ms); + peer.touch(packet.timestamp_ms); } - // Update address for roaming support - peer.set_current_addr(packet.transport_id, packet.remote_addr.clone()); - - // Update statistics - peer.link_stats_mut().record_recv(packet.data.len(), packet.timestamp_ms); - peer.touch(packet.timestamp_ms); - - // Dispatch to link message handler (msg_type + payload, inner header stripped) + // Dispatch to link message handler self.dispatch_link_message(&node_addr, link_message, ce_flag).await; } + + /// Log a decryption failure with replay suppression. + fn log_decrypt_failure( + &mut self, + node_addr: &crate::NodeAddr, + header: &EncryptedHeader, + error: &NoiseError, + ) { + if matches!(error, NoiseError::ReplayDetected(_)) { + if let Some(peer) = self.peers.get_mut(node_addr) { + let count = peer.increment_replay_suppressed(); + if count <= 3 { + debug!( + peer = %self.peer_display_name(node_addr), + counter = header.counter, + error = %error, + "Decryption failed" + ); + } else if count == 4 { + debug!( + peer = %self.peer_display_name(node_addr), + "Suppressing further replay detection messages" + ); + } + } else { + debug!( + peer = %self.peer_display_name(node_addr), + counter = header.counter, + error = %error, + "Decryption failed" + ); + } + } else { + debug!( + peer = %self.peer_display_name(node_addr), + counter = header.counter, + error = %error, + "Decryption failed" + ); + } + } + + /// Increment decrypt failure counter and force-remove peer if threshold exceeded. + fn handle_decrypt_failure(&mut self, node_addr: &crate::NodeAddr) { + if let Some(peer) = self.peers.get_mut(node_addr) { + let count = peer.increment_decrypt_failures(); + if count >= DECRYPT_FAILURE_THRESHOLD { + warn!( + peer = %self.peer_display_name(node_addr), + consecutive_failures = count, + "Excessive decryption failures, removing peer" + ); + self.remove_active_peer(node_addr); + } + } + } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index d0b65b4..d6525e5 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -96,13 +96,22 @@ impl Node { return; } } else { - // Outbound link to this address — cross-connection, allow msg1 - debug!( - transport_id = %packet.transport_id, - remote_addr = %packet.remote_addr, - existing_link_id = %existing_link_id, - "Cross-connection detected: have outbound, received inbound msg1" - ); + // Outbound link to this address. If it belongs to an active + // peer, this may be a rekey msg1 (same epoch) or a + // restart (different epoch). Set possible_restart to enable + // the epoch/rekey check below. + let is_active_peer = self.peers.values() + .any(|p| p.link_id() == existing_link_id); + if is_active_peer { + possible_restart = true; + } else { + debug!( + transport_id = %packet.transport_id, + remote_addr = %packet.remote_addr, + existing_link_id = %existing_link_id, + "Cross-connection detected: have outbound, received inbound msg1" + ); + } } } @@ -141,6 +150,13 @@ impl Node { let peer_node_addr = *peer_identity.node_addr(); + // Identity-based restart/rekey detection: if the peer is already + // active but addr_to_link didn't match (different source address, e.g., + // TCP from a different port), we still need to check for restart/rekey. + if !possible_restart && self.peers.contains_key(&peer_node_addr) { + possible_restart = true; + } + // Epoch-based restart detection and duplicate msg1 handling. // // If we fell through from the addr_to_link check above with @@ -163,8 +179,96 @@ impl Node { // Fall through to process as new connection } _ => { - // Same epoch (or no epoch stored) — duplicate msg1 from - // same session. Resend stored msg2. + // Same epoch (or no epoch stored). + // If the peer has an active session and rekey is enabled, + // this is a rekey msg1 (not a duplicate initial msg1). + // Guard: the session must be at least 30s old to avoid + // misidentifying a cross-connection msg1 as a rekey. + // During simultaneous connection, both sides promote + // within the same tick and the peer's msg1 arrives + // immediately — a genuine rekey can't fire that fast. + let session_age_secs = existing_peer + .session_established_at() + .elapsed() + .as_secs(); + if self.config.node.rekey.enabled + && existing_peer.has_session() + && existing_peer.is_healthy() + && session_age_secs >= 30 + { + // Rekey: process as responder, store new session as pending + let noise_session = conn.take_session(); + let our_new_index = match self.index_allocator.allocate() { + Ok(idx) => idx, + Err(e) => { + warn!(error = %e, "Failed to allocate index for rekey"); + self.msg1_rate_limiter.complete_handshake(); + return; + } + }; + + let noise_session = match noise_session { + Some(s) => s, + None => { + warn!("Rekey msg1: no session from handshake"); + let _ = self.index_allocator.free(our_new_index); + self.msg1_rate_limiter.complete_handshake(); + return; + } + }; + + // Send msg2 response using the new handshake + let wire_msg2 = build_msg2(our_new_index, header.sender_idx, &msg2_response); + if let Some(transport) = self.transports.get(&packet.transport_id) { + match transport.send(&packet.remote_addr, &wire_msg2).await { + Ok(_) => { + debug!( + peer = %self.peer_display_name(&peer_node_addr), + new_our_index = %our_new_index, + "Sent rekey msg2 response" + ); + } + Err(e) => { + warn!( + peer = %self.peer_display_name(&peer_node_addr), + error = %e, + "Failed to send rekey msg2" + ); + let _ = self.index_allocator.free(our_new_index); + self.msg1_rate_limiter.complete_handshake(); + return; + } + } + } + + // Store pending session on the existing peer + if let Some(peer) = self.peers.get_mut(&peer_node_addr) { + peer.set_pending_session( + noise_session, + our_new_index, + header.sender_idx, + ); + peer.record_peer_rekey(); + } + + // Register new index in peers_by_index + self.peers_by_index.insert( + (packet.transport_id, our_new_index.as_u32()), + peer_node_addr, + ); + + // Clean up: remove the temporary connection/link we created. + // Do NOT remove addr_to_link — the entry must remain pointing + // to the original link so future msg1s from this address are + // recognized as rekeys (not new connections). + self.connections.remove(&link_id); + self.links.remove(&link_id); + + self.msg1_rate_limiter.complete_handshake(); + return; + } + + // Not a rekey — duplicate msg1. Resend stored msg2. if let Some(msg2) = existing_peer.handshake_msg2().map(|m| m.to_vec()) && let Some(transport) = self.transports.get(&packet.transport_id) { @@ -384,14 +488,71 @@ impl Node { } }; - let conn = match self.connections.get_mut(&link_id) { - Some(c) => c, - None => { - // Connection removed, clean up pending_outbound + // Check if this is a rekey msg2: the handshake state is on the + // ActivePeer (not a PeerConnection), so self.connections won't have it. + // Look for a peer with matching rekey_our_index. + if !self.connections.contains_key(&link_id) { + let noise_msg2 = &packet.data[header.noise_msg2_offset..]; + + // Find peer with rekey in progress for this index + let peer_addr = self.peers.iter().find_map(|(addr, peer)| { + if peer.rekey_in_progress() + && peer.rekey_our_index() == Some(header.receiver_idx) + { + Some(*addr) + } else { + None + } + }); + + if let Some(peer_node_addr) = peer_addr { + let display_name = self.peer_display_name(&peer_node_addr); + + // Complete the rekey handshake on the ActivePeer + if let Some(peer) = self.peers.get_mut(&peer_node_addr) { + match peer.complete_rekey_msg2(noise_msg2) { + Ok(session) => { + let our_index = peer.rekey_our_index() + .unwrap_or(header.receiver_idx); + peer.set_pending_session(session, our_index, header.sender_idx); + + if let Some(transport_id) = peer.transport_id() { + self.peers_by_index.insert( + (transport_id, our_index.as_u32()), + peer_node_addr, + ); + } + + debug!( + peer = %display_name, + new_our_index = %our_index, + new_their_index = %header.sender_idx, + "Rekey completed (initiator), pending K-bit cutover" + ); + } + Err(e) => { + warn!( + peer = %display_name, + error = %e, + "Rekey msg2 processing failed" + ); + if let Some(idx) = peer.abandon_rekey() { + let _ = self.index_allocator.free(idx); + } + } + } + } + self.pending_outbound.remove(&key); return; } - }; + + // Not a rekey — stale pending_outbound entry + self.pending_outbound.remove(&key); + return; + } + + let conn = self.connections.get_mut(&link_id).unwrap(); // Process Noise msg2 let noise_msg2 = &packet.data[header.noise_msg2_offset..]; diff --git a/src/node/handlers/mod.rs b/src/node/handlers/mod.rs index ca16234..8c52643 100644 --- a/src/node/handlers/mod.rs +++ b/src/node/handlers/mod.rs @@ -6,6 +6,7 @@ mod encrypted; mod forwarding; mod handshake; mod mmp; +mod rekey; mod rx_loop; pub(in crate::node) mod session; mod timeout; diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs new file mode 100644 index 0000000..d15bd71 --- /dev/null +++ b/src/node/handlers/rekey.rs @@ -0,0 +1,413 @@ +//! Periodic rekey (key rotation) for FMP link sessions. +//! +//! Checks all active peers on each tick for: +//! 1. Rekey trigger (time elapsed or send counter exceeded) +//! 2. Drain window expiry (clean up previous session after cutover) +//! 3. Initiator-side cutover (first send after handshake completion) + +use crate::node::Node; +use crate::node::wire::build_msg1; +use crate::noise::HandshakeState; +use crate::protocol::{SessionDatagram, SessionSetup}; +use crate::NodeAddr; +use tracing::{debug, info, trace, warn}; + +/// Keep previous session alive for this long after cutover. +const DRAIN_WINDOW_SECS: u64 = 10; + +/// Suppress local rekey initiation for this long after receiving +/// a peer's rekey msg1. +const REKEY_DAMPENING_SECS: u64 = 30; + +impl Node { + /// Periodic rekey check. Called from the tick loop. + /// + /// For each active peer with a session: + /// - If the initiator has a pending session, perform K-bit cutover + /// - If the drain window has expired, clean up the previous session + /// - If the rekey timer/counter fires, initiate a new handshake + pub(in crate::node) async fn check_rekey(&mut self) { + if !self.config.node.rekey.enabled { + return; + } + + let rekey_after_secs = self.config.node.rekey.after_secs; + let rekey_after_messages = self.config.node.rekey.after_messages; + + // Collect peers that need action (to avoid borrow conflicts) + let mut peers_to_cutover: Vec = Vec::new(); + let mut peers_to_drain: Vec = Vec::new(); + let mut peers_to_rekey: Vec = Vec::new(); + + for (node_addr, peer) in &self.peers { + if !peer.has_session() || !peer.is_healthy() { + continue; + } + + // 1. Initiator-side cutover: we completed a rekey and have + // a pending session ready. Cut over on the next tick. + if peer.pending_new_session().is_some() && !peer.rekey_in_progress() { + peers_to_cutover.push(*node_addr); + continue; + } + + // 2. Drain window expiry + if peer.is_draining() && peer.drain_expired(DRAIN_WINDOW_SECS) { + peers_to_drain.push(*node_addr); + } + + // 3. Rekey trigger + if peer.rekey_in_progress() { + continue; + } + if peer.is_rekey_dampened(REKEY_DAMPENING_SECS) { + continue; + } + + let elapsed = peer.session_established_at().elapsed().as_secs(); + let counter = peer.noise_session() + .map(|s| s.current_send_counter()) + .unwrap_or(0); + + if elapsed >= rekey_after_secs || counter >= rekey_after_messages { + peers_to_rekey.push(*node_addr); + } + } + + // Execute cutover for initiator side + for node_addr in peers_to_cutover { + if let Some(peer) = self.peers.get_mut(&node_addr) + && let Some(_old_our_index) = peer.cutover_to_new_session() + { + if let (Some(transport_id), Some(new_our_index)) = + (peer.transport_id(), peer.our_index()) + { + self.peers_by_index.insert( + (transport_id, new_our_index.as_u32()), + node_addr, + ); + } + info!( + peer = %self.peer_display_name(&node_addr), + "Rekey cutover complete (initiator), K-bit flipped" + ); + } + } + + // Execute drain completion + for node_addr in peers_to_drain { + if let Some(peer) = self.peers.get_mut(&node_addr) + && let Some(old_our_index) = peer.complete_drain() + { + if let Some(transport_id) = peer.transport_id() { + self.peers_by_index.remove(&(transport_id, old_our_index.as_u32())); + } + let _ = self.index_allocator.free(old_our_index); + trace!( + peer = %self.peer_display_name(&node_addr), + old_index = %old_our_index, + "Drain complete, previous session erased" + ); + } + } + + // Initiate new rekeys + for node_addr in peers_to_rekey { + self.initiate_rekey(&node_addr).await; + } + } + + /// Initiate an outbound rekey to a peer. + /// + /// Creates a new IK handshake as initiator, sends msg1 over the existing + /// link (same transport, same remote address), and stores the handshake + /// state on the ActivePeer. No new Link or PeerConnection is created. + async fn initiate_rekey(&mut self, node_addr: &NodeAddr) { + let peer = match self.peers.get(node_addr) { + Some(p) => p, + None => return, + }; + + let transport_id = match peer.transport_id() { + Some(t) => t, + None => return, + }; + let remote_addr = match peer.current_addr() { + Some(a) => a.clone(), + None => return, + }; + let link_id = peer.link_id(); + let peer_pubkey = peer.identity().pubkey_full(); + + // Allocate a new session index for the rekey + let our_index = match self.index_allocator.allocate() { + Ok(idx) => idx, + Err(e) => { + warn!( + peer = %self.peer_display_name(node_addr), + error = %e, + "Failed to allocate index for rekey" + ); + return; + } + }; + + // Create IK initiator handshake directly (no PeerConnection) + let our_keypair = self.identity.keypair(); + let mut hs = HandshakeState::new_initiator(our_keypair, peer_pubkey); + hs.set_local_epoch(self.startup_epoch); + + let noise_msg1 = match hs.write_message_1() { + Ok(msg) => msg, + Err(e) => { + warn!( + peer = %self.peer_display_name(node_addr), + error = %e, + "Failed to generate rekey msg1" + ); + let _ = self.index_allocator.free(our_index); + return; + } + }; + + let wire_msg1 = build_msg1(our_index, &noise_msg1); + + // Send msg1 on the existing link (same transport + address) + if let Some(transport) = self.transports.get(&transport_id) { + match transport.send(&remote_addr, &wire_msg1).await { + Ok(_) => { + debug!( + peer = %self.peer_display_name(node_addr), + our_index = %our_index, + "Rekey initiated, sent msg1 on existing link" + ); + } + Err(e) => { + warn!( + peer = %self.peer_display_name(node_addr), + error = %e, + "Failed to send rekey msg1" + ); + let _ = self.index_allocator.free(our_index); + return; + } + } + } + + // Store handshake state on the ActivePeer (not a separate PeerConnection) + let resend_interval = self.config.node.rate_limit.handshake_resend_interval_ms; + let now_ms = Self::now_ms(); + if let Some(peer) = self.peers.get_mut(node_addr) { + peer.set_rekey_state(hs, our_index, wire_msg1, now_ms + resend_interval); + } + + // Register in pending_outbound for msg2 dispatch (maps to existing link) + self.pending_outbound.insert((transport_id, our_index.as_u32()), link_id); + } + + /// Resend pending rekey msg1s and abandon timed-out rekeys. + /// + /// Called from the tick loop. Uses the same resend interval and max + /// resend count as initial handshakes. + pub(in crate::node) async fn resend_pending_rekeys(&mut self, now_ms: u64) { + if !self.config.node.rekey.enabled { + return; + } + + let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms; + + // Collect peers needing action + let mut to_resend: Vec<(NodeAddr, Vec)> = Vec::new(); + + for (node_addr, peer) in &self.peers { + if !peer.rekey_in_progress() || peer.rekey_msg1().is_none() { + continue; + } + if peer.needs_msg1_resend(now_ms) { + to_resend.push((*node_addr, peer.rekey_msg1().unwrap().to_vec())); + } + } + + for (node_addr, msg1_bytes) in to_resend { + let (transport_id, remote_addr) = match self.peers.get(&node_addr) { + Some(p) => match (p.transport_id(), p.current_addr()) { + (Some(tid), Some(addr)) => (tid, addr.clone()), + _ => continue, + }, + None => continue, + }; + + let sent = if let Some(transport) = self.transports.get(&transport_id) { + transport.send(&remote_addr, &msg1_bytes).await.is_ok() + } else { + false + }; + + if sent { + if let Some(peer) = self.peers.get_mut(&node_addr) { + peer.set_msg1_next_resend(now_ms + interval_ms); + } + trace!( + peer = %self.peer_display_name(&node_addr), + "Resent rekey msg1" + ); + } + } + } + + /// Periodic session (FSP) rekey check. Called from the tick loop. + /// + /// For each established session: + /// - If the initiator has a pending session, perform K-bit cutover + /// - If the drain window has expired, clean up the previous session + /// - If the rekey timer/counter fires, initiate a new XK handshake + pub(in crate::node) async fn check_session_rekey(&mut self) { + if !self.config.node.rekey.enabled { + return; + } + + let rekey_after_secs = self.config.node.rekey.after_secs; + let rekey_after_messages = self.config.node.rekey.after_messages; + let now_ms = Self::now_ms(); + let drain_ms = DRAIN_WINDOW_SECS * 1000; + let dampening_ms = REKEY_DAMPENING_SECS * 1000; + + let mut sessions_to_cutover: Vec = Vec::new(); + let mut sessions_to_drain: Vec = Vec::new(); + let mut sessions_to_rekey: Vec = Vec::new(); + + for (node_addr, entry) in &self.sessions { + if !entry.is_established() { + continue; + } + + // 1. Initiator-side cutover: completed rekey, pending session ready + if entry.pending_new_session().is_some() + && !entry.has_rekey_in_progress() + && entry.is_rekey_initiator() + { + sessions_to_cutover.push(*node_addr); + continue; + } + + // 2. Drain window expiry + if entry.is_draining() && entry.drain_expired(now_ms, drain_ms) { + sessions_to_drain.push(*node_addr); + } + + // 3. Rekey trigger + if entry.has_rekey_in_progress() { + continue; + } + if entry.pending_new_session().is_some() { + continue; // Responder with pending session, wait for initiator's K-bit + } + if entry.is_rekey_dampened(now_ms, dampening_ms) { + continue; + } + + let elapsed_secs = now_ms.saturating_sub(entry.session_start_ms()) / 1000; + let counter = entry.send_counter(); + + if elapsed_secs >= rekey_after_secs || counter >= rekey_after_messages { + sessions_to_rekey.push(*node_addr); + } + } + + // Execute cutover for initiator side + for node_addr in sessions_to_cutover { + if let Some(entry) = self.sessions.get_mut(&node_addr) + && entry.cutover_to_new_session(now_ms) + { + info!( + peer = %self.peer_display_name(&node_addr), + "FSP rekey cutover complete (initiator), K-bit flipped" + ); + } + } + + // Execute drain completion + for node_addr in sessions_to_drain { + if let Some(entry) = self.sessions.get_mut(&node_addr) { + entry.complete_drain(); + trace!( + peer = %self.peer_display_name(&node_addr), + "FSP drain complete, previous session erased" + ); + } + } + + // Initiate new rekeys + for node_addr in sessions_to_rekey { + self.initiate_session_rekey(&node_addr).await; + } + } + + /// Initiate an FSP session rekey. + /// + /// Creates a new XK handshake as initiator, sends SessionSetup msg1 + /// through the mesh, and stores the handshake state on the existing entry. + async fn initiate_session_rekey(&mut self, dest_addr: &NodeAddr) { + // Check route availability before paying crypto cost + if self.find_next_hop(dest_addr).is_none() { + trace!( + peer = %self.peer_display_name(dest_addr), + "FSP rekey skipped: no route to destination" + ); + return; + } + + let entry = match self.sessions.get(dest_addr) { + Some(e) => e, + None => return, + }; + let dest_pubkey = *entry.remote_pubkey(); + + // Create Noise XK initiator handshake + let our_keypair = self.identity.keypair(); + let mut handshake = HandshakeState::new_xk_initiator(our_keypair, dest_pubkey); + handshake.set_local_epoch(self.startup_epoch); + + let msg1 = match handshake.write_xk_message_1() { + Ok(m) => m, + Err(e) => { + warn!( + peer = %self.peer_display_name(dest_addr), + error = %e, + "Failed to generate FSP rekey XK msg1" + ); + return; + } + }; + + // Build SessionSetup with coordinates + let our_coords = self.tree_state.my_coords().clone(); + let dest_coords = self.get_dest_coords(dest_addr); + let setup = SessionSetup::new(our_coords, dest_coords).with_handshake(msg1); + let setup_payload = setup.encode(); + + // Send through the mesh + let my_addr = *self.node_addr(); + let mut datagram = SessionDatagram::new(my_addr, *dest_addr, setup_payload) + .with_ttl(self.config.node.session.default_ttl); + + if let Err(e) = self.send_session_datagram(&mut datagram).await { + debug!( + peer = %self.peer_display_name(dest_addr), + error = %e, + "Failed to send FSP rekey SessionSetup" + ); + return; + } + + // Store rekey state on the existing session entry + if let Some(entry) = self.sessions.get_mut(dest_addr) { + entry.set_rekey_state(handshake, true); + } + + debug!( + peer = %self.peer_display_name(dest_addr), + "FSP rekey initiated, sent SessionSetup" + ); + } +} diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 928f40a..e961612 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -108,6 +108,7 @@ impl Node { .map(|d| d.as_millis() as u64) .unwrap_or(0); self.resend_pending_handshakes(now_ms).await; + self.resend_pending_rekeys(now_ms).await; self.resend_pending_session_handshakes(now_ms).await; self.purge_idle_sessions(now_ms); self.process_pending_retries(now_ms).await; @@ -116,6 +117,8 @@ impl Node { self.check_mmp_reports().await; self.check_session_mmp_reports().await; self.check_link_heartbeats().await; + self.check_rekey().await; + self.check_session_rekey().await; self.purge_stale_lookups(now_ms); self.poll_transport_discovery().await; self.sample_transport_congestion(); diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 44bdd0b..41c5669 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -9,8 +9,8 @@ use crate::node::session::{EndToEndState, SessionEntry}; use crate::node::session_wire::{ build_fsp_header, fsp_prepend_inner_header, fsp_strip_inner_header, parse_encrypted_coords, FspCommonPrefix, FspEncryptedHeader, FSP_COMMON_PREFIX_SIZE, - FSP_FLAG_CP, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2, - FSP_PHASE_MSG3, + FSP_FLAG_CP, FSP_FLAG_K, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, + FSP_PHASE_MSG2, FSP_PHASE_MSG3, }; use crate::protocol::{coords_wire_size, encode_coords}; use crate::upper::icmp::FIPS_OVERHEAD; @@ -162,6 +162,25 @@ impl Node { } } + // K-bit flip detection: peer has cut over to the new session. + let received_k_bit = header.flags & FSP_FLAG_K != 0; + { + let entry = self.sessions.get(src_addr).unwrap(); + let k_bit_flipped = received_k_bit != entry.current_k_bit() + && entry.pending_new_session().is_some(); + + if k_bit_flipped { + let display_name = self.peer_display_name(src_addr); + info!( + peer = %display_name, + "Peer FSP K-bit flip detected, promoting new session" + ); + let now_ms = Self::now_ms(); + let entry = self.sessions.get_mut(src_addr).unwrap(); + entry.handle_peer_kbit_flip(now_ms); + } + } + let mut entry = match self.sessions.remove(src_addr) { Some(e) => e, None => return, @@ -184,12 +203,31 @@ impl Node { ) { Ok(pt) => pt, Err(e) => { - debug!( - error = %e, src = %self.peer_display_name(src_addr), counter = header.counter, - "Session AEAD decryption failed" - ); - self.sessions.insert(*src_addr, entry); - return; + // Current session failed — try previous session (drain window) + if let Some(prev_session) = entry.previous_noise_session_mut() { + match prev_session.decrypt_with_replay_check_and_aad( + ciphertext, + header.counter, + &header.header_bytes, + ) { + Ok(pt) => pt, + Err(_) => { + debug!( + error = %e, src = %self.peer_display_name(src_addr), counter = header.counter, + "Session AEAD decryption failed (current and previous)" + ); + self.sessions.insert(*src_addr, entry); + return; + } + } + } else { + debug!( + error = %e, src = %self.peer_display_name(src_addr), counter = header.counter, + "Session AEAD decryption failed" + ); + self.sessions.insert(*src_addr, entry); + return; + } } }; @@ -338,6 +376,53 @@ impl Node { } return; } else if existing.is_established() { + // Rekey: if rekey enabled, treat as rekey for key rotation. + // The existing established session remains active for traffic. + if self.config.node.rekey.enabled && !existing.has_rekey_in_progress() { + let our_keypair = self.identity.keypair(); + let mut handshake = HandshakeState::new_xk_responder(our_keypair); + handshake.set_local_epoch(self.startup_epoch); + + if let Err(e) = handshake.read_xk_message_1(&setup.handshake_payload) { + debug!(error = %e, "Failed to process rekey XK msg1"); + return; + } + + // Generate msg2 + let msg2 = match handshake.write_xk_message_2() { + Ok(m) => m, + Err(e) => { + debug!(error = %e, "Failed to generate rekey XK msg2"); + return; + } + }; + + // Build and send SessionAck + let our_coords = self.tree_state.my_coords().clone(); + let ack = SessionAck::new(our_coords, setup.src_coords).with_handshake(msg2); + let ack_payload = ack.encode(); + let my_addr = *self.node_addr(); + let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack_payload) + .with_ttl(self.config.node.session.default_ttl); + + if let Err(e) = self.send_session_datagram(&mut datagram).await { + debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send rekey SessionAck"); + return; + } + + // Store rekey state on the existing entry + let now_ms = Self::now_ms(); + let entry = self.sessions.get_mut(src_addr).unwrap(); + entry.set_rekey_state(handshake, false); + entry.record_peer_rekey(now_ms); + + debug!( + src = %self.peer_display_name(src_addr), + "FSP rekey: processed peer's msg1, sent msg2, awaiting msg3" + ); + return; + } + // Re-establishment: replace existing session below debug!(src = %self.peer_display_name(src_addr), "Session re-establishment from peer"); } @@ -423,6 +508,70 @@ impl Node { } }; + // Rekey path: entry is Established with rekey_state + if entry.is_established() && entry.has_rekey_in_progress() && entry.is_rekey_initiator() { + let mut handshake = match entry.take_rekey_state() { + Some(hs) => hs, + None => { + self.sessions.insert(*src_addr, entry); + return; + } + }; + + // Process XK msg2 + if let Err(e) = handshake.read_xk_message_2(&ack.handshake_payload) { + debug!(error = %e, "Failed to process rekey XK msg2"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + + // Generate XK msg3 + let msg3 = match handshake.write_xk_message_3() { + Ok(m) => m, + Err(e) => { + debug!(error = %e, "Failed to generate rekey XK msg3"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + }; + + // Send SessionMsg3 + let msg3_wire = SessionMsg3::new(msg3); + let msg3_payload = msg3_wire.encode(); + let my_addr = *self.node_addr(); + let mut datagram = SessionDatagram::new(my_addr, *src_addr, msg3_payload) + .with_ttl(self.config.node.session.default_ttl); + + if let Err(e) = self.send_session_datagram(&mut datagram).await { + debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send rekey SessionMsg3"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + + // Complete handshake → store as pending new session + let session = match handshake.into_session() { + Ok(s) => s, + Err(e) => { + debug!(error = %e, "Failed to create session from rekey XK"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + }; + + entry.set_pending_session(session); + self.sessions.insert(*src_addr, entry); + + debug!( + src = %self.peer_display_name(src_addr), + "FSP rekey: completed XK as initiator, pending cutover" + ); + return; + } + // Must be in Initiating state — check before take to avoid poisoning if !entry.is_initiating() { debug!(src = %self.peer_display_name(src_addr), "SessionAck but session not in Initiating state"); @@ -518,6 +667,45 @@ impl Node { } }; + // Rekey path: entry is Established with rekey_state (responder side) + if entry.is_established() && entry.has_rekey_in_progress() && !entry.is_rekey_initiator() { + let mut handshake = match entry.take_rekey_state() { + Some(hs) => hs, + None => { + self.sessions.insert(*src_addr, entry); + return; + } + }; + + // Process XK msg3 + if let Err(e) = handshake.read_xk_message_3(&msg3.handshake_payload) { + debug!(error = %e, "Failed to process rekey XK msg3"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + + // Complete the handshake → store as pending new session + let session = match handshake.into_session() { + Ok(s) => s, + Err(e) => { + debug!(error = %e, "Failed to create session from rekey XK msg3"); + entry.abandon_rekey(); + self.sessions.insert(*src_addr, entry); + return; + } + }; + + entry.set_pending_session(session); + self.sessions.insert(*src_addr, entry); + + debug!( + src = %self.peer_display_name(src_addr), + "FSP rekey: completed XK as responder, pending cutover" + ); + return; + } + // Must be in AwaitingMsg3 state if !entry.is_awaiting_msg3() { debug!(src = %self.peer_display_name(src_addr), "SessionMsg3 but session not in AwaitingMsg3 state"); @@ -985,8 +1173,13 @@ impl Node { entry.set_coords_warmup_remaining(entry.coords_warmup_remaining() - 1); } - // Build FSP flags (CP flag only if coords will be piggybacked) - let flags = if include_coords { FSP_FLAG_CP } else { 0 }; + // Build FSP flags (CP flag if coords, K-bit for key epoch) + let mut flags = if include_coords { FSP_FLAG_CP } else { 0 }; + if let Some(entry) = self.sessions.get(dest_addr) + && entry.current_k_bit() + { + flags |= FSP_FLAG_K; + } // Borrow session for counter + encryption (after potential standalone send) let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { @@ -1074,6 +1267,10 @@ impl Node { node_addr: *dest_addr, reason: "no session".into(), })?; + + // Read K-bit before mutable borrow of session state + let k_flags = if entry.current_k_bit() { FSP_FLAG_K } else { 0 }; + let session = match entry.state_mut() { EndToEndState::Established(s) => s, _ => { @@ -1089,9 +1286,9 @@ impl Node { // FSP inner header + plaintext let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, payload); - // Build 12-byte FSP header (no flags — no CP for reports) + // Build 12-byte FSP header (K-bit for key epoch, no CP for reports) let payload_len = inner_plaintext.len() as u16; - let header = build_fsp_header(counter, 0, payload_len); + let header = build_fsp_header(counter, k_flags, payload_len); // Encrypt with AAD let ciphertext = session.encrypt_with_aad(&inner_plaintext, &header).map_err(|e| { @@ -1255,7 +1452,7 @@ impl Node { /// Returns our own coordinates as a fallback (the SessionSetup will /// carry src_coords for return path routing; empty dest_coords /// would fail wire encoding since TreeCoordinate requires ≥1 entry). - fn get_dest_coords(&self, dest: &NodeAddr) -> crate::tree::TreeCoordinate { + pub(in crate::node) fn get_dest_coords(&self, dest: &NodeAddr) -> crate::tree::TreeCoordinate { let now_ms = Self::now_ms(); if let Some(coords) = self.coord_cache.get(dest, now_ms) { return coords.clone(); diff --git a/src/node/mod.rs b/src/node/mod.rs index 43b0d02..f0221c5 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -35,7 +35,7 @@ use crate::transport::ethernet::EthernetTransport; use crate::tree::TreeState; use crate::upper::icmp_rate_limit::IcmpRateLimiter; use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; -use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_SP}; +use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP}; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity}; use rand::Rng; use std::collections::{HashMap, VecDeque}; @@ -1350,6 +1350,9 @@ impl Node { if ce_flag { flags |= FLAG_CE; } + if peer.current_k_bit() { + flags |= FLAG_KEY_EPOCH; + } let session = peer.noise_session_mut().ok_or_else(|| NodeError::SendFailed { node_addr: *node_addr, diff --git a/src/node/session.rs b/src/node/session.rs index 922a39d..8f1845a 100644 --- a/src/node/session.rs +++ b/src/node/session.rs @@ -91,6 +91,22 @@ pub(crate) struct SessionEntry { resend_count: u32, /// When the next resend should fire (Unix ms). 0 = no resend scheduled. next_resend_at_ms: u64, + + // === Rekey (Key Rotation) === + /// Current K-bit epoch value (alternates each rekey). + current_k_bit: bool, + /// Previous NoiseSession during drain window after cutover. + previous_noise_session: Option, + /// When drain window started (Unix ms). 0 = no drain. + drain_started_ms: u64, + /// In-progress rekey state (runs alongside Established session). + rekey_state: Option, + /// Pending completed session awaiting K-bit cutover. + pending_new_session: Option, + /// Whether we initiated the current rekey. + rekey_initiator: bool, + /// Dampening: last time peer sent us a rekey msg1 (Unix ms). + last_peer_rekey_ms: u64, } impl SessionEntry { @@ -119,6 +135,13 @@ impl SessionEntry { handshake_payload: None, resend_count: 0, next_resend_at_ms: 0, + current_k_bit: false, + previous_noise_session: None, + drain_started_ms: 0, + rekey_state: None, + pending_new_session: None, + rekey_initiator: false, + last_peer_rekey_ms: 0, } } @@ -292,4 +315,144 @@ impl SessionEntry { self.resend_count += 1; self.next_resend_at_ms = next_resend_at_ms; } + + // === Rekey (Key Rotation) === + + /// Current K-bit epoch value. + pub(crate) fn current_k_bit(&self) -> bool { + self.current_k_bit + } + + /// Whether a rekey is currently in progress. + pub(crate) fn has_rekey_in_progress(&self) -> bool { + self.rekey_state.is_some() + } + + /// Get the pending new session (completed rekey, not yet cut over). + pub(crate) fn pending_new_session(&self) -> Option<&NoiseSession> { + self.pending_new_session.as_ref() + } + + /// Get the previous session for decryption fallback during drain. + pub(crate) fn previous_noise_session_mut(&mut self) -> Option<&mut NoiseSession> { + self.previous_noise_session.as_mut() + } + + /// Whether we initiated the current rekey. + pub(crate) fn is_rekey_initiator(&self) -> bool { + self.rekey_initiator + } + + /// Check if rekey initiation is dampened. + pub(crate) fn is_rekey_dampened(&self, now_ms: u64, dampening_ms: u64) -> bool { + if self.last_peer_rekey_ms == 0 { + return false; + } + now_ms.saturating_sub(self.last_peer_rekey_ms) < dampening_ms + } + + /// Record that the peer initiated a rekey (for dampening). + pub(crate) fn record_peer_rekey(&mut self, now_ms: u64) { + self.last_peer_rekey_ms = now_ms; + } + + /// When the session transitioned to Established (for rekey timer). + pub(crate) fn session_start_ms(&self) -> u64 { + self.session_start_ms + } + + /// Get the current send counter from the established NoiseSession. + pub(crate) fn send_counter(&self) -> u64 { + match self.state.as_ref() { + Some(EndToEndState::Established(s)) => s.current_send_counter(), + _ => 0, + } + } + + /// Store a completed rekey session. + pub(crate) fn set_pending_session(&mut self, session: NoiseSession) { + self.pending_new_session = Some(session); + self.rekey_state = None; + } + + /// Set the rekey handshake state (in-progress XK handshake). + pub(crate) fn set_rekey_state(&mut self, state: HandshakeState, is_initiator: bool) { + self.rekey_state = Some(state); + self.rekey_initiator = is_initiator; + } + + /// Take the rekey state for processing. + pub(crate) fn take_rekey_state(&mut self) -> Option { + self.rekey_state.take() + } + + /// Cut over to the pending new session (initiator side). + /// + /// Moves current session to previous (for drain), promotes pending to current, + /// flips the K-bit. + pub(crate) fn cutover_to_new_session(&mut self, now_ms: u64) -> bool { + let new_session = match self.pending_new_session.take() { + Some(s) => s, + None => return false, + }; + + // Demote current to previous for drain + if let Some(EndToEndState::Established(old)) = self.state.take() { + self.previous_noise_session = Some(old); + } + self.drain_started_ms = now_ms; + + // Promote pending to current + self.state = Some(EndToEndState::Established(new_session)); + self.current_k_bit = !self.current_k_bit; + self.session_start_ms = now_ms; + self.rekey_state = None; + self.rekey_initiator = false; + true + } + + /// Handle receiving a K-bit flip from the peer (responder side). + pub(crate) fn handle_peer_kbit_flip(&mut self, now_ms: u64) -> bool { + let new_session = match self.pending_new_session.take() { + Some(s) => s, + None => return false, + }; + + // Demote current to previous for drain + if let Some(EndToEndState::Established(old)) = self.state.take() { + self.previous_noise_session = Some(old); + } + self.drain_started_ms = now_ms; + + // Promote pending to current + self.state = Some(EndToEndState::Established(new_session)); + self.current_k_bit = !self.current_k_bit; + self.session_start_ms = now_ms; + self.rekey_state = None; + self.rekey_initiator = false; + true + } + + /// Check if the drain window has expired. + pub(crate) fn drain_expired(&self, now_ms: u64, drain_ms: u64) -> bool { + self.drain_started_ms > 0 && now_ms.saturating_sub(self.drain_started_ms) >= drain_ms + } + + /// Whether a drain is in progress. + pub(crate) fn is_draining(&self) -> bool { + self.drain_started_ms > 0 + } + + /// Complete the drain: drop previous session. + pub(crate) fn complete_drain(&mut self) { + self.previous_noise_session = None; + self.drain_started_ms = 0; + } + + /// Abandon an in-progress rekey. + pub(crate) fn abandon_rekey(&mut self) { + self.rekey_state = None; + self.pending_new_session = None; + self.rekey_initiator = false; + } } diff --git a/src/peer/active.rs b/src/peer/active.rs index f6d5997..d66fb54 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -6,7 +6,7 @@ use crate::bloom::BloomFilter; use crate::mmp::{MmpConfig, MmpPeerState}; use crate::utils::index::SessionIndex; -use crate::noise::NoiseSession; +use crate::noise::{HandshakeState as NoiseHandshakeState, NoiseError, NoiseSession}; use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId}; use crate::tree::{ParentDeclaration, TreeCoordinate}; use crate::{FipsAddress, NodeAddr, PeerIdentity}; @@ -147,6 +147,38 @@ pub struct ActivePeer { // === Replay Detection Suppression === /// Number of replay detections suppressed since last session reset. replay_suppressed_count: u32, + /// Consecutive decryption failures (reset on any successful decrypt). + consecutive_decrypt_failures: u32, + + // === Rekey (Key Rotation) === + /// When the current Noise session was established (for rekey timer). + session_established_at: Instant, + /// Current K-bit epoch value (alternates each rekey). + current_k_bit: bool, + /// Previous session kept alive during drain window after cutover. + previous_session: Option, + /// Previous session's our_index (for peers_by_index cleanup on drain expiry). + previous_our_index: Option, + /// When the drain window started (None = no drain in progress). + drain_started: Option, + /// Pending new session from completed rekey (before K-bit cutover). + pending_new_session: Option, + /// Pending new session's our_index. + pending_our_index: Option, + /// Pending new session's their_index. + pending_their_index: Option, + /// Whether a rekey is currently in progress (handshake sent, not yet complete). + rekey_in_progress: bool, + /// When we last received a rekey msg1 from this peer (dampening). + last_peer_rekey: Option, + /// In-progress rekey: Noise handshake state (initiator only). + rekey_handshake: Option, + /// In-progress rekey: our new session index. + rekey_our_index: Option, + /// In-progress rekey: wire-format msg1 for resend. + rekey_msg1: Option>, + /// In-progress rekey: next resend timestamp (Unix ms). + rekey_msg1_next_resend: u64, } impl ActivePeer { @@ -155,6 +187,7 @@ impl ActivePeer { /// Called after successful authentication handshake. /// For peers with Noise sessions, use `with_session` instead. pub fn new(identity: PeerIdentity, link_id: LinkId, authenticated_at: u64) -> Self { + let now = Instant::now(); Self { identity, link_id, @@ -173,7 +206,7 @@ impl ActivePeer { filter_sequence: 0, filter_received_at: 0, pending_filter_update: true, // Send filter on new connection - session_start: Instant::now(), + session_start: now, link_stats: LinkStats::new(), authenticated_at, last_seen: authenticated_at, @@ -182,6 +215,21 @@ impl ActivePeer { last_heartbeat_sent: None, handshake_msg2: None, replay_suppressed_count: 0, + consecutive_decrypt_failures: 0, + session_established_at: now, + current_k_bit: false, + previous_session: None, + previous_our_index: None, + drain_started: None, + pending_new_session: None, + pending_our_index: None, + pending_their_index: None, + rekey_in_progress: false, + last_peer_rekey: None, + rekey_handshake: None, + rekey_our_index: None, + rekey_msg1: None, + rekey_msg1_next_resend: 0, } } @@ -219,6 +267,7 @@ impl ActivePeer { mmp_config: &MmpConfig, remote_epoch: Option<[u8; 8]>, ) -> Self { + let now = Instant::now(); Self { identity, link_id, @@ -237,7 +286,7 @@ impl ActivePeer { filter_sequence: 0, filter_received_at: 0, pending_filter_update: true, - session_start: Instant::now(), + session_start: now, link_stats, authenticated_at, last_seen: authenticated_at, @@ -246,6 +295,21 @@ impl ActivePeer { last_heartbeat_sent: None, handshake_msg2: None, replay_suppressed_count: 0, + consecutive_decrypt_failures: 0, + session_established_at: now, + current_k_bit: false, + previous_session: None, + previous_our_index: None, + drain_started: None, + pending_new_session: None, + pending_our_index: None, + pending_their_index: None, + rekey_in_progress: false, + last_peer_rekey: None, + rekey_handshake: None, + rekey_our_index: None, + rekey_msg1: None, + rekey_msg1_next_resend: 0, } } @@ -415,6 +479,19 @@ impl ActivePeer { self.replay_suppressed_count } + // === Decryption Failure Tracking === + + /// Increment consecutive decryption failure counter, returning new count. + pub fn increment_decrypt_failures(&mut self) -> u32 { + self.consecutive_decrypt_failures += 1; + self.consecutive_decrypt_failures + } + + /// Reset consecutive decryption failure counter. + pub fn reset_decrypt_failures(&mut self) { + self.consecutive_decrypt_failures = 0; + } + // === Epoch Accessors === /// Get the remote peer's startup epoch (from handshake). @@ -689,6 +766,258 @@ impl ActivePeer { pub fn clear_filter_update_needed(&mut self) { self.pending_filter_update = false; } + + // === Rekey (Key Rotation) === + + /// When the current Noise session was established. + pub fn session_established_at(&self) -> Instant { + self.session_established_at + } + + /// Current K-bit epoch value. + pub fn current_k_bit(&self) -> bool { + self.current_k_bit + } + + /// Whether a rekey is currently in progress. + pub fn rekey_in_progress(&self) -> bool { + self.rekey_in_progress + } + + /// Mark that a rekey has been initiated. + pub fn set_rekey_in_progress(&mut self) { + self.rekey_in_progress = true; + } + + /// Check if rekey initiation is dampened (peer recently sent us msg1). + pub fn is_rekey_dampened(&self, dampening_secs: u64) -> bool { + match self.last_peer_rekey { + Some(t) => t.elapsed().as_secs() < dampening_secs, + None => false, + } + } + + /// Record that the peer initiated a rekey (for dampening). + pub fn record_peer_rekey(&mut self) { + self.last_peer_rekey = Some(Instant::now()); + } + + /// Get the pending new session's our_index. + pub fn pending_our_index(&self) -> Option { + self.pending_our_index + } + + /// Get the pending new session's their_index. + pub fn pending_their_index(&self) -> Option { + self.pending_their_index + } + + /// Get the previous session's our_index (during drain). + pub fn previous_our_index(&self) -> Option { + self.previous_our_index + } + + /// Get the previous session for decryption fallback. + pub fn previous_session(&self) -> Option<&NoiseSession> { + self.previous_session.as_ref() + } + + /// Get mutable access to the previous session for decryption. + pub fn previous_session_mut(&mut self) -> Option<&mut NoiseSession> { + self.previous_session.as_mut() + } + + /// Get the pending new session (completed rekey, not yet cut over). + pub fn pending_new_session(&self) -> Option<&NoiseSession> { + self.pending_new_session.as_ref() + } + + /// Store a completed rekey session and its indices. + /// + /// Called when the rekey handshake completes. The session is held + /// as pending until the initiator flips the K-bit on the next outbound packet. + pub fn set_pending_session( + &mut self, + session: NoiseSession, + our_index: SessionIndex, + their_index: SessionIndex, + ) { + self.pending_new_session = Some(session); + self.pending_our_index = Some(our_index); + self.pending_their_index = Some(their_index); + self.rekey_in_progress = false; + // Clear initiator handshake state (index now lives in pending_our_index) + self.rekey_our_index = None; + self.rekey_handshake = None; + self.rekey_msg1 = None; + self.rekey_msg1_next_resend = 0; + } + + /// Cut over to the pending new session (initiator side). + /// + /// Moves current session to previous (for drain), promotes pending to current, + /// flips the K-bit. Returns the old our_index that should remain in peers_by_index + /// during the drain window. + pub fn cutover_to_new_session(&mut self) -> Option { + let new_session = self.pending_new_session.take()?; + let new_our_index = self.pending_our_index.take(); + let new_their_index = self.pending_their_index.take(); + + // Demote current to previous + self.previous_session = self.noise_session.take(); + self.previous_our_index = self.our_index; + self.drain_started = Some(Instant::now()); + + // Promote pending to current + self.noise_session = Some(new_session); + self.our_index = new_our_index; + self.their_index = new_their_index; + + // Flip K-bit and reset timing + self.current_k_bit = !self.current_k_bit; + self.session_established_at = Instant::now(); + self.session_start = Instant::now(); + self.rekey_in_progress = false; + self.reset_replay_suppressed(); + + self.previous_our_index + } + + /// Handle receiving a K-bit flip from the peer (responder side). + /// + /// Promotes pending_new_session to current, demotes current to previous. + /// Returns the old our_index for drain tracking. + pub fn handle_peer_kbit_flip(&mut self) -> Option { + let new_session = self.pending_new_session.take()?; + let new_our_index = self.pending_our_index.take(); + let new_their_index = self.pending_their_index.take(); + + // Demote current to previous + self.previous_session = self.noise_session.take(); + self.previous_our_index = self.our_index; + self.drain_started = Some(Instant::now()); + + // Promote pending to current + self.noise_session = Some(new_session); + self.our_index = new_our_index; + self.their_index = new_their_index; + + // Match peer's K-bit + self.current_k_bit = !self.current_k_bit; + self.session_established_at = Instant::now(); + self.session_start = Instant::now(); + self.rekey_in_progress = false; + self.reset_replay_suppressed(); + + self.previous_our_index + } + + /// Check if the drain window has expired. + pub fn drain_expired(&self, drain_secs: u64) -> bool { + match self.drain_started { + Some(t) => t.elapsed().as_secs() >= drain_secs, + None => false, + } + } + + /// Whether a drain is in progress. + pub fn is_draining(&self) -> bool { + self.drain_started.is_some() + } + + /// Complete the drain: drop previous session and free its index. + /// + /// Returns the previous our_index so the caller can remove it from + /// peers_by_index and free it from the IndexAllocator. + pub fn complete_drain(&mut self) -> Option { + self.previous_session = None; + self.drain_started = None; + self.previous_our_index.take() + } + + /// Abandon an in-progress rekey. + /// + /// Returns the rekey our_index so the caller can free it. + /// Also clears any pending session state if the handshake was completed + /// but not yet cut over. + pub fn abandon_rekey(&mut self) -> Option { + self.rekey_handshake = None; + self.rekey_msg1 = None; + self.rekey_msg1_next_resend = 0; + self.rekey_in_progress = false; + // Return whichever index needs freeing + self.rekey_our_index.take() + .or_else(|| { + self.pending_new_session = None; + self.pending_their_index = None; + self.pending_our_index.take() + }) + } + + // === Rekey Handshake State (Initiator) === + + /// Store rekey handshake state after sending msg1. + pub fn set_rekey_state( + &mut self, + handshake: NoiseHandshakeState, + our_index: SessionIndex, + wire_msg1: Vec, + next_resend_ms: u64, + ) { + self.rekey_handshake = Some(handshake); + self.rekey_our_index = Some(our_index); + self.rekey_msg1 = Some(wire_msg1); + self.rekey_msg1_next_resend = next_resend_ms; + self.rekey_in_progress = true; + } + + /// Get the rekey our_index (for msg2 dispatch lookup). + pub fn rekey_our_index(&self) -> Option { + self.rekey_our_index + } + + /// Complete the rekey by processing msg2 (initiator side). + /// + /// Takes the stored handshake state, reads msg2, and returns the + /// completed NoiseSession. Clears the handshake-related fields but + /// leaves rekey_our_index for set_pending_session to use. + pub fn complete_rekey_msg2( + &mut self, + msg2_bytes: &[u8], + ) -> Result { + let mut hs = self.rekey_handshake + .take() + .ok_or_else(|| NoiseError::WrongState { + expected: "rekey handshake in progress".to_string(), + got: "no handshake state".to_string(), + })?; + + hs.read_message_2(msg2_bytes)?; + let session = hs.into_session()?; + + // Clear msg1 resend state + self.rekey_msg1 = None; + self.rekey_msg1_next_resend = 0; + + Ok(session) + } + + /// Check if msg1 needs resending. + pub fn needs_msg1_resend(&self, now_ms: u64) -> bool { + self.rekey_in_progress + && self.rekey_msg1.is_some() + && now_ms >= self.rekey_msg1_next_resend + } + + /// Get msg1 bytes for resend (without consuming). + pub fn rekey_msg1(&self) -> Option<&[u8]> { + self.rekey_msg1.as_deref() + } + + /// Update next resend timestamp. + pub fn set_msg1_next_resend(&mut self, next_ms: u64) { + self.rekey_msg1_next_resend = next_ms; + } } #[cfg(test)] diff --git a/testing/static/configs/topologies/rekey.yaml b/testing/static/configs/topologies/rekey.yaml new file mode 100644 index 0000000..ce7e544 --- /dev/null +++ b/testing/static/configs/topologies/rekey.yaml @@ -0,0 +1,36 @@ +# Rekey Integration Test Topology +# +# Same sparse mesh as mesh.yaml but configs are post-processed to use +# aggressive rekey timers (35s) for CI testing. The rekey-test.sh script +# handles the config injection and multi-phase verification. + +nodes: + a: + nsec: "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20" + npub: "npub1sjlh2c3x9w7kjsqg2ay080n2lff2uvt325vpan33ke34rn8l5jcqawh57m" + docker_ip: "172.20.0.10" + peers: [d, e] + + b: + nsec: "b102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fb0" + npub: "npub1tdwa4vjrjl33pcjdpf2t4p027nl86xrx24g4d3avg4vwvayr3g8qhd84le" + docker_ip: "172.20.0.11" + peers: [c] + + c: + nsec: "c102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fc0" + npub: "npub1cld9yay0u24davpu6c35l4vldrhzvaq66pcqtg9a0j2cnjrn9rtsxx2pe6" + docker_ip: "172.20.0.12" + peers: [b, d, e] + + d: + nsec: "d102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fd0" + npub: "npub1n9lpnv0592cc2ps6nm0ca3qls642vx7yjsv35rkxqzj2vgds52sqgpverl" + docker_ip: "172.20.0.13" + peers: [a, c, e] + + e: + nsec: "e102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1fe0" + npub: "npub1wf8akf8lu2zdkjkmwhl75pqvven654mpv4sz2x2tprl5265mgrzq8nhak4" + docker_ip: "172.20.0.14" + peers: [a, c, d] diff --git a/testing/static/docker-compose.yml b/testing/static/docker-compose.yml index fbc2604..e673480 100644 --- a/testing/static/docker-compose.yml +++ b/testing/static/docker-compose.yml @@ -204,6 +204,67 @@ services: fips-net: ipv4_address: 172.20.0.14 + # ── Rekey integration test (mesh + aggressive rekey timers) ── + rekey-a: + <<: *fips-common + profiles: ["rekey"] + container_name: fips-node-a + hostname: node-a + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/rekey/node-a.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.10 + + rekey-b: + <<: *fips-common + profiles: ["rekey"] + container_name: fips-node-b + hostname: node-b + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/rekey/node-b.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.11 + + rekey-c: + <<: *fips-common + profiles: ["rekey"] + container_name: fips-node-c + hostname: node-c + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/rekey/node-c.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.12 + + rekey-d: + <<: *fips-common + profiles: ["rekey"] + container_name: fips-node-d + hostname: node-d + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/rekey/node-d.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.13 + + rekey-e: + <<: *fips-common + profiles: ["rekey"] + container_name: fips-node-e + hostname: node-e + volumes: + - ./resolv.conf:/etc/resolv.conf:ro + - ./generated-configs/rekey/node-e.yaml:/etc/fips/fips.yaml:ro + networks: + fips-net: + ipv4_address: 172.20.0.14 + # ── TCP chain topology (A-B-C) ─────────────────────────────── tcp-a: <<: *fips-common diff --git a/testing/static/scripts/ping-test.sh b/testing/static/scripts/ping-test.sh index 09716ec..6cb75eb 100755 --- a/testing/static/scripts/ping-test.sh +++ b/testing/static/scripts/ping-test.sh @@ -52,8 +52,8 @@ echo "=== FIPS Ping Test ($PROFILE topology) ===" echo "" # Wait for nodes to converge -echo "Waiting 3s for mesh convergence..." -sleep 3 +echo "Waiting 5s for mesh convergence..." +sleep 5 if [ "$PROFILE" = "mesh" ] || [ "$PROFILE" = "mesh-public" ]; then # Sparse mesh topology: A-B, B-C, C-D, D-E, E-A, A-D diff --git a/testing/static/scripts/rekey-test.sh b/testing/static/scripts/rekey-test.sh new file mode 100755 index 0000000..6c58a45 --- /dev/null +++ b/testing/static/scripts/rekey-test.sh @@ -0,0 +1,257 @@ +#!/bin/bash +# Integration test for Noise rekey (periodic key rotation). +# +# Verifies that FMP link rekey and FSP session rekey complete without +# disrupting connectivity. Uses aggressive rekey timers (35s) so that +# multiple rekey cycles complete within CI time budgets. +# +# Tested failure modes: +# - Cross-connection msg1 misidentified as rekey (session age guard) +# - K-bit cutover and drain window (old session cleanup) +# - FMP + FSP coordinated rekeying +# - Multi-hop session survival across rekey +# - Back-to-back rekey cycles (consecutive rekeys) +# - Link stability through rekey (no spurious link teardowns) +# +# Usage: +# ./rekey-test.sh Run the full test (containers must be up) +# ./rekey-test.sh inject-config Inject rekey config into generated configs +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +TOPOLOGY="rekey" +NODES="a b c d e" + +# Rekey timing configuration +REKEY_AFTER_SECS=35 + +# ── inject-config subcommand ────────────────────────────────────────── +# Inject rekey config into generated node configs. Called separately +# by CI before building Docker images. +if [ "${1:-}" = "inject-config" ]; then + echo "Injecting rekey config (after_secs=$REKEY_AFTER_SECS) into node configs..." + for node in $NODES; do + cfg="$SCRIPT_DIR/../generated-configs/$TOPOLOGY/node-$node.yaml" + if [ ! -f "$cfg" ]; then + echo " Error: $cfg not found" >&2 + exit 1 + fi + python3 -c " +import yaml +with open('$cfg') as f: + cfg = yaml.safe_load(f) +cfg.setdefault('node', {})['rekey'] = { + 'enabled': True, + 'after_secs': $REKEY_AFTER_SECS, + 'after_messages': 65536, +} +with open('$cfg', 'w') as f: + yaml.dump(cfg, f, default_flow_style=False, sort_keys=False) +" + echo " ✓ node-$node" + done + echo "✓ Config injection complete" + exit 0 +fi + +# ── Full test ───────────────────────────────────────────────────────── +trap 'echo ""; echo "Test interrupted"; exit 130' INT + +# Wait times derived from rekey timer +CONVERGE_WAIT=5 +FIRST_REKEY_WAIT=40 # > REKEY_AFTER_SECS, allow margin +SECOND_REKEY_WAIT=40 # wait for second cycle + +TIMEOUT=5 +PASSED=0 +FAILED=0 +TOTAL_PASSED=0 +TOTAL_FAILED=0 + +# Node identities +ENV_FILE="$SCRIPT_DIR/../generated-configs/npubs.env" +if [ ! -f "$ENV_FILE" ]; then + echo "Error: $ENV_FILE not found. Run generate-configs.sh first." >&2 + exit 1 +fi +source "$ENV_FILE" + +NPUBS=("$NPUB_A" "$NPUB_B" "$NPUB_C" "$NPUB_D" "$NPUB_E") +LABELS=(A B C D E) + +# ── Helpers ──────────────────────────────────────────────────────────── + +ping_one() { + local from="$1" + local to_npub="$2" + local label="$3" + local quiet="${4:-}" + + if output=$(docker exec "fips-$from" ping6 -c 1 -W "$TIMEOUT" "${to_npub}.fips" 2>&1); then + local rtt=$(echo "$output" | grep -oE 'time=[0-9.]+' | cut -d= -f2) + if [ -z "$quiet" ]; then + echo " $label ... OK (${rtt:-?}ms)" + fi + PASSED=$((PASSED + 1)) + else + if [ -z "$quiet" ]; then + echo " $label ... FAIL" + fi + FAILED=$((FAILED + 1)) + fi +} + +# Run all 20 directed pairs +ping_all() { + local quiet="${1:-}" + PASSED=0 + FAILED=0 + for i in 0 1 2 3 4; do + if [ -z "$quiet" ]; then + echo " From node-${LABELS[$i],,}:" + fi + for j in 0 1 2 3 4; do + [ "$i" -eq "$j" ] && continue + ping_one "node-${LABELS[$i],,}" "${NPUBS[$j]}" \ + "${LABELS[$i]} → ${LABELS[$j]}" "$quiet" + done + done +} + +phase_result() { + local phase="$1" + TOTAL_PASSED=$((TOTAL_PASSED + PASSED)) + TOTAL_FAILED=$((TOTAL_FAILED + FAILED)) + if [ "$FAILED" -eq 0 ]; then + echo " ✓ $phase: $PASSED/$((PASSED + FAILED)) passed" + else + echo " ✗ $phase: $PASSED passed, $FAILED FAILED" + fi +} + +# Count occurrences of a pattern across all node logs +count_log_pattern() { + local pattern="$1" + local total=0 + for node in $NODES; do + local count=$(docker logs "fips-node-$node" 2>&1 | grep -c "$pattern" || true) + total=$((total + count)) + done + echo "$total" +} + +# Check that a pattern appears at least N times across all logs +assert_min_count() { + local pattern="$1" + local min_count="$2" + local description="$3" + local count=$(count_log_pattern "$pattern") + if [ "$count" -ge "$min_count" ]; then + echo " ✓ $description: $count (>= $min_count)" + PASSED=$((PASSED + 1)) + else + echo " ✗ $description: $count (expected >= $min_count)" + FAILED=$((FAILED + 1)) + fi +} + +# Check that a pattern appears zero times across all logs +assert_zero_count() { + local pattern="$1" + local description="$2" + local count=$(count_log_pattern "$pattern") + if [ "$count" -eq 0 ]; then + echo " ✓ $description: 0" + PASSED=$((PASSED + 1)) + else + echo " ✗ $description: $count (expected 0)" + FAILED=$((FAILED + 1)) + fi +} + +# ── Main ─────────────────────────────────────────────────────────────── + +echo "=== FIPS Rekey Integration Test ===" +echo "" +echo "Config: rekey.after_secs=$REKEY_AFTER_SECS" +echo "" + +# ── Phase 1: Pre-rekey baseline ─────────────────────────────────────── +echo "Phase 1: Pre-rekey connectivity (waiting ${CONVERGE_WAIT}s for convergence)" +sleep "$CONVERGE_WAIT" +ping_all +phase_result "Pre-rekey baseline (all 20 pairs)" +echo "" + +# ── Phase 2: Wait for first FMP rekey cycle ─────────────────────────── +echo "Phase 2: First rekey cycle (waiting ${FIRST_REKEY_WAIT}s for rekey)" +sleep "$FIRST_REKEY_WAIT" + +# Verify rekey events fired +PASSED=0 +FAILED=0 +echo " Checking FMP rekey events..." +assert_min_count "Rekey cutover complete (initiator), K-bit flipped" 1 "FMP rekey initiator cutovers" +phase_result "FMP rekey events" +echo "" + +# Verify connectivity after first rekey +echo "Phase 3: Post-rekey connectivity" +ping_all +phase_result "Post-first-rekey (all 20 pairs)" +echo "" + +# ── Phase 4: Wait for second rekey cycle ────────────────────────────── +echo "Phase 4: Second rekey cycle (waiting ${SECOND_REKEY_WAIT}s)" +sleep "$SECOND_REKEY_WAIT" + +# Verify connectivity after second rekey (back-to-back) +echo "Phase 5: Post-second-rekey connectivity" +ping_all +phase_result "Post-second-rekey (all 20 pairs)" +echo "" + +# ── Phase 6: Log analysis ───────────────────────────────────────────── +echo "Phase 6: Log analysis" +PASSED=0 +FAILED=0 + +# Positive checks: rekey machinery worked +assert_min_count "Rekey cutover complete (initiator), K-bit flipped" 4 \ + "FMP rekey initiator cutovers (>= 2 cycles)" + +# FSP rekey checks (sessions between non-adjacent nodes) +assert_min_count "FSP rekey cutover complete" 1 \ + "FSP session rekey initiator cutovers" +assert_min_count "Peer FSP K-bit flip detected" 1 \ + "FSP session rekey responder cutovers" + +# Negative checks: no bad things happened +assert_zero_count "PANIC\|panicked" "Panics" +assert_zero_count "ERROR" "Errors" +assert_zero_count "MMP link teardown" "Spurious link teardowns" +assert_zero_count "Excessive decrypt failures" \ + "Excessive decrypt failure removals" +assert_zero_count "Rekey msg2 processing failed" "Rekey msg2 failures" + +phase_result "Log analysis" +echo "" + +# ── Summary ──────────────────────────────────────────────────────────── +echo "=== Results: $TOTAL_PASSED passed, $TOTAL_FAILED failed ===" + +if [ "$TOTAL_FAILED" -eq 0 ]; then + exit 0 +else + # Dump logs on failure for diagnostics + echo "" + echo "=== Node logs (rekey-related) ===" + for node in $NODES; do + echo "--- node-$node ---" + docker logs "fips-node-$node" 2>&1 | \ + grep -E "(rekey|Rekey|cross|Cross|teardown|ERROR|PANIC|K-bit)" | \ + head -30 + echo "" + done + exit 1 +fi