laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add periodic Noise rekey with fresh DH for forward secrecy (FMP + FSP)

Browse Source 
Implement periodic full rekey at both protocol layers using fresh DH
key exchanges. Uses the existing K-bit flag (FLAG_KEY_EPOCH /
FSP_FLAG_K) to coordinate cutover between peers.

FMP layer (IK pattern):
- ActivePeer gains rekey state: pending/previous sessions, K-bit epoch
  tracking, drain window, dampening timer
- Handshake state stored on ActivePeer with msg1 sent on existing link
- Encrypted frame handler detects K-bit flips, promotes pending
  sessions, falls back to previous session during drain
- Handshake handlers distinguish rekey from new connections using
  addr_to_link lookup with identity-based fallback
- Free all session indices (current, rekey, pending, previous) on
  peer removal

FSP layer (XK pattern):
- SessionEntry gains parallel rekey fields with XK-specific state
  for the 3-message handshake
- Route availability check before FSP rekey initiation
- Encrypted session handler adds K-bit flip detection and dual-session
  decrypt fallback
- SessionSetup/Ack/Msg3 handlers extended for rekey paths

Defense-in-depth:
- Consecutive decryption failure detector (threshold=20) triggers
  forced peer removal instead of waiting for link-dead timeout
- Identity-based rekey detection as fallback when addr_to_link
  doesn't match (e.g., TCP ephemeral ports)

Configuration: RekeyConfig with enabled flag, after_secs (default 120),
and after_messages (default 65536) thresholds.

Logging: info for successful K-bit cutover completions, warn for
failures, debug for intermediate handshake steps, trace for routine
operations (resends, drain cleanup).

Rekey lifecycle:
1. Timer/counter fires -> initiator starts new handshake
2. Old session continues handling traffic during handshake
3. Handshake completes -> initiator cuts over, flips K-bit
4. Responder sees flipped K-bit -> promotes new session
5. Both keep old session for 10s drain window
6. After drain, old session discarded

Integration test: Docker-based multi-phase test exercising both FMP
and FSP rekey with aggressive timers (35s). Verifies connectivity
across all 20 directed pairs survives two consecutive rekey cycles.
Includes rekey topology, docker-compose profile, and CI matrix entry.

Increase ping test convergence wait from 3s to 5s for CI reliability.
This commit is contained in:
Johnathan Corgan
2026-03-07 18:33:27 +00:00
parent 392572f821
commit bf117df0ca
17 changed files with 1927 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/dispatch.rs
+20 -4
View File
@@ -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
src/node/handlers/encrypted.rs
+161 -87
View File
@@ -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);
}
}
}
}
src/node/handlers/handshake.rs
+175 -14
View File
@@ -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..];
src/node/handlers/mod.rs
+1
View File
@@ -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;
src/node/handlers/rekey.rs
+413
View File
@@ -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<NodeAddr> = Vec::new();
let mut peers_to_drain: Vec<NodeAddr> = Vec::new();
let mut peers_to_rekey: Vec<NodeAddr> = 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<u8>)> = 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<NodeAddr> = Vec::new();
let mut sessions_to_drain: Vec<NodeAddr> = Vec::new();
let mut sessions_to_rekey: Vec<NodeAddr> = 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"
);
}
}
src/node/handlers/rx_loop.rs
+3
View File
@@ -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();
src/node/handlers/session.rs
+210 -13
View File
@@ -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();
src/node/mod.rs
+4 -1
View File
@@ -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,
src/node/session.rs
+163
View File
@@ -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<NoiseSession>,
/// When drain window started (Unix ms). 0 = no drain.
drain_started_ms: u64,
/// In-progress rekey state (runs alongside Established session).
rekey_state: Option<HandshakeState>,
/// Pending completed session awaiting K-bit cutover.
pending_new_session: Option<NoiseSession>,
/// 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<HandshakeState> {
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;
}
}