From d6785f29709c99417f4ce314e64e0c7e926239f6 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 02:54:00 +0000 Subject: [PATCH] node: make the inbound handshake machine persistent across its leg's life The inbound control machine was a msg3-time throwaway: handle_msg3 built a transient, stepped it once for the decision, and discarded it, with promote_connection birthing a fresh established() machine. Every handshake leg now gets one persistent machine for its whole life: - handle_msg1 births the machine parked in the sent-msg2 phase alongside the window leg (msg1 crypto and the msg2 build/send stay inline; the msg2-send-failure cleanup disposes it). - handle_msg3 steps that persistent machine instead of a transient. The decision path is unchanged: the msg3 handler never reads machine state and overwrites the identity-plane fields from the wire outcome before dispatching, so every arm's decision and actions are byte-identical. - promote_connection stops rebirthing: the executor feeds the promotion result back into the machine (the shape the IK line already uses) and the machine crystallizes to Established in place. - Every path that tears down an inbound window leg now disposes its machine through remove_peer_machine: the seven inline msg3 termination arms, the six executor swap/rekey-responder consumption sites, both promote-failure arms, the single-peer leaf reject, the losing side of a cross-connection, and the msg1 send-failure cleanup. The promote-failure disposal also fixes a pre-existing leak: an outbound leg whose promotion fails had its connections entry consumed before the error, so the stale reaper could never reach its dial machine. Index-free behavior is unchanged at every site. Two test helpers that hand-roll outbound dials now insert the dial machine production inserts, and new tests pin the machine's birth phase, its post-promote crystallized state, and disposal on the failure and rekey-responder arms. --- src/node/dataplane/peer_actions.rs | 80 ++++++++++--- src/node/handlers/handshake.rs | 138 +++++++++++++--------- src/node/tests/handshake.rs | 180 +++++++++++++++++++++++++++++ src/node/tests/spanning_tree.rs | 7 ++ src/peer/machine.rs | 20 ++++ 5 files changed, 352 insertions(+), 73 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index aa21273..b5f1d42 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -20,11 +20,13 @@ //! `LinkDeadSuspected`, driving `InvalidateSendState` → `remove_active_peer`). //! //! Inbound msg1 is not machine-driven here: `handle_msg1` builds and sends -//! msg2 inline, so `PeerEvent::InboundMsg1` is never dispatched and the -//! `SendHandshake` `their_index == Some` (msg2) branch stays dormant. Inbound -//! msg3 IS machine-driven: `handle_msg3` steps a throwaway decision machine and -//! this executor performs its verdict (`PromoteToActive`, -//! `SwapToInboundSession`, `RekeyRespondTrigger`). +//! msg2 inline (persisting the leg's machine parked at `SentMsg2` alongside), +//! so `PeerEvent::InboundMsg1` is never dispatched and the `SendHandshake` +//! `their_index == Some` (msg2) branch stays dormant. Inbound msg3 IS +//! machine-driven: `handle_msg3` steps the leg's persistent machine and this +//! executor performs its verdict (`PromoteToActive`, `SwapToInboundSession`, +//! `RekeyRespondTrigger`), disposing the machine on every path that consumes +//! the leg without promoting it. //! //! The genuine inert stubs remaining are `SendRekey`, `SendLinkMessage`, and //! the connected-UDP arms. `RegisterDecryptSession` is a deliberate no-op — @@ -230,13 +232,13 @@ impl Node { PeerAction::PromoteToActive { link: promote_link } => { // Establish promote, driven through the machine. Transcribes // `handle_msg3`'s shared inbound promote block verbatim, adapted - // to the executor's ambient context. Two XX-specific choices vs - // the IK-lineage executor: (1) the decrypt-worker register stays + // to the executor's ambient context. One XX-specific choice vs + // the IK-lineage executor: the decrypt-worker register stays // INSIDE `promote_connection` (NOT relocated here) — re- - // registering would double-register; (2) NO `PromotionResolved` - // is fed back — the persistent machine is born `established()` by - // `promote_connection`, so feeding it would perturb that ctor - // state that the rekey/reap folds read. + // registering would double-register. The promoted leg's machine + // (msg1-born inbound, dial-born outbound) survives the promotion + // and is crystallized in place by the `PromotionResolved` + // feedback fed back after the Ok handling below. // Capture msg2 BEFORE `promote_connection` removes the pending // connection, so a duplicate msg1 can be answered with it. Only @@ -274,12 +276,14 @@ impl Node { self.peers.contains_key(ambient.verified_identity.node_addr()), ); } - match self.promote_connection( + let promote_result = self.promote_connection( promote_link, ambient.verified_identity, ambient.now_ms, - ) { + ); + match &promote_result { Ok(PromotionResult::Promoted(node_addr)) => { + let node_addr = *node_addr; if ambient.is_outbound { // The outbound promote logs a second line here in // addition to `promote_connection`'s "Connection @@ -319,6 +323,7 @@ impl Node { loser_link_id, node_addr, }) => { + let (loser_link_id, node_addr) = (*loser_link_id, *node_addr); // UNREACHABLE on driven XX establish paths: `Promote` // and `RestartThenPromote` (which removes the old peer // first) both imply no existing peer at promote time, so @@ -358,6 +363,7 @@ impl Node { self.reset_lookup_backoff(); } Ok(PromotionResult::CrossConnectionLost { winner_link_id }) => { + let winner_link_id = *winner_link_id; // UNREACHABLE on driven XX establish paths (see the Won // arm). Body kept byte-equivalent to next; uses the // ambient transport/addr in place of next's `packet.*`. @@ -386,13 +392,18 @@ impl Node { // The outbound promote-failure path is warn-only: it // records the reject but performs no link/index teardown // and leaves the `pending_outbound` entry for the stale- - // connection reaper. + // connection reaper. The leg's machine must go with the + // leg, though: `promote_connection` consumed the + // `connections` entry before erring, so the reaper (which + // sweeps `connections`) can never reach this link's + // machine — dropping it here is the only disposal point. warn!( target: "fips::node::handlers::handshake", link_id = %promote_link, error = %e, "Failed to promote connection" ); + self.remove_peer_machine(promote_link); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } @@ -430,10 +441,32 @@ impl Node { if let Some(idx) = ambient.our_index { let _ = self.index_allocator.free(idx); } + self.remove_peer_machine(promote_link); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } } + + // Feed the promotion outcome back into the surviving machine + // and fold the follow-up actions onto the worklist (the + // `Promoted` arm crystallizes the machine in place and emits + // `RegisterDecryptSession`, a redundant no-op here — see its + // arm). Unconditional across Ok variants; on + // `CrossConnectionLost` the losing leg's machine was disposed + // inside `promote_connection`, so the lookup misses and no + // machine-side index free can double the inline one. Disjoint + // field borrow again. + if let Ok(result) = promote_result { + let follow = match self.peer_machines.get_mut(&promote_link) { + Some(machine) => machine.step( + PeerEvent::PromotionResolved { result }, + ambient.now_ms, + &mut self.index_allocator, + ), + None => Vec::new(), + }; + queue.extend(follow); + } } PeerAction::SwapSendState { .. } => { // Initiator cutover: the live authoritative rekey-cadence @@ -557,6 +590,7 @@ impl Node { None => { self.connections.remove(&link); self.remove_link(&link); + self.remove_peer_machine(link); self.stats_mut().record_reject(RejectReason::Handshake( HandshakeReject::BadState, )); @@ -569,6 +603,7 @@ impl Node { let Some(transport_id) = peer_ref.transport_id() else { self.connections.remove(&link); self.remove_link(&link); + self.remove_peer_machine(link); self.stats_mut().record_reject(RejectReason::Handshake( HandshakeReject::BadState, )); @@ -600,9 +635,11 @@ impl Node { } // Both branches tear down the temporary inbound link fully - // (including its `addr_to_link` mapping) via `remove_link`. + // (including its `addr_to_link` mapping) via `remove_link`, + // disposing the leg's machine with it. self.connections.remove(&link); self.remove_link(&link); + self.remove_peer_machine(link); return; } PeerAction::RekeyRespondTrigger { @@ -643,6 +680,7 @@ impl Node { let Some(conn) = self.connections.get_mut(&link) else { warn!(link_id = %link, "Connection removed during rekey msg3 processing"); self.links.remove(&link); + self.remove_peer_machine(link); self.stats_mut().record_reject(RejectReason::Handshake( HandshakeReject::UnknownConnection, )); @@ -658,6 +696,7 @@ impl Node { warn!("Rekey msg3: no session from handshake"); self.connections.remove(&link); self.links.remove(&link); + self.remove_peer_machine(link); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -674,12 +713,15 @@ impl Node { self.peers_by_index .insert((ambient.transport_id, our_new_index.as_u32()), peer); - // Clean up: remove the temporary connection/link. Do NOT remove - // addr_to_link — the entry must remain pointing to the original - // link so the established peer stays routable, so this uses the - // bare `links.remove` rather than the full `remove_link`. + // Clean up: remove the temporary connection/link and the leg's + // machine (the established peer keeps its own, keyed by its own + // link). Do NOT remove addr_to_link — the entry must remain + // pointing to the original link so the established peer stays + // routable, so this uses the bare `links.remove` rather than + // the full `remove_link`. self.connections.remove(&link); self.links.remove(&link); + self.remove_peer_machine(link); debug!( peer = %self.peer_display_name(&peer), diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index ad4b501..e0c4a87 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -300,6 +300,15 @@ impl Node { self.links.insert(link_id, link); self.addr_to_link.insert(addr_key, link_id); self.connections.insert(link_id, conn); + // The leg's persistent control machine is born alongside its pending + // connection, parked at `SentMsg2` awaiting msg3 (identity is unknown + // until then). Inserted before the msg2 send below so no suspension + // point observes a leg without a machine. `handle_msg3` steps this + // same machine; every teardown path disposes it with the leg. + self.peer_machines.insert( + link_id, + PeerMachine::inbound_msg2_sent(link_id, our_index, packet.timestamp_ms), + ); // Build and send msg2 response, storing for potential resend let wire_msg2 = build_msg2(our_index, header.sender_idx, &msg2_response); @@ -330,6 +339,7 @@ impl Node { self.addr_to_link .remove(&(packet.transport_id, packet.remote_addr)); let _ = self.index_allocator.free(our_index); + self.remove_peer_machine(link_id); self.msg1_rate_limiter.complete_handshake(); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); @@ -881,16 +891,18 @@ impl Node { // // Net-new outbound promote via the per-peer machine. Look up the machine // persisted at DIAL for an identified leg and step `OutboundMsg2 → - // [PromoteToActive]` in place; an anonymous-discovery leg never persisted - // a machine at dial and falls to the transient in the `None` arm. The - // transient is NOT inserted into `peer_machines` — the persistent - // Established machine is created inside `promote_connection` (next's - // invariant: `promote_connection` owns the Established machine). With no - // outbound decision core to run, the machine emits `PromoteToActive` - // unconditionally (the net-new-vs-cross-connection decision was the - // `peers.contains_key` test above, which returns on an existing peer). The - // promote tail (info log, tree/bloom/backoff, `pending_outbound` removal) - // lives in the executor's `PromoteToActive` arm. + // [PromoteToActive]` in place; the machine survives the promotion and the + // executor crystallizes it via the `PromotionResolved` feedback. An + // anonymous-discovery leg never persisted a machine at dial and falls to + // the transient in the `None` arm; the transient is NOT inserted into + // `peer_machines`, so an anonymous promote leaves no machine behind (the + // anonymous leg's machine birth lands with the outbound-side convention + // work). With no outbound decision core to run, the machine emits + // `PromoteToActive` unconditionally (the net-new-vs-cross-connection + // decision was the `peers.contains_key` test above, which returns on an + // existing peer). The promote tail (info log, tree/bloom/backoff, + // `pending_outbound` removal) lives in the executor's `PromoteToActive` + // arm. let promote_actions = match self.peer_machines.get_mut(&link_id) { Some(machine) => machine.step( PeerEvent::OutboundMsg2 { @@ -1022,6 +1034,7 @@ impl Node { self.connections.get(&link_id).and_then(|c| c.our_index()); self.connections.remove(&link_id); self.remove_link(&link_id); + self.remove_peer_machine(link_id); if let Some(idx) = our_idx_to_free { let _ = self.index_allocator.free(idx); } @@ -1039,6 +1052,7 @@ impl Node { warn!(link_id = %link_id, our_profile = %our_profile, error = %e, "FMP negotiation failed"); self.connections.remove(&link_id); self.remove_link(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -1053,6 +1067,7 @@ impl Node { warn!("Identity not learned from msg3"); self.connections.remove(&link_id); self.remove_link(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -1106,6 +1121,7 @@ impl Node { } self.connections.remove(&link_id); self.remove_link(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -1115,6 +1131,7 @@ impl Node { debug!(link_id = %link_id, "Received msg3 from self, dropping"); self.connections.remove(&link_id); self.remove_link(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -1213,6 +1230,7 @@ impl Node { let _ = self.index_allocator.free(our_index); self.connections.remove(&link_id); self.links.remove(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -1240,6 +1258,7 @@ impl Node { let _ = self.index_allocator.free(our_index); self.connections.remove(&link_id); self.links.remove(&link_id); + self.remove_peer_machine(link_id); return; } decision @ (InboundDecision::RestartThenPromote { .. } @@ -1255,8 +1274,8 @@ impl Node { "Peer restart detected (epoch mismatch), removing stale session" ); } - // Machine-driven inbound establish/rekey resolution. A TRANSIENT - // inbound machine, seeded with the msg1-allocated index, re-derives + // Machine-driven inbound establish/rekey resolution. The leg's + // PERSISTENT machine — born at msg1, parked `SentMsg2` — re-derives // the decision from the same snapshot and emits the action stream: // `[PromoteToActive]` for `Promote`; // `[InvalidateSendState, ReportLost, PromoteToActive]` for @@ -1264,22 +1283,47 @@ impl Node { // `remove_active_peer` / `note_link_dead`, in that order); // `[SwapToInboundSession]` for a simultaneous-init cross-connection; // `[RekeyRespondTrigger]` for a rekey-responder tie-break. - // The transient is never inserted into `peer_machines`; the - // persistent `established()` machine is created inside - // `promote_connection`, so there is no double-insert. The relocated + // On a promote the machine survives and crystallizes in place via + // the executor's `PromotionResolved` feedback; on the other arms the + // executor's teardown disposes it with the leg. The relocated // session-swap / promote / teardown bodies live in the executor's // `SwapToInboundSession` / `RekeyRespondTrigger` / `PromoteToActive` // / `InvalidateSendState` / `ReportLost` arms. - let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); - let actions = machine.step( - PeerEvent::InboundMsg3 { - wire, - est: snap, - our_index, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ); + let actions = match self.peer_machines.get_mut(&link_id) { + // Disjoint field borrow: `self.peer_machines` (the map entry) + // and `self.index_allocator` (the capability) are separate + // fields. + Some(machine) => machine.step( + PeerEvent::InboundMsg3 { + wire, + est: snap, + our_index, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ), + None => { + // Every inbound leg's machine is born at msg1, so a miss + // here means a teardown path dropped the machine but left + // the leg behind. Recover with a fresh machine seeded the + // way msg1 would have left it, so the step below behaves + // identically. + debug_assert!(false, "peer machine present for every pending inbound leg"); + let mut machine = + PeerMachine::inbound_msg2_sent(link_id, our_index, packet.timestamp_ms); + let actions = machine.step( + PeerEvent::InboundMsg3 { + wire, + est: snap, + our_index, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + self.peer_machines.insert(link_id, machine); + actions + } + }; let ambient = PeerActionCtx { verified_identity: peer_identity, transport_id: packet.transport_id, @@ -1316,13 +1360,14 @@ impl Node { link_id = %link_id, "Leaf node rejecting additional peer (single-peer enforcement)" ); - // Clean up the connection + // Clean up the connection and its control machine if let Some(conn) = self.connections.remove(&link_id) && let Some(idx) = conn.our_index() { let _ = self.index_allocator.free(idx); } self.remove_link(&link_id); + self.remove_peer_machine(link_id); return Err(NodeError::MaxPeersExceeded { max: 1 }); } @@ -1464,20 +1509,9 @@ impl Node { ); self.peers.insert(peer_node_addr, new_peer); - // Populate the inert per-peer machine so every - // established peer has exactly one machine, keyed by its - // link_id (the winner link here). Nothing drives it yet. - self.peer_machines.insert( - link_id, - PeerMachine::established( - link_id, - verified_identity, - our_index, - is_outbound, - remote_epoch, - current_time_ms, - ), - ); + // The winning leg's machine (keyed by the winner link) survives + // the promotion; the executor crystallizes it in place via the + // `PromotionResolved` feedback after this returns. self.peers_by_index .insert((transport_id, our_index.as_u32()), peer_node_addr); self.peering @@ -1514,6 +1548,13 @@ impl Node { // Free the index we allocated let _ = self.index_allocator.free(our_index); + // Dispose the losing leg's machine here, with the leg. The + // executor's post-promote `PromotionResolved` dispatch then + // misses on this link, so the machine-side `FreeIndex` for the + // lost leg never fires — the inline free above stays the only + // one. + self.remove_peer_machine(link_id); + debug!( peer = %self.peer_display_name(&peer_node_addr), winner_link = %existing_link_id, @@ -1595,20 +1636,9 @@ impl Node { } self.peers.insert(peer_node_addr, new_peer); - // Populate the inert per-peer machine so every - // established peer has exactly one machine, keyed by its link_id. - // Nothing drives it yet. - self.peer_machines.insert( - link_id, - PeerMachine::established( - link_id, - verified_identity, - our_index, - is_outbound, - remote_epoch, - current_time_ms, - ), - ); + // The promoted leg's machine (born at msg1 for inbound, at dial for + // outbound) survives the promotion; the executor crystallizes it in + // place via the `PromotionResolved` feedback after this returns. self.peers_by_index .insert((transport_id, our_index.as_u32()), peer_node_addr); self.peering diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index 8daeac4..de9947a 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -1482,6 +1482,13 @@ async fn drive_to_msg3( .addr_to_link .insert((initiator.transport_id, responder.addr.clone()), link_id); initiator.node.connections.insert(link_id, conn); + // Mirror the production dial path: an identified outbound leg persists + // its control machine at dial, and the promote feedback later + // crystallizes that same machine in place. + initiator.node.peer_machines.insert( + link_id, + crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, now_ms), + ); initiator .node .pending_outbound @@ -1564,6 +1571,12 @@ async fn test_msg3_dual_rekey_won_frees_index() { "DualRekeyWon must free the msg1-allocated index" ); assert_eq!(responder.node.peer_count(), 1, "active peer untouched"); + // The rejected leg's msg1-born machine goes with the leg; only the + // established peer's machine remains. + let peer_link = responder.node.get_peer(&peer_addr).unwrap().link_id(); + assert_eq!(responder.node.peer_machines.len(), 1); + assert!(responder.node.peer_machines.contains_key(&peer_link)); + responder.node.debug_assert_peer_maps_coherent(); assert!( responder .node @@ -1624,6 +1637,12 @@ async fn test_msg3_resend_msg2_frees_index() { "ResendMsg2 must free the msg1-allocated index" ); assert_eq!(responder.node.peer_count(), 1, "active peer untouched"); + // The duplicate leg's msg1-born machine goes with the leg; only the + // established peer's machine remains. + let peer_link = responder.node.get_peer(&peer_addr).unwrap().link_id(); + assert_eq!(responder.node.peer_machines.len(), 1); + assert!(responder.node.peer_machines.contains_key(&peer_link)); + responder.node.debug_assert_peer_maps_coherent(); assert!( responder .node @@ -1637,3 +1656,164 @@ async fn test_msg3_resend_msg2_frees_index() { stop_hs(&mut initiator).await; stop_hs(&mut responder).await; } + +// =========================================================================== +// Inbound machine lifecycle: every window leg carries a persistent machine +// from msg1 — parked `SentMsg2`, crystallized in place on promote, disposed +// with the leg on every terminating msg3 arm. +// =========================================================================== + +#[tokio::test] +async fn test_inbound_machine_born_at_msg1_and_crystallized_at_promote() { + use crate::peer::machine::{HandshakePhase, PeerState}; + + let mut initiator = make_hs_node(Config::new()).await; + let mut responder = make_hs_node(Config::new()).await; + + let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await; + + // After msg1 the responder's window leg carries a machine parked at + // `SentMsg2`, seeded with the leg's msg1-allocated index. + assert_eq!(responder.node.connections.len(), 1); + let leg_link = *responder.node.connections.keys().next().unwrap(); + let leg_index = responder + .node + .get_connection(&leg_link) + .unwrap() + .our_index(); + assert!(leg_index.is_some(), "msg1 allocated the leg index"); + { + let machine = responder + .node + .peer_machines + .get(&leg_link) + .expect("window leg carries a machine from msg1"); + assert!(matches!( + machine.state(), + PeerState::Handshaking { + phase: HandshakePhase::SentMsg2, + .. + } + )); + assert_eq!(machine.our_index(), leg_index); + } + responder.node.debug_assert_peer_maps_coherent(); + + // msg3 promotes; the SAME machine survives and crystallizes in place. + responder.node.handle_msg3(msg3).await; + assert_eq!(responder.node.peer_count(), 1); + let peer_addr = + *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr(); + let peer = responder.node.get_peer(&peer_addr).unwrap(); + assert_eq!(peer.link_id(), leg_link, "promote keeps the leg's link"); + let peer_index = peer.our_index(); + assert_eq!(peer_index, leg_index, "promote keeps the msg1 index"); + let machine = responder + .node + .peer_machines + .get(&leg_link) + .expect("machine survives promotion"); + assert_eq!(machine.state(), PeerState::Established { addr: peer_addr }); + assert_eq!(machine.our_index(), peer_index); + responder.node.debug_assert_peer_maps_coherent(); + + // The initiator's dial-persisted machine crystallized in place too. + let responder_addr = + *PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full()).node_addr(); + let init_link = initiator.node.get_peer(&responder_addr).unwrap().link_id(); + let init_machine = initiator + .node + .peer_machines + .get(&init_link) + .expect("dial machine survives promotion"); + assert_eq!( + init_machine.state(), + PeerState::Established { + addr: responder_addr + } + ); + initiator.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} + +#[tokio::test] +async fn test_msg3_crypto_fail_disposes_leg_machine() { + let mut initiator = make_hs_node(Config::new()).await; + let mut responder = make_hs_node(Config::new()).await; + + let mut msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await; + assert_eq!(responder.node.peer_machines.len(), 1, "msg1-born machine"); + assert_eq!(responder.node.index_allocator.count(), 1); + + // Corrupt the Noise payload so `complete_handshake_msg3` fails. + let last = msg3.data.len() - 1; + msg3.data[last] ^= 0xFF; + responder.node.handle_msg3(msg3).await; + + assert_eq!(responder.node.peer_count(), 0, "no promotion"); + assert!(responder.node.connections.is_empty(), "leg torn down"); + assert!( + responder.node.peer_machines.is_empty(), + "crypto-fail teardown disposes the leg's machine" + ); + assert_eq!( + responder.node.index_allocator.count(), + 0, + "msg1-allocated index returned" + ); + responder.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} + +#[tokio::test] +async fn test_msg3_rekey_respond_disposes_leg_machine() { + // Rekey enabled with a tiny interval so the rekey age floor collapses to + // its 5s minimum; with the session backdated past it and NO rekey of our + // own in flight, a fresh inbound msg3 classifies as rekey-responder. + let mut config = Config::new(); + config.node.rekey.enabled = true; + config.node.rekey.after_secs = 1; + + let mut initiator = make_hs_node(Config::new()).await; + let mut responder = make_hs_node(config).await; + + // First handshake establishes the active peer (and its machine). + let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await; + responder.node.handle_msg3(msg3).await; + assert_eq!(responder.node.peer_count(), 1); + + let peer_addr = + *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr(); + responder + .node + .get_peer_mut(&peer_addr) + .unwrap() + .test_backdate_session_established(std::time::Duration::from_secs(6)); + + // Second handshake lands on the rekey-responder arm: the pending session + // moves onto the established peer; the window leg and its msg1-born + // machine are consumed. + let msg3b = drive_to_msg3(&mut initiator, &mut responder, 2000).await; + responder.node.handle_msg3(msg3b).await; + + let peer = responder.node.get_peer(&peer_addr).unwrap(); + assert!( + peer.pending_new_session().is_some(), + "rekey-responder arm stores the pending session" + ); + let peer_link = peer.link_id(); + assert_eq!( + responder.node.peer_machines.len(), + 1, + "the rekey window leg's machine is disposed with the leg" + ); + assert!(responder.node.peer_machines.contains_key(&peer_link)); + responder.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} diff --git a/src/node/tests/spanning_tree.rs b/src/node/tests/spanning_tree.rs index e22160a..70bea98 100644 --- a/src/node/tests/spanning_tree.rs +++ b/src/node/tests/spanning_tree.rs @@ -163,6 +163,13 @@ pub(super) async fn initiate_handshake(nodes: &mut [TestNode], i: usize, j: usiz .addr_to_link .insert((transport_id, responder_addr.clone()), link_id); initiator.node.connections.insert(link_id, conn); + // Mirror the production dial path: an identified outbound leg persists + // its control machine at dial, and the promote feedback later + // crystallizes that same machine in place. + initiator.node.peer_machines.insert( + link_id, + crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, 1000), + ); initiator .node .pending_outbound diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 88ff76b..d2151b6 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -515,6 +515,26 @@ impl PeerMachine { } } + /// New inbound machine for a leg whose msg1 was already processed + /// shell-side: the index is allocated and msg2 has been built and sent, so + /// the machine is born parked at `Handshaking{SentMsg2}` awaiting msg3 + /// (where identity crystallizes). This is the birth ctor for the + /// msg1-inline path (`handle_msg1` runs the crypto and the msg2 send + /// itself, so no `InboundMsg1` event is dispatched); it seeds exactly the + /// state that event's handler would have left behind, minus the timers — + /// no timer is armed on inbound legs (the stale-connection reaper owns + /// their timeout). + pub(crate) fn inbound_msg2_sent(link: LinkId, our_index: SessionIndex, now: u64) -> Self { + let mut machine = Self::new_inbound(link, now); + machine.conn.set_our_index(our_index); + machine.our_index = Some(our_index); + machine.state = PeerState::Handshaking { + link, + phase: HandshakePhase::SentMsg2, + }; + machine + } + /// New machine for an ALREADY-established peer: the post-handshake state a /// promoted peer occupies before any rekey. The driver inserts one of these /// into `Node.peer_machines` at each `promote_connection` establishment site