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)