diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 8523dc3f..38a8f982 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -292,7 +292,13 @@ impl Node { // // With XX, we can't do identity-based checks in msg1 (no identity yet). // We can only detect duplicates by address: if we already have an inbound - // link from this address with a pending connection, resend msg2. + // link from this address with a pending connection that answered this + // same msg1, resend its msg2. A different msg1 is a new attempt (a + // rekey, a fresh dial after a restart, or a stale leg left by a replay + // or an abandoned attempt) and gets its own leg: XX msg2 is bound to + // the initiator's ephemeral in msg1, so the pending leg's msg2 cannot + // answer it. The pending leg is not replaced, so a msg1 from this + // address cannot discard a genuine leg awaiting its msg3. // If we have an active peer on this address, it could be a restart or // rekey — but we can't tell until msg3 reveals identity. For now, allow // the new handshake to proceed. Identity-based checks happen in handle_msg3. @@ -304,7 +310,25 @@ impl Node { // Check if this link belongs to an already-promoted active peer let is_active_peer = self.peers.values().any(|p| p.link_id() == existing_link_id); - if !is_active_peer { + if is_active_peer { + // Active peer on this address — allow the new handshake. + // Identity checks (restart, rekey) deferred to handle_msg3. + debug!( + transport_id = %packet.transport_id, + remote_addr = %packet.remote_addr, + existing_link_id = %existing_link_id, + "XX msg1 from address with active peer — proceeding (identity check deferred to msg3)" + ); + } else if self.is_new_msg1_attempt(existing_link_id, &packet.data) { + // A pending handshake answered a different msg1 — give + // this one its own leg and leave that one for its msg3. + debug!( + transport_id = %packet.transport_id, + remote_addr = %packet.remote_addr, + existing_link_id = %existing_link_id, + "Msg1 differs from the one the pending handshake at this address answered; starting a new handshake" + ); + } else { // Genuinely pending handshake — resend msg2 let msg2_bytes = self.find_stored_msg2(existing_link_id); if let Some(msg2) = msg2_bytes { @@ -332,14 +356,6 @@ impl Node { } return; } - // Active peer on this address — allow the new handshake. - // Identity checks (restart, rekey) deferred to handle_msg3. - debug!( - transport_id = %packet.transport_id, - remote_addr = %packet.remote_addr, - existing_link_id = %existing_link_id, - "XX msg1 from address with active peer — proceeding (identity check deferred to msg3)" - ); } else { // Outbound link to this address — cross-connection. // Allow the inbound handshake to proceed. @@ -446,8 +462,11 @@ impl Node { // the Noise handles it carries. machine.park_inbound_msg2_sent(our_index); // Store the framed msg2 on the surviving carrier for duplicate-msg1 - // resend while the handshake is still pending. + // resend while the handshake is still pending, and the msg1 it answers, + // which tells that resend apart from a new attempt. An inbound leg + // never resends msg1, so the resend deadline is moot. machine.set_conn_handshake_msg2(wire_msg2.clone()); + machine.set_conn_handshake_msg1(packet.data.clone(), 0); self.peer_machines.insert(link_id, machine); if let Some(transport) = self.transports.get(&packet.transport_id) { @@ -485,6 +504,17 @@ impl Node { .insert((packet.transport_id, our_index.as_u32()), link_id); } + /// Whether `msg1` is a different attempt from the one the pending inbound + /// leg on `link` answered. A leg with no machine or no recorded msg1 is + /// treated as answering it, which keeps the duplicate path's behaviour for + /// it. + fn is_new_msg1_attempt(&self, link: LinkId, msg1: &[u8]) -> bool { + self.peer_machines + .get(&link) + .and_then(|machine| machine.conn_handshake_msg1()) + .is_some_and(|answered| answered != msg1) + } + /// Find stored msg2 bytes for a given link (pre- or post-promotion). /// /// Checks the control machine's carrier (if still pending) and then the diff --git a/src/node/mod.rs b/src/node/mod.rs index 089109e5..20ac6574 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -2876,10 +2876,18 @@ impl Node { let displaced = self.displaced_links.remove(link_id); let link = self.links.remove(link_id)?; self.addr_to_link.retain(|_, mapped| *mapped != *link_id); - if let Some((key, prior)) = displaced - && self.links.contains_key(&prior) - { - self.addr_to_link.entry(key).or_insert(prior); + if let Some((key, prior)) = displaced { + // A newer leg at the same address displaced this one; it now + // displaces what this one did, or its disposal would find a dead + // link and drop the key from a link that is still live. + for (newer_key, newer_prior) in self.displaced_links.values_mut() { + if *newer_prior == *link_id && *newer_key == key { + *newer_prior = prior; + } + } + if self.links.contains_key(&prior) { + self.addr_to_link.entry(key).or_insert(prior); + } } Some(link) } diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index f3447743..a7f8104f 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -5225,3 +5225,183 @@ async fn an_unreadable_rekey_msg3_leaves_the_rekey_leg_for_the_resent_msg3() { stop_hs(&mut initiator).await; stop_hs(&mut responder).await; } + +// =========================================================================== +// A msg1 that differs from the one a pending inbound leg answered gets its own +// leg. XX msg2 is bound to the initiator's ephemeral in msg1, so answering a +// new msg1 with the pending leg's stored msg2 gives the sender a msg2 it can +// never read; a byte-identical msg1 is still a resend, answered from the +// stored msg2 (`test_xx_duplicate_msg1_resends_msg2`). +// =========================================================================== + +/// A bare XX initiator with `keypair`: its handshake and the framed msg1 it +/// sent under `idx`. Stands in for one dial attempt, so a test can put +/// several distinct msg1s from one identity on the wire. +fn bare_msg1( + keypair: secp256k1::Keypair, + epoch: [u8; 8], + idx: SessionIndex, +) -> (crate::noise::HandshakeState, Vec) { + let mut hs = crate::noise::HandshakeState::new_initiator(keypair); + hs.set_local_epoch(epoch); + let noise_msg1 = hs.write_message_1().unwrap(); + (hs, crate::proto::fmp::wire::build_msg1(idx, &noise_msg1)) +} + +/// Read a framed msg2 on a bare initiator's handshake and return the framed +/// msg3 it answers with, sent under `our_idx`. +fn bare_msg3( + mut hs: crate::noise::HandshakeState, + msg2: &[u8], + our_idx: SessionIndex, +) -> Result, crate::noise::NoiseError> { + use crate::proto::fmp::wire::{Msg2Header, build_msg3}; + + let header = Msg2Header::parse(msg2).expect("msg2 header"); + let noise = &msg2[header.noise_msg2_offset..]; + let (base, extra) = noise.split_at(crate::noise::HANDSHAKE_MSG2_SIZE.min(noise.len())); + hs.read_message_2(base)?; + if !extra.is_empty() { + hs.decrypt_payload(extra)?; + } + let noise_msg3 = hs.write_message_3()?; + Ok(build_msg3(our_idx, header.sender_idx, &noise_msg3)) +} + +/// Send `wire` from `from`'s socket to `to` and have `to` handle it as msg1. +async fn deliver_msg1(from: &mut HsNode, to: &mut HsNode, wire: &[u8]) { + from.node + .transports + .get(&from.transport_id) + .unwrap() + .send(&to.addr, wire) + .await + .expect("send msg1"); + let pkt = recv_phase(&mut to.packet_rx, 1, "msg1").await; + to.node.handle_msg1(pkt).await; +} + +#[tokio::test] +async fn a_different_msg1_from_an_address_with_a_pending_leg_gets_a_msg2_it_can_read() { + let mut initiator = make_hs_node(Config::new()).await; + let mut responder = make_hs_node(Config::new()).await; + let keypair = initiator.node.identity().keypair(); + let epoch = initiator.node.startup_epoch(); + + // A first attempt the initiator abandons: its leg stays pending at the + // responder. + let (_abandoned, first) = bare_msg1(keypair, epoch, SessionIndex::new(0x1001)); + deliver_msg1(&mut initiator, &mut responder, &first).await; + recv_phase(&mut initiator.packet_rx, 2, "first msg2").await; + + // The second attempt must get a msg2 of its own. + let second_idx = SessionIndex::new(0x1002); + let (hs, second) = bare_msg1(keypair, epoch, second_idx); + deliver_msg1(&mut initiator, &mut responder, &second).await; + let msg2 = recv_phase(&mut initiator.packet_rx, 2, "second msg2").await; + let msg3 = bare_msg3(hs, &msg2.data, second_idx) + .expect("the second attempt must be able to read the msg2 it gets"); + assert_eq!( + responder.node.connection_count(), + 2, + "the second attempt has its own leg" + ); + + responder + .node + .handle_msg3(ReceivedPacket { + transport_id: responder.transport_id, + remote_addr: initiator.addr.clone(), + data: msg3, + timestamp_ms: Node::now_ms(), + }) + .await; + let peer_addr = + *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr(); + assert!( + responder.node.get_peer(&peer_addr).is_some(), + "the second attempt completes" + ); + assert_eq!( + responder.node.connection_count(), + 1, + "the abandoned leg still waits for its msg3 or its reap" + ); + #[cfg(debug_assertions)] + responder.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} + +#[tokio::test] +async fn a_stale_leg_at_a_peer_address_does_not_block_the_peer_rekey() { + let mut initiator = make_hs_node(rekey_config()).await; + let mut responder = make_hs_node(rekey_config()).await; + let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await; + responder.node.handle_msg3(msg3).await; + let peer_addr = + *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr(); + let responder_addr = + *PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full()).node_addr(); + for (node, addr) in [ + (&mut initiator.node, responder_addr), + (&mut responder.node, peer_addr), + ] { + node.get_peer_mut(&addr) + .unwrap() + .test_backdate_session_established(std::time::Duration::from_secs(120)); + } + let peer_link = responder.node.get_peer(&peer_addr).unwrap().link_id(); + + // A stale leg at the peer's address: a msg1 from some other identity + // arriving from it. Its msg2 lands at the initiator and is drained here, + // so the rekey below cannot mistake it for its own. + let (_stale_hs, stale) = bare_msg1( + Identity::generate().keypair(), + [0x33; 8], + SessionIndex::new(0x2001), + ); + deliver_msg1(&mut initiator, &mut responder, &stale).await; + recv_phase(&mut initiator.packet_rx, 2, "stale msg2").await; + let stale_leg = link_at(&responder, &initiator.addr).unwrap(); + assert_ne!(stale_leg, peer_link); + + // The peer's rekey msg1 gets its own leg. + initiator.node.check_rekey().await; + let msg1 = recv_phase(&mut responder.packet_rx, 1, "rekey msg1").await; + responder.node.handle_msg1(msg1).await; + let rekey_leg = link_at(&responder, &initiator.addr).unwrap(); + + // The stale leg is reaped while the rekey is in flight. + responder + .node + .cleanup_stale_connection(stale_leg, Node::now_ms()) + .await; + assert!(!responder.node.links.contains_key(&stale_leg)); + + let msg2 = recv_phase(&mut initiator.packet_rx, 2, "rekey msg2").await; + initiator.node.handle_msg2(msg2).await; + let msg3 = recv_phase(&mut responder.packet_rx, 3, "rekey msg3").await; + responder.node.handle_msg3(msg3).await; + assert!( + responder + .node + .get_peer(&peer_addr) + .unwrap() + .pending_new_session() + .is_some(), + "the rekey completes despite the stale leg" + ); + assert_ne!(rekey_leg, stale_leg, "the rekey msg1 had its own leg"); + assert_eq!( + link_at(&responder, &initiator.addr), + Some(peer_link), + "the peer's address key returns to the peer's link past the reaped leg" + ); + #[cfg(debug_assertions)] + responder.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 131b441e..a27e7fb1 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -1184,6 +1184,7 @@ impl PeerMachine { /// Stored wire-format msg1 of the surviving carrier — the resend source for /// the outbound handshake retransmit, now that the leg no longer carries it. + /// On an inbound leg, the msg1 the leg answered. pub(crate) fn conn_handshake_msg1(&self) -> Option<&[u8]> { self.conn.handshake_msg1() } @@ -1196,6 +1197,7 @@ impl PeerMachine { /// Store the wire-format msg1 for resend on the surviving carrier and record /// the first resend deadline, mirroring the leg's start-of-handshake write. + /// An inbound leg stores the msg1 it answered here; it never resends it. pub(crate) fn set_conn_handshake_msg1(&mut self, msg1: Vec, first_resend_at_ms: u64) { self.conn.set_handshake_msg1(msg1, first_resend_at_ms); } diff --git a/src/proto/fmp/state.rs b/src/proto/fmp/state.rs index 606d022a..eb7e81ba 100644 --- a/src/proto/fmp/state.rs +++ b/src/proto/fmp/state.rs @@ -106,7 +106,11 @@ pub struct ConnectionState { peer_profile: Option, // === Handshake Resend === - /// Wire-format msg1 bytes for resend (initiator only). + /// Wire-format msg1 bytes. On an outbound leg, the msg1 we sent, kept for + /// resend. On an inbound leg, the msg1 the leg answered, compared with a + /// later msg1 from the same address to tell a resend from a new attempt; + /// every msg1 resend path runs only on a leg that sent msg1, so an inbound + /// leg's copy is never sent. handshake_msg1: Option>, /// Wire-format msg2 bytes for resend (responder only). @@ -372,7 +376,8 @@ impl ConnectionState { // === Handshake Resend === - /// Store the wire-format msg1 bytes for resend and reset the resend counter. + /// Store the wire-format msg1 bytes (see the field) and reset the resend + /// counter. /// The first-resend deadline is scheduled by the shell timer driver, not /// tracked here. pub fn set_handshake_msg1(&mut self, msg1: Vec, _first_resend_at_ms: u64) {