diff --git a/CHANGELOG.md b/CHANGELOG.md index 879bf7d6..3b884b93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +#### Data-plane / transports + +- A peer that moves to a new address now loses the per-peer `connect(2)`-ed UDP + socket pinned to the address it left. `set_current_addr` returns whether the + address actually changed so the caller can drop the stale socket, and the + 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. + #### Peering - A heartbeat whose send failed no longer counts as one that was delivered. The diff --git a/src/node/dataplane/encrypted.rs b/src/node/dataplane/encrypted.rs index e06880ab..0f58e0a7 100644 --- a/src/node/dataplane/encrypted.rs +++ b/src/node/dataplane/encrypted.rs @@ -264,6 +264,7 @@ impl Node { let ce_flag = header.flags & FLAG_CE != 0; let sp_flag = header.flags & FLAG_SP != 0; + let mut address_changed = false; if let Some(peer) = self.peers.get_mut(&node_addr) { if let Some(mmp) = peer.mmp_mut() { mmp.receiver.record_recv( @@ -275,12 +276,28 @@ impl Node { ); let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now_ms); } - peer.set_current_addr(packet.transport_id, packet.remote_addr.clone()); + address_changed = + peer.set_current_addr(packet.transport_id, packet.remote_addr.clone()); peer.link_stats_mut() .record_recv(packet.data.len(), packet.timestamp_ms); peer.touch(packet.timestamp_ms); } + // Address rotation invalidates the per-peer connect()-ed UDP + // socket. Drop the connected socket + drain so the wildcard + // listen socket takes over until the new 5-tuple settles. The + // decrypt-worker completion path in + // `process_authentic_fmp_plaintext` does the same thing with + // the same flag; this is the in-line decrypt twin of it. + #[cfg(any(target_os = "linux", target_os = "macos"))] + if address_changed { + self.clear_connected_udp_for_peer(&node_addr); + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + { + let _ = address_changed; + } + // Dispatch to link message handler self.dispatch_link_message(&node_addr, link_message, ce_flag) .await; diff --git a/src/node/tests/connected_udp.rs b/src/node/tests/connected_udp.rs new file mode 100644 index 00000000..a713eeef --- /dev/null +++ b/src/node/tests/connected_udp.rs @@ -0,0 +1,219 @@ +//! Per-peer connected UDP sockets against the in-line decrypt path. +//! +//! A `connect(2)`-ed UDP socket is pinned to one 5-tuple. When the peer +//! moves, the address the socket was opened against is gone, but the +//! socket is still installed and the send path prefers it over the +//! wildcard listen socket. `ActivePeer::set_current_addr` returns +//! whether the address actually changed precisely so the caller can +//! drop the stale socket, and both post-decrypt paths have to act on +//! that return: the decrypt-worker completion path +//! (`process_authentic_fmp_plaintext`) and the in-line one +//! (`handle_encrypted_frame`). These tests cover the in-line path, +//! which is the one the worker path's own coverage does not reach. +//! +//! `bool` carries no `#[must_use]`, so discarding the return here is +//! silent under `-D warnings`; the assertions below are what makes the +//! difference between binding it and dropping it observable. + +use super::*; +use crate::noise::NoiseSession; +use crate::proto::fmp::wire::{build_encrypted, build_established_header, prepend_inner_header}; + +/// The address `seed_completed_connection` promotes a peer on, and so +/// the peer's `current_addr` before anything rotates it. +const PROMOTED_ADDR: &str = "127.0.0.1:5000"; + +/// The address the peer is made to move to. +const ROAMED_ADDR: &str = "127.0.0.1:5001"; + +/// Build a promoted peer and hand back the far side's Noise session. +/// +/// [`seed_completed_connection`] runs both legs of the handshake and +/// then drops the responder, so nothing outside it can produce a frame +/// the node will actually authenticate. This is the same seeding with +/// the responder's session kept, which is what lets these tests reach +/// the post-decrypt side effects rather than stopping at the AEAD. +/// +/// Returns the node, the peer's `NodeAddr`, the session index an +/// inbound frame must name to be routed to that peer, and the session +/// to encrypt those frames with. +fn promoted_peer_with_the_far_side_session( + transport_id: TransportId, +) -> (Node, NodeAddr, SessionIndex, NoiseSession) { + let mut node = make_node(); + let link_id = LinkId::new(1); + + let peer_identity_full = Identity::generate(); + // from_pubkey_full, not from_pubkey: the ECDH needs the parity bit. + let peer_identity = PeerIdentity::from_pubkey_full(peer_identity_full.pubkey_full()); + + let our_index = node.index_allocator.allocate().unwrap(); + node.seed_handshake_machine( + HandshakeSeed::outbound(link_id, peer_identity, 1_000) + .with_our_index(our_index) + .with_their_index(SessionIndex::new(42)) + .with_transport_id(transport_id) + .with_source_addr(TransportAddr::from_string(PROMOTED_ADDR)), + ) + .unwrap(); + + let our_keypair = node.identity().keypair(); + let startup_epoch = node.startup_epoch(); + let msg1 = node + .peer_machines + .get_mut(&link_id) + .unwrap() + .start_handshake(our_keypair, startup_epoch, 1_000) + .unwrap(); + + let mut responder = inbound_leg(LinkId::new(999), 1_000); + let mut responder_epoch = [0u8; 8]; + rand::Rng::fill_bytes(&mut rand::rng(), &mut responder_epoch); + let msg2 = responder + .receive_handshake_init(peer_identity_full.keypair(), responder_epoch, &msg1, 1_000) + .unwrap(); + + node.peer_machines + .get_mut(&link_id) + .unwrap() + .complete_handshake(&msg2, 1_000) + .unwrap(); + + let far_side_session = responder + .take_session() + .expect("the responder holds a session once it has written msg2"); + + node.promote_connection(link_id, peer_identity, 2_000) + .unwrap(); + let node_addr = *peer_identity.node_addr(); + let our_index = node + .get_peer(&node_addr) + .and_then(|p| p.our_index()) + .expect("a promoted peer carries the index it was allocated"); + + (node, node_addr, our_index, far_side_session) +} + +/// Encrypt one well-formed established frame from the far side. +/// +/// The link message is a heartbeat (`0x51`), which the dispatcher +/// handles as a no-op — these tests are about the side effects that run +/// before the dispatch, so the message must not have any of its own. +fn far_side_frame(session: &mut NoiseSession, receiver_idx: SessionIndex) -> Vec { + let inner = prepend_inner_header(0, &[0x51]); + let counter = session.current_send_counter(); + let header = build_established_header(receiver_idx, counter, 0, inner.len() as u16); + let ciphertext = session.encrypt_with_aad(&inner, &header).unwrap(); + build_encrypted(&header, &ciphertext) +} + +/// Install a real `connect()`-ed UDP socket on a peer, the way the +/// tick-driven activation in `dataplane::connected_udp` does. +/// +/// The socket is opened against the loopback discard port: nothing is +/// ever sent through it, and the tests only care whether the handle is +/// still installed afterwards. +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn install_connected_udp(node: &mut Node, addr: &NodeAddr, transport_id: TransportId) { + let local: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + let peer_sa: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap(); + + let owned = crate::transport::udp::open_connected_fd(local, peer_sa, 65_536, 65_536) + .expect("open a connected UDP socket"); + let socket = std::sync::Arc::new(crate::transport::udp::ConnectedPeerSocket::from_fd( + owned, peer_sa, local, + )); + let (packet_tx, _packet_rx) = packet_channel(8); + let drain = crate::transport::udp::PeerRecvDrain::spawn( + socket.clone(), + transport_id, + peer_sa, + packet_tx, + ) + .expect("spawn the peer recv drain"); + + node.get_peer_mut(addr) + .expect("peer present") + .set_connected_udp(socket, drain); +} + +/// **The defect.** +/// +/// The in-line decrypt path called `set_current_addr` as a bare +/// statement and dropped its return, so a peer could roam, have its +/// `current_addr` updated, and keep a connected socket pinned to the +/// 5-tuple it had just left. The send path prefers that socket while it +/// is installed, so every frame after the move goes out to an address +/// the peer is no longer at. +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[tokio::test] +async fn a_peer_that_roams_loses_the_connected_socket_pinned_to_the_address_it_left() { + let transport_id = TransportId::new(1); + let (mut node, node_addr, our_index, mut far_side) = + promoted_peer_with_the_far_side_session(transport_id); + + install_connected_udp(&mut node, &node_addr, transport_id); + assert!( + node.get_peer(&node_addr).unwrap().connected_udp().is_some(), + "precondition: the peer holds a connected socket before it moves" + ); + + let frame = far_side_frame(&mut far_side, our_index); + node.handle_encrypted_frame(ReceivedPacket::new( + transport_id, + TransportAddr::from_string(ROAMED_ADDR), + frame, + )) + .await; + + let peer = node + .get_peer(&node_addr) + .expect("the peer survives an authentic frame"); + assert_eq!( + peer.current_addr(), + Some(&TransportAddr::from_string(ROAMED_ADDR)), + "precondition for the assertion below: the frame must have been \ + authenticated and the rotation recorded, or the test proves nothing" + ); + assert!( + peer.connected_udp().is_none(), + "a socket pinned to the address the peer has left must not survive \ + the rotation" + ); +} + +/// **The healthy path.** +/// +/// A frame from the address the peer is already on changes nothing, so +/// the connected socket has to stay. A fix that cleared unconditionally +/// would tear down and reopen the socket on every single frame. +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[tokio::test] +async fn a_frame_from_the_address_the_peer_is_already_on_keeps_the_connected_socket() { + let transport_id = TransportId::new(1); + let (mut node, node_addr, our_index, mut far_side) = + promoted_peer_with_the_far_side_session(transport_id); + + install_connected_udp(&mut node, &node_addr, transport_id); + + let frame = far_side_frame(&mut far_side, our_index); + node.handle_encrypted_frame(ReceivedPacket::new( + transport_id, + TransportAddr::from_string(PROMOTED_ADDR), + frame, + )) + .await; + + let peer = node + .get_peer(&node_addr) + .expect("the peer survives an authentic frame"); + assert_eq!( + peer.current_addr(), + Some(&TransportAddr::from_string(PROMOTED_ADDR)), + "the peer has not moved" + ); + assert!( + peer.connected_udp().is_some(), + "a frame from the address already in use must leave the socket alone" + ); +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index dd56ed0d..9b793f78 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -11,6 +11,7 @@ mod ble; mod bloom; mod bloom_poison; mod bootstrap; +mod connected_udp; mod control; mod decrypt_failure; mod disconnect;