From b37f7cdc21f32bc2ba32a2266d95abbc9262a92a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 14 Sep 2026 13:57:03 +0000 Subject: [PATCH 1/3] fix(rekey): keep the rekey cycle when a msg2 fails to authenticate The rekey initiator took its handshake off the peer before reading msg2 and abandoned the whole cycle when the read failed. Nothing authenticates a msg2 ahead of that read, and the index it is dispatched on travels in cleartext in rekey msg1, so anyone on the path who saw the msg1 could answer it first with a msg2 of the right size under that index. That costs more than one cycle. The IK responder commits its new keys when it answers msg1 and cuts over on its own next rekey tick, while the initiator, having abandoned, drops the responder's genuine msg2 at the dispatch lookup. Frames from the responder then miss the initiator's index table at once, frames to the responder fail once its drain window closes, and each end removes the other on the link-dead timeout. With the tick functions driven in-process at one-second steps, the responder removed the link at about 30 s and the initiator at about 31 s, and the initiator's last msg1 resend left the responder holding a peer the initiator no longer had, removed 30 s after that. Read msg2 through a rolling-back variant of the IK read, modelled on the existing XK one, and put the handshake back on the peer when it fails. The msg2 handler keeps the cycle and its dispatch entry in that case and still counts the reject, so the genuine msg2 completes the rekey. The other failures, a missing handshake or one after the read succeeded, still abandon, and if no readable msg2 ever arrives the msg1 resend budget abandons the cycle as before. Each forged msg2 carrying the live index now costs the initiator the msg2 key agreement until the cycle ends, where before only the first one did; the resend budget bounds that. No wire format changes. The new test ages a two-node link past both rekey gates, delivers the rekey msg1, holds the responder's msg2, injects a forged one under the live index, releases the real one, runs a rekey tick on both nodes and checks delivery in both directions. It fails on the unfixed tree at the responder-to-initiator delivery, which is the check that separates the two outcomes, and fails there again with each part of the fix reverted on its own: the handshake put-back, the symmetric-state rollback, and keeping the dispatch entry. --- CHANGELOG.md | 18 +++ src/node/handlers/handshake.rs | 31 ++++- src/node/tests/session.rs | 227 +++++++++++++++++++++++++++++++++ src/noise/handshake.rs | 42 +++++- src/peer/active.rs | 23 +++- 5 files changed, 332 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c75f428..3e4c662e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 interval instead. That retry interval gates only a peer whose last attempt failed, so it cannot clamp a `heartbeat_interval_secs` configured below it. +#### Link rekey + +- A forged rekey msg2 no longer takes the link down. The rekey initiator gave + up its handshake before reading msg2 and abandoned the cycle when the read + failed, although nothing authenticates a msg2 ahead of that read. Anyone on + the path who saw the rekey msg1 go out could answer first with a msg2 of the + right size under the index msg1 carries in cleartext. The responder has + already committed its new session by then and cuts over on its next tick, so + the two ends were left on different keys: frames from the responder were + dropped at once, frames to it failed once its drain window closed, and each + end removed the other on the link-dead timeout about 30 s later. A msg2 that + fails the read now leaves the handshake as it was before the read, along + with the msg1 resend schedule and the msg2 dispatch entry, so the + responder's genuine msg2 still completes the rekey. In exchange, every such + forgery now costs the initiator the msg2 key agreement until the cycle ends, + where before only the first one did; the msg1 resend budget bounds that. The + wire format is unchanged. + #### Control socket - `show_links` (`fipsctl show links`) now reports the traffic a link has diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 5139e536..34727180 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1183,6 +1183,7 @@ impl Node { // Complete the rekey handshake on the ActivePeer let mut rekey_completed = false; + let mut cycle_kept = false; if let Some(peer) = self.peers.get_mut(&peer_node_addr) { match peer.complete_rekey_msg2(noise_msg2) { Ok((session, remote_epoch)) => { @@ -1222,6 +1223,26 @@ impl Node { ); rekey_completed = true; } + // Nothing authenticated this msg2 before the read, and the + // index it names travels in cleartext in our msg1, so it + // may be a forgery. The responder committed its new + // session when it answered that msg1 and cuts over on its + // own tick, so abandoning here would leave the two ends on + // different keys. The read rolled the handshake back: + // keep the cycle and its dispatch entry so the genuine + // msg2 can still complete it. If no readable msg2 ever + // arrives, the msg1 resend budget abandons the cycle as + // it would for a lost one. + Err(e) if peer.awaits_msg2() => { + debug!( + peer = %display_name, + error = %e, + "Rekey msg2 did not authenticate, keeping the rekey cycle" + ); + cycle_kept = true; + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + } Err(e) => { warn!( peer = %display_name, @@ -1242,14 +1263,16 @@ impl Node { // Feed the control machine the completed-rekey observation so its // shadow index and rekey phase stay coherent. Only on success — - // the failure path above reverts the rekey and leaves the machine - // untouched. The crypto effect already ran inline; this emits no - // action. + // the failure paths above either keep the cycle as it was or + // revert it, and leave the machine untouched. The crypto effect + // already ran inline; this emits no action. if rekey_completed { self.observe_rekey_msg2(&peer_node_addr, header.sender_idx); } - self.pending_outbound.remove(&key); + if !cycle_kept { + self.pending_outbound.remove(&key); + } return; } diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 4e0255e6..46b5e8db 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -1222,6 +1222,233 @@ async fn rekey_cutover_preserves_data_plane() { cleanup_nodes(&mut nodes).await; } +/// A forged rekey msg2 that carries the initiator's live rekey index must not +/// split the link. +/// +/// An on-path observer sees the rekey msg1 go out, reads the initiator's +/// cleartext rekey index from it, and delivers a msg2 of the right size under +/// that index ahead of the responder's real reply. Under IK the forgery cannot +/// authenticate: only the responder's static key produces a msg2 the initiator +/// can read. The IK responder has already committed its new session when it +/// answered msg1 and cuts over on its own next rekey tick, so the initiator has +/// to keep the cycle through the forgery and complete it on the real msg2. An +/// initiator that gives the cycle up instead holds no session matching the one +/// the responder now sends on. +/// +/// Deterministic, no wall-clock wait: both sessions are backdated past node 0's +/// time trigger and node 1's rekey-acceptance floor, node 1 never initiates, +/// and every handshake message is delivered by hand. +#[tokio::test] +async fn forged_rekey_msg2_does_not_split_the_link() { + use crate::noise::HANDSHAKE_MSG2_SIZE; + use crate::proto::fmp::wire::{CommonPrefix, PHASE_MSG2, build_msg2}; + use crate::transport::ReceivedPacket; + use crate::utils::index::SessionIndex; + + const REKEY_AFTER_SECS: u64 = 60; + + // node 0 rekeys on time; node 1 only ever responds. + let mut cfg0 = crate::config::Config::new(); + cfg0.node.rekey.enabled = true; + cfg0.node.rekey.after_secs = REKEY_AFTER_SECS; + cfg0.node.rekey.after_messages = u64::MAX; + let mut cfg1 = crate::config::Config::new(); + cfg1.node.rekey.enabled = true; + cfg1.node.rekey.after_secs = u64::MAX; + cfg1.node.rekey.after_messages = u64::MAX; + + let mut nodes = vec![ + make_test_node_with_config(cfg0, 1280).await, + make_test_node_with_config(cfg1, 1280).await, + ]; + + // FMP peering + FSP session between the two loopback nodes. + initiate_handshake(&mut nodes, 0, 1).await; + drain_all_packets(&mut nodes, false).await; + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + assert!(nodes[0].node.get_peer(&node1_addr).is_some()); + assert!(nodes[1].node.get_peer(&node0_addr).is_some()); + populate_all_coord_caches(&mut nodes); + + let node1_pubkey = nodes[1].node.identity().pubkey_full(); + nodes[0] + .node + .initiate_session(node1_addr, node1_pubkey) + .await + .unwrap(); + for _ in 0..4 { + tokio::time::sleep(Duration::from_millis(10)).await; + process_available_packets(&mut nodes).await; + } + for (i, remote) in [(0, node1_addr), (1, node0_addr)] { + assert!( + nodes[i] + .node + .get_session(&remote) + .is_some_and(|s| s.state().is_established()), + "node {i} session established" + ); + } + + // Each node's TUN receiver observes the plaintext the other one sent. + let (tun0_tx, tun0_rx) = std::sync::mpsc::channel(); + nodes[0].node.supervisor.tun_tx = Some(tun0_tx); + let (tun1_tx, tun1_rx) = std::sync::mpsc::channel(); + nodes[1].node.supervisor.tun_tx = Some(tun1_tx); + let fips0 = crate::FipsAddress::from_node_addr(&node0_addr); + let fips1 = crate::FipsAddress::from_node_addr(&node1_addr); + + // Baseline: both directions decode before the rekey, so a failure below + // is the rekey's and not the harness's. + let pre_fwd = build_ipv6_packet(&fips0, &fips1, b"pre-rekey 0 to 1"); + let pre_rev = build_ipv6_packet(&fips1, &fips0, b"pre-rekey 1 to 0"); + nodes[0].node.handle_tun_outbound(pre_fwd.clone()).await; + nodes[1].node.handle_tun_outbound(pre_rev.clone()).await; + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(10)).await; + if process_available_packets(&mut nodes).await == 0 { + break; + } + } + let got: Vec> = std::iter::from_fn(|| tun1_rx.try_recv().ok()).collect(); + assert_eq!(got, vec![pre_fwd], "baseline node 0 to node 1 must decode"); + let got: Vec> = std::iter::from_fn(|| tun0_rx.try_recv().ok()).collect(); + assert_eq!(got, vec![pre_rev], "baseline node 1 to node 0 must decode"); + + // Age both sessions past both rekey gates: node 0's jittered time trigger, + // and node 1's 30 s floor below which a msg1 is a duplicate, not a rekey. + let age = Duration::from_secs(REKEY_AFTER_SECS + crate::node::REKEY_JITTER_SECS as u64 + 1); + nodes[0] + .node + .get_peer_mut(&node1_addr) + .unwrap() + .test_backdate_session_established(age); + nodes[1] + .node + .get_peer_mut(&node0_addr) + .unwrap() + .test_backdate_session_established(age); + let node0_idx_before = nodes[0].node.get_peer(&node1_addr).unwrap().our_index(); + let node1_idx_before = nodes[1].node.get_peer(&node0_addr).unwrap().our_index(); + + // node 0 starts the rekey; its msg1 lands in node 1's queue. + nodes[0].node.check_rekey().await; + let rekey_idx = nodes[0] + .node + .get_peer(&node1_addr) + .unwrap() + .rekey_our_index() + .expect("node 0 must have started a rekey"); + + // Deliver the msg1 to node 1 only. node 1 answers as the rekey responder + // and commits its new session at once. + assert_eq!( + process_available_packets(&mut nodes[1..]).await, + 1, + "node 1 must have exactly node 0's rekey msg1 queued" + ); + assert!( + nodes[1] + .node + .get_peer(&node0_addr) + .unwrap() + .pending_new_session() + .is_some(), + "node 1 must answer the msg1 as a rekey and hold its new session" + ); + + // Hold node 1's real msg2 back. + let mut held: Vec = + std::iter::from_fn(|| nodes[0].packet_rx.try_recv().ok()).collect(); + assert_eq!(held.len(), 1, "node 0 must have only node 1's msg2 queued"); + let real_msg2 = held.remove(0); + assert_eq!( + CommonPrefix::parse(&real_msg2.data).map(|p| p.phase), + Some(PHASE_MSG2), + "the held packet must be node 1's msg2" + ); + + // The forgery: a well-formed header naming node 0's live rekey index, a + // valid curve point as the ephemeral so the read gets as far as mixing it + // into the handshake, and an epoch ciphertext that cannot authenticate. + // The source is node 1's address, as a spoofed UDP source would be. + let mut forged_noise = Identity::generate().pubkey_full().serialize().to_vec(); + forged_noise.resize(HANDSHAKE_MSG2_SIZE, 0xA5); + let forged = ReceivedPacket::new( + nodes[0].transport_id, + nodes[1].addr.clone(), + build_msg2(SessionIndex::new(0x5EED_F00D), rekey_idx, &forged_noise), + ); + nodes[0].node.handle_msg2(forged).await; + + // Release the real msg2, then run one rekey tick on each node. + nodes[0].node.handle_msg2(real_msg2).await; + nodes[0].node.check_rekey().await; + nodes[1].node.check_rekey().await; + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(10)).await; + if process_available_packets(&mut nodes).await == 0 { + break; + } + } + + // node 1 cut over to the session it committed at msg1. Without this the + // delivery checks below could pass because no rekey happened at all. + assert_ne!( + nodes[1].node.get_peer(&node0_addr).unwrap().our_index(), + node1_idx_before, + "node 1 must have cut over to its new session" + ); + + let post_fwd = build_ipv6_packet(&fips0, &fips1, b"post-rekey 0 to 1"); + let post_rev = build_ipv6_packet(&fips1, &fips0, b"post-rekey 1 to 0"); + nodes[0].node.handle_tun_outbound(post_fwd.clone()).await; + nodes[1].node.handle_tun_outbound(post_rev.clone()).await; + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(10)).await; + if process_available_packets(&mut nodes).await == 0 { + break; + } + } + + let got: Vec> = std::iter::from_fn(|| tun1_rx.try_recv().ok()).collect(); + assert_eq!( + got, + vec![post_fwd], + "node 0 to node 1 must decode after the forged msg2" + ); + + // This is the assertion that tells the two outcomes apart; keep it. node 0 + // to node 1 passes either way inside this test, because node 1 keeps its + // previous session through the drain window and still decrypts node 0's + // old-session frames. node 1 to node 0 fails exactly when node 0 lost the + // cycle to the forgery: node 1 now sends on its new session, addressed to + // node 0's rekey index, and node 0 has no session registered under it. + let handshake = &nodes[0].node.stats().handshake; + let (bad_state, unknown) = (handshake.bad_state, handshake.unknown_connection); + let got: Vec> = std::iter::from_fn(|| tun0_rx.try_recv().ok()).collect(); + assert_eq!( + got, + vec![post_rev], + "node 1 to node 0 must decode after the forged msg2 \ + (node 0 handshake rejects: bad_state={bad_state}, unknown_connection={unknown})" + ); + + let peer = nodes[0].node.get_peer(&node1_addr).unwrap(); + assert_ne!( + peer.our_index(), + node0_idx_before, + "node 0 must have completed the rekey on the real msg2 and cut over" + ); + assert!( + !peer.rekey_in_progress(), + "node 0 must not be left mid-rekey" + ); + + cleanup_nodes(&mut nodes).await; +} + #[tokio::test] async fn test_tun_outbound_triggers_session_initiation() { // Two connected nodes, no session yet. diff --git a/src/noise/handshake.rs b/src/noise/handshake.rs index 5116707f..51dcbd49 100644 --- a/src/noise/handshake.rs +++ b/src/noise/handshake.rs @@ -15,9 +15,10 @@ use zeroize::{Zeroize, ZeroizeOnDrop}; /// /// Maintains the chaining key (ck), handshake hash (h), and current cipher. /// -/// `Clone` exists for [`HandshakeState::try_read_xk_message_2`], which has to -/// put the pre-read state back after a message that mixed material in before -/// failing to authenticate. +/// `Clone` exists for [`HandshakeState::try_read_message_2`] and +/// [`HandshakeState::try_read_xk_message_2`], which have to put the pre-read +/// state back after a message that mixed material in before failing to +/// authenticate. /// /// `ck` and `h` are cleared on drop, including on the clone above once it /// goes out of scope. `cipher` is skipped because [`CipherState`] clears its @@ -636,6 +637,41 @@ impl HandshakeState { Ok(()) } + /// Read message 2, leaving the handshake untouched when the message + /// does not authenticate. + /// + /// `read_message_2` mixes the sender's ephemeral into the symmetric state, + /// and both DH results into the key, before it authenticates the encrypted + /// epoch, so a message that fails partway leaves a handshake that can + /// never read the genuine msg2 afterwards. A caller that keeps its rekey + /// cycle across a failed read — because the message may be a forgery + /// rather than the responder's corrupt reply — needs the pre-read state + /// back. + /// + /// The saved set is exactly what `read_message_2` writes: `symmetric`, + /// `remote_ephemeral`, `remote_epoch` and `progress`. `remote_static` is + /// not in it, because IK pins the responder's static before msg1 and the + /// read only uses it. **That mirror is manual.** A later edit that adds a + /// write to `read_message_2` without adding it here silently reintroduces + /// the poisoning, and no caller can detect it. + pub fn try_read_message_2(&mut self, message: &[u8]) -> Result<(), NoiseError> { + let symmetric = self.symmetric.clone(); + let remote_ephemeral = self.remote_ephemeral; + let remote_epoch = self.remote_epoch; + let progress = self.progress; + + match self.read_message_2(message) { + Ok(()) => Ok(()), + Err(e) => { + self.symmetric = symmetric; + self.remote_ephemeral = remote_ephemeral; + self.remote_epoch = remote_epoch; + self.progress = progress; + Err(e) + } + } + } + // ======================================================================== // XK Pattern Methods (Session Layer) // ======================================================================== diff --git a/src/peer/active.rs b/src/peer/active.rs index 3216ea1f..fa819454 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -1246,11 +1246,27 @@ impl ActivePeer { self.rekey_our_index } + /// Whether this peer still holds its rekey initiator handshake, waiting + /// on msg2. + /// + /// [`complete_rekey_msg2`](Self::complete_rekey_msg2) keeps the handshake + /// when a msg2 fails to authenticate, so after a failed call this tells + /// the caller the cycle is still intact. + pub fn awaits_msg2(&self) -> bool { + self.rekey_handshake.is_some() + } + /// Complete the rekey by processing msg2 (initiator side). /// - /// Takes the stored handshake state, reads msg2, and returns the + /// Reads msg2 against the stored handshake state and returns the /// completed NoiseSession. Clears the handshake-related fields but /// leaves rekey_our_index for set_pending_session to use. + /// + /// A msg2 that fails the read changes nothing: the handshake goes back + /// in its pre-read state, and the msg1 resend schedule stays as it was. + /// Nothing authenticates a msg2 before this read, so the message may be a + /// forgery naming our rekey index, and the responder's genuine msg2 has + /// to remain readable when it arrives. pub fn complete_rekey_msg2( &mut self, msg2_bytes: &[u8], @@ -1263,7 +1279,10 @@ impl ActivePeer { got: "no handshake state".to_string(), })?; - hs.read_message_2(msg2_bytes)?; + if let Err(e) = hs.try_read_message_2(msg2_bytes) { + self.rekey_handshake = Some(hs); + return Err(e); + } let remote_epoch = hs.remote_epoch(); let session = hs.into_session()?; From ae9a574bc0bfdd6b5b71a5c68317fdd400265bb2 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 14 Sep 2026 14:32:23 +0000 Subject: [PATCH 2/3] fix(transport/udp): let connected sockets join an adopted traversal socket's port A NAT traversal binds its base socket with a plain port-zero bind, and a successful punch adopts that socket as the peer's transport. Connected-UDP activation then opens a per-peer connected socket on the transport's local address. Linux admits that second bind only when the socket already holding the port set a reuse flag, and the traversal socket set neither, so every activation attempt failed with EADDRINUSE at bind. The rx-loop tick retried it on every pass, and the peer never left the unconnected path. Set SO_REUSEPORT and SO_REUSEADDR inside UdpRawSocket::adopt, after the socket is already bound. The order matters. Flags set before a port-zero bind let the kernel hand out a port another flagged socket already holds, so two concurrent traversals could share a port and the kernel would silently split one peer's datagrams across two transports. Flags set after the bind cannot change which port the socket was given; they only let a later socket join it. The listen socket in UdpRawSocket::open already sets its flags after its bind for the same reason. adopt has one caller chain, which ends at traversal adoption, so the STUN probe and configured listeners are untouched. The flags are best-effort, as in open: if setting them fails, adoption still succeeds and only the connected fast path is refused, rather than a working punched peer being torn down. Two tests cover the change, and both fail without it. The first adopts a socket bound the way the traversal path binds, checks that it arrives with neither flag, and requires a connected socket to open on its port and both flags to read back; without the change the open fails with EADDRINUSE at bind. The second adopts a traversal socket into a node peered with a node on a configured listener, runs activation on both, and requires a connected socket on each side, the adopted side's on the adopted socket's own port; without the change the configured side activates, the adopted side does not, and the failure carries the bind error from a direct open. Each part of the change was also removed in turn. Without SO_REUSEPORT the first test fails on that readback while the connected open still succeeds, because either flag on the holder admits the join; without SO_REUSEADDR it fails on that readback the same way; without both, both tests fail at the bind. Flagging the socket before adoption fails the first test's precondition, and running the second with FIPS_CONNECTED_UDP=0 fails its configured-listener assertion first, so neither can pass vacuously. Not established: Darwin behaviour. adopt is shared by the Unix backends, so macOS builds get the flags too, and no measurement covers SO_REUSEPORT set after bind there. FIPS_MACOS_CONNECTED_UDP=0 or FIPS_CONNECTED_UDP=0 turns the fast path off without a rebuild. --- CHANGELOG.md | 6 ++ src/node/tests/bootstrap.rs | 116 +++++++++++++++++++++++++++++++++++ src/transport/udp/io/mod.rs | 44 +++++++++++++ src/transport/udp/io/unix.rs | 11 ++++ 4 files changed, 177 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e4c662e..2555bb44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 decrypt-worker completion path already acted on that return; the in-line decrypt path discarded it, so the socket stayed installed and the send path kept preferring it over the wildcard listen socket. +- A peer reached by NAT traversal now gets its per-peer connected UDP socket. + The adopted traversal socket carried no address-reuse flags, so the connected + socket's bind to the same port was refused with `EADDRINUSE` on every tick and + the peer never left the unconnected path. The flags are now set when the + socket is adopted, after its bind, so the traversal bind still receives a + port no other socket holds. #### Peering diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index 459181b1..3905f1e4 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -377,3 +377,119 @@ async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { transport.stop().await.ok(); } } + +/// A peer reached through an adopted traversal socket gets a per-peer +/// connected UDP socket on that socket's own port, as a peer on a configured +/// listener does. The traversal socket comes from a plain bind with no reuse +/// flags, and the kernel refuses the connected socket's bind to a port whose +/// holder did not opt in to sharing it. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn test_connected_udp_activates_on_an_adopted_traversal_transport_and_on_a_configured_one() { + let mut node_a = make_node(); + let mut node_b = make_node(); + + let transport_id_b = TransportId::new(1); + let udp_config = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + mtu: Some(1280), + ..Default::default() + }; + + let (packet_tx_a, packet_rx_a) = packet_channel(64); + let (packet_tx_b, packet_rx_b) = packet_channel(64); + + node_a.supervisor.packet_tx = Some(packet_tx_a.clone()); + node_a.packet_rx = Some(packet_rx_a); + node_a.supervisor.state = NodeState::Running; + + let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b.clone()); + transport_b.start_async().await.unwrap(); + + let addr_b = transport_b.local_addr().unwrap(); + node_b.supervisor.packet_tx = Some(packet_tx_b.clone()); + node_b.packet_rx = Some(packet_rx_b); + node_b.supervisor.state = NodeState::Running; + node_b + .transports + .insert(transport_id_b, TransportHandle::Udp(transport_b)); + + let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let handoff = + EstablishedTraversal::new("sess-connected", node_b.npub(), addr_b, adopted_socket) + .with_transport_name("nostr-punched"); + + let result = node_a.adopt_established_traversal(handoff).await.unwrap(); + + tokio::select! { + result = node_b.run_rx_loop() => { + panic!("node_b rx loop exited unexpectedly: {:?}", result); + } + _ = tokio::time::sleep(Duration::from_millis(500)) => {} + } + + tokio::select! { + result = node_a.run_rx_loop() => { + panic!("node_a rx loop exited unexpectedly: {:?}", result); + } + _ = tokio::time::sleep(Duration::from_millis(500)) => {} + } + + let peer_a_node_addr = + *PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr(); + let peer_b_node_addr = + *PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full()).node_addr(); + + // Preconditions: both sides are peered, and node_a reaches node_b over + // the adopted transport rather than some other one. + assert_eq!(node_a.peer_count(), 1, "node_a should promote node_b"); + assert_eq!(node_b.peer_count(), 1, "node_b should promote node_a"); + assert!(node_b.get_peer(&peer_a_node_addr).unwrap().has_session()); + let peer_on_a = node_a.get_peer(&peer_b_node_addr).unwrap(); + assert!(peer_on_a.has_session()); + assert_eq!( + peer_on_a.transport_id(), + Some(result.transport_id), + "node_a's peer must be on the adopted transport", + ); + assert!(peer_on_a.current_addr().is_some()); + + node_b.activate_connected_udp_sessions().await; + node_a.activate_connected_udp_sessions().await; + + assert!( + node_b + .get_peer(&peer_a_node_addr) + .unwrap() + .connected_udp() + .is_some(), + "node_b's peer on its configured listener must get a connected UDP socket; \ + if it does not, check whether FIPS_CONNECTED_UDP turns the fast path off here", + ); + + let Some(connected) = node_a.get_peer(&peer_b_node_addr).unwrap().connected_udp() else { + let direct = + match crate::transport::udp::open_connected_fd(result.local_addr, addr_b, 65536, 65536) + { + Ok(_) => "succeeds".to_string(), + Err(e) => format!("fails with {e}"), + }; + panic!( + "node_a's peer on the adopted transport must get a connected UDP socket; \ + opening one on the adopted socket's address directly {direct}" + ); + }; + assert_eq!( + connected.local_addr(), + result.local_addr, + "the connected socket must join the adopted socket's port, not another transport's", + ); + drop(connected); + + for (_, transport) in node_a.transports.iter_mut() { + transport.stop().await.ok(); + } + for (_, transport) in node_b.transports.iter_mut() { + transport.stop().await.ok(); + } +} diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs index 8acbb630..eae1001c 100644 --- a/src/transport/udp/io/mod.rs +++ b/src/transport/udp/io/mod.rs @@ -99,6 +99,50 @@ mod tests { ); } + /// The traversal path binds its socket plainly on port zero, so the + /// socket reaches `adopt` carrying neither reuse flag. Adoption has to + /// add them after the fact, or the per-peer connected socket's bind to + /// the same address is refused with `EADDRINUSE`. Either flag on the + /// holder admits that bind on Linux, so the successful open cannot tell + /// whether both were set; each flag is also read back. + #[cfg(target_os = "linux")] + #[test] + fn an_adopted_plain_socket_carries_both_reuse_flags_and_admits_a_connected_socket_on_its_port() + { + use std::os::fd::{AsRawFd, BorrowedFd}; + + let peer = std::net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind the peer"); + let peer_addr = peer.local_addr().expect("peer local address"); + + let plain = std::net::UdpSocket::bind(("0.0.0.0", 0)).expect("failed to bind the holder"); + { + let probe = socket2::SockRef::from(&plain); + assert!( + !probe.reuse_address().expect("read SO_REUSEADDR"), + "precondition: a plain bind must arrive without SO_REUSEADDR", + ); + assert!( + !probe.reuse_port().expect("read SO_REUSEPORT"), + "precondition: a plain bind must arrive without SO_REUSEPORT", + ); + } + + let adopted = UdpRawSocket::adopt(plain, 65536, 65536).expect("failed to adopt the holder"); + + let joined = super::open_connected_fd(adopted.local_addr(), peer_addr, 65536, 65536); + // SAFETY: `adopted` owns this fd and outlives every use of the borrow. + let fd = unsafe { BorrowedFd::borrow_raw(adopted.as_raw_fd()) }; + let flags = socket2::SockRef::from(&fd); + let reuse_address = flags.reuse_address().expect("read SO_REUSEADDR"); + let reuse_port = flags.reuse_port().expect("read SO_REUSEPORT"); + + if let Err(err) = &joined { + panic!("a connected socket must be able to bind the adopted socket's port: {err}"); + } + assert!(reuse_address, "the adopted socket must carry SO_REUSEADDR"); + assert!(reuse_port, "the adopted socket must carry SO_REUSEPORT"); + } + #[tokio::test] async fn test_async_udp_socket_send_recv() { let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) diff --git a/src/transport/udp/io/unix.rs b/src/transport/udp/io/unix.rs index 60a4cb43..32341810 100644 --- a/src/transport/udp/io/unix.rs +++ b/src/transport/udp/io/unix.rs @@ -125,6 +125,8 @@ impl UdpRawSocket { /// Adopt an existing bound UDP socket. /// /// This preserves socket identity/NAT mapping created by bootstrap code. + /// The adopted socket is also made joinable by per-peer connected + /// sockets, which bind its local address. pub fn adopt( socket: std::net::UdpSocket, recv_buf_size: usize, @@ -135,6 +137,15 @@ impl UdpRawSocket { sock.set_nonblocking(true) .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + // A per-peer connected socket later binds this socket's own address, + // and the kernel admits that joiner only when the holder carries a + // reuse flag too. The socket arrives already bound, so setting the + // flags here cannot change which port it was given, unlike flags set + // ahead of a port-zero bind; see the comment in `open`. Best-effort, + // as there: without them only the connected fast path is refused. + let _ = sock.set_reuse_port(true); + let _ = sock.set_reuse_address(true); + sock.set_recv_buffer_size(recv_buf_size) .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; sock.set_send_buffer_size(send_buf_size) From 17509445d8101ea2d6ff86433d78a1a9a346a8c8 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 14 Sep 2026 14:32:43 +0000 Subject: [PATCH 3/3] test(transport/udp): measure duplicate ephemeral ports for reuse flags set before and after bind A reuse flag set before a port-zero bind lets the kernel hand out a port that another flagged socket already holds; set after the bind it only lets a later socket join. The adoption fix depends on the second half, and the first half shows up only when hundreds of sockets are bound concurrently, which no ordinary test reaches. Add an ignored measurement that binds 500 sockets from eight threads released together, adopting each one as it is bound and holding all of them, over five trials, in three arms: flags set before the bind, a plain bind followed by adopt, and plain binds made while 200 sockets flagged before their own bind are held. It fails if the before-bind arm shows no duplicates, because such a run could not have seen the hazard, and it fails if either other arm shows any. On the development host, port range 32768-60999, three runs with the adoption fix gave before-bind duplicates of [5, 6, 4, 6, 5], [5, 7, 5, 1, 6] and [6, 2, 3, 4, 5] per 500 binds, and zero in both other arms in every trial. A run without the fix, where adopt sets no flags, gave [3, 2, 8, 6, 3] and zeros. Building the adopt arm with flags set before the bind turns it red, at [3, 5, 6, 4, 7]. It is ignored because the counts depend on load and nothing gating should run it. It measures the kernel behaviour the adoption path relies on, not the traversal binds themselves, so a later change that flagged those binds before binding would not turn it red. --- src/transport/udp/io/mod.rs | 125 ++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs index eae1001c..64a25cd3 100644 --- a/src/transport/udp/io/mod.rs +++ b/src/transport/udp/io/mod.rs @@ -168,6 +168,131 @@ mod tests { assert_eq!(src, addr1); } + /// Measurement: duplicate local ports among many concurrently held + /// port-zero UDP binds, with reuse flags set before the bind and with + /// reuse flags set after it by `adopt`. + /// + /// A reuse flag set before a port-zero bind lets the kernel hand out a + /// port another flagged socket already holds; set after the bind it only + /// lets a later socket join. Three arms, each run over several trials: + /// + /// - `pre` flags each socket and then binds it, which is the ordering + /// that must not be used for a traversal socket. It is built here from + /// socket2 because no such helper exists in the tree. It must show + /// duplicates: if it shows none, this run could not have seen the + /// hazard, and the measurement fails rather than passing. + /// - `post` binds exactly as the traversal path does and then adopts, + /// so `adopt` flags each socket while other threads are still binding. + /// - `orphan` holds sockets flagged before their bind, as a connected + /// socket is, and counts later plain binds handed one of their ports. + /// + /// Coverage gap: nothing that gates runs this, because the port + /// allocator is probabilistic. It measures the kernel behaviour the + /// adoption path relies on, not the traversal binds themselves, so a + /// change that flagged those binds before binding would not turn it red. + /// + /// Run with: + /// cargo test --lib transport::udp::io::tests::measure_duplicate_ephemeral_ports -- --ignored --nocapture + #[cfg(target_os = "linux")] + #[test] + #[ignore = "probabilistic kernel port-allocator measurement; run explicitly with --ignored --nocapture"] + fn measure_duplicate_ephemeral_ports_for_reuse_flags_set_before_and_after_bind() { + use socket2::{Domain, Protocol, Socket, Type}; + use std::collections::HashSet; + use std::sync::Barrier; + + const N: usize = 500; + const TRIALS: usize = 5; + const THREADS: usize = 8; + const HELD: usize = 200; + const BUF: usize = 65536; + + fn plain_bind() -> std::net::UdpSocket { + std::net::UdpSocket::bind(("0.0.0.0", 0)).expect("plain port-zero bind") + } + + fn flagged_bind() -> std::net::UdpSocket { + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)).expect("socket"); + sock.set_reuse_port(true).expect("set SO_REUSEPORT"); + sock.set_reuse_address(true).expect("set SO_REUSEADDR"); + let any: SocketAddr = "0.0.0.0:0".parse().unwrap(); + sock.bind(&any.into()).expect("flagged port-zero bind"); + sock.into() + } + + /// Bind `n` sockets across `THREADS` threads released together, + /// adopting each one as soon as it is bound, and hold every socket + /// until all the threads have finished. + fn bind_concurrently(n: usize, bind: fn() -> std::net::UdpSocket) -> Vec { + let barrier = Barrier::new(THREADS); + std::thread::scope(|s| { + let handles: Vec<_> = (0..THREADS) + .map(|t| { + let share = n / THREADS + usize::from(t < n % THREADS); + let barrier = &barrier; + s.spawn(move || { + barrier.wait(); + (0..share) + .map(|_| UdpRawSocket::adopt(bind(), BUF, BUF).expect("adopt")) + .collect::>() + }) + }) + .collect(); + handles + .into_iter() + .flat_map(|h| h.join().expect("bind thread panicked")) + .collect() + }) + } + + fn duplicates(socks: &[UdpRawSocket]) -> usize { + let distinct: HashSet = socks.iter().map(|s| s.local_addr().port()).collect(); + socks.len() - distinct.len() + } + + let mut pre = Vec::with_capacity(TRIALS); + let mut post = Vec::with_capacity(TRIALS); + let mut orphan = Vec::with_capacity(TRIALS); + for _ in 0..TRIALS { + pre.push(duplicates(&bind_concurrently(N, flagged_bind))); + post.push(duplicates(&bind_concurrently(N, plain_bind))); + + let held: Vec = (0..HELD).map(|_| flagged_bind()).collect(); + let held_ports: HashSet = held + .iter() + .map(|s| s.local_addr().expect("held local address").port()) + .collect(); + let later = bind_concurrently(N, plain_bind); + orphan.push( + later + .iter() + .filter(|s| held_ports.contains(&s.local_addr().port())) + .count(), + ); + } + + eprintln!("duplicate ports per {N} binds, {THREADS} threads, {TRIALS} trials"); + eprintln!(" pre (flags before bind): {pre:?}"); + eprintln!(" post (flags after bind): {post:?}"); + eprintln!(" orphan ({HELD} held pre-flagged, later plain binds): {orphan:?}"); + + assert!( + pre.iter().sum::() > 0, + "the before-bind arm showed no duplicates, so this run could not have seen the \ + hazard at N = {N}; raise N rather than reading the other arms as clean", + ); + assert_eq!( + post.iter().sum::(), + 0, + "flags set after bind: {post:?}" + ); + assert_eq!( + orphan.iter().sum::(), + 0, + "orphan collisions: {orphan:?}" + ); + } + /// Microbench: compare per-packet `recv_from` (single recvmsg syscall + /// task wakeup per datagram — the macOS pre-recvmsg_x baseline) vs /// `recv_batch` (the new recvmsg_x path, up to 32 datagrams per syscall).