Merge maint into master, carrying the rekey msg2 rollback and adopted-socket reuse flags

maint's changelog paragraph for the moved-peer connected socket fix is
dropped again here, as in the earlier merge that brought that fix across:
master already documents it in its own words. The new maint entries keep
their sections.

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

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;
             }
 

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();
+    }
+}

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<Vec<u8>> = 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<Vec<u8>> = 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<ReceivedPacket> =
+        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<Vec<u8>> = 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<Vec<u8>> = 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.

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)
     // ========================================================================

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()?;
 

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<UdpRawSocket> {
+            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::<Vec<_>>()
+                        })
+                    })
+                    .collect();
+                handles
+                    .into_iter()
+                    .flat_map(|h| h.join().expect("bind thread panicked"))
+                    .collect()
+            })
+        }
+
+        fn duplicates(socks: &[UdpRawSocket]) -> usize {
+            let distinct: HashSet<u16> = 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<std::net::UdpSocket> = (0..HELD).map(|_| flagged_bind()).collect();
+            let held_ports: HashSet<u16> = 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::<usize>() > 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::<usize>(),
+            0,
+            "flags set after bind: {post:?}"
+        );
+        assert_eq!(
+            orphan.iter().sum::<usize>(),
+            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).

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)