diff --git a/src/node/dataplane/mod.rs b/src/node/dataplane/mod.rs index e0643a1..bcf5d61 100644 --- a/src/node/dataplane/mod.rs +++ b/src/node/dataplane/mod.rs @@ -14,3 +14,5 @@ mod encrypted; mod forwarding; mod peer_actions; mod rx_loop; + +pub(in crate::node) use peer_actions::PeerActionCtx; diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 88f00c3..40d1891 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -20,6 +20,7 @@ //! `RegisterDecryptSession` is a deliberate no-op — see its arm for the C3-2 note. use crate::node::Node; +use crate::node::reject::{HandshakeReject, RejectReason}; use crate::peer::machine::{PeerAction, PeerEvent}; use crate::proto::fmp::wire::build_msg2; use crate::transport::{LinkId, TransportAddr, TransportId}; @@ -96,7 +97,6 @@ impl Node { ambient: &PeerActionCtx, actions: Vec, ) { - let _ = link; let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { @@ -115,8 +115,30 @@ impl Node { (ambient.our_index, ambient.their_index) { let frame = build_msg2(sender_idx, receiver_idx, &bytes); - if let Some(transport) = self.transports.get(&ambient.transport_id) { - let _ = transport.send(&ambient.remote_addr, &frame).await; + // GAP-5: surface the send Result. A missing transport skips + // the send and continues (mirrors `handle_msg1`'s + // `if let Some(transport)` guard); a send *error* runs the + // pre-refactor msg2-send-failure cleanup (`handle_msg1` + // L494-503) and ABORTS the remaining queue so the queued + // `PromoteToActive` never runs. + let send_err = match self.transports.get(&ambient.transport_id) { + Some(transport) => { + transport.send(&ambient.remote_addr, &frame).await.is_err() + } + None => false, + }; + if send_err { + self.connections.remove(&link); + self.links.remove(&link); + self.addr_to_link + .remove(&(ambient.transport_id, ambient.remote_addr.clone())); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&link); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; } } } @@ -151,8 +173,19 @@ impl Node { queue.extend(follow); } Err(_e) => { - // C3-2 realizes the promotion-failure cleanup tail - // (`handle_msg1:587` / `handle_msg2:1005`). + // GAP-4: promotion failed. `promote_connection` already + // removed `connections[link]`; mirror the pre-refactor + // cleanup (`handle_msg1` L587-591): drop the link + + // reverse map, free our index, discard the machine, and + // record the reject. The queue is drained (PromoteToActive + // is the last establish action), so no explicit abort. + self.remove_link(&promote_link); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&promote_link); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 8b6875f..9aff408 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -3,8 +3,10 @@ use crate::NodeAddr; use crate::PeerIdentity; use crate::node::acl::PeerAclContext; +use crate::node::dataplane::PeerActionCtx; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; +use crate::peer::machine::{FailReason, HandshakePhase, PeerEvent, PeerMachine, PeerState}; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; use crate::proto::fmp::{ @@ -272,13 +274,47 @@ impl Node { // shared authorize → allocate → send-msg2 → promote tail; the other // variants complete the rate-limiter and return here. match self.fmp.establish_inbound(&est, &wire) { - InboundDecision::Reject { reason } => { + InboundDecision::Reject { + reason: InboundReject::AtMaxPeers, + } => { + // C3-2a net-new arm: drive the reject through the machine to + // prove realization A — a net-new msg1 at the max-peers cap + // reaches `Failed{Rejected}` with the index allocator untouched + // (no allocate before the reject). The transient machine is + // discarded (never inserted into `peer_machines`); `conn`/ + // `link_id` were never inserted into the registry either. + debug!( + peer = %self.peer_display_name(&peer_node_addr), + max = self.max_peers(), + "Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)" + ); + let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + let _ = machine.step( + PeerEvent::InboundMsg1 { + link: link_id, + wire, + est, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + debug_assert!(matches!( + machine.state(), + PeerState::Failed { + reason: FailReason::Rejected + } + )); + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + InboundDecision::Reject { + reason: reason @ (InboundReject::PendingSession | InboundReject::DualRekeyWon), + } => { + // Existing-peer rekey rejects — still inline (C4). Byte-unchanged + // from the pre-refactor shared reject tail. match reason { - InboundReject::AtMaxPeers => debug!( - peer = %self.peer_display_name(&peer_node_addr), - max = self.max_peers(), - "Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)" - ), InboundReject::PendingSession => debug!( peer = %self.peer_display_name(&peer_node_addr), "Rekey msg1 received but already have pending session, dropping" @@ -287,6 +323,7 @@ impl Node { peer = %self.peer_display_name(&peer_node_addr), "Dual rekey initiation: we win (smaller addr), dropping their msg1" ), + InboundReject::AtMaxPeers => unreachable!(), } // `conn`/`link_id` were never inserted into the registry, so the // local drop suffices — no cleanup needed. @@ -419,7 +456,146 @@ impl Node { .unwrap_or(0); self.note_link_dead(peer, now_ms); } - InboundDecision::Promote => {} + InboundDecision::Promote => { + // === C3-2a: net-new inbound establish, driven by the machine. === + // Realization A (two-phase authorize): Phase 1 classifies with no + // allocation; the shell interposes the late-ACL gate here; Phase 2 + // allocates the single index and emits [SendHandshake, + // PromoteToActive]. A rejected/unauthorized msg1 therefore + // allocates NO index — matching the pre-refactor + // authorize-before-allocate ordering exactly. + + // Keep the shell's own copies of the msg2 framing inputs before + // the machine event consumes `wire` (WireOutcome is not Clone). + let msg2_payload = wire.msg2_payload.clone(); + let their_index = wire.their_index; + + let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + + // Phase 1: classify (no allocation, no actions for a net-new leg). + let phase1 = machine.step( + PeerEvent::InboundMsg1 { + link: link_id, + wire, + est, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + debug_assert!(phase1.is_empty()); + debug_assert!(matches!( + machine.state(), + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + )); + + // Shell interposition: late-ACL authorize BEFORE any allocation. + if self + .authorize_peer( + &peer_identity, + PeerAclContext::InboundHandshake, + packet.transport_id, + &packet.remote_addr, + ) + .is_err() + { + let _ = machine.step( + PeerEvent::Rejected, + packet.timestamp_ms, + &mut self.index_allocator, + ); + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + + // Phase 2: allocate our index + emit [SendHandshake, PromoteToActive]. + let promote_actions = machine.step( + PeerEvent::Authorized, + packet.timestamp_ms, + &mut self.index_allocator, + ); + let our_index = match machine.our_index() { + Some(idx) => idx, + None => { + // Allocation exhausted in Phase 2 (mirrors the pre-refactor + // allocate-failure path): no msg2, no promote. + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + }; + + // Shell registry surgery (Option A1), in the pre-refactor order: + // set indices on the shell connection, insert link / reverse map / + // connection, then build + store the framed msg2. + conn.set_our_index(our_index); + conn.set_their_index(their_index); + let link = Link::connectionless( + link_id, + packet.transport_id, + packet.remote_addr.clone(), + LinkDirection::Inbound, + Duration::from_millis(self.config().node.base_rtt_ms), + ); + self.links.insert(link_id, link); + self.addr_to_link.insert(addr_key, link_id); + self.connections.insert(link_id, conn); + let wire_msg2 = build_msg2(our_index, their_index, &msg2_payload); + if let Some(conn) = self.connections.get_mut(&link_id) { + conn.set_handshake_msg2(wire_msg2.clone()); + } + + // Register the machine (Promote tail only — discarded on every + // reject/resend/rekey arm per the insertion discipline). + self.peer_machines.insert(link_id, machine); + + // Execute [SendHandshake, PromoteToActive]. The executor frames + + // sends msg2 (bytes identical to `wire_msg2`), promotes via + // `promote_connection`, feeds PromotionResolved back, and runs the + // inert RegisterDecryptSession (R2 — register stays in + // `promote_connection`). Its send-failure / promote-failure arms + // run the pre-refactor cleanup and remove the machine, leaving it + // absent (not Established). + let ambient = PeerActionCtx { + verified_identity: peer_identity, + transport_id: packet.transport_id, + remote_addr: packet.remote_addr.clone(), + our_index: Some(our_index), + their_index: Some(their_index), + now_ms: packet.timestamp_ms, + }; + self.execute_peer_actions(link_id, &ambient, promote_actions) + .await; + + // Post-`Promoted` shell tail (byte-identical to the pre-refactor + // Promoted arm), reached only when promotion succeeded (the machine + // is now Established); a send/promote failure removed the machine + // and already cleaned up. + if matches!( + self.peer_machines.get(&link_id).map(|m| m.state()), + Some(PeerState::Established { .. }) + ) { + // Store msg2 on peer for resend on duplicate msg1 + if let Some(peer) = self.peers.get_mut(&peer_node_addr) { + peer.set_handshake_msg2(wire_msg2.clone()); + } + // Send initial tree announce to new peer + if let Err(e) = self.send_tree_announce_to_peer(&peer_node_addr).await { + debug!(peer = %self.peer_display_name(&peer_node_addr), error = %e, "Failed to send initial TreeAnnounce"); + } + // Schedule filter announce (sent on next tick via debounce) + self.bloom_state.mark_update_needed(peer_node_addr); + self.reset_lookup_backoff(); + } + + self.msg1_rate_limiter.complete_handshake(); + return; + } } if self diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 179b4a4..2b5e885 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -305,6 +305,10 @@ pub(crate) struct PeerMachine { conn: ConnectionState, /// Remote startup epoch (establish-path-only; NOT in send-state). remote_epoch: Option<[u8; 8]>, + /// Inbound two-phase authorize (realization A): the opaque Noise msg2 + /// payload stashed in Phase 1 (`InboundMsg1`) and emitted in Phase 2 + /// (`on_authorized`), so a rejected/unauthorized msg1 allocates no index. + pending_msg2_payload: Option>, // --- rekey negotiation sub-state (control tier; NOT the pending send slot) --- rekey_in_progress: bool, @@ -343,6 +347,7 @@ impl PeerMachine { identity: Some(identity), conn: ConnectionState::outbound(link, identity, now), remote_epoch: None, + pending_msg2_payload: None, rekey_in_progress: false, rekey_our_index: None, rekey_msg1: None, @@ -368,6 +373,7 @@ impl PeerMachine { identity: None, conn: ConnectionState::inbound(link, now), remote_epoch: None, + pending_msg2_payload: None, rekey_in_progress: false, rekey_our_index: None, rekey_msg1: None, @@ -387,6 +393,14 @@ impl PeerMachine { self.state } + /// The index we allocated for this peer's inbound session, once Phase 2 + /// (`on_authorized`) has run. `None` before allocation (and after a + /// rejected/unauthorized msg1). The inbound cutover reads this to perform + /// the shell registry surgery with the machine-owned index (Option A1). + pub(crate) fn our_index(&self) -> Option { + self.our_index + } + /// The crystallized node address, if identity is known. fn addr(&self) -> Option { self.identity.map(|id| *id.node_addr()) @@ -421,7 +435,7 @@ impl PeerMachine { PeerEvent::Msg2 { their_index, out } => { self.on_msg2(their_index, out, now, index_allocator) } - PeerEvent::Authorized => Vec::new(), + PeerEvent::Authorized => self.on_authorized(now, index_allocator), PeerEvent::Rejected => self.fail(FailReason::AclRejected), PeerEvent::PromotionResolved { result } => self.on_promotion_resolved(result, now), PeerEvent::RekeyMsg1 { wire, est } => { @@ -608,40 +622,67 @@ impl PeerMachine { actions.push(PeerAction::UnregisterDecryptSession { index: idx }); } actions.push(PeerAction::ReportLost { peer }); - actions.extend(self.inbound_promote(link, &wire, now, alloc)); + actions.extend(self.inbound_classify(link, &wire)); actions } - InboundDecision::Promote => self.inbound_promote(link, &wire, now, alloc), + InboundDecision::Promote => self.inbound_classify(link, &wire), } } - /// The inbound Promote tail: allocate our index, record indices/epoch/msg2, - /// emit msg2 + drive promotion. `RegisterDecryptSession` follows on the - /// `PromotionResolved{Promoted}` feedback (§3.2 "then on PromotionResult"). - fn inbound_promote( - &mut self, - link: LinkId, - wire: &WireOutcome, - _now: u64, - alloc: &mut IndexAllocator, - ) -> Vec { + /// Inbound **Phase 1** (realization A): classify the fresh leg *without* + /// allocating an index. Records identity/epoch/their-index and stashes the + /// opaque msg2 payload, parking at `Handshaking{ReceivedMsg1}` — the + /// "awaiting Authorized" marker. The index allocation and the msg2/promote + /// emission happen in Phase 2 ([`Self::on_authorized`]) only after the + /// shell's late-ACL gate passes, so a rejected/unauthorized msg1 allocates + /// nothing (preserving the pre-refactor global index-allocation sequence). + fn inbound_classify(&mut self, link: LinkId, wire: &WireOutcome) -> Vec { self.identity = Some(wire.peer_identity); self.remote_epoch = wire.remote_epoch; self.conn.set_their_index(wire.their_index); - let our_index = alloc.allocate().ok(); - if let Some(idx) = our_index { - self.conn.set_our_index(idx); - self.our_index = Some(idx); - } - self.conn.set_handshake_msg2(wire.msg2_payload.clone()); + self.pending_msg2_payload = Some(wire.msg2_payload.clone()); self.state = PeerState::Handshaking { link, phase: HandshakePhase::ReceivedMsg1, }; + Vec::new() + } + + /// Inbound **Phase 2** (realization A): the late-ACL gate passed shell-side. + /// Allocate our index NOW — the single inbound allocation point — record it + /// on `conn`, and emit the msg2 send + promotion. `RegisterDecryptSession` + /// follows on the `PromotionResolved{Promoted}` feedback (§3.2). Guarded to + /// the inbound `ReceivedMsg1` phase so the benign outbound `Authorized` + /// confirmation stays a no-op (state `Handshaking{SentMsg1}` and every other + /// state fall through to `Vec::new()`). + fn on_authorized(&mut self, _now: u64, alloc: &mut IndexAllocator) -> Vec { + if !matches!( + self.state, + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + ) { + return Vec::new(); + } + let our_index = match alloc.allocate() { + Ok(idx) => idx, + Err(_) => { + // Allocation exhausted: no index, no msg2, no promote. The shell + // records the reject + completes the rate-limiter bracket + // (mirrors the pre-refactor `handle_msg1` allocate-failure path). + self.state = PeerState::Failed { + reason: FailReason::Rejected, + }; + return Vec::new(); + } + }; + self.conn.set_our_index(our_index); + self.our_index = Some(our_index); + let bytes = self.pending_msg2_payload.take().unwrap_or_default(); + let link = self.link; vec![ - PeerAction::SendHandshake { - bytes: wire.msg2_payload.clone(), - }, + PeerAction::SendHandshake { bytes }, PeerAction::PromoteToActive { link }, ] } @@ -1352,22 +1393,37 @@ mod tests { &mut alloc, ); - // Restart tail: invalidate, unregister old, report lost, then Promote. - assert_eq!(actions[0], PeerAction::InvalidateSendState); + // Phase 1: restart tail only (invalidate, unregister old, report lost), + // then park at ReceivedMsg1 — no index allocated yet (realization A). assert_eq!( - actions[1], - PeerAction::UnregisterDecryptSession { - index: SessionIndex::new(0xDEAD) - } + actions, + vec![ + PeerAction::InvalidateSendState, + PeerAction::UnregisterDecryptSession { + index: SessionIndex::new(0xDEAD) + }, + PeerAction::ReportLost { peer: peer_addr }, + ] ); - assert_eq!(actions[2], PeerAction::ReportLost { peer: peer_addr }); + assert!(matches!( + m.state(), + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + )); + assert_eq!(m.our_index(), None); + assert_eq!(alloc.count(), 0); + + // Phase 2: late-ACL gate passed -> allocate + Promote tail. + let promote = m.step(PeerEvent::Authorized, 1_000, &mut alloc); assert!( - actions + promote .iter() .any(|a| matches!(a, PeerAction::SendHandshake { .. })) ); assert!( - actions + promote .iter() .any(|a| matches!(a, PeerAction::PromoteToActive { .. })) ); @@ -1400,7 +1456,7 @@ mod tests { let our = *peer_identity().node_addr(); let est_w = est_new_peer(our); let wire_w = wire_outcome(peer, Some([3u8; 8]), 0x77); - let wa = winner.step( + let wp1 = winner.step( PeerEvent::InboundMsg1 { link: LinkId::new(1), wire: wire_w, @@ -1409,6 +1465,8 @@ mod tests { 100, &mut alloc, ); + assert!(wp1.is_empty()); // Phase 1 classifies without emitting. + let wa = winner.step(PeerEvent::Authorized, 100, &mut alloc); assert!( wa.iter().any( |a| matches!(a, PeerAction::PromoteToActive { link } if *link == LinkId::new(1)) @@ -1432,7 +1490,7 @@ mod tests { let mut loser = PeerMachine::new_inbound(LinkId::new(2), 0); let est_l = est_new_peer(our); let wire_l = wire_outcome(peer, Some([3u8; 8]), 0x88); - let la = loser.step( + let lp1 = loser.step( PeerEvent::InboundMsg1 { link: LinkId::new(2), wire: wire_l, @@ -1441,6 +1499,8 @@ mod tests { 100, &mut alloc, ); + assert!(lp1.is_empty()); // Phase 1 classifies without emitting. + let la = loser.step(PeerEvent::Authorized, 100, &mut alloc); assert!( la.iter() .any(|a| matches!(a, PeerAction::PromoteToActive { .. })) @@ -1490,7 +1550,9 @@ mod tests { let est = est_new_peer(our); let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); - let mut actions = m.step( + // Phase 1 (InboundMsg1): classify only — no actions, no allocation, + // parked at ReceivedMsg1 (realization A). + let phase1 = m.step( PeerEvent::InboundMsg1 { link: LinkId::new(1), wire, @@ -1499,29 +1561,111 @@ mod tests { 200, &mut alloc, ); - actions.extend(m.step( + assert!(phase1.is_empty()); + assert_eq!( + m.state(), + PeerState::Handshaking { + link: LinkId::new(1), + phase: HandshakePhase::ReceivedMsg1 + } + ); + assert_eq!(m.our_index(), None); + assert_eq!(alloc.count(), 0); // allocator untouched pre-authorize + + // Phase 2 (Authorized): allocate + [SendHandshake, PromoteToActive]. + let phase2 = m.step(PeerEvent::Authorized, 200, &mut alloc); + assert!(matches!(phase2[0], PeerAction::SendHandshake { .. })); + assert_eq!( + phase2[1], + PeerAction::PromoteToActive { + link: LinkId::new(1) + } + ); + assert!(m.our_index().is_some()); + assert_eq!(alloc.count(), 1); // exactly one index allocated + + // Phase 3 (PromotionResolved{Promoted}): register + Established. + let phase3 = m.step( PeerEvent::PromotionResolved { result: PromotionResult::Promoted(peer_addr), }, 200, &mut alloc, - )); - - // Combined promote sequence (§3.2 "then on PromotionResult ..."). - assert!(matches!(actions[0], PeerAction::SendHandshake { .. })); - assert_eq!( - actions[1], - PeerAction::PromoteToActive { - link: LinkId::new(1) - } ); assert!(matches!( - actions[2], + phase3[0], PeerAction::RegisterDecryptSession { .. } )); assert_eq!(m.state(), PeerState::Established { addr: peer_addr }); } + // ---- Test 6b: inbound late-ACL rejected -> no allocation -------------- + #[test] + fn inbound_authorize_rejected_no_alloc() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + let mut m = PeerMachine::new_inbound(LinkId::new(1), 0); + let our = *peer_identity().node_addr(); + let est = est_new_peer(our); + let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); + + // Phase 1 classifies (no alloc). + let phase1 = m.step( + PeerEvent::InboundMsg1 { + link: LinkId::new(1), + wire, + est, + }, + 200, + &mut alloc, + ); + assert!(phase1.is_empty()); + assert_eq!(alloc.count(), 0); + + // Late-ACL rejects -> Failed{AclRejected}, still no allocation. + let rej = m.step(PeerEvent::Rejected, 200, &mut alloc); + assert!(rej.is_empty()); + assert_eq!( + m.state(), + PeerState::Failed { + reason: FailReason::AclRejected + } + ); + assert_eq!(alloc.count(), 0); + assert_eq!(m.our_index(), None); + } + + // ---- Test 6c: inbound reject at max_peers -> no allocation ------------ + #[test] + fn inbound_at_max_peers_reject_no_alloc() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + let mut m = PeerMachine::new_inbound(LinkId::new(1), 0); + let our = *peer_identity().node_addr(); + let mut est = est_new_peer(our); + est.at_max_peers = true; + let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); + + let actions = m.step( + PeerEvent::InboundMsg1 { + link: LinkId::new(1), + wire, + est, + }, + 200, + &mut alloc, + ); + assert!(actions.is_empty()); + assert_eq!( + m.state(), + PeerState::Failed { + reason: FailReason::Rejected + } + ); + assert_eq!(alloc.count(), 0); + assert_eq!(m.our_index(), None); + } + // ---- Test 7: outbound establish (+ cross-connection) ------------------ #[test] fn outbound_establish() {