From ba3c552dce5b1f4ce9508f8dab292567b71c6958 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 17 Jul 2026 03:25:28 +0000 Subject: [PATCH] node: give anonymous outbound legs a persistent control machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The anonymous-discovery dial paths drove no control machine: the leg was born bare in start_handshake and a throwaway machine appeared only at msg2. Anonymous legs now birth a persistent identity-less machine alongside the leg (new_outbound widened to Option, backed by the existing identity-less outbound connection state), and handle_msg2 crystallizes the learned identity onto it before stepping. The msg2 missing-machine arm becomes defensive recovery (debug-assert + insert, matching the IK line), and the self-connect drop arm disposes the machine it previously leaked — reachable from identified dials too, since the handshake completion overwrites the expected identity without comparing it to the dial-time expectation. The self-connect arm's index/link/pending-entry leak is pre-existing and stays as-is. With every leg now born with a machine, the leg-to-machine direction of the debug coherence check switches on for this line. The add_connection test seam seeds direction-faithful machines (an outbound-anonymous seeded connection previously got an inbound-shaped one), and the hand-rolled dial sites in tests seed dial machines mirroring production. New tests pin the identity-less birth, the msg2 identity crystallization through promote, and the self-connect disposal. --- src/node/handlers/handshake.rs | 68 ++++++++----- src/node/lifecycle/mod.rs | 51 +++++++--- src/node/mod.rs | 21 ++-- src/node/tests/acl.rs | 12 +++ src/node/tests/handshake.rs | 172 +++++++++++++++++++++++++++++++- src/node/tests/spanning_tree.rs | 2 +- src/node/tests/unit.rs | 6 ++ src/peer/machine.rs | 70 +++++++++---- 8 files changed, 330 insertions(+), 72 deletions(-) diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index e0c4a87..0fa5dec 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -682,8 +682,16 @@ impl Node { } if peer_node_addr == *self.identity().node_addr() { + // Reachable by any outbound leg whose msg2 static key turns out to + // be our own — usually an anonymous shared-media beacon, but an + // identified dial misdirected at ourselves lands here too (the + // learned identity overwrites the dial-time expectation and is + // never compared against it). This leg never promotes; its machine + // goes with it. The index, link, and `pending_outbound` entry are + // deliberately NOT freed here (pre-existing shape). debug!(link_id = %link_id, "Discovered self via shared-media beacon, dropping"); self.connections.remove(&link_id); + self.remove_peer_machine(link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -889,31 +897,22 @@ impl Node { // an existing peer), so this is the net-new path: promote_connection hits // its normal-promotion branch and returns Promoted. // - // 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; 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. + // Every outbound leg carries a persistent machine by now — identified + // dials persist one at dial, anonymous-discovery legs at leg birth in + // `start_handshake` — so the lookup is expected to hit. For an anonymous + // machine this is where its identity crystallizes: msg2 revealed who + // answered, and the learned identity lands on the machine before the + // step (a no-op for identified machines). The machine survives the + // promotion and the executor crystallizes its state via the + // `PromotionResolved` feedback. 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 { - their_index: header.sender_idx, - }, - packet.timestamp_ms, - &mut self.index_allocator, - ), - None => { - let mut machine = - PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); + Some(machine) => { + machine.crystallize_identity(peer_identity); machine.step( PeerEvent::OutboundMsg2 { their_index: header.sender_idx, @@ -922,6 +921,27 @@ impl Node { &mut self.index_allocator, ) } + None => { + // A miss is a state-machine inconsistency (e.g. a test seeding + // `connections`/`pending_outbound` directly): rebuild the + // machine defensively and persist it, so the promotion feedback + // below still finds it and the promoted peer keeps a machine. + debug_assert!( + false, + "outbound leg {link_id} reached msg2 without a control machine" + ); + let mut machine = + PeerMachine::new_outbound(link_id, Some(peer_identity), packet.timestamp_ms); + let actions = machine.step( + PeerEvent::OutboundMsg2 { + their_index: header.sender_idx, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + self.peer_machines.insert(link_id, machine); + actions + } }; // The outbound Msg2 Promote step cancels the two dial-armed handshake // timers (the machine survives promotion, so they would otherwise linger diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 62828a8..bdf478f 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -496,7 +496,7 @@ impl Node { // a spurious `UnregisterDecryptSession`. It is removed on every // failure path in the dial window (`prepare_outbound_msg1` / // `poll_pending_connects`), mirroring the connection's lifetime. - let machine = PeerMachine::new_outbound(link_id, identity, Self::now_ms()); + let machine = PeerMachine::new_outbound(link_id, Some(identity), Self::now_ms()); self.peer_machines.insert(link_id, machine); if !is_connection_oriented { @@ -537,9 +537,14 @@ impl Node { Ok(()) } None => { - // Anonymous-discovery dial: next's inline path, no control machine - // (the transient is born at `handle_msg2` when the peer's identity - // crystallizes from `conn.expected_identity()`). + // Anonymous-discovery dial: the inline path. The leg's control + // machine is born inside `start_handshake` at leg creation + // (identity-less; `handle_msg2` crystallizes the identity the XX + // handshake learns). The connection-oriented connect window below + // deliberately has no leg and no machine yet — the Failed arm of + // `poll_pending_connects` only disposes machines for identified + // legs, and stays correct because anonymous machines don't exist + // until connect resolution reaches `start_handshake`. if is_connection_oriented { // Connection-oriented: start non-blocking connect, defer handshake. if let Some(transport) = self.transports.get(&transport_id) { @@ -584,8 +589,9 @@ impl Node { /// connection for `send_stored_msg1` to transmit. Does NOT send — so the /// fallible setup can propagate its error synchronously before any machine /// drive. Anonymous-discovery legs use the monolithic `start_handshake` - /// instead; only identified legs persist a machine, so this is the only - /// caller path that cleans up `peer_machines`. + /// instead, whose machine is born after its fallible setup — so this is + /// the only caller path whose Err arms clean up `peer_machines` (the + /// dial-time machine predates the failure). pub(in crate::node) fn prepare_outbound_msg1( &mut self, link_id: LinkId, @@ -658,11 +664,12 @@ impl Node { } /// Start an outbound Noise handshake inline (allocate index, run the Noise - /// leaf, arm the resend, track `pending_outbound`, and send msg1). Used by - /// the anonymous-discovery dial paths (connectionless dial and the - /// connection-oriented connect-resolution), which drive no control machine. - /// Anonymous discovery (no `peer_identity`) leaves identity to be learned - /// from the XX msg2. + /// leaf, arm the resend, track `pending_outbound`, persist the leg's + /// control machine, and send msg1). Used by the anonymous-discovery dial + /// paths (connectionless dial and the connection-oriented + /// connect-resolution). Anonymous discovery (no `peer_identity`) leaves + /// identity to be learned from the XX msg2, which crystallizes it onto the + /// leg-born machine. pub(super) async fn start_handshake( &mut self, link_id: LinkId, @@ -739,6 +746,22 @@ impl Node { self.pending_outbound .insert((transport_id, our_index.as_u32()), link_id); self.connections.insert(link_id, connection); + // The leg's persistent control machine is born alongside its pending + // connection, after all fallible setup (the index-alloc and Noise Err + // returns above predate it and need no disposal). Both production + // callers dial anonymously (identified dials persist their machine at + // dial and go through `prepare_outbound_msg1`), so the machine starts + // identity-less; `handle_msg2` crystallizes the identity the XX + // handshake learns. Inserted before the send below so no suspension + // point observes a leg without a machine. The send-failure arm retains + // the failed connection, and the machine stays with it — the stale- + // connection reaper disposes both together. No timers are armed and no + // event is dispatched: this path sends msg1 inline, so the machine + // parks at `Discovered` until msg2. + self.peer_machines.insert( + link_id, + PeerMachine::new_outbound(link_id, peer_identity, current_time_ms), + ); // Send the wire format handshake message if let Some(transport) = self.transports.get(&transport_id) { @@ -1422,8 +1445,10 @@ impl Node { } } None => { - // Anonymous-discovery leg: next's inline handshake, no - // control machine. + // Anonymous-discovery leg: the inline handshake persists + // the leg's control machine at leg birth. Its Err returns + // all precede that birth, so the cleanup below has no + // machine to dispose. if let Err(e) = self .start_handshake( pending.link_id, diff --git a/src/node/mod.rs b/src/node/mod.rs index 128149e..193d79b 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -2317,13 +2317,13 @@ impl Node { /// Gates the leg→machine direction of `debug_assert_peer_maps_coherent`. /// Asserting it requires the convention that every `connections` insert is /// paired with a `peer_machines` insert before the next await point — - /// which does NOT hold here yet: the inbound msg1→msg3 window legs and the - /// anonymous-discovery outbound legs legitimately run machine-less (their - /// decision machines are msg3/msg2-time transients). Flip this to `true` - /// once every leg is born with a persistent machine; until then only the - /// machine→carrier leak-tripwire direction asserts. + /// which now holds everywhere: identified dials persist their machine + /// before the leg exists (`initiate_connection`), anonymous-discovery legs + /// are born with one inside `start_handshake` alongside the connection + /// insert, inbound legs get theirs in `handle_msg1` at the same point, and + /// the `add_connection` test seam seeds one for directly-inserted legs. #[cfg(debug_assertions)] - const CHECK_LEGS_HAVE_MACHINES: bool = false; + const CHECK_LEGS_HAVE_MACHINES: bool = true; /// Debug-build coherence sweep over the peer-lifecycle maps, run once per /// rx-loop tick and invoked directly by unit tests. @@ -2444,11 +2444,10 @@ impl Node { self.peer_machines.entry(link_id).or_insert_with(|| { let now = connection.started_at(); - match connection.expected_identity() { - Some(identity) if connection.is_outbound() => { - PeerMachine::new_outbound(link_id, *identity, now) - } - _ => PeerMachine::new_inbound(link_id, now), + if connection.is_outbound() { + PeerMachine::new_outbound(link_id, connection.expected_identity().copied(), now) + } else { + PeerMachine::new_inbound(link_id, now) } }); diff --git a/src/node/tests/acl.rs b/src/node/tests/acl.rs index dae2d52..711fc9d 100644 --- a/src/node/tests/acl.rs +++ b/src/node/tests/acl.rs @@ -80,6 +80,12 @@ async fn test_outbound_msg2_denied_after_acl_reload() { node_a .pending_outbound .insert((transport_id, our_index_a.as_u32()), link_id_a); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a, + crate::peer::machine::PeerMachine::new_outbound(link_id_a, Some(peer_b_identity), 1000), + ); let mut conn_b = PeerConnection::inbound(LinkId::new(2), 1000); let responder_epoch = [0x11; 8]; @@ -200,6 +206,12 @@ async fn test_inbound_msg3_denied_triggers_disconnect() { node_a .pending_outbound .insert((transport_id_a, our_index_a.as_u32()), link_id_a); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a, + crate::peer::machine::PeerMachine::new_outbound(link_id_a, Some(peer_b_identity), 1000), + ); let transport = node_a.transports.get(&transport_id_a).unwrap(); transport diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index de9947a..3aafe28 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -85,6 +85,12 @@ async fn test_two_node_handshake_udp() { node_a .pending_outbound .insert((transport_id_a, our_index_a.as_u32()), link_id_a); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a, + crate::peer::machine::PeerMachine::new_outbound(link_id_a, Some(peer_b_identity), 1000), + ); // Send msg1 from A to B over UDP let transport = node_a.transports.get(&transport_id_a).unwrap(); @@ -344,6 +350,12 @@ async fn test_run_rx_loop_handshake() { node_a .pending_outbound .insert((transport_id_a, our_index_a.as_u32()), link_id_a); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a, + crate::peer::machine::PeerMachine::new_outbound(link_id_a, Some(peer_b_identity), 1000), + ); // Send msg1 from A to B over real UDP let transport = node_a.transports.get(&transport_id_a).unwrap(); @@ -532,6 +544,12 @@ async fn test_cross_connection_both_initiate() { node_a .pending_outbound .insert((transport_id_a, our_index_a.as_u32()), link_id_a_out); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a_out, + crate::peer::machine::PeerMachine::new_outbound(link_id_a_out, Some(peer_b_identity), 1000), + ); // Node B initiates to Node A let link_id_b_out = node_b.allocate_link_id(); @@ -562,6 +580,12 @@ async fn test_cross_connection_both_initiate() { node_b .pending_outbound .insert((transport_id_b, our_index_b.as_u32()), link_id_b_out); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_b.peer_machines.insert( + link_id_b_out, + crate::peer::machine::PeerMachine::new_outbound(link_id_b_out, Some(peer_a_identity), 1000), + ); // Both send msg1 over UDP let transport = node_a.transports.get(&transport_id_a).unwrap(); @@ -906,7 +930,7 @@ async fn test_resend_scheduling() { // Dial it to `SentMsg1` (connectionless: no connect step) and arm its // retransmit timer at now + 1000ms, mirroring what a real dial arms. let mut machine = - crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, now_ms); + crate::peer::machine::PeerMachine::new_outbound(link_id, Some(peer_identity), now_ms); let _ = machine.step( crate::peer::machine::PeerEvent::Dial { transport_id, @@ -981,7 +1005,7 @@ async fn test_handshake_timeout_drive() { // Machine in SentMsg1 with a HandshakeTimeout timer armed at dial + 30s. let mut machine = - crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, dial_ms); + crate::peer::machine::PeerMachine::new_outbound(link_id, Some(peer_identity), dial_ms); let _ = machine.step( crate::peer::machine::PeerEvent::Dial { transport_id, @@ -1487,7 +1511,7 @@ async fn drive_to_msg3( // 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), + crate::peer::machine::PeerMachine::new_outbound(link_id, Some(peer_identity), now_ms), ); initiator .node @@ -1817,3 +1841,145 @@ async fn test_msg3_rekey_respond_disposes_leg_machine() { stop_hs(&mut initiator).await; stop_hs(&mut responder).await; } + +// =========================================================================== +// Anonymous-discovery outbound lifecycle: the leg's persistent machine is +// born identity-less at leg birth inside `start_handshake`, learns its +// identity from XX msg2 (crystallization), survives the promote, and is +// disposed with the leg when the dial turns out to target ourselves. +// =========================================================================== + +#[tokio::test] +async fn test_anonymous_dial_births_identityless_machine_at_leg_birth() { + use crate::peer::machine::PeerState; + + let mut initiator = make_hs_node(Config::new()).await; + let responder = make_hs_node(Config::new()).await; + + // Anonymous dial (no peer identity): the connectionless path runs the + // inline handshake, which creates the leg and its machine together. + initiator + .node + .initiate_connection(initiator.transport_id, responder.addr.clone(), None) + .await + .expect("anonymous dial"); + + assert_eq!(initiator.node.connections.len(), 1); + let leg_link = *initiator.node.connections.keys().next().unwrap(); + let machine = initiator + .node + .peer_machines + .get(&leg_link) + .expect("anonymous leg carries a machine from leg birth"); + assert!( + machine.identity().is_none(), + "anonymous machine is born without an identity" + ); + assert_eq!( + machine.state(), + PeerState::Discovered, + "no event is dispatched on the inline dial path" + ); + initiator.node.debug_assert_peer_maps_coherent(); + + let mut initiator = initiator; + let mut responder = responder; + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} + +#[tokio::test] +async fn test_anonymous_msg2_crystallizes_identity_and_promotes() { + use crate::peer::machine::PeerState; + + let mut initiator = make_hs_node(Config::new()).await; + let mut responder = make_hs_node(Config::new()).await; + + initiator + .node + .initiate_connection(initiator.transport_id, responder.addr.clone(), None) + .await + .expect("anonymous dial"); + let leg_link = *initiator.node.connections.keys().next().unwrap(); + initiator.node.debug_assert_peer_maps_coherent(); + + // Responder answers msg1 with msg2; the initiator's msg2 processing learns + // who answered, crystallizes the identity onto the leg-born machine, and + // promotes through it. + let msg1_pkt = recv_phase(&mut responder.packet_rx, 1, "msg1").await; + responder.node.handle_msg1(msg1_pkt).await; + let msg2_pkt = recv_phase(&mut initiator.packet_rx, 2, "msg2").await; + initiator.node.handle_msg2(msg2_pkt).await; + + let responder_identity = + PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full()); + let responder_addr = *responder_identity.node_addr(); + assert_eq!(initiator.node.peer_count(), 1); + let peer = initiator.node.get_peer(&responder_addr).expect("promoted"); + assert_eq!(peer.link_id(), leg_link, "promote keeps the leg's link"); + + // The SAME machine survived the promote, with the learned identity and + // the established state crystallized in place. + let machine = initiator + .node + .peer_machines + .get(&leg_link) + .expect("machine survives the anonymous promote"); + assert_eq!( + machine.identity().map(|id| *id.node_addr()), + Some(responder_addr), + "msg2 crystallized the learned identity onto the machine" + ); + assert_eq!( + machine.state(), + PeerState::Established { + addr: responder_addr + } + ); + initiator.node.debug_assert_peer_maps_coherent(); + + // Complete the exchange so the responder promotes too, and both sides + // stay coherent across the full anonymous establish path. + let msg3_pkt = recv_phase(&mut responder.packet_rx, 3, "msg3").await; + responder.node.handle_msg3(msg3_pkt).await; + assert_eq!(responder.node.peer_count(), 1); + responder.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut initiator).await; + stop_hs(&mut responder).await; +} + +#[tokio::test] +async fn test_anonymous_self_connect_drop_disposes_machine() { + let mut node = make_hs_node(Config::new()).await; + let self_addr = node.addr.clone(); + + // Anonymously dial our own bound address (a shared-media beacon can echo + // ourselves back at us). + node.node + .initiate_connection(node.transport_id, self_addr, None) + .await + .expect("anonymous self dial"); + let leg_link = *node.node.connections.keys().next().unwrap(); + assert_eq!(node.node.peer_machines.len(), 1); + + // We answer our own msg1, then our msg2 processing discovers the learned + // identity is our own and drops the leg — machine included. + let msg1_pkt = recv_phase(&mut node.packet_rx, 1, "msg1").await; + node.node.handle_msg1(msg1_pkt).await; + let msg2_pkt = recv_phase(&mut node.packet_rx, 2, "msg2").await; + node.node.handle_msg2(msg2_pkt).await; + + assert_eq!(node.node.peer_count(), 0, "no promotion"); + assert!( + !node.node.connections.contains_key(&leg_link), + "self-connect drop removes the outbound leg" + ); + assert!( + !node.node.peer_machines.contains_key(&leg_link), + "self-connect drop disposes the outbound leg's machine" + ); + node.node.debug_assert_peer_maps_coherent(); + + stop_hs(&mut node).await; +} diff --git a/src/node/tests/spanning_tree.rs b/src/node/tests/spanning_tree.rs index 70bea98..069805b 100644 --- a/src/node/tests/spanning_tree.rs +++ b/src/node/tests/spanning_tree.rs @@ -168,7 +168,7 @@ pub(super) async fn initiate_handshake(nodes: &mut [TestNode], i: usize, j: usiz // crystallizes that same machine in place. initiator.node.peer_machines.insert( link_id, - crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, 1000), + crate::peer::machine::PeerMachine::new_outbound(link_id, Some(peer_identity), 1000), ); initiator .node diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 7bc71f7..712220d 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -2031,6 +2031,12 @@ async fn drive_xx_handshake( node_a .pending_outbound .insert((transport_id, our_index_a.as_u32()), link_id_a); + // Mirror the production dial path: an outbound leg persists its control + // machine at dial. + node_a.peer_machines.insert( + link_id_a, + crate::peer::machine::PeerMachine::new_outbound(link_id_a, Some(peer_b_identity), 1000), + ); let transport = node_a.transports.get(&transport_id).unwrap(); transport diff --git a/src/peer/machine.rs b/src/peer/machine.rs index d2151b6..1233e62 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -464,14 +464,21 @@ pub(crate) struct PeerMachine { impl PeerMachine { /// New outbound machine (we dial). Starts at `Discovered`; the reconciler's - /// `Dial` event drives the first transition. - pub(crate) fn new_outbound(link: LinkId, identity: PeerIdentity, now: u64) -> Self { + /// `Dial` event drives the first transition. `identity` is `None` for an + /// anonymous-discovery leg (a shared-media beacon carries no identity): the + /// machine is born identity-less and [`crystallize_identity`] + /// (Self::crystallize_identity) fills it in when the XX msg2 reveals who + /// answered. + pub(crate) fn new_outbound(link: LinkId, identity: Option, now: u64) -> Self { Self { state: PeerState::Discovered, link, - identity: Some(identity), - node_addr: Some(*identity.node_addr()), - conn: ConnectionState::outbound(link, identity, now), + identity, + node_addr: identity.map(|id| *id.node_addr()), + conn: match identity { + Some(id) => ConnectionState::outbound(link, id, now), + None => ConnectionState::outbound_anonymous(link, now), + }, remote_epoch: None, rekey_in_progress: false, rekey_our_index: None, @@ -595,6 +602,29 @@ impl PeerMachine { self.our_index } + /// The peer identity this machine was born with (outbound identified dial) + /// or crystallized onto it (anonymous discovery at msg2, inbound at msg3 + /// promote). `None` while the peer is still anonymous. + #[cfg(test)] + pub(crate) fn identity(&self) -> Option<&PeerIdentity> { + self.identity.as_ref() + } + + /// Crystallize a learned identity onto an anonymous machine: XX msg2 is + /// where an anonymous-discovery leg first learns who answered its dial. + /// Sets the identity, the node address, and the conn's expected identity — + /// control-tier fields only; no action is emitted and no state transition + /// runs. A no-op when the identity is already known (identified dials fix + /// it at birth), so callers on shared paths need not distinguish the two. + pub(crate) fn crystallize_identity(&mut self, identity: PeerIdentity) { + if self.identity.is_some() { + return; + } + self.identity = Some(identity); + self.node_addr = Some(*identity.node_addr()); + self.conn.set_expected_identity(identity); + } + /// The msg1 resend count for this outbound handshake leg. The per-peer /// machine is the home for this counter — the timer driver advances it via /// [`record_resend`](Self::record_resend) on each successful resend, and the @@ -1555,7 +1585,7 @@ mod tests { let id = peer_identity(); let addr = *id.node_addr(); let link = LinkId::new(3); - let mut m = PeerMachine::new_outbound(link, id, 0); + let mut m = PeerMachine::new_outbound(link, Some(id), 0); let their_index = SessionIndex::new(0x77); let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc); @@ -1585,7 +1615,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let link = LinkId::new(4); - let mut m = PeerMachine::new_outbound(link, id, 0); + let mut m = PeerMachine::new_outbound(link, Some(id), 0); let their_index = SessionIndex::new(0x88); let actions = m.step(PeerEvent::OutboundMsg2 { their_index }, 0, &mut alloc); @@ -1729,7 +1759,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); // Arrange: a completed rekey pending cutover. m.state = PeerState::Maintaining { addr, @@ -1791,7 +1821,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Active { addr }; let actions = m.step( @@ -1826,7 +1856,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let our = *smaller.node_addr(); let peer = larger; - let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0); let peer_addr = *peer.node_addr(); m.state = PeerState::Maintaining { addr: peer_addr, @@ -1874,7 +1904,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let our = *larger.node_addr(); let peer = smaller; - let mut m = PeerMachine::new_outbound(LinkId::new(2), peer, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(2), Some(peer), 0); let peer_addr = *peer.node_addr(); m.state = PeerState::Maintaining { addr: peer_addr, @@ -2391,7 +2421,7 @@ mod tests { let our = *peer_identity().node_addr(); // Persisted at dial: Discovered, conn.our_index deliberately NOT set. - let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0); assert_eq!(m.our_index(), None); // Promote via msg2 from Discovered (the production path — the former @@ -2479,7 +2509,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let peer = peer_identity(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0); // Connectionless dial: no OpenTransport, straight to Handshaking{SentMsg1}. let dial = m.step( PeerEvent::Dial { @@ -2547,7 +2577,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let peer = peer_identity(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0); // Connection-oriented dial: open the transport first, no msg1 yet. let dial = m.step( PeerEvent::Dial { @@ -2605,7 +2635,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Active { addr }; m.our_index = Some(SessionIndex::new(0x4242)); @@ -2649,7 +2679,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Maintaining { addr, kind: MaintainKind::Rekey(RekeyPhase::PendingCutover), @@ -2708,7 +2738,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Established { addr }; let acts = m.step(PeerEvent::RekeyInitiated, 5_000, &mut alloc); @@ -2734,7 +2764,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Maintaining { addr, kind: MaintainKind::Rekey(RekeyPhase::Msg1Sent), @@ -2768,7 +2798,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Active { addr }; m.conn.set_our_index(SessionIndex::new(0x1111)); m.conn.set_their_index(SessionIndex::new(0x2222)); @@ -2797,7 +2827,7 @@ mod tests { let mut alloc = IndexAllocator::new(); let id = peer_identity(); let addr = *id.node_addr(); - let mut m = PeerMachine::new_outbound(LinkId::new(1), id, 0); + let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0); m.state = PeerState::Active { addr }; m.conn.set_our_index(SessionIndex::new(0x1111)); m.conn.set_their_index(SessionIndex::new(0x2222));