mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-22 07:48:26 +00:00
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.
This commit is contained in:
Johnathan Corgan
@@ -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),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user