diff --git a/easytier/src/instance/shared_virtual_nic.rs b/easytier/src/instance/shared_virtual_nic.rs index 6f5a4213..518b08ab 100644 --- a/easytier/src/instance/shared_virtual_nic.rs +++ b/easytier/src/instance/shared_virtual_nic.rs @@ -310,8 +310,21 @@ impl SharedVirtualNic { self.ensure_valid()?; let mut next_ifcfg = self.ifcfg.clone(); + let old_claims = self.ifcfg.claims_of(member_id); + let next_claims = claims.clone(); let delta = next_ifcfg.apply_member_claims(member_id, claims); - self.apply_ifcfg_delta(&delta).await?; + self.sync_dispatcher_sources_for_ifcfg_update(member_id, &old_claims, &next_claims) + .await?; + + if let Err(err) = self.apply_ifcfg_delta(&delta).await { + let _ = self + .sync_dispatcher_sources_for_member(member_id, &old_claims) + .await; + return Err(err); + } + + self.sync_dispatcher_sources_for_member(member_id, &next_claims) + .await?; self.ifcfg = next_ifcfg; Ok(()) @@ -328,6 +341,7 @@ impl SharedVirtualNic { return Ok(()); }; self.apply_ifcfg_delta(&delta).await?; + self.remove_dispatcher_sources_for_member(member_id).await?; self.ifcfg = next_ifcfg; Ok(()) @@ -397,11 +411,13 @@ impl SharedVirtualNic { } let tunnel = self.nic.lock().await.create_dev().await?; - self.dispatcher = Some(SharedVirtualNicDispatcher::start( + let dispatcher = SharedVirtualNicDispatcher::start( tunnel, self.member_tunnel_table.clone(), self.valid.clone(), - )); + ); + self.sync_dispatcher_sources(&dispatcher).await?; + self.dispatcher = Some(dispatcher); Ok(()) } @@ -417,11 +433,81 @@ impl SharedVirtualNic { } let tunnel = self.nic.lock().await.create_dev_for_mobile(tun_fd).await?; - self.dispatcher = Some(SharedVirtualNicDispatcher::start( + let dispatcher = SharedVirtualNicDispatcher::start( tunnel, self.member_tunnel_table.clone(), self.valid.clone(), - )); + ); + self.sync_dispatcher_sources(&dispatcher).await?; + self.dispatcher = Some(dispatcher); + Ok(()) + } + + async fn sync_dispatcher_sources( + &self, + dispatcher: &SharedVirtualNicDispatcher, + ) -> Result<(), Error> { + for (member_id, claims) in &self.ifcfg.member_claims { + dispatcher + .update_sources(*member_id, &claims.ipv4_addresses, &claims.ipv6_addresses) + .await?; + } + Ok(()) + } + + async fn sync_dispatcher_sources_for_ifcfg_update( + &self, + member_id: SharedVirtualNicMemberId, + old_claims: &SharedIfConfigClaims, + next_claims: &SharedIfConfigClaims, + ) -> Result<(), Error> { + let mut active_ipv4_addresses = old_claims.ipv4_addresses.clone(); + active_ipv4_addresses.extend(next_claims.ipv4_addresses.iter().copied()); + let mut active_ipv6_addresses = old_claims.ipv6_addresses.clone(); + active_ipv6_addresses.extend(next_claims.ipv6_addresses.iter().copied()); + + self.sync_dispatcher_sources_for_addresses( + member_id, + &active_ipv4_addresses, + &active_ipv6_addresses, + ) + .await + } + + async fn sync_dispatcher_sources_for_member( + &self, + member_id: SharedVirtualNicMemberId, + claims: &SharedIfConfigClaims, + ) -> Result<(), Error> { + self.sync_dispatcher_sources_for_addresses( + member_id, + &claims.ipv4_addresses, + &claims.ipv6_addresses, + ) + .await + } + + async fn sync_dispatcher_sources_for_addresses( + &self, + member_id: SharedVirtualNicMemberId, + ipv4_addresses: &BTreeSet, + ipv6_addresses: &BTreeSet, + ) -> Result<(), Error> { + if let Some(dispatcher) = &self.dispatcher { + dispatcher + .update_sources(member_id, ipv4_addresses, ipv6_addresses) + .await?; + } + Ok(()) + } + + async fn remove_dispatcher_sources_for_member( + &self, + member_id: SharedVirtualNicMemberId, + ) -> Result<(), Error> { + if let Some(dispatcher) = &self.dispatcher { + dispatcher.remove_sources(member_id).await?; + } Ok(()) } } diff --git a/easytier/src/instance/shared_virtual_nic/dispatcher.rs b/easytier/src/instance/shared_virtual_nic/dispatcher.rs index ba8484b2..83093e4b 100644 --- a/easytier/src/instance/shared_virtual_nic/dispatcher.rs +++ b/easytier/src/instance/shared_virtual_nic/dispatcher.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeMap, HashMap, VecDeque}, + collections::{BTreeMap, BTreeSet, HashMap, VecDeque}, pin::Pin, sync::{ Arc, Mutex as StdMutex, @@ -7,6 +7,7 @@ use std::{ }, }; +use cidr::{Ipv4Inet, Ipv6Inet}; use futures::{SinkExt, StreamExt}; use tokio::sync::{Notify, mpsc, oneshot}; use tokio_util::task::AbortOnDropHandle; @@ -40,6 +41,11 @@ enum SharedVirtualNicControl { Unregister { member_id: SharedVirtualNicMemberId, }, + UpdateSources { + member_id: SharedVirtualNicMemberId, + sources: BTreeSet, + ack: oneshot::Sender<()>, + }, } #[derive(Clone, Default)] @@ -168,7 +174,7 @@ impl SharedVirtualNicMemberTunnelTable { } } -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] enum SharedVirtualNicFlowAddr { V4(u32), V6([u8; 16]), @@ -302,6 +308,7 @@ impl SharedVirtualNicFlowTable { pub(super) struct SharedVirtualNicDispatcher { _task: AbortOnDropHandle<()>, + control_sender: mpsc::UnboundedSender, } impl SharedVirtualNicDispatcher { @@ -313,7 +320,7 @@ impl SharedVirtualNicDispatcher { let (tun_stream, tun_sink) = tunnel.split(); let (to_tun_sender, to_tun_receiver) = mpsc::channel(MEMBER_TUNNEL_BUFFER_SIZE); let (control_sender, control_receiver) = mpsc::unbounded_channel(); - member_tunnel_table.attach_dispatcher(to_tun_sender, control_sender); + member_tunnel_table.attach_dispatcher(to_tun_sender, control_sender.clone()); let task = SharedVirtualNicDispatcherTask { tun_stream, @@ -327,8 +334,43 @@ impl SharedVirtualNicDispatcher { Self { _task: AbortOnDropHandle::new(tokio::spawn(task.run())), + control_sender, } } + + pub(super) async fn update_sources( + &self, + member_id: SharedVirtualNicMemberId, + ipv4_addresses: &BTreeSet, + ipv6_addresses: &BTreeSet, + ) -> Result<(), Error> { + let (ack, rx) = oneshot::channel(); + self.control_sender + .send(SharedVirtualNicControl::UpdateSources { + member_id, + sources: sources_from_addresses(ipv4_addresses, ipv6_addresses), + ack, + }) + .map_err(|_| anyhow::anyhow!("shared virtual nic dispatcher is not running"))?; + rx.await + .map_err(|_| anyhow::anyhow!("shared virtual nic dispatcher is not running").into()) + } + + pub(super) async fn remove_sources( + &self, + member_id: SharedVirtualNicMemberId, + ) -> Result<(), Error> { + let (ack, rx) = oneshot::channel(); + self.control_sender + .send(SharedVirtualNicControl::UpdateSources { + member_id, + sources: BTreeSet::new(), + ack, + }) + .map_err(|_| anyhow::anyhow!("shared virtual nic dispatcher is not running"))?; + rx.await + .map_err(|_| anyhow::anyhow!("shared virtual nic dispatcher is not running").into()) + } } struct SharedVirtualNicDispatcherTask { @@ -398,6 +440,7 @@ impl SharedVirtualNicDispatcherTask { struct SharedVirtualNicDispatcherState { members: BTreeMap, flow_table: SharedVirtualNicFlowTable, + source_table: SharedVirtualNicSourceTable, } impl SharedVirtualNicDispatcherState { @@ -409,6 +452,14 @@ impl SharedVirtualNicDispatcherState { SharedVirtualNicControl::Unregister { member_id } => { self.unregister(member_id); } + SharedVirtualNicControl::UpdateSources { + member_id, + sources, + ack, + } => { + self.source_table.update_member_sources(member_id, sources); + let _ = ack.send(()); + } } } @@ -430,6 +481,7 @@ impl SharedVirtualNicDispatcherState { fn close_all(&mut self) { let members = std::mem::take(&mut self.members); self.flow_table.clear(); + self.source_table.clear(); for entry in members.into_values() { entry.close_notifier.notify_one(); @@ -441,19 +493,15 @@ impl SharedVirtualNicDispatcherState { } async fn forward_tun_packet_to_member(&mut self, packet: ZCPacket) { - let member_id = self.flow_table.owner_of(&packet); - if !self.send_packet(member_id, packet).await { + if !self.send_packet(packet).await { tracing::trace!("shared virtual nic dropped packet without active member"); } } - async fn send_packet( - &mut self, - preferred_member_id: Option, - packet: ZCPacket, - ) -> bool { + async fn send_packet(&mut self, packet: ZCPacket) -> bool { let mut packet = packet; - if let Some(member_id) = preferred_member_id { + + if let Some(member_id) = self.flow_table.owner_of(&packet) { match self.send_packet_to_member(member_id, packet).await { Ok(()) => return true, Err(packet_on_failure) => { @@ -462,6 +510,14 @@ impl SharedVirtualNicDispatcherState { } } + match self.source_table.owner_of_source(&packet, &self.members) { + SourceOwner::Active(member_id) => { + return self.send_packet_to_member(member_id, packet).await.is_ok(); + } + SourceOwner::Inactive => return false, + SourceOwner::None => {} + } + let Some(member_id) = self.members.keys().next().copied() else { return false; }; @@ -492,6 +548,108 @@ impl SharedVirtualNicDispatcherState { } } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SourceOwner { + Active(SharedVirtualNicMemberId), + Inactive, + None, +} + +#[derive(Default)] +struct SharedVirtualNicSourceTable { + member_sources: BTreeMap>, + source_owners: BTreeMap>, +} + +impl SharedVirtualNicSourceTable { + fn update_member_sources( + &mut self, + member_id: SharedVirtualNicMemberId, + sources: BTreeSet, + ) { + let old_sources = self.member_sources.remove(&member_id).unwrap_or_default(); + + for source in old_sources.difference(&sources) { + self.remove_source_owner(*source, member_id); + } + for source in sources.difference(&old_sources) { + self.source_owners + .entry(*source) + .or_default() + .insert(member_id); + } + + if !sources.is_empty() { + self.member_sources.insert(member_id, sources); + } + } + + fn remove_owner(&mut self, member_id: SharedVirtualNicMemberId) { + let Some(sources) = self.member_sources.remove(&member_id) else { + return; + }; + + for source in sources { + self.remove_source_owner(source, member_id); + } + } + + fn clear(&mut self) { + self.member_sources.clear(); + self.source_owners.clear(); + } + + fn owner_of_source( + &self, + packet: &ZCPacket, + active_members: &BTreeMap, + ) -> SourceOwner { + let Some(source) = SharedVirtualNicFlowKey::from_packet(packet).map(|key| key.src) else { + return SourceOwner::None; + }; + let Some(owners) = self.source_owners.get(&source) else { + return SourceOwner::None; + }; + + owners + .iter() + .find(|member_id| active_members.contains_key(member_id)) + .copied() + .map(SourceOwner::Active) + .unwrap_or(SourceOwner::Inactive) + } + + fn remove_source_owner( + &mut self, + source: SharedVirtualNicFlowAddr, + member_id: SharedVirtualNicMemberId, + ) { + let Some(owners) = self.source_owners.get_mut(&source) else { + return; + }; + + owners.remove(&member_id); + if owners.is_empty() { + self.source_owners.remove(&source); + } + } +} + +fn sources_from_addresses( + ipv4_addresses: &BTreeSet, + ipv6_addresses: &BTreeSet, +) -> BTreeSet { + ipv4_addresses + .iter() + .map(|addr| SharedVirtualNicFlowAddr::V4(u32::from_be_bytes(addr.address().octets()))) + .chain( + ipv6_addresses + .iter() + .map(|addr| SharedVirtualNicFlowAddr::V6(addr.address().octets())), + ) + .collect() +} + fn transport_ports(protocol: u8, payload: &[u8]) -> Option { let min_len = match protocol { TCP_PROTOCOL => TCP_HEADER_MIN_LEN, @@ -514,3 +672,110 @@ fn read_ipv6_addr(payload: &[u8], start: usize) -> [u8; 16] { addr.copy_from_slice(&payload[start..start + 16]); addr } + +#[cfg(test)] +mod tests { + use std::net::Ipv6Addr; + + use super::*; + + fn ipv6_packet(src: Ipv6Addr, dst: Ipv6Addr) -> ZCPacket { + let mut payload = vec![0; IPV6_HEADER_LEN]; + payload[0] = 0x60; + payload[6] = 58; + payload[8..24].copy_from_slice(&src.octets()); + payload[24..40].copy_from_slice(&dst.octets()); + ZCPacket::new_with_payload(&payload) + } + + fn member_entry(sender: mpsc::Sender) -> SharedVirtualNicMemberTunnelEntry { + SharedVirtualNicMemberTunnelEntry { + sender, + close_notifier: Arc::new(Notify::new()), + _tasks: Vec::new(), + } + } + + #[test] + fn source_table_selects_ipv6_source_owner() { + let first = uuid::Uuid::from_u128(1); + let second = uuid::Uuid::from_u128(2); + let first_addr = "2001:db8::1".parse::().unwrap(); + let second_addr = "2001:db8::2".parse::().unwrap(); + let dst = "2001:db8:ffff::1".parse::().unwrap(); + let mut table = SharedVirtualNicSourceTable::default(); + let (first_sender, _first_receiver) = mpsc::channel(1); + let (second_sender, _second_receiver) = mpsc::channel(1); + let mut members = BTreeMap::new(); + members.insert(first, member_entry(first_sender)); + members.insert(second, member_entry(second_sender)); + + table.update_member_sources( + first, + BTreeSet::from([SharedVirtualNicFlowAddr::V6(first_addr.octets())]), + ); + table.update_member_sources( + second, + BTreeSet::from([SharedVirtualNicFlowAddr::V6(second_addr.octets())]), + ); + + assert_eq!( + table.owner_of_source(&ipv6_packet(second_addr, dst), &members), + SourceOwner::Active(second) + ); + + table.remove_owner(second); + assert_eq!( + table.owner_of_source(&ipv6_packet(second_addr, dst), &members), + SourceOwner::None + ); + } + + #[tokio::test] + async fn dispatcher_prefers_source_owner_over_fallback_member() { + let fallback = uuid::Uuid::from_u128(1); + let owner = uuid::Uuid::from_u128(2); + let source = "2001:db8::2".parse::().unwrap(); + let dst = "2001:db8:ffff::1".parse::().unwrap(); + let (fallback_sender, mut fallback_receiver) = mpsc::channel(1); + let (owner_sender, mut owner_receiver) = mpsc::channel(1); + let mut state = SharedVirtualNicDispatcherState::default(); + + state.register(fallback, member_entry(fallback_sender)); + state.register(owner, member_entry(owner_sender)); + state.source_table.update_member_sources( + owner, + BTreeSet::from([SharedVirtualNicFlowAddr::V6(source.octets())]), + ); + state + .forward_tun_packet_to_member(ipv6_packet(source, dst)) + .await; + + assert!(fallback_receiver.try_recv().is_err()); + assert!(owner_receiver.try_recv().is_ok()); + } + + #[tokio::test] + async fn dispatcher_drops_inactive_source_owner_without_fallback() { + let fallback = uuid::Uuid::from_u128(1); + let owner = uuid::Uuid::from_u128(2); + let source = "2001:db8::2".parse::().unwrap(); + let dst = "2001:db8:ffff::1".parse::().unwrap(); + let (fallback_sender, mut fallback_receiver) = mpsc::channel(1); + let (owner_sender, _owner_receiver) = mpsc::channel(1); + let mut state = SharedVirtualNicDispatcherState::default(); + + state.register(fallback, member_entry(fallback_sender)); + state.register(owner, member_entry(owner_sender)); + state.source_table.update_member_sources( + owner, + BTreeSet::from([SharedVirtualNicFlowAddr::V6(source.octets())]), + ); + state.unregister(owner); + state + .forward_tun_packet_to_member(ipv6_packet(source, dst)) + .await; + + assert!(fallback_receiver.try_recv().is_err()); + } +} diff --git a/easytier/src/tests/three_node.rs b/easytier/src/tests/three_node.rs index c5ac99b9..c577f08d 100644 --- a/easytier/src/tests/three_node.rs +++ b/easytier/src/tests/three_node.rs @@ -258,6 +258,356 @@ pub async fn drop_insts(insts: Vec) { while set.join_next().await.is_some() {} } +#[cfg(feature = "tun")] +fn shared_tun_test_dev_name() -> String { + format!("st{:08x}", rand::random::()) +} + +#[cfg(feature = "tun")] +fn shared_tun_test_netns(prefix: &str) -> String { + format!("{prefix}{:04x}", rand::random::()) +} + +#[cfg(feature = "tun")] +fn shared_tun_test_config( + inst_name: &str, + network_name: &str, + network_secret: &str, + netns: Option<&str>, + dev_name: Option<&str>, + ipv4: &str, + no_tun: bool, +) -> TomlConfigLoader { + let config = TomlConfigLoader::default(); + config.set_inst_name(inst_name.to_owned()); + config.set_network_identity(NetworkIdentity::new( + network_name.to_string(), + network_secret.to_string(), + )); + config.set_netns(netns.map(str::to_owned)); + config.set_ipv4(Some(ipv4.parse().unwrap())); + config.set_ipv6(None); + config.set_dhcp(false); + config.set_listeners(vec![]); + config.set_socks5_portal(None); + + let mut flags = config.get_flags(); + flags.dev_name = dev_name.unwrap_or_default().to_string(); + flags.enable_ipv6 = false; + flags.no_tun = no_tun; + config.set_flags(flags); + + config +} + +#[cfg(feature = "tun")] +async fn wait_tun_ready( + receiver: &mut tokio::sync::broadcast::Receiver, +) -> String { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + match receiver.recv().await.unwrap() { + crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) => return ifname, + crate::common::global_ctx::GlobalCtxEvent::TunDeviceError(err) => { + panic!("tun device error: {err}") + } + _ => {} + } + } + }) + .await + .expect("timed out waiting for tun ready") +} + +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn shared_tun_instances_with_same_dev_name_can_communicate() { + let dev_name = shared_tun_test_dev_name(); + let ns_a = shared_tun_test_netns("stna"); + let ns_b = shared_tun_test_netns("stnb"); + let _ns_a_guard = TestNetnsGuard::new(&ns_a, "10.231.1.2/24", "fd31::2/64"); + let _ns_b_guard = TestNetnsGuard::new(&ns_b, "10.231.2.2/24", "fd32::2/64"); + + let mut first = Instance::new(shared_tun_test_config( + "shared_tun_first", + "shared_tun_network_a", + "shared_tun_secret_a", + None, + Some(&dev_name), + "10.144.250.1/24", + false, + )); + let mut second = Instance::new(shared_tun_test_config( + "shared_tun_second", + "shared_tun_network_b", + "shared_tun_secret_b", + None, + Some(&dev_name), + "10.144.251.1/24", + false, + )); + let mut first_peer = Instance::new(shared_tun_test_config( + "shared_tun_first_peer", + "shared_tun_network_a", + "shared_tun_secret_a", + Some(&ns_a), + None, + "10.144.250.2/24", + false, + )); + let mut second_peer = Instance::new(shared_tun_test_config( + "shared_tun_second_peer", + "shared_tun_network_b", + "shared_tun_secret_b", + Some(&ns_b), + None, + "10.144.251.2/24", + false, + )); + let mut first_events = first.get_global_ctx().subscribe(); + let mut second_events = second.get_global_ctx().subscribe(); + + first.run().await.unwrap(); + let first_ifname = wait_tun_ready(&mut first_events).await; + second.run().await.unwrap(); + let second_ifname = wait_tun_ready(&mut second_events).await; + first_peer.run().await.unwrap(); + second_peer.run().await.unwrap(); + + assert_eq!(first_ifname, dev_name); + assert_eq!(second_ifname, dev_name); + + first_peer + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", first.id()).parse().unwrap(), + )); + second_peer + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", second.id()).parse().unwrap(), + )); + + wait_for_condition( + || async { + first.get_peer_manager().list_routes().await.len() == 1 + && first_peer.get_peer_manager().list_routes().await.len() == 1 + && second.get_peer_manager().list_routes().await.len() == 1 + && second_peer.get_peer_manager().list_routes().await.len() == 1 + }, + Duration::from_secs(5), + ) + .await; + + wait_for_condition( + || async { ping_test(&ns_a, "10.144.250.1", None).await }, + Duration::from_secs(5), + ) + .await; + wait_for_condition( + || async { ping_test(&ns_b, "10.144.251.1", None).await }, + Duration::from_secs(5), + ) + .await; + + drop_insts(vec![first, second, first_peer, second_peer]).await; +} + +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn no_tun_instance_does_not_block_shared_tun_communication() { + let dev_name = shared_tun_test_dev_name(); + let ns = shared_tun_test_netns("stnc"); + let _ns_guard = TestNetnsGuard::new(&ns, "10.231.3.2/24", "fd33::2/64"); + + let mut no_tun_inst = Instance::new(shared_tun_test_config( + "shared_tun_no_tun", + "shared_tun_no_tun_network", + "shared_tun_no_tun_secret", + None, + Some(&dev_name), + "10.144.252.1/24", + true, + )); + let mut tun_inst = Instance::new(shared_tun_test_config( + "shared_tun_enabled", + "shared_tun_enabled_network", + "shared_tun_enabled_secret", + None, + Some(&dev_name), + "10.144.253.1/24", + false, + )); + let mut tun_peer = Instance::new(shared_tun_test_config( + "shared_tun_enabled_peer", + "shared_tun_enabled_network", + "shared_tun_enabled_secret", + Some(&ns), + None, + "10.144.253.2/24", + false, + )); + let mut tun_events = tun_inst.get_global_ctx().subscribe(); + + no_tun_inst.run().await.unwrap(); + tun_inst.run().await.unwrap(); + let ifname = wait_tun_ready(&mut tun_events).await; + tun_peer.run().await.unwrap(); + + assert_eq!(ifname, dev_name); + + tun_peer + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", tun_inst.id()).parse().unwrap(), + )); + + wait_for_condition( + || async { + tun_inst.get_peer_manager().list_routes().await.len() == 1 + && tun_peer.get_peer_manager().list_routes().await.len() == 1 + }, + Duration::from_secs(5), + ) + .await; + wait_for_condition( + || async { ping_test(&ns, "10.144.253.1", None).await }, + Duration::from_secs(5), + ) + .await; + + drop_insts(vec![no_tun_inst, tun_inst, tun_peer]).await; +} + +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn shared_tun_proxy_cidr_reaches_member_network() { + prepare_linux_namespaces(); + + let dev_name = shared_tun_test_dev_name(); + let center_cfg = get_inst_config( + "shared_proxy_center", + Some("net_a"), + "10.144.254.1", + "fd00:254::1/64", + ); + center_cfg.set_ipv6(None); + center_cfg.set_listeners(vec![]); + center_cfg.set_socks5_portal(None); + let mut center = Instance::new(center_cfg); + + let shared_cfg_1 = get_inst_config( + "shared_proxy_first", + Some("net_c"), + "10.144.254.2", + "fd00:254::2/64", + ); + shared_cfg_1.set_ipv6(None); + shared_cfg_1.set_listeners(vec![]); + shared_cfg_1.set_socks5_portal(None); + shared_cfg_1 + .add_proxy_cidr("10.1.2.0/24".parse().unwrap(), None) + .unwrap(); + let mut shared_flags = shared_cfg_1.get_flags(); + shared_flags.dev_name = dev_name.clone(); + shared_cfg_1.set_flags(shared_flags.clone()); + let mut shared_1 = Instance::new(shared_cfg_1); + + let shared_cfg_2 = get_inst_config( + "shared_proxy_second", + Some("net_c"), + "10.144.254.3", + "fd00:254::3/64", + ); + shared_cfg_2.set_ipv6(None); + shared_cfg_2.set_listeners(vec![]); + shared_cfg_2.set_socks5_portal(None); + shared_cfg_2.set_flags(shared_flags); + let mut shared_2 = Instance::new(shared_cfg_2); + + let remote_cfg = get_inst_config( + "shared_proxy_remote", + Some("net_b"), + "10.144.254.4", + "fd00:254::4/64", + ); + remote_cfg.set_ipv6(None); + remote_cfg.set_listeners(vec![]); + remote_cfg.set_socks5_portal(None); + let mut remote = Instance::new(remote_cfg); + + let mut shared_1_events = shared_1.get_global_ctx().subscribe(); + let mut shared_2_events = shared_2.get_global_ctx().subscribe(); + + center.run().await.unwrap(); + shared_1.run().await.unwrap(); + shared_2.run().await.unwrap(); + remote.run().await.unwrap(); + + let shared_1_ifname = wait_tun_ready(&mut shared_1_events).await; + let shared_2_ifname = wait_tun_ready(&mut shared_2_events).await; + assert_eq!(shared_1_ifname, dev_name); + assert_eq!(shared_2_ifname, dev_name); + + shared_1 + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", center.id()).parse().unwrap(), + )); + shared_2 + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", center.id()).parse().unwrap(), + )); + remote + .get_conn_manager() + .add_connector(RingTunnelConnector::new( + format!("ring://{}", center.id()).parse().unwrap(), + )); + + wait_for_condition( + || async { + center.get_peer_manager().list_routes().await.len() == 3 + && shared_1.get_peer_manager().list_routes().await.len() == 3 + && shared_2.get_peer_manager().list_routes().await.len() == 3 + && remote.get_peer_manager().list_routes().await.len() == 3 + }, + Duration::from_secs(8), + ) + .await; + + wait_proxy_route_appear( + ¢er.get_peer_manager(), + "10.144.254.2/24", + shared_1.peer_id(), + "10.1.2.0/24", + ) + .await; + wait_proxy_route_appear( + &remote.get_peer_manager(), + "10.144.254.2/24", + shared_1.peer_id(), + "10.1.2.0/24", + ) + .await; + + wait_for_condition( + || async { ping_test("net_a", "10.1.2.4", None).await }, + Duration::from_secs(8), + ) + .await; + wait_for_condition( + || async { ping_test("net_b", "10.1.2.4", None).await }, + Duration::from_secs(8), + ) + .await; + + drop_insts(vec![center, shared_1, shared_2, remote]).await; +} + mod direct_connector_mapped_listener_tests { use std::sync::Arc; @@ -924,6 +1274,123 @@ pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() { drop_insts(vec![provider, client]).await; } +#[cfg(feature = "tun")] +#[tokio::test] +#[serial_test::serial] +pub async fn shared_tun_public_ipv6_auto_addr_end_to_end() { + let lab = PublicIpv6Lab::setup(); + let provider_dev = shared_tun_test_dev_name(); + let client_dev = shared_tun_test_dev_name(); + + let provider_cfg = get_public_ipv6_config( + "provider_shared_public_ipv6", + PublicIpv6Lab::PROVIDER_NS, + "10.144.144.1", + &provider_dev, + uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap(), + ); + provider_cfg.set_ipv6_public_addr_provider(true); + + let client_cfg = get_public_ipv6_config( + "client_shared_public_ipv6", + PublicIpv6Lab::CLIENT_NS, + "10.144.144.2", + &client_dev, + uuid::Uuid::parse_str("55555555-5555-5555-5555-555555555555").unwrap(), + ); + client_cfg.set_ipv6_public_addr_auto(true); + + let client_peer_cfg = get_public_ipv6_config( + "client_shared_public_ipv6_peer", + PublicIpv6Lab::CLIENT_NS, + "10.144.144.3", + &client_dev, + uuid::Uuid::parse_str("66666666-6666-6666-6666-666666666666").unwrap(), + ); + client_peer_cfg.set_listeners(vec![]); + + let mut provider = Instance::new(provider_cfg); + let mut client = Instance::new(client_cfg); + let mut client_peer = Instance::new(client_peer_cfg); + let mut client_events = client.get_global_ctx().subscribe(); + let mut client_peer_events = client_peer.get_global_ctx().subscribe(); + + provider.run().await.unwrap(); + client.run().await.unwrap(); + client_peer.run().await.unwrap(); + + let shared_ifname = wait_tun_ready(&mut client_events).await; + assert_eq!(shared_ifname, wait_tun_ready(&mut client_peer_events).await); + assert_eq!(shared_ifname, client_dev); + + provider + .get_conn_manager() + .add_connector(TcpTunnelConnector::new( + "tcp://10.1.1.2:11010".parse().unwrap(), + )); + + wait_for_condition( + || async { + provider.get_peer_manager().list_routes().await.len() == 1 + && client.get_peer_manager().list_routes().await.len() == 1 + }, + Duration::from_secs(8), + ) + .await; + + wait_for_condition( + || async { + provider + .get_global_ctx() + .get_advertised_ipv6_public_addr_prefix() + == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap()) + }, + Duration::from_secs(10), + ) + .await; + + let leased = wait_for_public_ipv6_addr(&client).await; + wait_for_public_ipv6_route(&provider, leased).await; + + wait_for_condition( + || async { + addr_exists_in_ns(PublicIpv6Lab::CLIENT_NS, &client_dev, &leased.to_string()) + && route_exists_in_ns( + PublicIpv6Lab::CLIENT_NS, + &format!("default dev {client_dev}"), + ) + && route_exists_in_ns( + PublicIpv6Lab::PROVIDER_NS, + &format!("{} dev {provider_dev}", leased.address()), + ) + }, + Duration::from_secs(10), + ) + .await; + + wait_for_condition( + || async { ping6_test(PublicIpv6Lab::CLIENT_NS, PublicIpv6Lab::SERVER_IP, None).await }, + Duration::from_secs(10), + ) + .await; + + wait_for_condition( + || async { + ping6_test( + PublicIpv6Lab::SERVER_NS, + leased.address().to_string().as_str(), + None, + ) + .await + }, + Duration::from_secs(10), + ) + .await; + + drop_insts(vec![provider, client, client_peer]).await; + drop(lab); +} + #[rstest::rstest] #[tokio::test] #[serial_test::serial]