diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 20f7161..2e3a12f 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -50,7 +50,19 @@ impl Node { 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. + // + // The header K-bit is NOT a sufficient gating event on its own. + // Under jitter the FMP rekey interval shrinks and the two + // directions' rekeys interleave, so a node can hold a `pending` + // session from rekey N while the peer's observed K-bit flip + // actually belongs to rekey N+1. Promoting on the bare bit then + // installs the WRONG Noise session as current — the two endpoints + // diverge, every subsequent frame fails AEAD on the far side, the + // receiver starves, and the link is declared dead at the 30s + // heartbeat timeout (Phase-5 routing failure, green crypto). This + // mirrors the FSP fix (node/session.rs): the authenticated decrypt, + // not the header bit, is the cutover signal. Trial-decrypt the + // frame against `pending` first; only promote if it authenticates. { let Some(peer) = self.peers.get(&node_addr) else { return; @@ -59,45 +71,84 @@ impl Node { received_k_bit != peer.current_k_bit() && peer.pending_new_session().is_some(); if k_bit_flipped { + let ciphertext = &packet.data[header.ciphertext_offset()..]; let display_name = self.peer_display_name(&node_addr); - let pending_our = peer.pending_our_index(); - let pending_their = peer.pending_their_index(); - info!( - peer = %display_name, - our_addr = %self.identity().node_addr(), - their_addr = %node_addr, - pending_our_index = ?pending_our, - pending_their_index = ?pending_their, - "Peer K-bit flip detected, promoting new session" - ); - + let our_addr = *self.identity().node_addr(); let Some(peer) = self.peers.get_mut(&node_addr) else { return; }; - let did_flip = peer.handle_peer_kbit_flip().is_some(); - if did_flip { - // New index was pre-registered in peers_by_index during - // msg1 handling (handshake.rs). Verify, don't duplicate. - debug_assert!( - peer.transport_id().is_some() - && peer.our_index().is_some() - && self.peers_by_index.contains_key(&( - peer.transport_id().unwrap(), - peer.our_index().unwrap().as_u32() - )), - "peers_by_index should contain pre-registered new index after K-bit flip" + // Authenticate the frame against the pending session. + let pending_plaintext = peer.pending_new_session_mut().and_then(|pending| { + pending + .decrypt_with_replay_check_and_aad( + ciphertext, + header.counter, + &header.header_bytes, + ) + .ok() + }); + + if let Some(plaintext) = pending_plaintext { + let pending_our = peer.pending_our_index(); + let pending_their = peer.pending_their_index(); + info!( + peer = %display_name, + our_addr = %our_addr, + their_addr = %node_addr, + pending_our_index = ?pending_our, + pending_their_index = ?pending_their, + "Peer new-epoch frame authenticated, K-bit flip promoting new session" ); + // The peer authenticated a frame on the new epoch, so it + // derived the new session (it received our rekey msg3). + // If we are the rekey initiator still retransmitting msg3, + // stop — the responder is confirmed. (No-op for the + // responder side, which never retained a msg3 payload.) + peer.clear_rekey_msg3_payload(); + // The trial-decrypt already advanced the pending + // session's replay window; handle_peer_kbit_flip moves + // that same session object to current, so no re-decrypt. + let did_flip = peer.handle_peer_kbit_flip().is_some(); + if did_flip { + debug_assert!( + peer.transport_id().is_some() + && peer.our_index().is_some() + && self.peers_by_index.contains_key(&( + peer.transport_id().unwrap(), + peer.our_index().unwrap().as_u32() + )), + "peers_by_index should contain pre-registered new index after K-bit flip" + ); + } + // Re-register the promoted session with the decrypt + // worker (cache_key changed at the flip). + #[cfg(unix)] + if did_flip { + self.register_decrypt_worker_session(&node_addr); + } + + // Deliver the frame we just authenticated via the + // canonical post-decrypt path, then return — it must + // not fall through to a second decrypt attempt. + let ce_flag = header.flags & FLAG_CE != 0; + self.process_authentic_fmp_plaintext( + &node_addr, + packet.transport_id, + &packet.remote_addr, + packet.timestamp_ms, + packet.data.len(), + header.counter, + ce_flag, + &plaintext, + ) + .await; + return; } - // Re-register the (now-promoted) session with the decrypt - // worker: cache_key = (transport_id, our_index) changed at - // the flip, so the old worker entry is stranded and every - // packet on the new session would miss the worker's - // HashMap lookup. Without this, throughput drops back to - // the inline-decrypt path after each rekey. - #[cfg(unix)] - if did_flip { - self.register_decrypt_worker_session(&node_addr); - } + // Pending did NOT authenticate this frame: the flip belongs + // to a different rekey epoch (stale pending). Do not + // promote. Fall through to the normal current/previous + // decrypt; the genuine cutover is recognized when a frame + // that authenticates against `pending` arrives. } } @@ -210,6 +261,13 @@ impl Node { let ce_flag = header.flags & FLAG_CE != 0; if let Some(peer) = self.peers.get_mut(&node_addr) { + // Initiator-side msg3 confirm (see process_authentic_fmp_plaintext): + // a frame authenticated against post-cutover `current` (no pending) + // proves the responder reached the new epoch. Inline-decrypt path + // mirror of the worker-bounce confirm. + if peer.rekey_msg3_payload().is_some() && peer.pending_new_session().is_none() { + peer.clear_rekey_msg3_payload(); + } if let Some(mmp) = peer.mmp_mut() { mmp.receiver.record_recv( header.counter, @@ -303,6 +361,18 @@ impl Node { let mut address_changed = false; if let Some(peer) = self.peers.get_mut(node_addr) { peer.reset_decrypt_failures(); + // If we are the rekey initiator that already cut over on its + // own timer (no `pending`, `current` is the new session) but + // still retain a msg3 retransmission payload, an authenticated + // peer frame here decrypts against the post-cutover `current` + // session — proof the responder reached the new epoch. Stop + // retransmitting. Mirrors the FSP Current-slot confirm in + // handle_encrypted_session_msg. Works in both the inline and + // worker-bounce paths since both funnel through here, and the + // only session registered for the new index is the new one. + if peer.rekey_msg3_payload().is_some() && peer.pending_new_session().is_none() { + peer.clear_rekey_msg3_payload(); + } address_changed = peer.set_current_addr(transport_id, remote_addr.clone()); peer.link_stats_mut() .record_recv(packet_len, packet_timestamp_ms); diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index b7448dc..7b7dd60 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -369,6 +369,9 @@ impl Node { .peers .get(&peer_node_addr) .and_then(|p| p.current_addr().cloned()); + let msg3_resend_interval = + self.config().node.rate_limit.handshake_resend_interval_ms; + let msg3_now_ms = Self::now_ms(); if let Some(peer) = self.peers.get_mut(&peer_node_addr) { match peer.complete_rekey_msg2(noise_msg2) { @@ -415,6 +418,19 @@ impl Node { if msg3_sent { peer.set_pending_session(session, our_index, header.sender_idx); + // Retain msg3 for retransmission until the + // responder is confirmed on the new epoch. + // FMP sends msg3 exactly once otherwise; a + // lost datagram leaves the responder without + // the new session, so when the initiator cuts + // over its new-epoch frames silently miss at + // the peer → 30s link-dead. Mirrors FSP's + // resend_pending_session_msg3 liveness path. + peer.set_rekey_msg3_payload( + wire_msg3.clone(), + msg3_now_ms + msg3_resend_interval, + ); + if let Some(tid) = transport_id { self.peers_by_index .insert((tid, our_index.as_u32()), peer_node_addr); @@ -1087,6 +1103,24 @@ impl Node { let session_age_secs = existing_peer.session_established_at().elapsed().as_secs(); + // The minimum plausible age for an inbound XX handshake to + // be a scheduled REKEY rather than an initial-handshake + // cross-connection. Derived from the configured interval + // and jitter so it tracks the real minimum rekey spacing + // (see the rekey-responder gate below, which uses the same + // value). A session at least this old that receives an + // inbound msg3 on a different link is a rekey, NOT an + // initial cross-connection. + let rekey_age_floor_secs = { + let min_interval = self + .config() + .node + .rekey + .after_secs + .saturating_sub(crate::node::REKEY_JITTER_SECS.unsigned_abs()); + min_interval.saturating_sub(5).max(5) + }; + // Simultaneous-init cross-connection (msg2-then-msg3 ordering). // // When both sides initiate XX in parallel (typical in @@ -1100,7 +1134,26 @@ impl Node { // already promoted at the same epoch — apply the same // tie-breaker handle_msg2 uses for the inverse ordering, so // both sides converge on a single Noise session pair. - if existing_peer.link_id() != link_id && session_age_secs < 30 { + // + // CRITICAL (jitter × XX rekey): the upper age bound MUST sit + // below the rekey floor. An initial cross-connection always + // resolves within ~1 RTT of promotion (both handshakes race + // in the same sub-second burst); a session old enough to be + // rekeying that receives a concurrent rekey msg3 must NOT be + // routed here. The old fixed `< 30` bound overlapped the + // jittered rekey floor (as low as 15s): under jitter a recent + // cutover resets `session_established_at`, so a concurrent + // rekey msg3 (always on a different temp link_id) landed in + // this branch and, on the "our outbound wins" side, the + // peer's rekey session was DISCARDED (index freed, no pending + // slot) — yet the peer cut over to it regardless, leaving the + // discarding node unable to decrypt the peer until the 30s + // dead-timer fired (Phase-5 link death, green crypto). Gating + // on the rekey floor makes any rekey-aged msg3 fall through to + // the rekey-responder path below, which converges both sides + // (dual-init tie-break) AND installs a `pending` slot. + if existing_peer.link_id() != link_id && session_age_secs < rekey_age_floor_secs + { let our_inbound_wins = cross_connection_winner( self.identity().node_addr(), &peer_node_addr, @@ -1169,11 +1222,26 @@ impl Node { return; } - // Check for rekey: session must be at least 30s old. + // Check for rekey: session must be old enough that an + // inbound XX handshake is plausibly a scheduled rekey + // rather than a fresh duplicate/restart. + // + // The floor (`rekey_age_floor_secs`, computed above) sits + // BELOW the minimum possible rekey interval, or jittered + // rekeys are wrongly rejected. The initiator's effective + // interval is `after_secs - REKEY_JITTER_SECS` at its lowest + // (20s for 35s ± 15s). A fixed 30s floor exceeds that 20s + // minimum, so under jitter the responder rejects a legitimate + // rekey as a "duplicate handshake" (resends msg2), the + // initiator cuts over anyway, and the endpoints diverge by + // one epoch → receiver starves → 30s link-dead. The same + // floor also bounds the cross-connection branch above, so the + // two paths partition cleanly: `< floor` → initial + // cross-connection, `>= floor` → rekey responder. if self.config().node.rekey.enabled && existing_peer.has_session() && existing_peer.is_healthy() - && session_age_secs >= 30 + && session_age_secs >= rekey_age_floor_secs { // Dual-initiation detection: both sides initiated rekey // simultaneously. Two states can reach this point: diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index d8b02bc..09d5ab4 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -71,6 +71,19 @@ impl Node { if peer.rekey_in_progress() { continue; } + if peer.pending_new_session().is_some() { + // Completed rekey awaiting cutover; don't stack another. + continue; + } + if peer.rekey_msg3_payload().is_some() { + // Initiator already cut over on its timer but is still + // retransmitting msg3 to a responder not yet confirmed on + // the new epoch. Don't start another rekey (which would + // overwrite the retained payload) until this cycle's msg3 + // is delivered or its budget exhausted. Mirrors FSP + // check_session_rekey. + continue; + } if peer.is_rekey_dampened(REKEY_DAMPENING_SECS) { continue; } @@ -311,6 +324,93 @@ impl Node { } } + /// Retransmit FMP rekey msg3 until the responder is confirmed on the + /// new epoch. + /// + /// FMP sends the rekey msg3 exactly once (in `handle_msg2`). If that + /// single datagram is lost, the responder never derives the new + /// session: when the initiator later cuts over on its own timer, its + /// new-epoch frames land on a `peers_by_index` index the responder + /// never built a session for, so they silently miss and the link dies + /// at the 30s heartbeat timeout. This driver retransmits the retained + /// msg3 over the existing link until the responder confirms. + /// + /// Liveness only — overlapping-epoch / pending-authenticated decrypt + /// covers cutover skew. The retained payload is cleared (confirmed) + /// when an inbound peer frame authenticates against `pending` (peer cut + /// over first, in `handle_encrypted_frame`) or against the post-cutover + /// `current` session (initiator already cut over, responder reached the + /// new epoch, in `process_authentic_fmp_plaintext`). After + /// `handshake_max_resends` with no confirmation the cycle is abandoned. + pub(in crate::node) async fn resend_pending_fmp_rekey_msg3(&mut self, now_ms: u64) { + if !self.config().node.rekey.enabled { + return; + } + + let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; + let backoff = self.config().node.rate_limit.handshake_resend_backoff; + let max_resends = self.config().node.rate_limit.handshake_max_resends; + + let mut to_resend: Vec<(NodeAddr, Vec)> = Vec::new(); + let mut to_abandon: Vec = Vec::new(); + + for (node_addr, peer) in &self.peers { + let payload = match peer.rekey_msg3_payload() { + Some(p) => p, + None => continue, + }; + if peer.rekey_msg3_next_resend_ms() == 0 || now_ms < peer.rekey_msg3_next_resend_ms() { + continue; + } + if peer.rekey_msg3_resend_count() >= max_resends { + to_abandon.push(*node_addr); + continue; + } + to_resend.push((*node_addr, payload.to_vec())); + } + + // Clear retained payload on cycles that exhausted their budget. + // The new session may still cut over via the normal path; the + // responder will recover on its own next rekey if it never got + // msg3. Stopping retransmission just bounds the effort. + for node_addr in to_abandon { + if let Some(peer) = self.peers.get_mut(&node_addr) { + peer.clear_rekey_msg3_payload(); + } + trace!( + peer = %self.peer_display_name(&node_addr), + "FMP rekey msg3 unconfirmed after max retransmissions, stopping resend" + ); + } + + for (node_addr, payload) 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, &payload).await.is_ok() + } else { + false + }; + + if sent && let Some(peer) = self.peers.get_mut(&node_addr) { + let count = peer.rekey_msg3_resend_count() + 1; + let next = now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64; + peer.record_rekey_msg3_resend(next); + trace!( + peer = %self.peer_display_name(&node_addr), + resend = count, + "Resent FMP rekey msg3" + ); + } + } + } + /// Retransmit FSP rekey msg3 until the responder is confirmed on the /// new epoch. /// diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 9f78543..1bc3875 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -265,6 +265,7 @@ impl Node { self.poll_lan_discovery().await; self.resend_pending_handshakes(now_ms).await; self.resend_pending_rekeys(now_ms).await; + self.resend_pending_fmp_rekey_msg3(now_ms).await; self.resend_pending_session_handshakes(now_ms).await; self.resend_pending_session_msg3(now_ms).await; self.purge_idle_sessions(now_ms); diff --git a/src/node/mod.rs b/src/node/mod.rs index e62b1fc..180745c 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -40,14 +40,19 @@ use self::routing_error_rate_limit::RoutingErrorRateLimiter; /// dual-initiation in symmetric-start meshes; the configured /// `node.rekey.after_secs` remains the nominal interval (mean preserved). /// -/// Disabled (set to 0) on next pending investigation: the jitter mechanism -/// was authored against the IK FMP rekey path on maint/master and works -/// cleanly there, but on next's XX FMP rekey path it produces reproducible -/// post-cutover routing-convergence failures (~50% Phase 5 ping loss in -/// the `rekey` integration suite). Restoring jitter on next requires -/// understanding why the XX cutover state cleanup doesn't absorb -/// variable-interval rekeys the way the IK path does. See CHANGELOG. -pub(crate) const REKEY_JITTER_SECS: i64 = 0; +/// Re-enabled on next after the jitter × XX rekey interaction was closed. +/// The jitter was previously disabled (set to 0) on next because applying it +/// to the XX FMP rekey path produced reproducible post-cutover +/// routing-convergence failures (~50% Phase 5 ping loss in the `rekey` +/// integration suite). Root cause: under jitter a recent cutover resets the +/// session-age clock, so a concurrent rekey msg3 (always on a temp link) was +/// caught by the initial-handshake cross-connection tie-breaker in +/// `handle_msg3` and, on the "our outbound wins" side, the peer's rekey +/// session was discarded with no `pending` slot — yet the peer cut over to it +/// regardless, starving the discarding node until the 30s dead-timer. Fixed by +/// bounding that cross-connection branch below the rekey age floor so a +/// rekey-aged msg3 falls through to the rekey-responder path. See CHANGELOG. +pub(crate) const REKEY_JITTER_SECS: i64 = 15; use self::wire::{ ESTABLISHED_HEADER_SIZE, FLAG_CE, FLAG_KEY_EPOCH, build_encrypted, build_established_header, prepend_inner_header, diff --git a/src/peer/active.rs b/src/peer/active.rs index a9a15f7..e4f68a8 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -215,6 +215,19 @@ pub struct ActivePeer { /// In-progress rekey responder: our new session index. rekey_responder_our_index: Option, + // === Rekey msg3 retransmission (initiator liveness) === + /// Retained wire-format rekey msg3, resent until the responder is + /// confirmed on the new epoch. Mirrors the FSP + /// `rekey_msg3_payload` mechanism (node/session.rs). Liveness only: + /// overlapping-epoch decrypt covers cutover skew; retransmission + /// guarantees the responder eventually derives the new session even + /// if the first msg3 datagram is lost. + rekey_msg3_payload: Option>, + /// Next msg3 resend timestamp (Unix ms; 0 = none retained). + rekey_msg3_next_resend_ms: u64, + /// Number of msg3 retransmissions performed this rekey cycle. + rekey_msg3_resend_count: u32, + /// Unix UDP fast-path: per-peer `connect()`-ed socket (paired with /// the listen socket via `SO_REUSEPORT`). The kernel demux prefers /// the connected 5-tuple, so inbound packets land here; the @@ -288,6 +301,9 @@ impl ActivePeer { rekey_msg1_next_resend: 0, rekey_responder_handshake: None, rekey_responder_our_index: None, + rekey_msg3_payload: None, + rekey_msg3_next_resend_ms: 0, + rekey_msg3_resend_count: 0, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_udp: None, #[cfg(any(target_os = "linux", target_os = "macos"))] @@ -386,6 +402,9 @@ impl ActivePeer { rekey_msg1_next_resend: 0, rekey_responder_handshake: None, rekey_responder_our_index: None, + rekey_msg3_payload: None, + rekey_msg3_next_resend_ms: 0, + rekey_msg3_resend_count: 0, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_udp: None, #[cfg(any(target_os = "linux", target_os = "macos"))] @@ -998,6 +1017,12 @@ impl ActivePeer { self.pending_new_session.as_ref() } + /// Mutable access to the pending new session, for trial-decrypt of an + /// inbound frame before promoting it on a peer K-bit flip. + pub fn pending_new_session_mut(&mut self) -> Option<&mut NoiseSession> { + self.pending_new_session.as_mut() + } + /// Store a completed rekey session and its indices. /// /// Called when the rekey handshake completes. The session is held @@ -1125,6 +1150,7 @@ impl ActivePeer { self.rekey_msg1 = None; self.rekey_msg1_next_resend = 0; self.rekey_in_progress = false; + self.clear_rekey_msg3_payload(); // Return whichever index needs freeing self.rekey_our_index.take().or_else(|| { self.pending_new_session = None; @@ -1240,6 +1266,47 @@ impl ActivePeer { Ok(session) } + // === Rekey msg3 retransmission (initiator liveness) === + + /// Retain the rekey msg3 wire payload for retransmission until the + /// responder is confirmed on the new epoch. Called by the initiator + /// right after the first successful msg3 send. Mirrors the FSP + /// `SessionEntry::set_rekey_msg3_payload`. + pub fn set_rekey_msg3_payload(&mut self, payload: Vec, next_resend_at_ms: u64) { + self.rekey_msg3_payload = Some(payload); + self.rekey_msg3_next_resend_ms = next_resend_at_ms; + self.rekey_msg3_resend_count = 0; + } + + /// Get the retained rekey msg3 payload for retransmission. + pub fn rekey_msg3_payload(&self) -> Option<&[u8]> { + self.rekey_msg3_payload.as_deref() + } + + /// Get the next msg3 resend timestamp (Unix ms; 0 = none retained). + pub fn rekey_msg3_next_resend_ms(&self) -> u64 { + self.rekey_msg3_next_resend_ms + } + + /// Get the number of msg3 retransmissions this cycle. + pub fn rekey_msg3_resend_count(&self) -> u32 { + self.rekey_msg3_resend_count + } + + /// Record a msg3 retransmission and schedule the next one. + pub fn record_rekey_msg3_resend(&mut self, next_resend_at_ms: u64) { + self.rekey_msg3_resend_count += 1; + self.rekey_msg3_next_resend_ms = next_resend_at_ms; + } + + /// Clear the retained rekey msg3 payload (responder confirmed on the + /// new epoch, or the rekey cycle was abandoned). + pub fn clear_rekey_msg3_payload(&mut self) { + self.rekey_msg3_payload = None; + self.rekey_msg3_next_resend_ms = 0; + self.rekey_msg3_resend_count = 0; + } + /// 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