diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c3fb34d..7d562842 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -431,6 +431,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 platforms with the connected-socket fast path); elsewhere the heartbeat alone carries the new address. +- 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. + +#### 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 213db09f..18d5e09b 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1199,6 +1199,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)) => { @@ -1238,6 +1239,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, @@ -1258,14 +1279,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/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/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 4369481b..81ba9b4e 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -1186,11 +1186,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], @@ -1203,7 +1219,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()?; diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs index 8acbb630..64a25cd3 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) @@ -124,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). 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)