From 10122d7d87a87bbc7dd34f19f87948e42e3f474e Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Apr 2026 13:14:17 +0000 Subject: [PATCH] Production hardening: unwrap safety, 14 new tests, diagnostics Harden unwrap() calls in handler hot paths (handshake, encrypted, rekey, session handlers) with proper error propagation. Add 14 tests: profile rejection (6), MMP forward-compat (4), discovery min_mtu pruning (3), XX duplicate msg1 dedup (1). Add NodeProfile Display impl with structured tracing for profile mismatch logging. Expose bloom compression diagnostics (total_compressed_bytes, total_raw_bytes) in control socket. Add module documentation for XX identity timing, profile decision tree, and bloom codec strategy. --- src/bloom/codec.rs | 17 +++++ src/mmp/report.rs | 52 ++++++++++++++ src/node/bloom.rs | 2 + src/node/handlers/encrypted.rs | 12 +++- src/node/handlers/handshake.rs | 30 +++++--- src/node/handlers/rekey.rs | 6 +- src/node/handlers/session.rs | 22 ++++-- src/node/stats.rs | 7 ++ src/node/tests/discovery.rs | 128 ++++++++++++++++++++++++++++++++- src/node/tests/handshake.rs | 119 ++++++++++++++++++++++++++++++ src/noise/mod.rs | 20 ++++++ src/protocol/negotiation.rs | 31 +++++++- 12 files changed, 424 insertions(+), 22 deletions(-) diff --git a/src/bloom/codec.rs b/src/bloom/codec.rs index fefd5b9..bf31533 100644 --- a/src/bloom/codec.rs +++ b/src/bloom/codec.rs @@ -3,6 +3,23 @@ //! Encodes a sequence of `u64` words using run-length encoding. //! Each run is encoded as `[count:2 LE][word:8 LE]` (10 bytes per run). //! Sparse data (XOR diffs with mostly zero words) compresses well. +//! +//! ## Delta vs Full Strategy +//! +//! The sender tracks `last_sent_filter` per peer. When a new filter is +//! ready, the sender XORs it with the last-sent filter to produce a diff. +//! The diff is mostly zero words (only changed bits set), which RLE +//! compresses efficiently. If no previous filter exists (first send, +//! size class change, or NACK recovery), a full filter is sent instead. +//! +//! The same RLE codec handles both cases — full filters at ~25% fill +//! still benefit from zero-word runs between set regions. +//! +//! ## NACK Recovery +//! +//! If the receiver detects a sequence gap (missed delta), it sends a +//! NACK. The sender responds with a full filter, resetting the delta +//! baseline for that peer. /// Statistics from a compression operation. #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/src/mmp/report.rs b/src/mmp/report.rs index a2b32b8..e44aeff 100644 --- a/src/mmp/report.rs +++ b/src/mmp/report.rs @@ -408,4 +408,56 @@ mod tests { let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); assert_eq!(rr, decoded); } + + #[test] + fn test_sender_report_v1_parsed_by_v0_decoder() { + let sr = sample_sender_report(); + let mut encoded = sr.encode(); + // Set format_version = 1 + encoded[1] = 1; + // Extend with hypothetical v1 fields (8 extra bytes) + let new_total_len = SENDER_REPORT_PAYLOAD + 8; + encoded[2..4].copy_from_slice(&new_total_len.to_le_bytes()); + encoded.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF, 0xCA, 0xFE, 0xBA, 0xBE]); + // v0 decoder parses known fields correctly + let decoded = SenderReport::decode(&encoded[1..]).unwrap(); + assert_eq!(sr, decoded); + } + + #[test] + fn test_receiver_report_v1_parsed_by_v0_decoder() { + let rr = sample_receiver_report(); + let mut encoded = rr.encode(); + // Set format_version = 1 + encoded[1] = 1; + // Extend with hypothetical v1 fields (12 extra bytes) + let new_total_len = RECEIVER_REPORT_PAYLOAD + 12; + encoded[2..4].copy_from_slice(&new_total_len.to_le_bytes()); + encoded.extend_from_slice(&[0xAB; 12]); + // v0 decoder parses known fields correctly + let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); + assert_eq!(rr, decoded); + } + + #[test] + fn test_sender_report_v1_total_length_too_short() { + let sr = sample_sender_report(); + let mut encoded = sr.encode(); + // Set format_version = 1 but total_length < v0 payload size + encoded[1] = 1; + let short_len: u16 = SENDER_REPORT_PAYLOAD - 2; + encoded[2..4].copy_from_slice(&short_len.to_le_bytes()); + assert!(SenderReport::decode(&encoded[1..]).is_err()); + } + + #[test] + fn test_receiver_report_v1_total_length_too_short() { + let rr = sample_receiver_report(); + let mut encoded = rr.encode(); + // Set format_version = 1 but total_length < v0 payload size + encoded[1] = 1; + let short_len: u16 = RECEIVER_REPORT_PAYLOAD - 4; + encoded[2..4].copy_from_slice(&short_len.to_le_bytes()); + assert!(ReceiverReport::decode(&encoded[1..]).is_err()); + } } diff --git a/src/node/bloom.rs b/src/node/bloom.rs index e40864c..bfcffc3 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -113,6 +113,8 @@ impl Node { } else { self.stats_mut().bloom.full_sends += 1; } + self.stats_mut().bloom.total_compressed_bytes += stats.compressed_bytes as u64; + self.stats_mut().bloom.total_raw_bytes += (stats.raw_words * 8) as u64; // Record send and store the filter for change detection debug!( diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 67c7416..179ca91 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -52,7 +52,9 @@ impl Node { // 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 Some(peer) = self.peers.get(&node_addr) else { + return; + }; let k_bit_flipped = received_k_bit != peer.current_k_bit() && peer.pending_new_session().is_some(); @@ -63,7 +65,9 @@ impl Node { "Peer K-bit flip detected, promoting new session" ); - let peer = self.peers.get_mut(&node_addr).unwrap(); + let Some(peer) = self.peers.get_mut(&node_addr) else { + return; + }; if let Some(_old_our_index) = peer.handle_peer_kbit_flip() { // New index was pre-registered in peers_by_index during // msg1 handling (handshake.rs). Verify, don't duplicate. @@ -83,7 +87,9 @@ impl Node { // 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 Some(peer) = self.peers.get_mut(&node_addr) else { + return; + }; let session = match peer.noise_session_mut() { Some(s) => s, None => { diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index e85b035..5c695d9 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -377,7 +377,11 @@ impl Node { return; } - let conn = self.connections.get_mut(&link_id).unwrap(); + let Some(conn) = self.connections.get_mut(&link_id) else { + warn!(link_id = %link_id, "Connection removed during msg2 processing"); + self.pending_outbound.remove(&key); + return; + }; // Create FMP negotiation payload for msg3 (includes profile, MMP bits, bloom TLV) let neg_payload = NegotiationPayload::fmp(1, 1, self.node_profile).encode(); @@ -403,7 +407,7 @@ impl Node { match process_fmp_negotiation(self.node_profile, conn, neg_bytes) { Ok(()) => {} Err(e) => { - warn!(link_id = %link_id, error = %e, "FMP negotiation failed"); + warn!(link_id = %link_id, our_profile = %self.node_profile, error = %e, "FMP negotiation failed"); conn.mark_failed(); return; } @@ -524,7 +528,11 @@ impl Node { ); // Update peers_by_index: remove old inbound index, add outbound - let transport_id = peer.transport_id().unwrap(); + let Some(transport_id) = peer.transport_id() else { + warn!(peer = %self.peer_display_name(&peer_node_addr), "Active peer missing transport_id during cross-connection"); + self.pending_outbound.remove(&key); + return; + }; if let Some(old_idx) = old_our_index { self.peers_by_index .remove(&(transport_id, old_idx.as_u32())); @@ -746,7 +754,7 @@ impl Node { match process_fmp_negotiation(self.node_profile, conn, neg_bytes) { Ok(()) => {} Err(e) => { - warn!(link_id = %link_id, error = %e, "FMP negotiation failed"); + warn!(link_id = %link_id, our_profile = %self.node_profile, error = %e, "FMP negotiation failed"); self.connections.remove(&link_id); self.remove_link(&link_id); return; @@ -857,7 +865,11 @@ impl Node { // Rekey: process as responder, store new session as pending let noise_session = { - let conn = self.connections.get_mut(&link_id).unwrap(); + let Some(conn) = self.connections.get_mut(&link_id) else { + warn!(link_id = %link_id, "Connection removed during rekey msg3 processing"); + self.links.remove(&link_id); + return; + }; conn.take_session() }; let our_new_index = our_index; @@ -1181,7 +1193,9 @@ impl Node { if this_wins { // This connection wins, replace the existing peer - let old_peer = self.peers.remove(&peer_node_addr).unwrap(); + let Some(old_peer) = self.peers.remove(&peer_node_addr) else { + return Err(NodeError::PeerNotFound(peer_node_addr)); + }; let loser_link_id = old_peer.link_id(); // Clean up old peer's index from peers_by_index @@ -1361,8 +1375,8 @@ fn process_fmp_negotiation( debug!( link_id = %conn.link_id(), - our_profile = ?our_profile, - peer_profile = ?their_profile, + our_profile = %our_profile, + peer_profile = %their_profile, "FMP negotiation complete" ); diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index 116586c..5c9e1e1 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -232,8 +232,10 @@ impl Node { 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())); + if peer.needs_msg1_resend(now_ms) + && let Some(msg1) = peer.rekey_msg1() + { + to_resend.push((*node_addr, msg1.to_vec())); } } diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index fd16e40..a05fa22 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -179,7 +179,9 @@ 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 Some(entry) = self.sessions.get(src_addr) else { + return; + }; let k_bit_flipped = received_k_bit != entry.current_k_bit() && entry.pending_new_session().is_some(); @@ -190,7 +192,9 @@ impl Node { "Peer FSP K-bit flip detected, promoting new session" ); let now_ms = Self::now_ms(); - let entry = self.sessions.get_mut(src_addr).unwrap(); + let Some(entry) = self.sessions.get_mut(src_addr) else { + return; + }; entry.handle_peer_kbit_flip(now_ms); } } @@ -445,8 +449,9 @@ impl Node { 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(); + if let Some(entry) = self.sessions.get_mut(src_addr) { + entry.abandon_rekey(); + } } else if has_pending { // Guard: already have a pending session waiting for K-bit cutover debug!( @@ -488,9 +493,10 @@ impl Node { // 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); + if let Some(entry) = self.sessions.get_mut(src_addr) { + entry.set_rekey_state(handshake, false); + entry.record_peer_rekey(now_ms); + } debug!( src = %self.peer_display_name(src_addr), @@ -1809,6 +1815,7 @@ impl Node { if original_packet.len() < 40 { return; } + // SAFETY: slice is exactly 16 bytes; length validated above (>= 40) let src_addr = Ipv6Addr::from(<[u8; 16]>::try_from(&original_packet[8..24]).unwrap()); // Rate limit ICMP PTB messages per source @@ -1824,6 +1831,7 @@ impl Node { // kernel sees the PTB coming from a remote router, not from itself. // Linux ignores PTBs whose source matches a local address, which // causes a PMTUD blackhole when both src and ICMP-src are local. + // SAFETY: slice is exactly 16 bytes; length validated above (>= 40) let dest_addr = Ipv6Addr::from(<[u8; 16]>::try_from(&original_packet[24..40]).unwrap()); if let Some(response) = build_packet_too_big(original_packet, mtu, dest_addr) && let Some(tun_tx) = &self.tun_tx diff --git a/src/node/stats.rs b/src/node/stats.rs index 1ffd3c2..1a1b24e 100644 --- a/src/node/stats.rs +++ b/src/node/stats.rs @@ -222,6 +222,9 @@ pub struct BloomStats { pub nacks_received: u64, // Adaptive sizing pub size_changes: u64, + // Compression tracking + pub total_compressed_bytes: u64, + pub total_raw_bytes: u64, } impl BloomStats { @@ -241,6 +244,8 @@ impl BloomStats { nacks_sent: self.nacks_sent, nacks_received: self.nacks_received, size_changes: self.size_changes, + total_compressed_bytes: self.total_compressed_bytes, + total_raw_bytes: self.total_raw_bytes, } } } @@ -415,6 +420,8 @@ pub struct BloomStatsSnapshot { pub nacks_sent: u64, pub nacks_received: u64, pub size_changes: u64, + pub total_compressed_bytes: u64, + pub total_raw_bytes: u64, } #[derive(Clone, Debug, Default, Serialize)] diff --git a/src/node/tests/discovery.rs b/src/node/tests/discovery.rs index 6f81dd3..cdee9a7 100644 --- a/src/node/tests/discovery.rs +++ b/src/node/tests/discovery.rs @@ -10,7 +10,7 @@ use crate::protocol::{LookupRequest, LookupResponse}; use crate::tree::TreeCoordinate; use spanning_tree::{ cleanup_nodes, generate_random_edges, process_available_packets, run_tree_test, - verify_tree_convergence, + run_tree_test_with_mtus, verify_tree_convergence, }; // ============================================================================ @@ -882,3 +882,129 @@ async fn test_originator_stores_path_mtu_in_cache() { "Originator should store path_mtu from LookupResponse in cache" ); } + +// ============================================================================ +// Integration Tests — min_mtu transit pruning +// ============================================================================ + +#[tokio::test] +async fn test_transit_prunes_lookup_by_min_mtu() { + // Topology: node0(1280) — node1(800) — node2(1280) + // Node0 initiates lookup for node2 with min_mtu=1280 (default TUN MTU). + // Node1's transport MTU is 800 < 1280, so node1 should NOT forward + // the request to node2. The lookup should fail (no cache entry). + let mtus = [1280, 800, 1280]; + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await; + + let node2_addr = *nodes[2].node.node_addr(); + let node2_pubkey = nodes[2].node.identity().pubkey_full(); + nodes[0].node.register_identity(node2_addr, node2_pubkey); + + nodes[0].node.initiate_lookup(&node2_addr, 8).await; + + for _ in 0..10 { + tokio::time::sleep(Duration::from_millis(100)).await; + process_available_packets(&mut nodes).await; + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + assert!( + !nodes[0].node.coord_cache().contains(&node2_addr, now_ms), + "Node0 should NOT have cached node2 route (transit pruned by min_mtu)" + ); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_transit_forwards_when_mtu_sufficient() { + // Topology: node0(1280) — node1(1400) — node2(1280) + // Node0 initiates lookup for node2 with min_mtu=1280 (default TUN MTU). + // Node1's transport MTU is 1400 >= 1280, so the request passes through. + // Node1 annotates path_mtu = min(u16::MAX, 1400) = 1400 on response. + let mtus = [1280, 1400, 1280]; + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await; + + let node2_addr = *nodes[2].node.node_addr(); + let node2_pubkey = nodes[2].node.identity().pubkey_full(); + nodes[0].node.register_identity(node2_addr, node2_pubkey); + + nodes[0].node.initiate_lookup(&node2_addr, 8).await; + + for _ in 0..10 { + tokio::time::sleep(Duration::from_millis(100)).await; + process_available_packets(&mut nodes).await; + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + assert!( + nodes[0].node.coord_cache().contains(&node2_addr, now_ms), + "Node0 should have cached node2 route (MTU sufficient)" + ); + + let entry = nodes[0].node.coord_cache().get_entry(&node2_addr).unwrap(); + let path_mtu = entry.path_mtu().expect("path_mtu should be set"); + assert_eq!( + path_mtu, 1400, + "path_mtu should reflect transit node's transport MTU (1400)" + ); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_response_path_mtu_four_node_chain() { + // Topology: node0(1280) — node1(1400) — node2(900) — node3(1280) + // Node0 initiates lookup for node3. Response travels node3→node2→node1→node0. + // Transit nodes apply min(): node2 sees min(u16::MAX, 900) = 900, + // node1 sees min(900, 1400) = 900. + // Final path_mtu at node0 should be 900 (bottleneck at node2). + // + // Note: min_mtu=1280 from TUN config. Node2's MTU (900) < 1280 would prune + // the forward request at node2, so node3 would never be reached. To test + // path_mtu annotation we need all transit links to pass the min_mtu check. + // Use MTUs above 1280 to avoid pruning but with different values to verify min(). + let mtus = [1280, 1500, 1350, 1280]; + let edges = vec![(0, 1), (1, 2), (2, 3)]; + let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await; + + let node3_addr = *nodes[3].node.node_addr(); + let node3_pubkey = nodes[3].node.identity().pubkey_full(); + nodes[0].node.register_identity(node3_addr, node3_pubkey); + + nodes[0].node.initiate_lookup(&node3_addr, 8).await; + + for _ in 0..15 { + tokio::time::sleep(Duration::from_millis(100)).await; + process_available_packets(&mut nodes).await; + } + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + assert!( + nodes[0].node.coord_cache().contains(&node3_addr, now_ms), + "Node0 should have cached node3 route" + ); + + let entry = nodes[0].node.coord_cache().get_entry(&node3_addr).unwrap(); + let path_mtu = entry.path_mtu().expect("path_mtu should be set"); + assert_eq!( + path_mtu, 1350, + "Four-node chain path_mtu should be min of transit MTUs (1350)" + ); + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index bea4a82..6775443 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -1,6 +1,7 @@ //! Integration tests for end-to-end Noise XX handshake scenarios. use super::*; +use super::spanning_tree::{cleanup_nodes, drain_all_packets, initiate_handshake, make_test_node}; #[tokio::test] async fn test_two_node_handshake_udp() { @@ -989,3 +990,121 @@ async fn test_duplicate_msg2_dropped() { assert_eq!(node.connection_count(), 0); assert_eq!(node.peer_count(), 0); } + +// ===== Profile Rejection Tests ===== + +/// Helper: create two test nodes, set their profiles, attempt a handshake, +/// and return whether they successfully peered. +async fn attempt_profile_handshake( + profile_a: crate::protocol::NodeProfile, + profile_b: crate::protocol::NodeProfile, +) -> (usize, usize) { + let mut nodes = vec![make_test_node().await, make_test_node().await]; + nodes[0].node.node_profile = profile_a; + nodes[1].node.node_profile = profile_b; + + initiate_handshake(&mut nodes, 0, 1).await; + drain_all_packets(&mut nodes, false).await; + + let peers = (nodes[0].node.peer_count(), nodes[1].node.peer_count()); + cleanup_nodes(&mut nodes).await; + peers +} + +#[tokio::test] +async fn test_nonrouting_nonrouting_rejected() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::NonRouting).await; + assert_eq!(a, 0, "NonRouting↔NonRouting should reject: node A"); + assert_eq!(b, 0, "NonRouting↔NonRouting should reject: node B"); +} + +#[tokio::test] +async fn test_leaf_leaf_rejected() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::Leaf).await; + assert_eq!(a, 0, "Leaf↔Leaf should reject: node A"); + assert_eq!(b, 0, "Leaf↔Leaf should reject: node B"); +} + +#[tokio::test] +async fn test_nonrouting_leaf_rejected() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::Leaf).await; + assert_eq!(a, 0, "NonRouting↔Leaf should reject: node A"); + assert_eq!(b, 0, "NonRouting↔Leaf should reject: node B"); +} + +#[tokio::test] +async fn test_leaf_nonrouting_rejected() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::NonRouting).await; + assert_eq!(a, 0, "Leaf↔NonRouting should reject: node A"); + assert_eq!(b, 0, "Leaf↔NonRouting should reject: node B"); +} + +#[tokio::test] +async fn test_full_nonrouting_accepted() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::NonRouting).await; + assert_eq!(a, 1, "Full↔NonRouting should accept: node A"); + assert_eq!(b, 1, "Full↔NonRouting should accept: node B"); +} + +#[tokio::test] +async fn test_full_leaf_accepted() { + use crate::protocol::NodeProfile; + let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::Leaf).await; + assert_eq!(a, 1, "Full↔Leaf should accept: node A"); + assert_eq!(b, 1, "Full↔Leaf should accept: node B"); +} + +// ===== XX Address-Based Dedup Tests ===== + +#[tokio::test] +async fn test_xx_duplicate_msg1_resends_msg2() { + use crate::node::wire::build_msg1; + use crate::transport::ReceivedPacket; + + // Node B with NO transport — msg2 send silently skips (if let Some check), + // but the pending connection and link are created. + let mut node_b = make_node(); + let transport_id = TransportId::new(1); + + // Build a valid XX msg1 from an external initiator + let initiator = Identity::generate(); + let mut hs = crate::noise::HandshakeState::new_initiator(initiator.keypair()); + let noise_msg1 = hs.write_message_1().unwrap(); + let sender_idx = SessionIndex::new(42); + let wire_msg1 = build_msg1(sender_idx, &noise_msg1); + + let remote_addr = TransportAddr::from_string("10.0.0.1:2121"); + + // First msg1 → B creates pending inbound connection + let first_packet = ReceivedPacket { + transport_id, + remote_addr: remote_addr.clone(), + data: wire_msg1.clone(), + timestamp_ms: 1000, + }; + node_b.handle_msg1(first_packet).await; + + assert_eq!(node_b.connection_count(), 1, "B: 1 connection after first msg1"); + assert_eq!(node_b.peer_count(), 0, "B: 0 peers (XX, no promotion at msg1)"); + + // Duplicate msg1 from same address → dedup triggers msg2 resend, not new handshake + let dup_packet = ReceivedPacket { + transport_id, + remote_addr: remote_addr.clone(), + data: wire_msg1.clone(), + timestamp_ms: 1100, + }; + node_b.handle_msg1(dup_packet).await; + + assert_eq!( + node_b.connection_count(), + 1, + "B: still 1 connection after duplicate msg1 (dedup, not new handshake)" + ); + assert_eq!(node_b.peer_count(), 0, "B: still 0 peers"); +} diff --git a/src/noise/mod.rs b/src/noise/mod.rs index 466790e..ff2afd1 100644 --- a/src/noise/mod.rs +++ b/src/noise/mod.rs @@ -16,6 +16,26 @@ //! The XX pattern handles both **link-layer peer authentication** (securing the //! direct link between neighboring nodes) and **session-layer end-to-end //! encryption** between arbitrary network addresses. +//! +//! ## Identity Timing +//! +//! Unlike IK (where the initiator's identity was in msg1), XX defers all +//! identity disclosure: +//! +//! - **msg1**: Ephemeral only. No identity, no DH with static keys. +//! - **msg2**: Responder reveals its static key to the initiator. +//! - **msg3**: Initiator reveals its static key to the responder. +//! +//! Consequence: all identity-based checks that previously ran during msg1 +//! processing (restart detection, rekey detection, allow/deny lists, +//! cross-connection resolution) are now deferred: +//! +//! - **Initiator** performs identity checks in `handle_msg2` after +//! decrypting the responder's static key. +//! - **Responder** performs identity checks in `handle_msg3` after +//! decrypting the initiator's static key. +//! - **msg1 handler** can only do address-based duplicate detection (same +//! transport + address). Identity-dependent decisions happen later. mod handshake; mod replay; diff --git a/src/protocol/negotiation.rs b/src/protocol/negotiation.rs index 9a3031d..7d1c974 100644 --- a/src/protocol/negotiation.rs +++ b/src/protocol/negotiation.rs @@ -13,6 +13,25 @@ //! Bytes 10+: TLV entries, each: //! [field_num:2 LE][length:2 LE][value:N] //! ``` +//! +//! ## Node Profile Decision Tree +//! +//! Profiles are self-declared (bits 0-2 of the feature bitfield): +//! +//! - **Full** (0): Full routing. Combines bloom filters from children, +//! forwards transit traffic, participates in spanning tree. +//! - **NonRouting** (1): Tree participation but no transit forwarding. +//! Receives bloom filters (one-way: F→N) but does not send them. +//! The full peer inserts N's identity via `leaf_dependents`. +//! - **Leaf** (2): Single upstream peer, no tree/bloom/transit. +//! Full peer inserts L's identity via `leaf_dependents`. +//! +//! **Link pairing rule**: at least one side must be Full. Invalid +//! pairings (N↔N, N↔L, L↔L) are rejected during FMP negotiation. +//! +//! **Routing implications**: `forward_lookup_request()` only considers +//! Full peers as transit. `peer_inbound_filters()` excludes non-Full +//! peers from bloom filter merging. use super::ProtocolError; @@ -58,6 +77,16 @@ pub enum NodeProfile { Leaf = 2, } +impl std::fmt::Display for NodeProfile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Full => write!(f, "full"), + Self::NonRouting => write!(f, "non-routing"), + Self::Leaf => write!(f, "leaf"), + } + } +} + impl TryFrom for NodeProfile { type Error = ProtocolError; @@ -274,7 +303,7 @@ impl NegotiationPayload { pub fn validate_profiles(ours: NodeProfile, theirs: NodeProfile) -> Result<(), ProtocolError> { if ours != NodeProfile::Full && theirs != NodeProfile::Full { return Err(ProtocolError::Malformed(format!( - "invalid profile pairing: {:?} <-> {:?} (at least one must be Full)", + "invalid profile pairing: {} <-> {} (at least one must be full)", ours, theirs ))); }