laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

peer: source the handshake identity and index family from the control machine

Browse Source 
The pending connection and the per-peer control machine have carried
duplicate copies of the handshake-phase fields since the machine gained
its own connection state. Read them from the machine and drop the
connection's projections.

The peer identity is the sharp one. The connection learned it from msg1
and the machine's carrier did not, so the two views genuinely disagreed
for inbound connections until the Noise operations moved onto the
machine and began recording each result on both. That is now in place,
so every reader can take the machine's copy unchanged: the establish
snapshot, the promotion sweep for competing connections, the dial and
path in-progress checks, the peering observations and per-peer in-flight
budget, and the stale-connection sweep's retry address.

Also repointed: our_index, their_index, transport_id, started_at, the
stored handshake message bytes, and the idle-timeout check. Two
duplicate index writes on the connection are dropped, both immediately
preceded by the machine-side write of the same value. Reads that
previously came off a connection detached just before its machine was
disposed now capture the machine's value first.

Test seeding follows the establish paths: the seed builders write
our_index to the carrier, which promotion now reads. A new test pins
the inbound identity learn on the carrier and the retry address a
failed inbound connection reports to the sweep, so a silent regression
to a blank identity cannot pass.

ConnectionState::duration loses its last non-test caller with the
connection accessor and is marked test-only.
This commit is contained in:
Johnathan Corgan
2026-07-19 00:25:17 +00:00
parent e7537929ba
commit e21e09d7e6
9 changed files with 182 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/handshake.rs
+26 -40
View File
@@ -39,14 +39,12 @@ impl EstablishView for Node {
rekey_in_progress: existing.map(|p| p.rekey_in_progress()).unwrap_or(false), rekey_in_progress: existing.map(|p| p.rekey_in_progress()).unwrap_or(false),
existing_msg2: existing.and_then(|p| p.handshake_msg2().map(|m| m.to_vec())), existing_msg2: existing.and_then(|p| p.handshake_msg2().map(|m| m.to_vec())),
at_max_peers: max_peers > 0 && self.peers.len() >= max_peers, at_max_peers: max_peers > 0 && self.peers.len() >= max_peers,
has_pending_outbound_to_peer: self has_pending_outbound_to_peer: self.connections().any(|(_, machine)| {
.connections() machine
.filter_map(|(_, machine)| machine.leg()) .conn_expected_identity()
.any(|conn| { .map(|id| id.node_addr() == peer_addr)
conn.expected_identity() .unwrap_or(false)
.map(|id| id.node_addr() == peer_addr) }),
.unwrap_or(false)
}),
rekey_enabled: self.config().node.rekey.enabled, rekey_enabled: self.config().node.rekey.enabled,
our_node_addr: *self.identity().node_addr(), our_node_addr: *self.identity().node_addr(),
} }
@@ -310,11 +308,8 @@ impl Node {
}; };
// Learn peer identity from msg1 // Learn peer identity from msg1
let peer_identity = match machine assert!(machine.leg().is_some(), "pending connection attached above");
.leg() let peer_identity = match machine.conn_expected_identity() {
.expect("pending connection attached above")
.expected_identity()
{
Some(id) => *id, Some(id) => *id,
None => { None => {
self.msg1_rate_limiter.complete_handshake(); self.msg1_rate_limiter.complete_handshake();
@@ -639,10 +634,6 @@ impl Node {
// connection, then build + store the framed msg2. The old index was // connection, then build + store the framed msg2. The old index was
// already freed by `remove_active_peer` above, BEFORE this fresh // already freed by `remove_active_peer` above, BEFORE this fresh
// allocation — matching the pre-refactor allocation sequence. // allocation — matching the pre-refactor allocation sequence.
machine
.leg_mut()
.expect("pending connection attached above")
.set_our_index(our_index);
let link = Link::connectionless( let link = Link::connectionless(
link_id, link_id,
packet.transport_id, packet.transport_id,
@@ -789,10 +780,6 @@ impl Node {
// Shell registry surgery, in the pre-refactor order: // Shell registry surgery, in the pre-refactor order:
// set indices on the shell connection, insert link / reverse map / // set indices on the shell connection, insert link / reverse map /
// connection, then build + store the framed msg2. // connection, then build + store the framed msg2.
machine
.leg_mut()
.expect("pending connection attached above")
.set_our_index(our_index);
let link = Link::connectionless( let link = Link::connectionless(
link_id, link_id,
packet.transport_id, packet.transport_id,
@@ -1045,11 +1032,12 @@ impl Node {
} }
machine.set_conn_source_addr(packet.remote_addr.clone()); machine.set_conn_source_addr(packet.remote_addr.clone());
let conn = machine assert!(
.leg_mut() machine.leg().is_some(),
.expect("pending connection present for msg2 completion"); "pending connection present for msg2 completion"
);
let peer_identity = match conn.expected_identity() { let peer_identity = match machine.conn_expected_identity() {
Some(id) => *id, Some(id) => *id,
None => { None => {
warn!(link_id = %link_id, "No identity after handshake"); warn!(link_id = %link_id, "No identity after handshake");
@@ -1059,7 +1047,7 @@ impl Node {
} }
}; };
(peer_identity, conn.our_index()) (peer_identity, machine.our_index())
}; };
if self if self
@@ -1166,10 +1154,10 @@ impl Node {
// right after the take — unconditionally, whether or not a // right after the take — unconditionally, whether or not a
// connection was carried — so none of this block's exits leave a // connection was carried — so none of this block's exits leave a
// dangling machine. // dangling machine.
let taken_conn = self let (taken_conn, carrier_our_index) = match self.peer_machines.get_mut(&link_id) {
.peer_machines Some(machine) => (machine.take_leg(), machine.our_index()),
.get_mut(&link_id) None => (None, None),
.and_then(|machine| machine.take_leg()); };
self.remove_peer_machine(link_id); self.remove_peer_machine(link_id);
let mut conn = match taken_conn { let mut conn = match taken_conn {
Some(c) => c, Some(c) => c,
@@ -1185,7 +1173,7 @@ impl Node {
if swap { if swap {
// We're the smaller node. Swap to outbound session + indices. // We're the smaller node. Swap to outbound session + indices.
// The peer will keep their inbound session (complement of ours). // The peer will keep their inbound session (complement of ours).
let outbound_our_index = conn.our_index(); let outbound_our_index = carrier_our_index;
let outbound_session = conn.noise_session.take(); let outbound_session = conn.noise_session.take();
let (outbound_session, outbound_our_index) = match ( let (outbound_session, outbound_our_index) = match (
@@ -1250,7 +1238,7 @@ impl Node {
// their outbound session, that index is exactly what they'll use. // their outbound session, that index is exactly what they'll use.
// The msg2 sender_idx we see here is the peer's INBOUND our_index, // The msg2 sender_idx we see here is the peer's INBOUND our_index,
// which becomes stale after the peer swaps. // which becomes stale after the peer swaps.
let outbound_our_index = conn.our_index(); let outbound_our_index = carrier_our_index;
if let Some(peer) = self.peers.get(&peer_node_addr) { if let Some(peer) = self.peers.get(&peer_node_addr) {
debug!( debug!(
@@ -1399,6 +1387,7 @@ impl Node {
let mut connection = machine let mut connection = machine
.take_leg() .take_leg()
.ok_or(NodeError::ConnectionNotFound(link_id))?; .ok_or(NodeError::ConnectionNotFound(link_id))?;
let carrier_our_index = machine.our_index();
let carrier_their_index = machine.conn_their_index(); let carrier_their_index = machine.conn_their_index();
let carrier_transport_id = machine.conn_transport_id(); let carrier_transport_id = machine.conn_transport_id();
let carrier_source_addr = machine.conn_source_addr().cloned(); let carrier_source_addr = machine.conn_source_addr().cloned();
@@ -1415,12 +1404,10 @@ impl Node {
.take() .take()
.ok_or(NodeError::NoSession(link_id))?; .ok_or(NodeError::NoSession(link_id))?;
let our_index = connection let our_index = carrier_our_index.ok_or_else(|| NodeError::PromotionFailed {
.our_index() link_id,
.ok_or_else(|| NodeError::PromotionFailed { reason: "missing our_index".into(),
link_id, })?;
reason: "missing our_index".into(),
})?;
let their_index = carrier_their_index.ok_or_else(|| NodeError::PromotionFailed { let their_index = carrier_their_index.ok_or_else(|| NodeError::PromotionFailed {
link_id, link_id,
reason: "missing their_index".into(), reason: "missing their_index".into(),
@@ -1570,8 +1557,7 @@ impl Node {
.connections() .connections()
.filter(|(_, machine)| { .filter(|(_, machine)| {
machine machine
.leg() .conn_expected_identity()
.and_then(|conn| conn.expected_identity())
.map(|id| *id.node_addr() == peer_node_addr) .map(|id| *id.node_addr() == peer_node_addr)
.unwrap_or(false) .unwrap_or(false)
}) })
src/node/handlers/timeout.rs
+17 -14
View File
@@ -19,18 +19,18 @@ impl LifecycleView for Node {
// reap. // reap.
self.peer_machines self.peer_machines
.iter() .iter()
.filter_map(|(link_id, machine)| machine.leg().map(|conn| (link_id, machine, conn))) .filter(|(_, machine)| machine.leg().is_some())
.filter(|(link_id, machine, _conn)| { .filter(|(link_id, machine)| {
machine.is_failed() machine.is_failed()
|| (machine.conn_is_timed_out(now_ms, timeout_ms) || (machine.conn_is_timed_out(now_ms, timeout_ms)
&& !self.peer_timers.get(*link_id).is_some_and(|timers| { && !self.peer_timers.get(*link_id).is_some_and(|timers| {
timers.contains_key(&TimerKind::HandshakeTimeout) timers.contains_key(&TimerKind::HandshakeTimeout)
})) }))
}) })
.map(|(link_id, machine, conn)| ConnSnapshot { .map(|(link_id, machine)| ConnSnapshot {
link: *link_id, link: *link_id,
is_outbound: machine.conn_is_outbound(), is_outbound: machine.conn_is_outbound(),
retry_addr: conn.expected_identity().map(|id| *id.node_addr()), retry_addr: machine.conn_expected_identity().map(|id| *id.node_addr()),
resend_count: 0, resend_count: 0,
msg1: Vec::new(), msg1: Vec::new(),
}) })
@@ -120,7 +120,7 @@ impl Node {
// dangling machine. A no-op for promoted peers — `promote_connection` // dangling machine. A no-op for promoted peers — `promote_connection`
// already consumed their connection, so this reaper never runs for // already consumed their connection, so this reaper never runs for
// them. // them.
let conn = match self let _detached_leg = match self
.peer_machines .peer_machines
.get_mut(&link_id) .get_mut(&link_id)
.and_then(|machine| machine.take_leg()) .and_then(|machine| machine.take_leg())
@@ -128,16 +128,16 @@ impl Node {
Some(c) => c, Some(c) => c,
None => return, None => return,
}; };
// Read the transport ID off the surviving carrier before disposing the // Read the transport ID and session index off the surviving carrier
// machine (the leg no longer projects it). // before disposing the machine (the leg no longer projects them).
let transport_id = self let (transport_id, our_index) = match self.peer_machines.get(&link_id) {
.peer_machines Some(machine) => (machine.conn_transport_id(), machine.our_index()),
.get(&link_id) None => (None, None),
.and_then(|machine| machine.conn_transport_id()); };
self.remove_peer_machine(link_id); self.remove_peer_machine(link_id);
// Free session index and pending_outbound if allocated // Free session index and pending_outbound if allocated
if let Some(idx) = conn.our_index() { if let Some(idx) = our_index {
if let Some(tid) = transport_id { if let Some(tid) = transport_id {
self.pending_outbound.remove(&(tid, idx.as_u32())); self.pending_outbound.remove(&(tid, idx.as_u32()));
} }
@@ -200,13 +200,16 @@ impl Node {
.get(&link) .get(&link)
.is_some_and(|machine| machine.conn_is_timed_out(now_ms, timeout_ms)); .is_some_and(|machine| machine.conn_is_timed_out(now_ms, timeout_ms));
let (reap, retry_peer) = match self.leg(&link) { let (reap, retry_peer) = match self.leg(&link) {
Some(conn) if timed_out => { Some(_) if timed_out => {
let retry_peer = if self let retry_peer = if self
.peer_machines .peer_machines
.get(&link) .get(&link)
.is_some_and(|machine| machine.conn_is_outbound()) .is_some_and(|machine| machine.conn_is_outbound())
{ {
conn.expected_identity().map(|id| *id.node_addr()) self.peer_machines
.get(&link)
.and_then(|machine| machine.conn_expected_identity())
.map(|id| *id.node_addr())
} else { } else {
None None
}; };
src/node/lifecycle/mod.rs
+25 -34
View File
@@ -372,13 +372,12 @@ impl Node {
} }
fn is_connecting_to_peer(&self, peer_node_addr: &NodeAddr) -> bool { fn is_connecting_to_peer(&self, peer_node_addr: &NodeAddr) -> bool {
self.connections() self.connections().any(|(_, machine)| {
.filter_map(|(_, machine)| machine.leg()) machine
.any(|conn| { .conn_expected_identity()
conn.expected_identity() .map(|id| id.node_addr() == peer_node_addr)
.map(|id| id.node_addr() == peer_node_addr) .unwrap_or(false)
.unwrap_or(false) })
})
} }
fn is_connecting_to_peer_on_path( fn is_connecting_to_peer_on_path(
@@ -388,13 +387,13 @@ impl Node {
remote_addr: &TransportAddr, remote_addr: &TransportAddr,
) -> bool { ) -> bool {
self.peer_machines.values().any(|machine| { self.peer_machines.values().any(|machine| {
machine.leg().is_some_and(|conn| { machine.leg().is_some()
conn.expected_identity() && machine
.conn_expected_identity()
.map(|id| id.node_addr() == peer_node_addr) .map(|id| id.node_addr() == peer_node_addr)
.unwrap_or(false) .unwrap_or(false)
&& machine.conn_transport_id() == Some(transport_id) && machine.conn_transport_id() == Some(transport_id)
&& machine.conn_source_addr() == Some(remote_addr) && machine.conn_source_addr() == Some(remote_addr)
})
}) || self.peering.pending_connects.iter().any(|pending| { }) || self.peering.pending_connects.iter().any(|pending| {
pending.peer_identity.node_addr() == peer_node_addr pending.peer_identity.node_addr() == peer_node_addr
&& pending.transport_id == transport_id && pending.transport_id == transport_id
@@ -635,14 +634,6 @@ impl Node {
}; };
// Set index and transport info on the connection // Set index and transport info on the connection
{
let conn = self
.peer_machines
.get_mut(&link_id)
.and_then(|machine| machine.leg_mut())
.expect("dial-time machine carries the connection");
conn.set_our_index(our_index);
}
self.peer_machines self.peer_machines
.get_mut(&link_id) .get_mut(&link_id)
.expect("dial-time machine carries the connection") .expect("dial-time machine carries the connection")
@@ -683,10 +674,8 @@ impl Node {
// projects it to the promotion hand-off); holds even if a direct caller // projects it to the promotion hand-off); holds even if a direct caller
// reached here without the dial-time `on_dial` write. // reached here without the dial-time `on_dial` write.
machine.set_conn_transport_id(transport_id); machine.set_conn_transport_id(transport_id);
// Record our session index on the surviving carrier — the same index // Record our session index on the surviving carrier, the single index
// just written on the connection above — so the carrier is the single // home on the outbound path (the inbound path writes it at authorize).
// index home on the outbound path (the inbound path writes it at
// authorize).
machine.set_conn_our_index(our_index); machine.set_conn_our_index(our_index);
// Store the msg1 wire on the surviving carrier (the connection does not // Store the msg1 wire on the surviving carrier (the connection does not
// hold the resend source); the retransmit driver reads it from here. // hold the resend source); the retransmit driver reads it from here.
@@ -717,7 +706,11 @@ impl Node {
Some(w) => w.to_vec(), Some(w) => w.to_vec(),
None => return, None => return,
}; };
let our_index = self.leg(&link_id).and_then(|c| c.our_index()); let our_index = self
.peer_machines
.get(&link_id)
.filter(|machine| machine.leg().is_some())
.and_then(|machine| machine.our_index());
// Send the wire format handshake message // Send the wire format handshake message
if let Some(transport) = self.transports.get(&transport_id) { if let Some(transport) = self.transports.get(&transport_id) {
@@ -952,8 +945,7 @@ impl Node {
.connections() .connections()
.filter(|(_, machine)| { .filter(|(_, machine)| {
machine machine
.leg() .conn_expected_identity()
.and_then(|conn| conn.expected_identity())
.map(|id| id.node_addr() == &peer_addr) .map(|id| id.node_addr() == &peer_addr)
.unwrap_or(false) .unwrap_or(false)
}) })
@@ -2744,12 +2736,11 @@ impl Node {
let connected: HashSet<NodeAddr> = self.peers.keys().copied().collect(); let connected: HashSet<NodeAddr> = self.peers.keys().copied().collect();
let connecting: HashSet<NodeAddr> = self let connecting: HashSet<NodeAddr> = self
.connections() .connections()
.filter_map(|(_, machine)| machine.leg()) .filter_map(|(_, machine)| machine.conn_expected_identity().map(|id| *id.node_addr()))
.filter_map(|conn| conn.expected_identity().map(|id| *id.node_addr()))
.collect(); .collect();
let mut in_flight_by_peer: HashMap<NodeAddr, usize> = HashMap::new(); let mut in_flight_by_peer: HashMap<NodeAddr, usize> = HashMap::new();
for conn in self.connections().filter_map(|(_, machine)| machine.leg()) { for (_, machine) in self.connections() {
if let Some(id) = conn.expected_identity() { if let Some(id) = machine.conn_expected_identity() {
*in_flight_by_peer.entry(*id.node_addr()).or_default() += 1; *in_flight_by_peer.entry(*id.node_addr()).or_default() += 1;
} }
} }
@@ -2818,9 +2809,9 @@ impl Node {
let in_flight_for_peer = self let in_flight_for_peer = self
.connections() .connections()
.filter_map(|(_, machine)| machine.leg()) .filter(|(_, machine)| {
.filter(|conn| { machine
conn.expected_identity() .conn_expected_identity()
.map(|identity| identity.node_addr() == peer_node_addr) .map(|identity| identity.node_addr() == peer_node_addr)
.unwrap_or(false) .unwrap_or(false)
}) })
src/node/mod.rs
+22 -15
View File
@@ -2444,7 +2444,7 @@ impl Node {
} }
let machine = self.peer_machines.entry(link_id).or_insert_with(|| { let machine = self.peer_machines.entry(link_id).or_insert_with(|| {
let now = connection.started_at(); let now = connection.state().started_at();
match connection.state().expected_identity() { match connection.state().expected_identity() {
Some(identity) if connection.state().is_outbound() => { Some(identity) if connection.state().is_outbound() => {
PeerMachine::new_outbound(link_id, *identity, now) PeerMachine::new_outbound(link_id, *identity, now)
@@ -2455,10 +2455,13 @@ impl Node {
// Seed the surviving carrier's peer index and transport from the // Seed the surviving carrier's peer index and transport from the
// pre-built leg so the promotion hand-off reads them from the machine, // pre-built leg so the promotion hand-off reads them from the machine,
// matching the establish paths that write them on the machine directly. // matching the establish paths that write them on the machine directly.
if let Some(their) = connection.their_index() { if let Some(ours) = connection.state().our_index() {
machine.set_conn_our_index(ours);
}
if let Some(their) = connection.state().their_index() {
machine.set_conn_their_index(their); machine.set_conn_their_index(their);
} }
if let Some(tid) = connection.transport_id() { if let Some(tid) = connection.state().transport_id() {
machine.set_conn_transport_id(tid); machine.set_conn_transport_id(tid);
} }
if let Some(addr) = connection.state().source_addr() { if let Some(addr) = connection.state().source_addr() {
@@ -2473,12 +2476,13 @@ impl Node {
/// free-standing leg first. /// free-standing leg first.
/// ///
/// The carrier seeding below is a verbatim copy of `add_connection`'s: the /// The carrier seeding below is a verbatim copy of `add_connection`'s: the
/// two conditional writes (`their_index`, `transport_id`) and `set_leg`, /// conditional writes (`our_index`, `their_index`, `transport_id`,
/// built through the same `entry(..).or_insert_with(..)` so an existing /// `source_addr`) and `set_leg`, built through the same
/// leg-less machine keeps its constructor-side fields. Nothing else is /// `entry(..).or_insert_with(..)` so an existing leg-less machine keeps its
/// written to the carrier — `our_index`, `source_addr`, post-construction /// constructor-side fields. The seeded carrier matches what the establish
/// `started_at`, and the stored handshake bytes stay leg-only, exactly as /// paths write: every field a promotion reads is present. Post-construction
/// they do for a test that goes through `add_connection` today. /// `started_at` and the stored handshake bytes are not seeded here, exactly
/// as they are not for a test that goes through `add_connection` today.
/// ///
/// The duplication is deliberate: keeping the carrier writes visible here /// The duplication is deliberate: keeping the carrier writes visible here
/// is what lets each later step of the leg dissolution revise them at a /// is what lets each later step of the leg dissolution revise them at a
@@ -2494,14 +2498,14 @@ impl Node {
None => PeerConnection::inbound(link_id, seed.started_at_ms), None => PeerConnection::inbound(link_id, seed.started_at_ms),
}; };
if let Some(id) = seed.transport_id { if let Some(id) = seed.transport_id {
connection.set_transport_id(id); connection.state_mut().set_transport_id(id);
} }
let seeded_source_addr = seed.source_addr.clone(); let seeded_source_addr = seed.source_addr.clone();
if let Some(index) = seed.our_index { if let Some(index) = seed.our_index {
connection.set_our_index(index); connection.state_mut().set_our_index(index);
} }
if let Some(index) = seed.their_index { if let Some(index) = seed.their_index {
connection.set_their_index(index); connection.state_mut().set_their_index(index);
} }
if self if self
@@ -2519,7 +2523,7 @@ impl Node {
} }
let machine = self.peer_machines.entry(link_id).or_insert_with(|| { let machine = self.peer_machines.entry(link_id).or_insert_with(|| {
let now = connection.started_at(); let now = connection.state().started_at();
match connection.state().expected_identity() { match connection.state().expected_identity() {
Some(identity) if connection.state().is_outbound() => { Some(identity) if connection.state().is_outbound() => {
PeerMachine::new_outbound(link_id, *identity, now) PeerMachine::new_outbound(link_id, *identity, now)
@@ -2527,10 +2531,13 @@ impl Node {
_ => PeerMachine::new_inbound(link_id, now), _ => PeerMachine::new_inbound(link_id, now),
} }
}); });
if let Some(their) = connection.their_index() { if let Some(ours) = connection.state().our_index() {
machine.set_conn_our_index(ours);
}
if let Some(their) = connection.state().their_index() {
machine.set_conn_their_index(their); machine.set_conn_their_index(their);
} }
if let Some(tid) = connection.transport_id() { if let Some(tid) = connection.state().transport_id() {
machine.set_conn_transport_id(tid); machine.set_conn_transport_id(tid);
} }
if let Some(addr) = seeded_source_addr { if let Some(addr) = seeded_source_addr {
src/node/tests/handshake.rs
+13 -18
View File
@@ -866,19 +866,17 @@ async fn test_msg1_stored_for_resend() {
let noise_msg1 = conn let noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch(), now_ms) .start_handshake(our_keypair, node.startup_epoch(), now_ms)
.unwrap(); .unwrap();
conn.leg_mut().unwrap().set_our_index(our_index); conn.set_conn_our_index(our_index);
conn.leg_mut().unwrap().set_transport_id(transport_id); conn.set_conn_transport_id(transport_id);
conn.set_conn_source_addr(remote_addr.clone()); conn.set_conn_source_addr(remote_addr.clone());
// Build wire msg1 and store it (as initiate_peer_connection does) // Build wire msg1 and store it (as initiate_peer_connection does)
let wire_msg1 = build_msg1(our_index, &noise_msg1); let wire_msg1 = build_msg1(our_index, &noise_msg1);
let resend_interval = node.config().node.rate_limit.handshake_resend_interval_ms; let resend_interval = node.config().node.rate_limit.handshake_resend_interval_ms;
conn.leg_mut() conn.set_conn_handshake_msg1(wire_msg1.clone(), now_ms + resend_interval);
.unwrap()
.set_handshake_msg1(wire_msg1.clone(), now_ms + resend_interval);
// Verify stored msg1 matches what was built // Verify stored msg1 matches what was built
assert_eq!(conn.leg().unwrap().handshake_msg1().unwrap(), &wire_msg1); assert_eq!(conn.conn_handshake_msg1().unwrap(), &wire_msg1);
} }
/// Test that resend scheduling respects max_resends and backoff. /// Test that resend scheduling respects max_resends and backoff.
@@ -899,15 +897,10 @@ async fn test_resend_scheduling() {
let noise_msg1 = conn let noise_msg1 = conn
.start_handshake(our_keypair, node.startup_epoch(), now_ms) .start_handshake(our_keypair, node.startup_epoch(), now_ms)
.unwrap(); .unwrap();
conn.leg_mut().unwrap().set_our_index(our_index);
conn.leg_mut().unwrap().set_transport_id(transport_id);
conn.set_conn_source_addr(remote_addr.clone()); conn.set_conn_source_addr(remote_addr.clone());
// Store msg1 with first resend at now + 1000ms // Store msg1 with first resend at now + 1000ms
let wire_msg1 = crate::proto::fmp::wire::build_msg1(our_index, &noise_msg1); let wire_msg1 = crate::proto::fmp::wire::build_msg1(our_index, &noise_msg1);
conn.leg_mut()
.unwrap()
.set_handshake_msg1(wire_msg1.clone(), now_ms + 1000);
let link = Link::connectionless( let link = Link::connectionless(
link_id, link_id,
@@ -941,6 +934,8 @@ async fn test_resend_scheduling() {
// The msg1 wire lives on the machine's carrier (the retransmit driver's // The msg1 wire lives on the machine's carrier (the retransmit driver's
// resend source), mirroring `prepare_outbound_msg1`. // resend source), mirroring `prepare_outbound_msg1`.
machine.set_conn_handshake_msg1(wire_msg1, now_ms + 1000); machine.set_conn_handshake_msg1(wire_msg1, now_ms + 1000);
machine.set_conn_our_index(our_index);
machine.set_conn_transport_id(transport_id);
machine.set_leg(conn.take_leg().unwrap()); machine.set_leg(conn.take_leg().unwrap());
node.peer_machines.insert(link_id, machine); node.peer_machines.insert(link_id, machine);
node.peer_timers.entry(link_id).or_default().insert( node.peer_timers.entry(link_id).or_default().insert(
@@ -986,8 +981,6 @@ async fn test_handshake_timeout_drive() {
let _ = conn let _ = conn
.start_handshake(our_keypair, node.startup_epoch(), dial_ms) .start_handshake(our_keypair, node.startup_epoch(), dial_ms)
.unwrap(); .unwrap();
conn.leg_mut().unwrap().set_our_index(our_index);
conn.leg_mut().unwrap().set_transport_id(transport_id);
conn.set_conn_source_addr(remote_addr.clone()); conn.set_conn_source_addr(remote_addr.clone());
let link = Link::connectionless( let link = Link::connectionless(
@@ -1017,6 +1010,8 @@ async fn test_handshake_timeout_drive() {
dial_ms, dial_ms,
&mut node.index_allocator, &mut node.index_allocator,
); );
machine.set_conn_our_index(our_index);
machine.set_conn_transport_id(transport_id);
machine.set_leg(conn.take_leg().unwrap()); machine.set_leg(conn.take_leg().unwrap());
node.peer_machines.insert(link_id, machine); node.peer_machines.insert(link_id, machine);
node.peer_timers.entry(link_id).or_default().insert( node.peer_timers.entry(link_id).or_default().insert(
@@ -1045,17 +1040,17 @@ async fn test_handshake_timeout_drive() {
); );
} }
/// Test that msg2 is stored on PeerConnection for responder resend. /// Test that msg2 is stored on the control machine's carrier for responder resend.
#[test] #[test]
fn test_msg2_stored_on_connection() { fn test_msg2_stored_on_connection() {
let mut conn = PeerConnection::inbound(LinkId::new(1), 1000); let mut machine = crate::peer::machine::PeerMachine::new_inbound(LinkId::new(1), 1000);
assert!(conn.handshake_msg2().is_none()); assert!(machine.conn_handshake_msg2().is_none());
let msg2_bytes = vec![0x01, 0x02, 0x03, 0x04]; let msg2_bytes = vec![0x01, 0x02, 0x03, 0x04];
conn.set_handshake_msg2(msg2_bytes.clone()); machine.set_conn_handshake_msg2(msg2_bytes.clone());
assert_eq!(conn.handshake_msg2().unwrap(), &msg2_bytes); assert_eq!(machine.conn_handshake_msg2().unwrap(), &msg2_bytes);
} }
/// Test that duplicate msg2 is silently dropped when pending_outbound is already cleared. /// Test that duplicate msg2 is silently dropped when pending_outbound is already cleared.
src/node/tests/unit.rs
+62
View File
@@ -2405,6 +2405,68 @@ fn test_failed_connection_is_retained_and_reaped() {
assert_eq!(node.connection_count(), 0); assert_eq!(node.connection_count(), 0);
} }
/// The identity a responder discovers in msg1 must land on the surviving
/// carrier, not only on the pending leg. Everything that names an inbound
/// peer mid-handshake reads the carrier: the stale-connection sweep's
/// `retry_addr` decides whether a reaped leg is retried or torn down, and a
/// blank identity there silently changes that choreography.
#[test]
fn inbound_msg1_records_the_learned_identity_on_the_carrier() {
use crate::proto::fmp::LifecycleView;
let mut node = make_node();
let link_id = LinkId::new(77);
// A genuine IK msg1 addressed to this node, from a known sender.
let sender = Identity::generate();
let sender_identity = PeerIdentity::from_pubkey_full(sender.pubkey_full());
let node_identity = PeerIdentity::from_pubkey_full(node.identity().pubkey_full());
let initiator_link = LinkId::new(78);
let mut initiator =
crate::peer::machine::PeerMachine::new_outbound(initiator_link, node_identity, 1000);
initiator.set_leg(crate::peer::PeerConnection::outbound(
initiator_link,
node_identity,
1000,
));
let noise_msg1 = initiator
.start_handshake(sender.keypair(), [9u8; 8], 1000)
.unwrap();
// Drive the responder half over an inbound leg that stays pending.
node.seed_handshake_machine(HandshakeSeed::inbound(link_id, 1000))
.unwrap();
let our_keypair = node.identity().keypair();
let startup_epoch = node.startup_epoch();
let machine = node.peer_machines.get_mut(&link_id).unwrap();
machine
.receive_handshake_init(our_keypair, startup_epoch, &noise_msg1, 1000)
.unwrap();
assert_eq!(
machine.conn_expected_identity(),
Some(&sender_identity),
"msg1 identity learn must be recorded on the surviving carrier"
);
// The send of the responder's msg2 fails: the leg is retained, empty, for
// the sweep to reclaim.
machine.mark_failed();
machine.mark_send_failed();
let stale = node.stale_connections(2000, 30_000);
assert_eq!(
stale.len(),
1,
"the failed inbound leg must reach the sweep"
);
assert_eq!(
stale[0].retry_addr,
Some(*sender_identity.node_addr()),
"a failed inbound leg still names the peer it learned from msg1"
);
}
/// A msg1 that fails Noise processing must leave no trace in the registry. /// A msg1 that fails Noise processing must leave no trace in the registry.
/// The control machine is built above the crypto so it can drive the /// The control machine is built above the crypto so it can drive the
/// handshake, but it stays a local until a promote tail inserts it — a /// handshake, but it stays a local until a promote tail inserts it — a
src/peer/connection.rs
+4 -91
View File
@@ -11,7 +11,6 @@ use crate::PeerIdentity;
use crate::noise::{self, NoiseSession}; use crate::noise::{self, NoiseSession};
use crate::proto::fmp::ConnectionState; use crate::proto::fmp::ConnectionState;
use crate::transport::{LinkId, TransportAddr, TransportId}; use crate::transport::{LinkId, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
use std::fmt; use std::fmt;
/// A connection in the handshake phase, before authentication completes. /// A connection in the handshake phase, before authentication completes.
@@ -89,63 +88,6 @@ impl PeerConnection {
} }
} }
// === Accessors (delegated to the pure ConnectionState) ===
/// Get the expected/learned peer identity, if known.
pub fn expected_identity(&self) -> Option<&PeerIdentity> {
self.state.expected_identity()
}
/// When the connection started. Retained only to seed a control machine's
/// carrier from a pre-built leg (`Node::add_connection`); the operator-facing
/// `started_at_ms`/`last_activity_ms` telemetry now reads the machine carrier,
/// not the leg.
pub fn started_at(&self) -> u64 {
self.state.started_at()
}
/// Connection duration so far.
pub fn duration(&self, current_time_ms: u64) -> u64 {
self.state.duration(current_time_ms)
}
/// Time since last activity.
pub fn idle_time(&self, current_time_ms: u64) -> u64 {
self.state.idle_time(current_time_ms)
}
// === Index Accessors ===
/// Get our session index (if set).
pub fn our_index(&self) -> Option<SessionIndex> {
self.state.our_index()
}
/// Set our session index.
pub fn set_our_index(&mut self, index: SessionIndex) {
self.state.set_our_index(index);
}
/// Get their session index (if known).
pub fn their_index(&self) -> Option<SessionIndex> {
self.state.their_index()
}
/// Set their session index.
pub fn set_their_index(&mut self, index: SessionIndex) {
self.state.set_their_index(index);
}
/// Get the transport ID (if set).
pub fn transport_id(&self) -> Option<TransportId> {
self.state.transport_id()
}
/// Set the transport ID.
pub fn set_transport_id(&mut self, id: TransportId) {
self.state.set_transport_id(id);
}
// === Epoch Accessors === // === Epoch Accessors ===
/// Get the remote peer's startup epoch (available after handshake). /// Get the remote peer's startup epoch (available after handshake).
@@ -153,28 +95,6 @@ impl PeerConnection {
self.state.remote_epoch() self.state.remote_epoch()
} }
// === Handshake Resend ===
/// Store the wire-format msg1 bytes for resend and schedule the first resend.
pub fn set_handshake_msg1(&mut self, msg1: Vec<u8>, first_resend_at_ms: u64) {
self.state.set_handshake_msg1(msg1, first_resend_at_ms);
}
/// Store the wire-format msg2 bytes for resend on duplicate msg1.
pub fn set_handshake_msg2(&mut self, msg2: Vec<u8>) {
self.state.set_handshake_msg2(msg2);
}
/// Get the stored msg1 bytes (if any).
pub fn handshake_msg1(&self) -> Option<&[u8]> {
self.state.handshake_msg1()
}
/// Get the stored msg2 bytes (if any).
pub fn handshake_msg2(&self) -> Option<&[u8]> {
self.state.handshake_msg2()
}
// === Crypto handle plumbing (the control machine drives the handshake) === // === Crypto handle plumbing (the control machine drives the handshake) ===
/// Mutable access to the pure bookkeeping, so the control machine's /// Mutable access to the pure bookkeeping, so the control machine's
@@ -187,13 +107,6 @@ impl PeerConnection {
pub(crate) fn state_mut(&mut self) -> &mut ConnectionState { pub(crate) fn state_mut(&mut self) -> &mut ConnectionState {
&mut self.state &mut self.state
} }
// === Validation ===
/// Check if the connection has timed out.
pub fn is_timed_out(&self, current_time_ms: u64, timeout_ms: u64) -> bool {
self.state.is_timed_out(current_time_ms, timeout_ms)
}
} }
impl fmt::Debug for PeerConnection { impl fmt::Debug for PeerConnection {
@@ -228,9 +141,9 @@ mod tests {
let identity = make_peer_identity(); let identity = make_peer_identity();
let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000);
assert_eq!(conn.duration(1500), 500); assert_eq!(conn.state().duration(1500), 500);
assert_eq!(conn.idle_time(1500), 500); assert_eq!(conn.state().idle_time(1500), 500);
assert!(!conn.is_timed_out(1500, 1000)); assert!(!conn.state().is_timed_out(1500, 1000));
assert!(conn.is_timed_out(2500, 1000)); assert!(conn.state().is_timed_out(2500, 1000));
} }
} }
src/peer/machine.rs
+12 -10
View File
@@ -574,6 +574,7 @@ impl PeerMachine {
) -> Result<Vec<u8>, NoiseError> { ) -> Result<Vec<u8>, NoiseError> {
let msg1 = { let msg1 = {
let direction = self.conn.direction(); let direction = self.conn.direction();
let expected_identity = self.conn.expected_identity().copied();
let leg = self.leg.as_mut().ok_or_else(no_pending_connection)?; let leg = self.leg.as_mut().ok_or_else(no_pending_connection)?;
if direction != LinkDirection::Outbound { if direction != LinkDirection::Outbound {
@@ -583,8 +584,7 @@ impl PeerMachine {
}); });
} }
let remote_static = leg let remote_static = expected_identity
.expected_identity()
.expect("outbound must have expected identity") .expect("outbound must have expected identity")
.pubkey_full(); .pubkey_full();
@@ -776,9 +776,11 @@ impl PeerMachine {
/// Expected peer identity of the surviving carrier — the home for the /// Expected peer identity of the surviving carrier — the home for the
/// operator-visible `expected_peer` now that the leg no longer projects it. /// operator-visible `expected_peer` now that the leg no longer projects it.
/// Outbound carries the dial identity from construction; inbound stays `None` /// Outbound carries the dial identity from construction; inbound records
/// in this view (the identity learned mid-handshake never rests here, as the /// the identity discovered in msg1, written here by `receive_handshake_init`
/// leg is consumed by promotion within the same message-handling step). /// at the same point it reaches the pending connection. Everything that
/// names a peer mid-handshake reads this, including the stale-connection
/// sweep's retry address.
pub(crate) fn conn_expected_identity(&self) -> Option<&PeerIdentity> { pub(crate) fn conn_expected_identity(&self) -> Option<&PeerIdentity> {
self.conn.expected_identity() self.conn.expected_identity()
} }
@@ -3240,8 +3242,8 @@ mod tests {
assert!(conn.conn_is_outbound()); assert!(conn.conn_is_outbound());
assert!(!conn.conn_is_inbound()); assert!(!conn.conn_is_inbound());
assert!(!conn.has_session()); assert!(!conn.has_session());
assert!(conn.leg().unwrap().expected_identity().is_some()); assert!(conn.conn_expected_identity().is_some());
assert_eq!(conn.leg().unwrap().started_at(), 1000); assert_eq!(conn.conn_started_at(), 1000);
} }
#[test] #[test]
@@ -3251,8 +3253,8 @@ mod tests {
assert!(conn.conn_is_inbound()); assert!(conn.conn_is_inbound());
assert!(!conn.conn_is_outbound()); assert!(!conn.conn_is_outbound());
assert!(!conn.has_session()); assert!(!conn.has_session());
assert!(conn.leg().unwrap().expected_identity().is_none()); assert!(conn.conn_expected_identity().is_none());
assert_eq!(conn.leg().unwrap().started_at(), 2000); assert_eq!(conn.conn_started_at(), 2000);
} }
#[test] #[test]
@@ -3288,7 +3290,7 @@ mod tests {
assert!(responder_conn.has_session()); assert!(responder_conn.has_session());
// Responder learned initiator's identity // Responder learned initiator's identity
let discovered = responder_conn.leg().unwrap().expected_identity().unwrap(); let discovered = responder_conn.conn_expected_identity().unwrap();
assert_eq!(discovered.pubkey(), initiator_identity.pubkey()); assert_eq!(discovered.pubkey(), initiator_identity.pubkey());
// Responder learned initiator's epoch // Responder learned initiator's epoch
src/proto/fmp/state.rs
+1
View File
@@ -210,6 +210,7 @@ impl ConnectionState {
} }
/// Connection duration so far. /// Connection duration so far.
#[cfg(test)]
pub fn duration(&self, current_time_ms: u64) -> u64 { pub fn duration(&self, current_time_ms: u64) -> u64 {
current_time_ms.saturating_sub(self.started_at) current_time_ms.saturating_sub(self.started_at)
} }