diff --git a/packaging/debian/fips.service b/packaging/debian/fips.service index 77d237d..19c09ca 100644 --- a/packaging/debian/fips.service +++ b/packaging/debian/fips.service @@ -10,11 +10,10 @@ Restart=on-failure RestartSec=5 # Logging: RUST_LOG controls verbosity. -# Use "info" for production, "debug" for troubleshooting. -Environment=RUST_LOG=info +Environment=RUST_LOG=debug # Control socket directory (/run/fips/). -# Group-readable so 'fips' group members can use fipsctl/fipstop. +# Group-accessible so 'fips' group members can use fipsctl/fipstop. RuntimeDirectory=fips RuntimeDirectoryMode=0750 diff --git a/packaging/systemd/fips.service b/packaging/systemd/fips.service index 93bf08b..54d7862 100644 --- a/packaging/systemd/fips.service +++ b/packaging/systemd/fips.service @@ -10,11 +10,10 @@ Restart=on-failure RestartSec=5 # Logging: RUST_LOG controls verbosity. -# Use "info" for production, "debug" for troubleshooting. -Environment=RUST_LOG=info +Environment=RUST_LOG=debug # Control socket directory (/run/fips/). -# Group-readable so 'fips' group members can use fipsctl/fipstop. +# Group-accessible so 'fips' group members can use fipsctl/fipstop. RuntimeDirectory=fips RuntimeDirectoryMode=0750 diff --git a/src/control/mod.rs b/src/control/mod.rs index b09c11d..3112529 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -54,6 +54,16 @@ impl ControlSocket { } let listener = UnixListener::bind(&socket_path)?; + + // Make the socket and its parent directory group-accessible so + // 'fips' group members can use fipsctl/fipstop without root. + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&socket_path, std::fs::Permissions::from_mode(0o770))?; + Self::chown_to_fips_group(&socket_path); + if let Some(parent) = socket_path.parent() { + Self::chown_to_fips_group(parent); + } + info!(path = %socket_path.display(), "Control socket listening"); Ok(Self { @@ -85,6 +95,34 @@ impl ControlSocket { } } + /// Set group ownership of a path to the 'fips' group (best-effort). + fn chown_to_fips_group(path: &Path) { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt; + + // Look up the 'fips' group + let group_name = CString::new("fips").unwrap(); + let grp = unsafe { libc::getgrnam(group_name.as_ptr()) }; + if grp.is_null() { + debug!("'fips' group not found, skipping chown for {}", path.display()); + return; + } + let gid = unsafe { (*grp).gr_gid }; + + let c_path = match CString::new(path.as_os_str().as_bytes()) { + Ok(p) => p, + Err(_) => return, + }; + let ret = unsafe { libc::chown(c_path.as_ptr(), u32::MAX, gid) }; + if ret != 0 { + warn!( + path = %path.display(), + error = %std::io::Error::last_os_error(), + "Failed to chown control socket to 'fips' group" + ); + } + } + /// Run the accept loop, forwarding requests to the main event loop via mpsc. /// /// Each accepted connection is handled in a spawned task: diff --git a/src/mmp/metrics.rs b/src/mmp/metrics.rs index 3562c8b..b83eebf 100644 --- a/src/mmp/metrics.rs +++ b/src/mmp/metrics.rs @@ -89,7 +89,12 @@ impl MmpMetrics { /// /// `our_timestamp_ms` is the current session-relative time in ms (for RTT). /// `now` is the current monotonic time (for goodput rate computation). - pub fn process_receiver_report(&mut self, rr: &ReceiverReport, our_timestamp_ms: u32, now: Instant) { + /// + /// Returns `true` if this report produced the first SRTT measurement + /// (transition from uninitialized to initialized). + pub fn process_receiver_report(&mut self, rr: &ReceiverReport, our_timestamp_ms: u32, now: Instant) -> bool { + let had_srtt = self.srtt.initialized(); + // --- RTT from timestamp echo --- // RTT = now - echoed_timestamp - dwell_time if rr.timestamp_echo > 0 { @@ -159,6 +164,8 @@ impl MmpMetrics { self.prev_rr_ecn_ce = rr.ecn_ce_count; self.prev_rr_reorder = rr.cumulative_reorder_count; self.prev_rr_time = Some(now); + + !had_srtt && self.srtt.initialized() } /// Update the reverse delivery ratio (from our own receiver state about the peer's traffic). diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index 284c774..4f9f9d9 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -27,7 +27,7 @@ impl Node { } 0x02 => { // ReceiverReport - self.handle_receiver_report(from, payload); + self.handle_receiver_report(from, payload).await; } 0x10 => { // TreeAnnounce diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 219927a..e836247 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -5,7 +5,7 @@ use crate::node::Node; 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, info, warn}; +use tracing::{debug, info, trace, warn}; /// Force-remove a peer after this many consecutive decryption failures. const DECRYPT_FAILURE_THRESHOLD: u32 = 20; @@ -32,7 +32,7 @@ impl Node { let node_addr = match self.peers_by_index.get(&key) { Some(id) => *id, None => { - debug!( + trace!( receiver_idx = %header.receiver_idx, transport_id = %packet.transport_id, "Unknown session index, dropping" @@ -221,7 +221,13 @@ impl Node { consecutive_failures = count, "Excessive decryption failures, removing peer" ); + let addr = *node_addr; self.remove_active_peer(node_addr); + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + self.schedule_reconnect(addr, now_ms); } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 70f6a3f..37e6a7d 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -176,6 +176,11 @@ impl Node { "Peer restart detected (epoch mismatch), removing stale session" ); self.remove_active_peer(&peer_node_addr); + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + self.schedule_reconnect(peer_node_addr, now_ms); // Fall through to process as new connection } _ => { @@ -196,6 +201,55 @@ impl Node { && existing_peer.is_healthy() && session_age_secs >= 30 { + // Guard: already have a pending session from a completed + // rekey (waiting for K-bit cutover). Don't overwrite it + // with a new handshake — drop this msg1. + if existing_peer.pending_new_session().is_some() { + debug!( + peer = %self.peer_display_name(&peer_node_addr), + "Rekey msg1 received but already have pending session, dropping" + ); + self.connections.remove(&link_id); + self.links.remove(&link_id); + self.msg1_rate_limiter.complete_handshake(); + return; + } + + // Dual-initiation detection: both sides sent msg1 + // simultaneously. Apply tie-breaker — smaller NodeAddr + // wins as initiator (same as cross-connection resolution). + if existing_peer.rekey_in_progress() { + let our_addr = self.identity.node_addr(); + if our_addr < &peer_node_addr { + // We win as initiator — drop their msg1. + // Our msg2 will arrive at peer, who completes + // as our responder. + debug!( + peer = %self.peer_display_name(&peer_node_addr), + "Dual rekey initiation: we win (smaller addr), dropping their msg1" + ); + self.connections.remove(&link_id); + self.links.remove(&link_id); + self.msg1_rate_limiter.complete_handshake(); + return; + } + // We lose — abandon our rekey, become responder below. + debug!( + peer = %self.peer_display_name(&peer_node_addr), + "Dual rekey initiation: we lose (larger addr), abandoning ours" + ); + if let Some(peer) = self.peers.get_mut(&peer_node_addr) + && let Some(idx) = peer.abandon_rekey() + { + if let Some(tid) = peer.transport_id() { + self.peers_by_index.remove(&(tid, idx.as_u32())); + self.pending_outbound.remove(&(tid, idx.as_u32())); + } + let _ = self.index_allocator.free(idx); + } + // Fall through to respond as responder + } + // Rekey: process as responder, store new session as pending let noise_session = conn.take_session(); let our_new_index = match self.index_allocator.allocate() { diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 328f62e..a9b7ec4 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -72,7 +72,7 @@ impl Node { /// /// The peer is telling us about what they received from us. We feed /// this to our metrics to compute RTT, loss rate, and trend indicators. - pub(in crate::node) fn handle_receiver_report(&mut self, from: &NodeAddr, payload: &[u8]) { + pub(in crate::node) async fn handle_receiver_report(&mut self, from: &NodeAddr, payload: &[u8]) { let rr = match ReceiverReport::decode(payload) { Ok(rr) => rr, Err(e) => { @@ -101,7 +101,7 @@ impl Node { // Process the report: computes RTT from timestamp echo, updates // loss rate, goodput rate, jitter trend, and ETX. let now = Instant::now(); - mmp.metrics.process_receiver_report(&rr, our_timestamp_ms, now); + let first_rtt = mmp.metrics.process_receiver_report(&rr, our_timestamp_ms, now); // Feed SRTT back to sender/receiver report interval tuning if let Some(srtt_ms) = mmp.metrics.srtt_ms() { @@ -126,6 +126,47 @@ impl Node { etx = format_args!("{:.2}", mmp.metrics.etx), "Processed ReceiverReport" ); + + // First RTT sample — peer is now eligible for parent selection. + // Trigger re-evaluation so the node doesn't wait for the next + // periodic tick or TreeAnnounce. + if first_rtt { + let peer_costs: std::collections::HashMap = self.peers.iter() + .filter(|(_, p)| p.has_srtt()) + .map(|(a, p)| (*a, p.link_cost())) + .collect(); + if let Some(new_parent) = self.tree_state.evaluate_parent(&peer_costs) { + let new_seq = self.tree_state.my_declaration().sequence() + 1; + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let flap_dampened = self.tree_state.set_parent(new_parent, new_seq, timestamp); + if let Err(e) = self.tree_state.sign_declaration(&self.identity) { + warn!(error = %e, "Failed to sign declaration after first-RTT parent eval"); + return; + } + self.tree_state.recompute_coords(); + self.coord_cache.clear(); + self.stats_mut().tree.parent_switched += 1; + self.stats_mut().tree.parent_switches += 1; + info!( + new_parent = %self.peer_display_name(&new_parent), + new_seq = new_seq, + new_root = %self.tree_state.root(), + depth = self.tree_state.my_coords().depth(), + trigger = "first-rtt", + "Parent switched after first RTT measurement" + ); + if flap_dampened { + self.stats_mut().tree.flap_dampened += 1; + warn!("Flap dampening engaged: excessive parent switches detected"); + } + self.send_tree_announce_to_all().await; + let all_peers: Vec = self.peers.keys().copied().collect(); + self.bloom_state.mark_all_updates_needed(all_peers); + } + } } /// Check all peers for pending MMP reports and send them. diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 52d1e95..ecf06e5 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -410,7 +410,37 @@ impl Node { } 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() { + if self.config.node.rekey.enabled { + let rekey_in_progress = existing.has_rekey_in_progress(); + let has_pending = existing.pending_new_session().is_some(); + + // Dual-initiation detection: both sides sent SessionSetup + // simultaneously. Apply tie-breaker — smaller NodeAddr + // wins as initiator (same as initial session setup). + if rekey_in_progress { + if self.identity.node_addr() < src_addr { + // We win as initiator — drop their msg1. + debug!( + src = %self.peer_display_name(src_addr), + "Dual FSP rekey initiation: we win (smaller addr), dropping their msg1" + ); + return; + } + // We lose — abandon our rekey, become responder below. + debug!( + src = %self.peer_display_name(src_addr), + "Dual FSP rekey initiation: we lose (larger addr), abandoning ours" + ); + let entry = self.sessions.get_mut(src_addr).unwrap(); + entry.abandon_rekey(); + } else if has_pending { + // Guard: already have a pending session waiting for K-bit cutover + debug!( + src = %self.peer_display_name(src_addr), + "FSP rekey msg1 received but already have pending session, dropping" + ); + return; + } let our_keypair = self.identity.keypair(); let mut handshake = HandshakeState::new_xk_responder(our_keypair); handshake.set_local_epoch(self.startup_epoch); diff --git a/src/node/tree.rs b/src/node/tree.rs index 7605449..e4850c3 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -211,8 +211,11 @@ impl Node { self.bloom_state.mark_update_needed(*from); } - // Re-evaluate parent selection with current link costs + // Re-evaluate parent selection with current link costs. + // Exclude peers without MMP RTT data — they are not yet eligible + // as parent candidates (prevents oscillation from optimistic defaults). let peer_costs: HashMap = self.peers.iter() + .filter(|(_, peer)| peer.has_srtt()) .map(|(addr, peer)| (*addr, peer.link_cost())) .collect(); if let Some(new_parent) = self.tree_state.evaluate_parent(&peer_costs) { @@ -263,6 +266,7 @@ impl Node { "Parent ancestry contains us — loop detected, dropping parent" ); let peer_costs: HashMap = self.peers.iter() + .filter(|(_, peer)| peer.has_srtt()) .map(|(addr, peer)| (*addr, peer.link_cost())) .collect(); if self.tree_state.handle_parent_lost(&peer_costs) { @@ -354,6 +358,7 @@ impl Node { self.last_parent_reeval = Some(now); let peer_costs: HashMap = self.peers.iter() + .filter(|(_, peer)| peer.has_srtt()) .map(|(addr, peer)| (*addr, peer.link_cost())) .collect(); @@ -410,6 +415,7 @@ impl Node { if was_parent { self.stats_mut().tree.parent_losses += 1; let peer_costs: HashMap = self.peers.iter() + .filter(|(_, peer)| peer.has_srtt()) .map(|(addr, peer)| (*addr, peer.link_cost())) .collect(); let changed = self.tree_state.handle_parent_lost(&peer_costs); diff --git a/src/peer/active.rs b/src/peer/active.rs index a99622c..b8ee9ea 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -593,6 +593,12 @@ impl ActivePeer { } } + /// Whether this peer has at least one MMP RTT measurement. + pub fn has_srtt(&self) -> bool { + self.mmp() + .is_some_and(|mmp| mmp.metrics.srtt_ms().is_some()) + } + /// When this peer was authenticated. pub fn authenticated_at(&self) -> u64 { self.authenticated_at diff --git a/src/tree/state.rs b/src/tree/state.rs index 4402b76..dceb4bb 100644 --- a/src/tree/state.rs +++ b/src/tree/state.rs @@ -340,7 +340,15 @@ impl TreeState { if coords.contains(&self.my_node_addr) { continue; } - let cost = peer_costs.get(peer_id).copied().unwrap_or(1.0); + // If any peer has MMP cost data, only consider measured peers. + // This prevents freshly connected peers (no SRTT, default cost 1.0) + // from appearing artificially cheap. During cold start (no peer has + // MMP data, peer_costs is empty), fall back to default cost 1.0. + let cost = match peer_costs.get(peer_id) { + Some(&c) => c, + None if peer_costs.is_empty() => 1.0, + None => continue, + }; let eff_depth = coords.depth() as f64 + cost; match &best_peer { None => best_peer = Some((*peer_id, eff_depth)), @@ -396,11 +404,14 @@ impl TreeState { // --- Same root, cost-aware comparison with hysteresis --- - // Current parent's effective_depth + // Current parent's effective_depth. + // If peer_costs is non-empty but current parent has no entry, + // treat as maximally expensive so any measured candidate can win. + // If peer_costs is empty (cold start), use default cost 1.0. let current_parent_cost = peer_costs .get(self.my_declaration.parent_id()) .copied() - .unwrap_or(1.0); + .unwrap_or(if peer_costs.is_empty() { 1.0 } else { f64::INFINITY }); let current_parent_coords = self.peer_ancestry.get(self.my_declaration.parent_id()); let current_parent_eff = match current_parent_coords { Some(coords) => coords.depth() as f64 + current_parent_cost,