diff --git i/src/node/handlers/session.rs w/src/node/handlers/session.rs index 853adb1..98f2225 100644 --- i/src/node/handlers/session.rs +++ w/src/node/handlers/session.rs @@ -456,7 +456,8 @@ impl Node { self.handle_session_ack(src_addr, inner).await; } FSP_PHASE_MSG3 => { - self.handle_session_msg3(src_addr, inner).await; + self.handle_session_msg3_from(src_addr, link_peer, inner) + .await; } FSP_PHASE_ESTABLISHED if prefix.is_unencrypted() => { // Plaintext error signals: read msg_type from first byte after prefix @@ -483,8 +484,10 @@ impl Node { } } FSP_PHASE_ESTABLISHED => { - self.handle_encrypted_session_msg(src_addr, payload, path_mtu, ce_flag) - .await; + self.handle_encrypted_session_msg_from( + src_addr, link_peer, payload, path_mtu, ce_flag, + ) + .await; } _ => { debug!(phase = prefix.phase, "Unknown FSP phase"); @@ -501,9 +504,10 @@ impl Node { /// 4. AEAD decrypt with AAD = header_bytes /// 5. Strip FSP inner header → timestamp, msg_type, inner_flags /// 6. Dispatch by msg_type - async fn handle_encrypted_session_msg( + async fn handle_encrypted_session_msg_from( &mut self, src_addr: &NodeAddr, + link_peer: &NodeAddr, payload: &[u8], path_mtu: u16, ce_flag: bool, @@ -665,6 +669,7 @@ impl Node { } } + entry.note_authenticated_ingress(*link_peer, crate::time::mono_ms()); self.sessions.insert(*src_addr, entry); // Strip FSP inner header (6 bytes) @@ -1061,7 +1066,10 @@ impl Node { let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack_payload) .with_ttl(self.config().node.session.default_ttl); - if let Err(e) = self.send_session_datagram(&mut datagram).await { + if let Err(e) = self + .send_session_datagram_via(&mut datagram, *link_peer) + .await + { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send rekey SessionAck"); return; } @@ -1097,7 +1105,10 @@ impl Node { let my_addr = *self.node_addr(); let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack.to_vec()) .with_ttl(self.config().node.session.default_ttl); - if let Err(e) = self.send_session_datagram(&mut datagram).await { + if let Err(e) = self + .send_session_datagram_via(&mut datagram, *link_peer) + .await + { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to resend SessionAck"); } return; @@ -1193,7 +1204,10 @@ impl Node { .with_ttl(self.config().node.session.default_ttl); // Route the ack back to the initiator - if let Err(e) = self.send_session_datagram(&mut datagram).await { + if let Err(e) = self + .send_session_datagram_via(&mut datagram, *link_peer) + .await + { debug!(error = %e, dest = %self.peer_display_name(src_addr), "Failed to send SessionAck"); return; } @@ -1556,7 +1570,12 @@ impl Node { /// The initiator reveals their encrypted static key. The responder /// processes msg3, learns the initiator's identity, and transitions /// to Established. - async fn handle_session_msg3(&mut self, src_addr: &NodeAddr, inner: &[u8]) { + async fn handle_session_msg3_from( + &mut self, + src_addr: &NodeAddr, + link_peer: &NodeAddr, + inner: &[u8], + ) { let msg3 = match SessionMsg3::decode(inner) { Ok(m) => m, Err(e) => { @@ -1808,6 +1827,8 @@ impl Node { new_entry.mark_established(now_ms); new_entry.init_mmp(&self.config().node.session_mmp); new_entry.touch(now_ms); + // Only after msg3 proves the static key and its source-address binding. + new_entry.note_authenticated_ingress(*link_peer, crate::time::mono_ms()); self.sessions.insert(*src_addr, new_entry); // Flush any pending packets @@ -3163,8 +3184,7 @@ impl Node { return Ok(false); }; - let Some(next_hop_addr) = self.find_next_hop(dest_addr).map(|peer| *peer.node_addr()) - else { + let Some(next_hop_addr) = self.session_next_hop(dest_addr) else { return Err(NodeError::SendFailed { node_addr: *dest_addr, reason: "no route to destination".into(), @@ -3630,12 +3650,47 @@ impl Node { /// /// Finds the next hop for the destination, seeds path_mtu from the /// first-hop transport MTU, and sends as an encrypted link message. + /// Preserve normal tree routing. Only bypass a stale direct destination + /// edge, and only with fresh end-to-end authenticated session evidence. + pub(in crate::node) fn session_next_hop(&mut self, dest: &NodeAddr) -> Option { + let normal = self.find_next_hop(dest).map(|peer| *peer.node_addr())?; + if normal != *dest { + return Some(normal); + } + let Some((hop, heard)) = self + .sessions + .get(dest) + .and_then(|entry| entry.authenticated_return_hop()) + else { + return Some(normal); + }; + let lifetime = self + .config() + .node + .link_dead_timeout_secs + .saturating_mul(1000); + // A fresh authenticated frame on the direct link wins again, without + // tearing down either peering or changing its liveness thresholds. + let direct_heard = self + .peers + .get(dest) + .and_then(|peer| peer.mmp()) + .and_then(|mmp| mmp.receiver.last_recv_ms()); + if !return_path_is_fresh(crate::time::mono_ms(), heard, direct_heard, lifetime) { + return Some(normal); + } + match self.peers.get(&hop) { + Some(peer) if peer.can_send() && peer.is_heard() => Some(hop), + _ => Some(normal), + } + } + pub(in crate::node) async fn send_session_datagram( &mut self, datagram: &mut SessionDatagram, ) -> Result<(), NodeError> { - let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) { - Some(peer) => *peer.node_addr(), + let next_hop_addr = match self.session_next_hop(&datagram.dest_addr) { + Some(peer) => peer, None => { return Err(NodeError::SendFailed { node_addr: datagram.dest_addr, @@ -3644,6 +3699,15 @@ impl Node { } }; + self.send_session_datagram_via(datagram, next_hop_addr) + .await + } + + async fn send_session_datagram_via( + &mut self, + datagram: &mut SessionDatagram, + next_hop_addr: NodeAddr, + ) -> Result<(), NodeError> { // Seed path_mtu from the first-hop transport MTU (same as forwarding path) if let Some(peer) = self.peers.get(&next_hop_addr) && let Some(tid) = peer.transport_id() @@ -4119,3 +4183,21 @@ impl Node { } } } + +// All times here are receiver-local monotonic observations, never wire values. +fn return_path_is_fresh(now: u64, heard: u64, direct_heard: Option, lifetime: u64) -> bool { + now.saturating_sub(heard) < lifetime && !direct_heard.is_some_and(|last| last > heard) +} + +#[cfg(test)] +mod authenticated_return_path_tests { + use super::return_path_is_fresh; + #[test] + fn hints_expire_and_new_direct_traffic_wins() { + assert!(return_path_is_fresh(100, 95, Some(90), 30)); + assert!(return_path_is_fresh(100, 95, Some(95), 30)); + assert!(!return_path_is_fresh(100, 95, Some(96), 30)); + assert!(!return_path_is_fresh(125, 95, Some(90), 30)); + assert!(!return_path_is_fresh(95, 95, None, 0)); + } +} diff --git i/src/node/session/mod.rs w/src/node/session/mod.rs index bd4c8d9..53833b0 100644 --- i/src/node/session/mod.rs +++ w/src/node/session/mod.rs @@ -233,6 +233,9 @@ pub(crate) struct SessionEntry { /// Only updated for DataPacket send/receive and session establishment. /// MMP reports do not update this field. Used for idle session timeout. last_activity: u64, + /// Last end-to-end authenticated ingress; monotonic receive time. Scoped + /// to this session, never learned from a claimed envelope or bare msg1. + authenticated_return_hop: Option<(NodeAddr, u64)>, /// When the session transitioned to Established (Unix milliseconds). /// Used to compute session-relative timestamps for the FSP inner header. /// Set to 0 until the session is established. @@ -351,6 +354,18 @@ pub(crate) struct SessionEntry { } impl SessionEntry { + pub(crate) fn note_authenticated_ingress(&mut self, hop: NodeAddr, now_ms: u64) { + self.authenticated_return_hop = if hop == self.remote_addr { + None + } else { + Some((hop, now_ms)) + }; + } + + pub(crate) fn authenticated_return_hop(&self) -> Option<(NodeAddr, u64)> { + self.authenticated_return_hop + } + /// Create a new session entry. pub(crate) fn new( remote_addr: NodeAddr, @@ -382,6 +397,7 @@ impl SessionEntry { initiator_paused: false, created_at: now_ms, last_activity: now_ms, + authenticated_return_hop: None, session_start_ms: 0, coords_warmup_remaining: 0, is_initiator, diff --git i/src/node/tests/forwarding.rs w/src/node/tests/forwarding.rs index 9f9f97d..c2eb01d 100644 --- i/src/node/tests/forwarding.rs +++ w/src/node/tests/forwarding.rs @@ -2442,3 +2442,211 @@ async fn a_foreign_root_write_is_not_counted_as_a_key_mismatch() { "a write refused for a foreign root was also counted as a key mismatch" ); } + +// Local Archipelago diagnostic only: not a change to the pinned dependency. +// The server/phone LAN edge goes silent while both retain their anchor edge. +#[tokio::test] +async fn archy_lost_lan_peer_recovers_via_authenticated_return_path() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .with_test_writer() + .try_init(); + let acl_dir = tempfile::tempdir().unwrap(); + std::fs::write(acl_dir.path().join("allow"), "").unwrap(); + std::fs::write(acl_dir.path().join("deny"), "").unwrap(); + let mut nodes = vec![ + spanning_tree::make_test_node().await, + spanning_tree::make_test_node().await, + spanning_tree::make_test_node().await, + ]; + for node in &mut nodes { + node.node.peer_acl = crate::node::acl::PeerAclReloader::with_paths( + acl_dir.path().join("allow"), + acl_dir.path().join("deny"), + ); + } + for (a, b) in [(0, 1), (0, 2), (1, 2)] { + spanning_tree::initiate_handshake(&mut nodes, a, b).await; + } + spanning_tree::drain_all_packets(&mut nodes, true).await; + nodes.sort_by_key(|n| *n.node.node_addr()); + const ANCHOR: usize = 0; + const SERVER: usize = 1; + const PHONE: usize = 2; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + let server = *nodes[SERVER].node.node_addr(); + let phone = *nodes[PHONE].node.node_addr(); + let anchor = *nodes[ANCHOR].node.node_addr(); + let server_key = nodes[SERVER].node.identity().pubkey_full(); + let server_transport = nodes[SERVER].addr.clone(); + let phone_transport = nodes[PHONE].addr.clone(); + + // Android can immediately discard its vanished local edge; the remote + // still considers that edge alive until its liveness check removes it. + nodes[PHONE].node.remove_active_peer(&server); + populate_all_coord_caches(&mut nodes); + assert_eq!( + *nodes[PHONE] + .node + .find_next_hop(&server) + .unwrap() + .node_addr(), + anchor + ); + assert_eq!( + *nodes[SERVER] + .node + .find_next_hop(&phone) + .unwrap() + .node_addr(), + phone + ); + nodes[PHONE] + .node + .initiate_session(server, server_key) + .await + .unwrap(); + + async fn deliver_except_lost_lan( + nodes: &mut [TestNode], + server_transport: &TransportAddr, + phone_transport: &TransportAddr, + ) -> usize { + let mut discarded = 0; + for _ in 0..24 { + let queued: Vec<_> = nodes.iter().map(|n| n.packet_rx.len()).collect(); + if queued.iter().all(|n| *n == 0) { + break; + } + for (index, count) in queued.into_iter().enumerate() { + for _ in 0..count { + let packet = nodes[index].packet_rx.try_recv().unwrap(); + if (index == PHONE && &packet.remote_addr == server_transport) + || (index == SERVER && &packet.remote_addr == phone_transport) + { + discarded += 1; + continue; + } + use crate::proto::fmp::wire::{ + CommonPrefix, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2, + }; + if let Some(prefix) = CommonPrefix::parse(&packet.data) { + match prefix.phase { + PHASE_MSG1 => nodes[index].node.handle_msg1(packet).await, + PHASE_MSG2 => nodes[index].node.handle_msg2(packet).await, + PHASE_ESTABLISHED => { + nodes[index].node.handle_encrypted_frame(packet).await + } + _ => {} + } + } + } + } + } + discarded + } + let lost = deliver_except_lost_lan(&mut nodes, &server_transport, &phone_transport).await; + assert_eq!(lost, 0, "handshake replies must use the arriving path"); + assert!( + nodes[PHONE] + .node + .get_session(&server) + .unwrap() + .is_established() + ); + assert!( + nodes[SERVER] + .node + .get_session(&phone) + .unwrap() + .is_established() + ); + assert!( + nodes[SERVER].node.get_peer(&phone).is_some(), + "do not tear down a link on an indirect session" + ); + // The global route still retains the old direct edge. Only authenticated + // session replies bypass it, including subsequent encrypted payloads. + assert_eq!( + *nodes[SERVER] + .node + .find_next_hop(&phone) + .unwrap() + .node_addr(), + phone + ); + nodes[SERVER] + .node + .send_session_msg( + &phone, + crate::proto::fsp::SessionMessageType::CoordsWarmup.to_byte(), + &[], + ) + .await + .unwrap(); + assert_eq!( + deliver_except_lost_lan(&mut nodes, &server_transport, &phone_transport).await, + 0 + ); + + assert_eq!(nodes[SERVER].node.session_next_hop(&phone), Some(anchor)); + // A new, well-formed but unauthenticated XK setup naming this session + // must not refresh the return hint or destroy the established session. + nodes[SERVER] + .node + .get_session_mut(&phone) + .unwrap() + .note_authenticated_ingress(anchor, 0); + let before = nodes[SERVER] + .node + .get_session(&phone) + .unwrap() + .authenticated_return_hop(); + let forger = Identity::generate(); + let mut hs = crate::noise::HandshakeState::new_xk_initiator(forger.keypair(), server_key); + hs.set_local_epoch([42; 8]); + let setup = SessionSetup::new( + nodes[PHONE].node.tree_state.my_coords().clone(), + nodes[SERVER].node.tree_state.my_coords().clone(), + ) + .with_handshake(hs.write_xk_message_1().unwrap()) + .encode(); + nodes[SERVER] + .node + .handle_session_payload(&phone, &anchor, &setup, 1280, false) + .await; + assert_eq!( + nodes[SERVER] + .node + .get_session(&phone) + .unwrap() + .authenticated_return_hop(), + before + ); + assert!( + nodes[SERVER] + .node + .get_session(&phone) + .unwrap() + .is_established() + ); + // Corrupted encrypted input arriving on the direct peer cannot clear it. + let forged = vec![0u8; 64]; + nodes[SERVER] + .node + .handle_session_payload(&phone, &phone, &forged, 1280, false) + .await; + assert_eq!( + nodes[SERVER] + .node + .get_session(&phone) + .unwrap() + .authenticated_return_hop(), + before + ); + // A dead return-hop cannot strand the session behind a stale hint. + nodes[SERVER].node.remove_active_peer(&anchor); + assert_eq!(nodes[SERVER].node.session_next_hop(&phone), Some(phone)); + cleanup_nodes(&mut nodes).await; +}