From 46ded33f8e39112d265b6b45b52e2115de5fd8d5 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 2 Oct 2026 16:14:26 +0000 Subject: [PATCH] Answer a different msg1 from an address with a pending handshake with its own handshake A msg1 from an address whose entry named a pending inbound handshake was treated as a duplicate and answered with that handshake's stored msg2, whatever the msg1 said. XX msg2 is bound to the initiator's ephemeral key from msg1, so a msg2 made for another msg1 cannot be read by the sender. A handshake left pending at a peer's address by a replayed msg1, or by the peer's own abandoned attempt, therefore answered every genuine msg1 from that peer (a rekey, or a fresh dial after a restart) with a msg2 it could not read, until the stale handshake was reaped at the handshake timeout. Each inbound handshake now records the msg1 it answered, in the carrier slot an outbound handshake uses for its own msg1 (no msg1 resend path runs on an inbound handshake). A byte-identical msg1, which is what an initiator's resend sends, is still answered from the stored msg2. A different msg1 gets its own handshake, and the pending one is left for its msg3 or its reap: replacing it instead would let anyone able to send from the address discard a genuine handshake in the window between msg2 and msg3. When a handshake that another at the same address displaced is removed first, the newer one inherits the link it displaced, so the address entry still returns to a live peer link when the newer one goes. The cost is a new handshake per distinct msg1 where there was one per address. A sender spoofing an established peer's address is admitted even with inbound connections off and is metered on the established-link msg1 bucket; before, it held one handshake per spoofed address until the reap and further msg1s were answered from the stored msg2 at no crypto cost. Now each distinct msg1 costs the msg2 crypto, a session index, a link and a handshake machine, held until the handshake timeout, bounded by that bucket's burst and rate: at defaults (128 burst, 6.4 per second, 30 s timeout) about 320 handshakes at once, the same order as the stranger bucket already admits from arbitrary addresses. Tests send two distinct msg1s from one identity and check the second gets a msg2 it can read and completes, and check a stale handshake at a peer's address no longer blocks the peer's rekey, including the address entry once the stale handshake is reaped mid-rekey. Both fail without the fix; the existing duplicate-msg1 resend test is unchanged. No wire format change. --- src/node/handlers/handshake.rs | 52 ++++++++-- src/node/mod.rs | 16 ++- src/node/tests/handshake.rs | 180 +++++++++++++++++++++++++++++++++ src/peer/machine.rs | 2 + src/proto/fmp/state.rs | 9 +- 5 files changed, 242 insertions(+), 17 deletions(-) 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) {