From fe18f32c8b3862598d96b7e549e09ecfdcc934ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BC=9A=E5=81=9A=E9=A5=AD=E7=9A=84=E4=BA=8C=E5=93=88?= Date: Sun, 25 Oct 2026 13:13:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(peer=5Fmanager):=20=E6=8C=89=20NIC=20?= =?UTF-8?q?=E8=BF=87=E6=BB=A4=E5=99=A8=E6=94=B9=E5=86=99=E5=90=8E=E7=9A=84?= =?UTF-8?q?=E7=9B=AE=E7=9A=84=E5=9C=B0=E5=9D=80=E8=B7=AF=E7=94=B1=20(#2740?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(peer_manager): route by destination rewritten in the nic packet pipeline NIC packet filters receive &mut ZCPacket and may rewrite the IP header (e.g. a NAT filter that maps a virtual peer address to an address on the local subnet). send_msg_by_ip parsed the destination before running the filter pipeline and then routed by that stale address, so rewritten packets were looked up by the pre-filter destination and dropped as unreachable. Re-parse the destination after run_nic_packet_process_pipeline and route by the rewritten address. Expose instance::packet_io as pub(crate) so peers can reuse the parser. * test(peer-manager): cover NIC destination rewrite routing Exercise IPv4 and IPv6 delivery through three peers connected by memory tunnels. A NIC filter rewrites the destination without setting a peer ID; verify delivery to the rewritten peer and no delivery to the original. Cover original destinations with and without an existing route. Wait for route convergence before sending each packet once, without data retries. Both regression tests fail with the PR routing fix removed and pass with it restored. All 37 peer-manager tests and formatting checks pass. Scoped Clippy completes with warnings in pre-existing code. --------- Co-authored-by: ldoubil Co-authored-by: KKRainbow <5665404+KKRainbow@users.noreply.github.com> --- easytier-core/src/instance/mod.rs | 2 +- easytier-core/src/peers/peer_manager.rs | 175 +++++++++++++++++++++++- 2 files changed, 175 insertions(+), 2 deletions(-) diff --git a/easytier-core/src/instance/mod.rs b/easytier-core/src/instance/mod.rs index 5367c138..f080ef44 100644 --- a/easytier-core/src/instance/mod.rs +++ b/easytier-core/src/instance/mod.rs @@ -10,7 +10,7 @@ mod management; mod management_extension; mod management_state; pub mod manager; -mod packet_io; +pub(crate) mod packet_io; mod packet_plane; #[cfg(feature = "proxy-packet")] mod packet_proxy_extension; diff --git a/easytier-core/src/peers/peer_manager.rs b/easytier-core/src/peers/peer_manager.rs index f2804b35..eee26382 100644 --- a/easytier-core/src/peers/peer_manager.rs +++ b/easytier-core/src/peers/peer_manager.rs @@ -2805,7 +2805,7 @@ impl PeerOutboundPacketRouter { pub async fn send_msg_by_ip( &self, mut msg: ZCPacket, - ip_addr: IpAddr, + mut ip_addr: IpAddr, not_send_to_self: bool, ) -> Result<(), Error> { tracing::trace!( @@ -2815,9 +2815,23 @@ impl PeerOutboundPacketRouter { ); msg.fill_peer_manager_hdr(self.my_peer_id, 0, PacketType::Data as u8); + let filters_registered = !self.nic_packet_process_pipeline.load().is_empty(); if !self.run_nic_packet_process_pipeline(&mut msg).await { return Ok(()); } + + if filters_registered + && let Ok(meta) = crate::instance::packet_io::parse_ip_packet(msg.payload()) + && meta.destination != ip_addr + { + tracing::trace!( + original_dst = %ip_addr, + new_dst = %meta.destination, + "destination rewritten by nic packet filter" + ); + ip_addr = meta.destination; + } + let packet_policy = self.context.packet_policy(); let cur_to_peer_id = msg.peer_manager_header().unwrap().to_peer_id.into(); if cur_to_peer_id != 0 { @@ -3912,6 +3926,165 @@ mod tests { assert!(!is_self); } + fn set_packet_destination(packet: &mut [u8], destination: IpAddr) { + match destination { + IpAddr::V4(destination) => { + assert_eq!(packet[0] >> 4, 4); + packet[16..20].copy_from_slice(&destination.octets()); + packet[10..12].fill(0); + let mut checksum: u32 = packet[..20] + .chunks_exact(2) + .map(|word| u32::from(u16::from_be_bytes([word[0], word[1]]))) + .sum(); + while checksum >> 16 != 0 { + checksum = (checksum & 0xffff) + (checksum >> 16); + } + packet[10..12].copy_from_slice(&(!(checksum as u16)).to_be_bytes()); + } + IpAddr::V6(destination) => { + assert_eq!(packet[0] >> 4, 6); + packet[24..40].copy_from_slice(&destination.octets()); + } + } + } + + struct DestinationRewriteNicFilter(IpAddr); + + #[async_trait::async_trait] + impl super::super::NicPacketFilter for DestinationRewriteNicFilter { + async fn try_process_packet_from_nic(&self, packet: &mut ZCPacket) -> bool { + // Leave to_peer_id unset: the router must use the rewritten IP header. + assert_eq!(packet.peer_manager_header().unwrap().to_peer_id.get(), 0); + set_packet_destination(packet.mut_payload(), self.0); + true + } + } + + async fn check_nic_filter_destination_routing(addresses: [IpAddr; 4]) { + let [source, original, rewritten, unrouted] = addresses; + let mut peers = Vec::new(); + let mut receivers = Vec::new(); + for address in [source, original, rewritten] { + let mut runtime = portable_runtime_config("nic-filter-routing"); + if address.is_ipv4() { + runtime.core.routes.ipv4 = Some(IpPrefix::new(address, 24).unwrap()); + } else { + runtime.core.routes.ipv6 = Some(IpPrefix::new(address, 64).unwrap()); + } + let (sender, receiver) = host_packet_channel(); + let peer = PeerManagerCore::new_portable_for_test( + PortablePeerManagerConfig::new(runtime), + sender, + ) + .unwrap(); + peer.run().await.unwrap(); + peers.push(peer); + receivers.push(receiver); + } + let sender = &peers[0]; + for peer in &peers[1..] { + let (client, server) = crate::tunnel::ring::create_ring_tunnel_pair(); + let (client_result, server_result) = tokio::join!( + sender.add_client_tunnel(client), + peer.add_tunnel_as_server(server), + ); + client_result.unwrap(); + server_result.unwrap(); + } + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if sender.get_msg_dst_peer(&original).await == (vec![peers[1].my_peer_id()], false) + && sender.get_msg_dst_peer(&rewritten).await + == (vec![peers[2].my_peer_id()], false) + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("both destination routes must be ready before sending"); + assert_eq!( + sender.get_msg_dst_peer(&unrouted).await, + (Vec::new(), false) + ); + sender + .add_nic_packet_process_pipeline(Box::new(DestinationRewriteNicFilter(rewritten))) + .await; + + // Exercise both misdelivery to an existing peer and silent loss for an + // address with no route. Send once per case, without retrying data packets. + for destination in [original, unrouted] { + let mut payload = match source { + IpAddr::V4(source) => { + let mut payload = vec![0u8; 20]; + payload[0] = 0x45; + payload[2..4].copy_from_slice(&20u16.to_be_bytes()); + payload[8] = 64; + payload[9] = 253; // Experimental protocol; no transport header. + payload[12..16].copy_from_slice(&source.octets()); + payload + } + IpAddr::V6(source) => { + let mut payload = vec![0u8; 40]; + payload[0] = 0x60; + payload[6] = 59; // No next header. + payload[7] = 64; + payload[8..24].copy_from_slice(&source.octets()); + payload + } + }; + set_packet_destination(&mut payload, destination); + sender + .send_msg_by_ip(ZCPacket::new_with_payload(&payload), destination, false) + .await + .unwrap(); + + let packet = tokio::time::timeout(Duration::from_secs(5), receivers[2].recv()) + .await + .expect("rewritten destination peer did not receive the packet") + .unwrap() + .into_core_packet(); + set_packet_destination(&mut payload, rewritten); + assert_eq!(packet.payload(), payload); + let header = packet.peer_manager_header().unwrap(); + assert_eq!(header.from_peer_id.get(), sender.my_peer_id()); + assert_eq!(header.to_peer_id.get(), peers[2].my_peer_id()); + assert!( + tokio::time::timeout(Duration::from_millis(100), receivers[1].recv()) + .await + .is_err(), + "packet reached the old destination" + ); + } + + for peer in &peers { + peer.clear_resources().await; + } + } + + #[tokio::test] + async fn nic_filter_rewritten_ipv4_destination_controls_routing() { + check_nic_filter_destination_routing([ + "10.82.0.1".parse().unwrap(), + "10.82.0.2".parse().unwrap(), + "10.82.0.3".parse().unwrap(), + "10.82.0.99".parse().unwrap(), + ]) + .await; + } + + #[tokio::test] + async fn nic_filter_rewritten_ipv6_destination_controls_routing() { + check_nic_filter_destination_routing([ + "fd00::1".parse().unwrap(), + "fd00::2".parse().unwrap(), + "fd00::3".parse().unwrap(), + "fd00::99".parse().unwrap(), + ]) + .await; + } + #[tokio::test] async fn foreign_network_stop_waits_for_inflight_admission() { let core = build_portable_for_test(portable_runtime_config("portable-net")).unwrap();