From 7ef023fcdd21cad34903c0d8f8b05facc5fb34df Mon Sep 17 00:00:00 2001 From: "sijie.sun" Date: Sun, 14 Jun 2026 02:20:34 +0800 Subject: [PATCH] test: cover shared tun dynamic proxy routes Cover shared tun duplicate proxy CIDR failover and runtime proxy CIDR add/remove through netns integration tests. Start IP proxy from config patch events so a node that did not have proxy CIDRs at startup can serve a later proxy network patch. Ignore NotFound for shared NIC remove-side ifcfg cleanup so member teardown does not poison shared owner state when the OS item is already gone. Add dispatcher coverage for TUN read failure invalidation and member close notification. --- easytier/src/instance/instance.rs | 89 ++++- easytier/src/instance/shared_virtual_nic.rs | 17 +- .../instance/shared_virtual_nic/dispatcher.rs | 39 +- easytier/src/tests/three_node.rs | 333 +++++++++++++++++- 4 files changed, 456 insertions(+), 22 deletions(-) diff --git a/easytier/src/instance/instance.rs b/easytier/src/instance/instance.rs index dacdd4ba..96f10bd1 100644 --- a/easytier/src/instance/instance.rs +++ b/easytier/src/instance/instance.rs @@ -4,7 +4,6 @@ use std::collections::HashSet; use std::net::{IpAddr, Ipv4Addr}; #[cfg(feature = "tun")] use std::sync::OnceLock; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Weak}; #[cfg(all(feature = "tun", not(mobile)))] use std::time::Duration; @@ -85,7 +84,14 @@ struct IpProxy { icmp_proxy: Arc, udp_proxy: Arc, global_ctx: ArcGlobalCtx, - started: Arc, + start_state: Arc>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum IpProxyStartState { + Idle, + Started, + Failed, } impl IpProxy { @@ -100,27 +106,44 @@ impl IpProxy { icmp_proxy, udp_proxy, global_ctx, - started: Arc::new(AtomicBool::new(false)), + start_state: Arc::new(Mutex::new(IpProxyStartState::Idle)), }) } - async fn start(&self) -> Result<(), Error> { - if (self.global_ctx.config.get_proxy_cidrs().is_empty() - || self.started.load(Ordering::Relaxed)) - && !self.global_ctx.enable_exit_node() - && !self.global_ctx.no_tun() - { - return Ok(()); - } - - // Actually, if this node is enabled as an exit node, - // we still can use the system stack to forward packets. + fn should_start(&self) -> bool { if self.global_ctx.proxy_forward_by_system() && !self.global_ctx.no_tun() { + return false; + } + + self.global_ctx.enable_exit_node() + || self.global_ctx.no_tun() + || !self.global_ctx.config.get_proxy_cidrs().is_empty() + } + + async fn start(&self) -> Result<(), Error> { + if !self.should_start() { return Ok(()); } - self.started.store(true, Ordering::Relaxed); - self.tcp_proxy.start(true).await?; + let mut start_state = self.start_state.lock().await; + match *start_state { + IpProxyStartState::Idle => {} + IpProxyStartState::Started => return Ok(()), + IpProxyStartState::Failed => { + return Err(anyhow::anyhow!("ip proxy start previously failed").into()); + } + } + + if let Err(err) = self.start_components().await { + *start_state = IpProxyStartState::Failed; + return Err(err); + } + + *start_state = IpProxyStartState::Started; + Ok(()) + } + + async fn start_components(&self) -> Result<(), Error> { if let Err(e) = self.icmp_proxy.start().await { tracing::error!("start icmp proxy failed: {:?}", e); if cfg!(not(any( @@ -135,9 +158,37 @@ impl IpProxy { return Err(e); } } + self.tcp_proxy.start(true).await?; self.udp_proxy.start().await?; Ok(()) } + + fn watch_config_patches(&self) -> AbortOnDropHandle<()> { + let ip_proxy = self.clone(); + let mut event_receiver = self.global_ctx.subscribe(); + AbortOnDropHandle::new(tokio::spawn(async move { + loop { + match event_receiver.recv().await { + Ok(GlobalCtxEvent::ConfigPatched(_)) => { + if let Err(err) = ip_proxy.start().await { + tracing::warn!(?err, "failed to start ip proxy after config patch"); + } + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + event_receiver = event_receiver.resubscribe(); + if let Err(err) = ip_proxy.start().await { + tracing::warn!( + ?err, + "failed to start ip proxy after missed config patch" + ); + } + } + } + } + })) + } } #[cfg(feature = "tun")] @@ -645,6 +696,7 @@ pub struct Instance { tcp_hole_puncher: Arc>, ip_proxy: Option, + ip_proxy_config_watcher: Option>, #[cfg(feature = "kcp")] kcp_proxy_src: Option, @@ -735,6 +787,7 @@ impl Instance { tcp_hole_puncher, ip_proxy: None, + ip_proxy_config_watcher: None, #[cfg(feature = "kcp")] kcp_proxy_src: None, #[cfg(feature = "kcp")] @@ -1165,6 +1218,10 @@ impl Instance { self.get_peer_manager(), )?); self.run_ip_proxy().await?; + self.ip_proxy_config_watcher = self + .ip_proxy + .as_ref() + .map(|proxy| proxy.watch_config_patches()); self.udp_hole_puncher.lock().await.run().await?; self.tcp_hole_puncher.lock().await.run().await?; diff --git a/easytier/src/instance/shared_virtual_nic.rs b/easytier/src/instance/shared_virtual_nic.rs index 518b08ab..b1766a5e 100644 --- a/easytier/src/instance/shared_virtual_nic.rs +++ b/easytier/src/instance/shared_virtual_nic.rs @@ -351,16 +351,18 @@ impl SharedVirtualNic { let nic = self.nic.lock().await; for route in &delta.ipv4_routes.removed { - nic.remove_route(route.address, route.prefix).await?; + ignore_removed_ifcfg_not_found(nic.remove_route(route.address, route.prefix).await)?; } for route in &delta.ipv6_routes.removed { - nic.remove_ipv6_route(route.address, route.prefix).await?; + ignore_removed_ifcfg_not_found( + nic.remove_ipv6_route(route.address, route.prefix).await, + )?; } for ip in &delta.ipv4_addresses.removed { - nic.remove_ip(Some(*ip)).await?; + ignore_removed_ifcfg_not_found(nic.remove_ip(Some(*ip)).await)?; } for ip in &delta.ipv6_addresses.removed { - nic.remove_ipv6(Some(*ip)).await?; + ignore_removed_ifcfg_not_found(nic.remove_ipv6(Some(*ip)).await)?; } for ip in &delta.ipv4_addresses.added { @@ -512,6 +514,13 @@ impl SharedVirtualNic { } } +fn ignore_removed_ifcfg_not_found(result: Result<(), Error>) -> Result<(), Error> { + match result { + Err(Error::NotFound) => Ok(()), + other => other, + } +} + struct SharedVirtualNicMemberRegistration { member_id: SharedVirtualNicMemberId, shared_nic: Arc>, diff --git a/easytier/src/instance/shared_virtual_nic/dispatcher.rs b/easytier/src/instance/shared_virtual_nic/dispatcher.rs index 83093e4b..93bee0f9 100644 --- a/easytier/src/instance/shared_virtual_nic/dispatcher.rs +++ b/easytier/src/instance/shared_virtual_nic/dispatcher.rs @@ -675,9 +675,10 @@ fn read_ipv6_addr(payload: &[u8], start: usize) -> [u8; 16] { #[cfg(test)] mod tests { - use std::net::Ipv6Addr; + use std::{net::Ipv6Addr, time::Duration}; use super::*; + use crate::tunnel::{TunnelError, common::TunnelWrapper, ring::create_ring_tunnel_pair}; fn ipv6_packet(src: Ipv6Addr, dst: Ipv6Addr) -> ZCPacket { let mut payload = vec![0; IPV6_HEADER_LEN]; @@ -778,4 +779,40 @@ mod tests { assert!(fallback_receiver.try_recv().is_err()); } + + #[tokio::test] + async fn dispatcher_invalidates_shared_nic_when_tun_read_fails() { + let member_id = uuid::Uuid::from_u128(1); + let (tun_tx, tun_rx) = mpsc::unbounded_channel(); + let tun_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(tun_rx); + let tun_sink = futures::sink::unfold((), |(), _packet: ZCPacket| async { + Ok::<(), TunnelError>(()) + }); + let tunnel = TunnelWrapper::new(tun_stream, tun_sink, None); + let member_tunnel_table = SharedVirtualNicMemberTunnelTable::default(); + let valid = Arc::new(AtomicBool::new(true)); + let dispatcher = SharedVirtualNicDispatcher::start( + Box::new(tunnel), + member_tunnel_table.clone(), + valid.clone(), + ); + let close_notifier = Arc::new(Notify::new()); + let (_member_tunnel, shared_tunnel) = create_ring_tunnel_pair(); + + member_tunnel_table + .register(member_id, shared_tunnel, close_notifier.clone()) + .unwrap(); + dispatcher + .update_sources(member_id, &BTreeSet::new(), &BTreeSet::new()) + .await + .unwrap(); + + tun_tx.send(Err(TunnelError::Shutdown)).unwrap(); + + tokio::time::timeout(Duration::from_secs(1), close_notifier.notified()) + .await + .unwrap(); + assert!(!valid.load(Ordering::Acquire)); + assert!(member_tunnel_table.dispatcher_channels().is_none()); + } } diff --git a/easytier/src/tests/three_node.rs b/easytier/src/tests/three_node.rs index 4a81efa3..8d160b11 100644 --- a/easytier/src/tests/three_node.rs +++ b/easytier/src/tests/three_node.rs @@ -322,6 +322,76 @@ async fn wait_tun_ready( .expect("timed out waiting for tun ready") } +#[cfg(feature = "tun")] +fn proxy_route_exists( + routes: &[crate::proto::api::instance::Route], + peer_id: PeerId, + proxy_cidr: &str, +) -> bool { + let proxy_cidr = proxy_cidr.to_owned(); + routes + .iter() + .any(|route| route.peer_id == peer_id && route.proxy_cidrs.contains(&proxy_cidr)) +} + +#[cfg(feature = "tun")] +async fn wait_proxy_route_to_peer( + mgr: &std::sync::Arc, + ipv4: &str, + dst_peer_id: PeerId, + proxy_cidr: &str, +) { + let proxy_cidr = proxy_cidr.to_owned(); + wait_for_condition( + || async { + let routes = mgr.list_routes().await; + let route_found = routes.iter().any(|route| { + route.peer_id == dst_peer_id + && route.ipv4_addr == Some(ipv4.parse().unwrap()) + && route.proxy_cidrs.contains(&proxy_cidr) + }); + route_found + }, + Duration::from_secs(8), + ) + .await; +} + +#[cfg(feature = "tun")] +async fn wait_proxy_route_absent( + mgr: &std::sync::Arc, + dst_peer_id: PeerId, + proxy_cidr: &str, +) { + wait_for_condition( + || async { + let routes = mgr.list_routes().await; + !proxy_route_exists(&routes, dst_peer_id, proxy_cidr) + }, + Duration::from_secs(8), + ) + .await; +} + +#[cfg(feature = "tun")] +async fn patch_proxy_cidr( + inst: &Instance, + action: crate::proto::api::config::ConfigPatchAction, + cidr: &str, +) { + inst.get_config_patcher() + .apply_patch(crate::proto::api::config::InstanceConfigPatch { + proxy_networks: vec![crate::proto::api::config::ProxyNetworkPatch { + action: action as i32, + cidr: Some(cidr.parse().unwrap()), + mapped_cidr: None, + }], + ..Default::default() + }) + .await + .unwrap(); +} + #[cfg(feature = "tun")] struct SharedTunProxyCidrTopology { insts: Vec, @@ -826,6 +896,268 @@ pub async fn shared_tun_proxy_cidr_reaches_member_network() { drop_insts(vec![center, shared_1, shared_2, remote]).await; } +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn shared_tun_duplicate_proxy_cidr_survives_member_departure() { + prepare_linux_namespaces(); + + let source_dev = shared_tun_test_dev_name(); + let mut destination_dev = shared_tun_test_dev_name(); + while destination_dev == source_dev { + destination_dev = shared_tun_test_dev_name(); + } + let network_name = "shared_duplicate_proxy_network"; + let network_secret = "shared_duplicate_proxy_secret"; + let proxy_cidr = "10.1.2.0/24"; + let target_ip = "10.1.2.4"; + + let mut source = Instance::new(shared_tun_test_config( + "shared_duplicate_proxy_source", + network_name, + network_secret, + Some("net_a"), + Some(&source_dev), + "10.144.245.1/24", + false, + )); + + let primary_cfg = shared_tun_test_config( + "shared_duplicate_proxy_primary", + network_name, + network_secret, + Some("net_c"), + Some(&destination_dev), + "10.144.245.2/24", + false, + ); + primary_cfg + .add_proxy_cidr(proxy_cidr.parse().unwrap(), None) + .unwrap(); + let mut primary = Instance::new(primary_cfg); + + let backup_cfg = shared_tun_test_config( + "shared_duplicate_proxy_backup", + network_name, + network_secret, + Some("net_c"), + Some(&destination_dev), + "10.144.245.3/24", + false, + ); + backup_cfg + .add_proxy_cidr(proxy_cidr.parse().unwrap(), None) + .unwrap(); + let mut backup = Instance::new(backup_cfg); + + let mut source_events = source.get_global_ctx().subscribe(); + let mut primary_events = primary.get_global_ctx().subscribe(); + let mut backup_events = backup.get_global_ctx().subscribe(); + + source.run().await.unwrap(); + primary.run().await.unwrap(); + backup.run().await.unwrap(); + + assert_eq!(wait_tun_ready(&mut source_events).await, source_dev); + assert_eq!(wait_tun_ready(&mut primary_events).await, destination_dev); + assert_eq!(wait_tun_ready(&mut backup_events).await, destination_dev); + + primary + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", source.id()).parse().unwrap(), + )); + backup + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", source.id()).parse().unwrap(), + )); + + wait_for_condition( + || async { + source.get_peer_manager().list_routes().await.len() == 2 + && primary.get_peer_manager().list_routes().await.len() == 2 + && backup.get_peer_manager().list_routes().await.len() == 2 + }, + Duration::from_secs(8), + ) + .await; + + wait_proxy_route_to_peer( + &source.get_peer_manager(), + "10.144.245.2/24", + primary.peer_id(), + proxy_cidr, + ) + .await; + wait_proxy_route_to_peer( + &source.get_peer_manager(), + "10.144.245.3/24", + backup.peer_id(), + proxy_cidr, + ) + .await; + wait_for_condition( + || async { ping_test("net_a", target_ip, None).await }, + Duration::from_secs(8), + ) + .await; + + let primary_peer_id = primary.peer_id(); + primary.clear_resources().await; + drop(primary); + + wait_proxy_route_absent(&source.get_peer_manager(), primary_peer_id, proxy_cidr).await; + wait_proxy_route_to_peer( + &source.get_peer_manager(), + "10.144.245.3/24", + backup.peer_id(), + proxy_cidr, + ) + .await; + wait_for_condition( + || async { + ipv4_route_exists_in_ns("net_a", &format!("{proxy_cidr} dev {source_dev}")) + && ping_test("net_a", target_ip, None).await + }, + Duration::from_secs(8), + ) + .await; + + drop_insts(vec![source, backup]).await; +} + +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn shared_tun_proxy_cidr_runtime_patch_adds_and_removes_route() { + prepare_linux_namespaces(); + + let source_dev = shared_tun_test_dev_name(); + let mut destination_dev = shared_tun_test_dev_name(); + while destination_dev == source_dev { + destination_dev = shared_tun_test_dev_name(); + } + let network_name = "shared_dynamic_proxy_network"; + let network_secret = "shared_dynamic_proxy_secret"; + let proxy_cidr = "10.1.2.0/24"; + let target_ip = "10.1.2.4"; + + let mut source = Instance::new(shared_tun_test_config( + "shared_dynamic_proxy_source", + network_name, + network_secret, + Some("net_a"), + Some(&source_dev), + "10.144.244.1/24", + false, + )); + let mut destination = Instance::new(shared_tun_test_config( + "shared_dynamic_proxy_destination", + network_name, + network_secret, + Some("net_c"), + Some(&destination_dev), + "10.144.244.2/24", + false, + )); + + let mut source_events = source.get_global_ctx().subscribe(); + let mut destination_events = destination.get_global_ctx().subscribe(); + + source.run().await.unwrap(); + destination.run().await.unwrap(); + + assert_eq!(wait_tun_ready(&mut source_events).await, source_dev); + assert_eq!( + wait_tun_ready(&mut destination_events).await, + destination_dev + ); + + destination + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", source.id()).parse().unwrap(), + )); + + wait_for_condition( + || async { + source.get_peer_manager().list_routes().await.len() == 1 + && destination.get_peer_manager().list_routes().await.len() == 1 + }, + Duration::from_secs(8), + ) + .await; + assert!( + !proxy_route_exists( + &source.get_peer_manager().list_routes().await, + destination.peer_id(), + proxy_cidr, + ), + "proxy route should not exist before runtime patch" + ); + + patch_proxy_cidr( + &destination, + crate::proto::api::config::ConfigPatchAction::Add, + proxy_cidr, + ) + .await; + wait_proxy_route_to_peer( + &source.get_peer_manager(), + "10.144.244.2/24", + destination.peer_id(), + proxy_cidr, + ) + .await; + wait_for_condition( + || async { + source + .get_peer_manager() + .list_proxy_cidrs() + .await + .contains(&proxy_cidr.parse().unwrap()) + }, + Duration::from_secs(10), + ) + .await; + wait_for_condition( + || async { ipv4_route_exists_in_ns("net_a", &format!("{proxy_cidr} dev {source_dev}")) }, + Duration::from_secs(10), + ) + .await; + wait_for_condition( + || async { ping_test("net_a", target_ip, None).await }, + Duration::from_secs(10), + ) + .await; + + patch_proxy_cidr( + &destination, + crate::proto::api::config::ConfigPatchAction::Remove, + proxy_cidr, + ) + .await; + wait_proxy_route_absent( + &source.get_peer_manager(), + destination.peer_id(), + proxy_cidr, + ) + .await; + wait_for_condition( + || async { !ipv4_route_exists_in_ns("net_a", &format!("{proxy_cidr} dev {source_dev}")) }, + Duration::from_secs(10), + ) + .await; + wait_for_condition( + || async { !ping_test("net_a", target_ip, None).await }, + Duration::from_secs(8), + ) + .await; + + drop_insts(vec![source, destination]).await; +} + #[cfg(feature = "tun")] #[tokio::test] #[serial_test::serial] @@ -1542,7 +1874,6 @@ fn route_exists_in_ns(ns: &str, needle: &str) -> bool { .any(|line| line.contains(needle)) } -#[cfg(all(feature = "tun", feature = "magic-dns"))] fn ipv4_route_exists_in_ns(ns: &str, needle: &str) -> bool { run_ip_in_ns_output(ns, &["route", "show"]) .lines()