feat: extend shared tun support and coverage

This commit is contained in:
sijie.sun
2026-04-20 00:38:44 +08:00
parent 0ee551a285
commit 7afbd52fa6
7 changed files with 763 additions and 65 deletions
+5 -7
View File
@@ -158,14 +158,12 @@ async fn set_tun_fd(fd: i32) -> Result<(), String> {
let Some(instance_manager) = INSTANCE_MANAGER.read().await.clone() else {
return Err("set_tun_fd is not supported in remote mode".to_string());
};
if let Some(uuid) = get_client_manager!()?
let instance_ids: Vec<_> = get_client_manager!()?
.get_enabled_instances_with_tun_ids()
.next()
{
instance_manager
.set_tun_fd(&uuid, fd)
.map_err(|e| e.to_string())?;
}
.collect();
instance_manager
.set_tun_fd_for_instances(instance_ids, fd)
.map_err(|e| e.to_string())?;
Ok(())
}
@@ -536,6 +536,7 @@ impl MagicDnsServerInstance {
None
};
let ifcfg = IfConfiger {};
let _g = peer_mgr.get_global_ctx().net_ns.guard();
ifcfg
.add_ipv4_route(tun_dev_name, fake_ip, 32, cost)
.await?;
@@ -590,6 +591,7 @@ impl MagicDnsServerInstance {
&& let Some(tun_dev_name) = &self.data.tun_dev
{
let ifcfg = IfConfiger {};
let _g = self.peer_mgr.get_global_ctx().net_ns.guard();
let _ = ifcfg
.remove_ipv4_route(tun_dev_name, self.data.fake_ip, 32)
.await;
+3 -4
View File
@@ -844,10 +844,9 @@ impl Instance {
Ok(attached) => {
global_ctx.issue_event(GlobalCtxEvent::TunDeviceReady(attached.ifname.clone()));
#[cfg(feature = "magic-dns")]
let dns_runner = ipv4_addr
.and_then(|ip| {
Self::create_magic_dns_runner(peer_mgr, Some(attached.ifname.clone()), ip)
});
let dns_runner = ipv4_addr.and_then(|ip| {
Self::create_magic_dns_runner(peer_mgr, Some(attached.ifname.clone()), ip)
});
Self::use_new_shared_nic_ctx(
arc_nic_ctx,
attached.handle,
+194 -51
View File
@@ -126,6 +126,90 @@ impl MemberClaims {
}
}
fn ipv4_prefixes_overlap(left: cidr::Ipv4Cidr, right: cidr::Ipv4Cidr) -> bool {
left.contains(&right.first_address()) || right.contains(&left.first_address())
}
fn ipv6_prefixes_overlap(left: cidr::Ipv6Cidr, right: cidr::Ipv6Cidr) -> bool {
left.contains(&right.first_address()) || right.contains(&left.first_address())
}
fn format_v4_prefix_conflict(left: cidr::Ipv4Cidr, right: cidr::Ipv4Cidr) -> String {
if left == right {
format!("shared tun conflict: duplicated IPv4 route prefix {}", left)
} else {
format!(
"shared tun conflict: overlapping IPv4 route prefix {} with {}",
left, right
)
}
}
fn format_v6_prefix_conflict(left: cidr::Ipv6Cidr, right: cidr::Ipv6Cidr) -> String {
if left == right {
format!("shared tun conflict: duplicated IPv6 route prefix {}", left)
} else {
format!(
"shared tun conflict: overlapping IPv6 route prefix {} with {}",
left, right
)
}
}
fn validate_claim_conflicts(
claims: &MemberClaims,
other_claims: &MemberClaims,
) -> Result<(), String> {
if let (Some(left), Some(right)) = (claims.ipv4, other_claims.ipv4)
&& left.address() == right.address()
{
return Err(format!(
"shared tun conflict: duplicated IPv4 address {}",
left.address()
));
}
if let (Some(left), Some(right)) = (claims.ipv6, other_claims.ipv6)
&& left.address() == right.address()
{
return Err(format!(
"shared tun conflict: duplicated IPv6 address {}",
left.address()
));
}
for prefix in claims.shared_route_v4_prefixes() {
for other_prefix in other_claims.dispatch_v4_prefixes() {
if ipv4_prefixes_overlap(prefix, other_prefix) {
return Err(format_v4_prefix_conflict(prefix, other_prefix));
}
}
}
if let Some(prefix) = claims.local_v4_prefix() {
for other_prefix in other_claims.shared_route_v4_prefixes() {
if ipv4_prefixes_overlap(prefix, other_prefix) {
return Err(format_v4_prefix_conflict(prefix, other_prefix));
}
}
}
for prefix in claims.shared_route_v6_prefixes() {
for other_prefix in other_claims.dispatch_v6_prefixes() {
if ipv6_prefixes_overlap(prefix, other_prefix) {
return Err(format_v6_prefix_conflict(prefix, other_prefix));
}
}
}
if let Some(prefix) = claims.local_v6_prefix() {
for other_prefix in other_claims.shared_route_v6_prefixes() {
if ipv6_prefixes_overlap(prefix, other_prefix) {
return Err(format_v6_prefix_conflict(prefix, other_prefix));
}
}
}
Ok(())
}
#[derive(Clone)]
struct MemberRuntimeContext {
device: Arc<SharedTunDevice>,
@@ -541,56 +625,7 @@ impl SharedTunDevice {
for other in others {
let other_claims = other.claims.read().await.clone();
if let (Some(left), Some(right)) = (claims.ipv4, other_claims.ipv4)
&& left.address() == right.address()
{
return Err(format!(
"shared tun conflict: duplicated IPv4 address {}",
left.address()
));
}
if let (Some(left), Some(right)) = (claims.ipv6, other_claims.ipv6)
&& left.address() == right.address()
{
return Err(format!(
"shared tun conflict: duplicated IPv6 address {}",
left.address()
));
}
for prefix in claims.shared_route_v4_prefixes() {
if other_claims.dispatch_v4_prefixes().contains(&prefix) {
return Err(format!(
"shared tun conflict: duplicated IPv4 route prefix {}",
prefix
));
}
}
if let Some(prefix) = claims.local_v4_prefix()
&& other_claims.shared_route_v4_prefixes().contains(&prefix)
{
return Err(format!(
"shared tun conflict: duplicated IPv4 route prefix {}",
prefix
));
}
for prefix in claims.shared_route_v6_prefixes() {
if other_claims.dispatch_v6_prefixes().contains(&prefix) {
return Err(format!(
"shared tun conflict: duplicated IPv6 route prefix {}",
prefix
));
}
}
if let Some(prefix) = claims.local_v6_prefix()
&& other_claims.shared_route_v6_prefixes().contains(&prefix)
{
return Err(format!(
"shared tun conflict: duplicated IPv6 route prefix {}",
prefix
));
}
validate_claim_conflicts(claims, &other_claims)?;
}
Ok(())
@@ -938,7 +973,7 @@ fn collect_owned_proxy_v4_routes(global_ctx: &ArcGlobalCtx) -> BTreeSet<cidr::Ip
mod tests {
use std::str::FromStr;
use super::MemberClaims;
use super::{MemberClaims, validate_claim_conflicts};
#[test]
fn nested_prefixes_are_distinct() {
@@ -999,4 +1034,112 @@ mod tests {
.contains(&cidr::Ipv4Cidr::from_str("10.1.2.0/24").unwrap())
);
}
#[test]
fn duplicated_ipv4_addresses_are_rejected() {
let claims = MemberClaims {
ipv4: Some(cidr::Ipv4Inet::from_str("10.144.144.2/24").unwrap()),
..Default::default()
};
let other = MemberClaims {
ipv4: Some(cidr::Ipv4Inet::from_str("10.144.144.2/24").unwrap()),
..Default::default()
};
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("duplicated IPv4 address"));
}
#[test]
fn duplicated_ipv6_addresses_are_rejected() {
let claims = MemberClaims {
ipv6: Some(cidr::Ipv6Inet::from_str("fd00::2/64").unwrap()),
..Default::default()
};
let other = MemberClaims {
ipv6: Some(cidr::Ipv6Inet::from_str("fd00::2/64").unwrap()),
..Default::default()
};
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("duplicated IPv6 address"));
}
#[test]
fn duplicated_proxy_v4_routes_are_rejected() {
let mut claims = MemberClaims::default();
claims
.owned_proxy_v4_routes
.insert(cidr::Ipv4Cidr::from_str("10.1.2.0/24").unwrap());
let mut other = MemberClaims::default();
other
.owned_proxy_v4_routes
.insert(cidr::Ipv4Cidr::from_str("10.1.2.0/24").unwrap());
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("duplicated IPv4 route prefix"));
}
#[test]
fn duplicated_default_ipv6_routes_are_rejected() {
let claims = MemberClaims {
default_route_v6: true,
..Default::default()
};
let other = MemberClaims {
default_route_v6: true,
..Default::default()
};
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("duplicated IPv6 route prefix"));
}
#[test]
fn overlapping_proxy_v4_routes_are_rejected() {
let mut claims = MemberClaims::default();
claims
.owned_proxy_v4_routes
.insert(cidr::Ipv4Cidr::from_str("10.1.0.0/16").unwrap());
let mut other = MemberClaims::default();
other
.owned_proxy_v4_routes
.insert(cidr::Ipv4Cidr::from_str("10.1.2.0/24").unwrap());
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("overlapping IPv4 route prefix"));
}
#[test]
fn default_ipv6_route_overlapping_local_subnet_is_rejected() {
let claims = MemberClaims {
default_route_v6: true,
..Default::default()
};
let other = MemberClaims {
ipv6: Some(cidr::Ipv6Inet::from_str("fd00::2/64").unwrap()),
..Default::default()
};
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("overlapping IPv6 route prefix"));
}
#[test]
fn proxy_v4_routes_overlapping_local_subnet_are_rejected() {
let mut claims = MemberClaims::default();
claims
.owned_proxy_v4_routes
.insert(cidr::Ipv4Cidr::from_str("10.144.144.0/23").unwrap());
let other = MemberClaims {
ipv4: Some(cidr::Ipv4Inet::from_str("10.144.145.2/24").unwrap()),
..Default::default()
};
let err = validate_claim_conflicts(&claims, &other).unwrap_err();
assert!(err.contains("overlapping IPv4 route prefix"));
}
}
+76
View File
@@ -241,6 +241,17 @@ impl NetworkInstanceManager {
Ok(())
}
pub fn set_tun_fd_for_instances<I>(&self, instance_ids: I, fd: i32) -> Result<(), anyhow::Error>
where
I: IntoIterator<Item = uuid::Uuid>,
{
for instance_id in instance_ids {
self.set_tun_fd(&instance_id, fd)?;
}
Ok(())
}
pub fn get_config_dir(&self) -> Option<&PathBuf> {
self.config_dir.as_ref()
}
@@ -465,6 +476,7 @@ impl Display for proto::api::instance::PeerConnInfo {
mod tests {
use super::*;
use crate::common::config::*;
use crate::launcher::{EasyTierLauncher, NetworkInstance};
#[tokio::test]
#[serial_test::serial]
@@ -616,6 +628,70 @@ mod tests {
assert_eq!(manager.instance_stop_tasks.len(), 0);
}
#[test]
fn set_tun_fd_for_instances_allows_empty_target_list() {
let manager = NetworkInstanceManager::new();
assert!(manager.set_tun_fd_for_instances(Vec::new(), 1234).is_ok());
}
#[test]
fn set_tun_fd_for_instances_errors_on_missing_instance() {
let manager = NetworkInstanceManager::new();
let err = manager
.set_tun_fd_for_instances(vec![uuid::Uuid::new_v4()], 1234)
.unwrap_err();
assert!(err.to_string().contains("instance not found"));
}
#[test]
fn set_tun_fd_for_instances_broadcasts_to_all_targets() {
let manager = NetworkInstanceManager::new();
let instance_ids = [
uuid::Uuid::new_v4(),
uuid::Uuid::new_v4(),
uuid::Uuid::new_v4(),
];
for instance_id in instance_ids {
let cfg = TomlConfigLoader::default();
cfg.set_id(instance_id);
cfg.set_inst_name(format!("inst-{instance_id}"));
cfg.set_listeners(vec![]);
manager.instance_map.insert(
instance_id,
NetworkInstance::with_launcher_for_test(
cfg,
ConfigFileControl::STATIC_CONFIG,
EasyTierLauncher::new(),
),
);
}
let mut receivers = instance_ids
.iter()
.map(|instance_id| {
manager
.instance_map
.get(instance_id)
.unwrap()
.take_tun_fd_receiver_for_test()
.unwrap()
})
.collect::<Vec<_>>();
manager
.set_tun_fd_for_instances(instance_ids[..2].to_vec(), 1234)
.unwrap();
assert_eq!(receivers[0].try_recv().unwrap(), Some(1234));
assert_eq!(receivers[1].try_recv().unwrap(), Some(1234));
assert!(matches!(
receivers[2].try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Empty)
));
}
#[tokio::test]
#[serial_test::serial]
async fn test_single_instance_failed() {
+25
View File
@@ -260,6 +260,11 @@ impl EasyTierLauncher {
}
}
}
#[cfg(test)]
pub(crate) fn take_tun_fd_receiver_for_test(&self) -> Option<mpsc::Receiver<TunFd>> {
self.data.tun_fd.1.lock().unwrap().take()
}
}
impl Default for EasyTierLauncher {
@@ -447,6 +452,26 @@ impl NetworkInstance {
.as_ref()
.and_then(|launcher| launcher.get_api_service())
}
#[cfg(test)]
pub(crate) fn with_launcher_for_test(
config: TomlConfigLoader,
config_file_control: ConfigFileControl,
launcher: EasyTierLauncher,
) -> Self {
Self {
config,
launcher: Some(launcher),
config_file_control,
}
}
#[cfg(test)]
pub(crate) fn take_tun_fd_receiver_for_test(&self) -> Option<mpsc::Receiver<TunFd>> {
self.launcher
.as_ref()
.and_then(|launcher| launcher.take_tun_fd_receiver_for_test())
}
}
pub fn add_proxy_network_to_config(
+458 -3
View File
@@ -13,6 +13,9 @@ use x25519_dalek::StaticSecret;
use super::*;
#[cfg(feature = "magic-dns")]
use crate::instance::dns_server::MAGIC_DNS_INSTANCE_ADDR;
// TODO: 需要加一个单测,确保 socks5 + exit node == self || proxy_cidr == 0.0.0.0/0 时,可以实现出口节点的能力。
use crate::{
@@ -263,7 +266,9 @@ async fn wait_for_tun_ready_event(
) -> String {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) = receiver.recv().await.unwrap() {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) =
receiver.recv().await.unwrap()
{
return ifname;
}
}
@@ -278,7 +283,9 @@ async fn assert_no_tun_ready_event(
) {
tokio::time::timeout(timeout, async {
loop {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) = receiver.recv().await.unwrap() {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceReady(ifname) =
receiver.recv().await.unwrap()
{
panic!("unexpected TunDeviceReady event: {ifname}");
}
}
@@ -293,7 +300,9 @@ async fn assert_no_tun_fallback_event(
) {
tokio::time::timeout(timeout, async {
loop {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceFallback(reason) = receiver.recv().await.unwrap() {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceFallback(reason) =
receiver.recv().await.unwrap()
{
panic!("unexpected TunDeviceFallback event: {reason}");
}
}
@@ -302,6 +311,50 @@ async fn assert_no_tun_fallback_event(
.ok();
}
async fn wait_for_tun_fallback_event(
receiver: &mut tokio::sync::broadcast::Receiver<crate::common::global_ctx::GlobalCtxEvent>,
) -> String {
tokio::time::timeout(Duration::from_secs(8), async {
loop {
if let crate::common::global_ctx::GlobalCtxEvent::TunDeviceFallback(reason) =
receiver.recv().await.unwrap()
{
return reason;
}
}
})
.await
.unwrap()
}
async fn wait_for_dhcp_ipv4_changed_event(
receiver: &mut tokio::sync::broadcast::Receiver<crate::common::global_ctx::GlobalCtxEvent>,
) -> cidr::Ipv4Inet {
tokio::time::timeout(Duration::from_secs(15), async {
loop {
if let crate::common::global_ctx::GlobalCtxEvent::DhcpIpv4Changed(_, Some(ip)) =
receiver.recv().await.unwrap()
{
return ip;
}
}
})
.await
.unwrap()
}
async fn link_exists_in_netns(netns: &str, ifname: &str) -> bool {
let _g = NetNS::new(Some(ROOT_NETNS_NAME.to_owned())).guard();
let code = tokio::process::Command::new("ip")
.args(["netns", "exec", netns, "ip", "link", "show", "dev", ifname])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.await
.unwrap();
code.success()
}
fn assert_tcp_proxy_metric_has_protocol(
inst: &Instance,
protocol: TcpProxyEntryTransportType,
@@ -1279,10 +1332,113 @@ pub async fn shared_tun_same_namespace_real_tun() {
Duration::from_secs(8),
)
.await;
wait_for_condition(
|| async { ping6_test("net_c", "fd00::2", None).await },
Duration::from_secs(8),
)
.await;
wait_for_condition(
|| async { ping6_test("net_c", "fd00::3", None).await },
Duration::from_secs(8),
)
.await;
wait_for_condition(
|| async { ping6_test("net_b", "fd00::4", None).await },
Duration::from_secs(8),
)
.await;
drop_insts(vec![center, shared_1, shared_2, remote]).await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_dev_name_mismatch_falls_back_to_dedicated_tun() {
prepare_linux_namespaces();
let shared_cfg_1 = get_inst_config("shared_1", Some("net_b"), "10.144.144.2", "fd00::2/64");
shared_cfg_1.set_listeners(vec![]);
shared_cfg_1.set_socks5_portal(None);
let mut flags_1 = shared_cfg_1.get_flags();
flags_1.dev_name = "et_sdm0".to_string();
shared_cfg_1.set_flags(flags_1);
let mut shared_1 = Instance::new(shared_cfg_1);
let shared_cfg_2 = get_inst_config("shared_2", Some("net_b"), "10.144.144.3", "fd00::3/64");
shared_cfg_2.set_listeners(vec![]);
shared_cfg_2.set_socks5_portal(None);
let mut flags_2 = shared_cfg_2.get_flags();
flags_2.dev_name = "et_sdm1".to_string();
shared_cfg_2.set_flags(flags_2);
let mut shared_2 = Instance::new(shared_cfg_2);
let mut shared_1_events = shared_1.get_global_ctx().subscribe();
let mut shared_2_events = shared_2.get_global_ctx().subscribe();
shared_1.run().await.unwrap();
shared_2.run().await.unwrap();
let ifname = wait_for_tun_ready_event(&mut shared_1_events).await;
assert_eq!(ifname, "et_sdm0");
let reason = wait_for_tun_fallback_event(&mut shared_2_events).await;
assert!(reason.contains("does not match requested dev_name"));
wait_for_condition(
|| async { link_exists_in_netns("net_b", "et_sdm1").await },
Duration::from_secs(8),
)
.await;
assert_no_tun_fallback_event(&mut shared_1_events, Duration::from_secs(2)).await;
drop_insts(vec![shared_1, shared_2]).await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_cleans_up_device_after_last_member_leaves() {
prepare_linux_namespaces();
let shared_cfg_1 = get_inst_config("shared_1", Some("net_b"), "10.144.144.2", "fd00::2/64");
shared_cfg_1.set_listeners(vec![]);
shared_cfg_1.set_socks5_portal(None);
let mut shared_flags = shared_cfg_1.get_flags();
shared_flags.dev_name = "et_shared_gc0".to_string();
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_2", Some("net_b"), "10.144.144.3", "fd00::3/64");
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 mut shared_1_events = shared_1.get_global_ctx().subscribe();
let mut shared_2_events = shared_2.get_global_ctx().subscribe();
shared_1.run().await.unwrap();
shared_2.run().await.unwrap();
let ifname = wait_for_tun_ready_event(&mut shared_1_events).await;
assert_eq!(ifname, wait_for_tun_ready_event(&mut shared_2_events).await);
assert!(link_exists_in_netns("net_b", &ifname).await);
shared_1.clear_resources().await;
drop(shared_1);
wait_for_condition(
|| async { link_exists_in_netns("net_b", &ifname).await },
Duration::from_secs(5),
)
.await;
shared_2.clear_resources().await;
drop(shared_2);
wait_for_condition(
|| async { !link_exists_in_netns("net_b", &ifname).await },
Duration::from_secs(8),
)
.await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_proxy_cidr_same_namespace_real_tun() {
@@ -1383,6 +1539,305 @@ pub async fn shared_tun_proxy_cidr_same_namespace_real_tun() {
drop_insts(vec![center, shared_1, shared_2, remote]).await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_dhcp_same_namespace_real_tun() {
prepare_linux_namespaces();
let center_cfg = get_inst_config("center", Some("net_a"), "10.144.144.1", "fd00::1/64");
center_cfg.set_listeners(vec![]);
let mut center = Instance::new(center_cfg);
let dhcp_cfg_1 = get_inst_config("dhcp_1", Some("net_b"), "10.144.144.2", "fd00::2/64");
dhcp_cfg_1.set_listeners(vec![]);
dhcp_cfg_1.set_socks5_portal(None);
dhcp_cfg_1.set_ipv4(None);
dhcp_cfg_1.set_dhcp(true);
let mut dhcp_flags = dhcp_cfg_1.get_flags();
dhcp_flags.dev_name = "et_shdh0".to_string();
dhcp_cfg_1.set_flags(dhcp_flags.clone());
let mut dhcp_1 = Instance::new(dhcp_cfg_1);
let dhcp_cfg_2 = get_inst_config("dhcp_2", Some("net_b"), "10.144.144.3", "fd00::3/64");
dhcp_cfg_2.set_listeners(vec![]);
dhcp_cfg_2.set_socks5_portal(None);
dhcp_cfg_2.set_ipv4(None);
dhcp_cfg_2.set_dhcp(true);
dhcp_cfg_2.set_flags(dhcp_flags);
let mut dhcp_2 = Instance::new(dhcp_cfg_2);
let remote_cfg = get_inst_config("remote", Some("net_c"), "10.144.144.4", "fd00::4/64");
remote_cfg.set_listeners(vec![]);
let mut remote = Instance::new(remote_cfg);
let mut dhcp_1_tun_events = dhcp_1.get_global_ctx().subscribe();
let mut dhcp_2_tun_events = dhcp_2.get_global_ctx().subscribe();
let mut dhcp_1_dhcp_events = dhcp_1.get_global_ctx().subscribe();
let mut dhcp_2_dhcp_events = dhcp_2.get_global_ctx().subscribe();
center.run().await.unwrap();
dhcp_1.run().await.unwrap();
dhcp_1
.get_conn_manager()
.add_connector(RingTunnelConnector::new(
format!("ring://{}", center.id()).parse().unwrap(),
));
let dhcp_1_tun = wait_for_tun_ready_event(&mut dhcp_1_tun_events).await;
let dhcp_1_ip = wait_for_dhcp_ipv4_changed_event(&mut dhcp_1_dhcp_events).await;
wait_for_condition(
|| async { dhcp_1.get_peer_manager().list_routes().await.len() == 1 },
Duration::from_secs(8),
)
.await;
dhcp_2.run().await.unwrap();
remote.run().await.unwrap();
dhcp_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(),
));
let dhcp_2_tun = wait_for_tun_ready_event(&mut dhcp_2_tun_events).await;
let dhcp_2_ip = wait_for_dhcp_ipv4_changed_event(&mut dhcp_2_dhcp_events).await;
assert_eq!(dhcp_1_tun, dhcp_2_tun);
assert_ne!(dhcp_1_ip.address(), dhcp_2_ip.address());
assert_eq!(dhcp_1_ip.network(), dhcp_2_ip.network());
wait_for_condition(
|| async {
dhcp_1.get_peer_manager().list_routes().await.len() == 3
&& dhcp_2.get_peer_manager().list_routes().await.len() == 3
&& remote.get_peer_manager().list_routes().await.len() == 3
},
Duration::from_secs(12),
)
.await;
let dhcp_1_ip_str = dhcp_1_ip.address().to_string();
let dhcp_2_ip_str = dhcp_2_ip.address().to_string();
wait_for_condition(
|| async { ping_test("net_c", &dhcp_1_ip_str, None).await },
Duration::from_secs(12),
)
.await;
wait_for_condition(
|| async { ping_test("net_c", &dhcp_2_ip_str, None).await },
Duration::from_secs(12),
)
.await;
wait_for_condition(
|| async { ping_test("net_b", "10.144.144.4", None).await },
Duration::from_secs(12),
)
.await;
assert_no_tun_fallback_event(&mut dhcp_1_tun_events, Duration::from_secs(2)).await;
assert_no_tun_fallback_event(&mut dhcp_2_tun_events, Duration::from_secs(2)).await;
drop_insts(vec![center, dhcp_1, dhcp_2, remote]).await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_dynamic_proxy_conflict_falls_back() {
use crate::proto::api::config::{ConfigPatchAction, InstanceConfigPatch, ProxyNetworkPatch};
prepare_linux_namespaces();
let center_cfg = get_inst_config("center", Some("net_a"), "10.144.144.1", "fd00::1/64");
center_cfg.set_listeners(vec![]);
let mut center = Instance::new(center_cfg);
let shared_cfg_1 = get_inst_config("shared_1", Some("net_c"), "10.144.144.2", "fd00::2/64");
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 = "et_srf0".to_string();
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_2", Some("net_c"), "10.144.144.3", "fd00::3/64");
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("remote", Some("net_b"), "10.144.144.4", "fd00::4/64");
remote_cfg.set_listeners(vec![]);
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_tun = wait_for_tun_ready_event(&mut shared_1_events).await;
let shared_2_tun = wait_for_tun_ready_event(&mut shared_2_events).await;
assert_eq!(shared_1_tun, shared_2_tun);
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 {
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;
shared_2
.get_config_patcher()
.apply_patch(InstanceConfigPatch {
proxy_networks: vec![ProxyNetworkPatch {
action: ConfigPatchAction::Add as i32,
cidr: Some("10.1.2.0/24".parse().unwrap()),
mapped_cidr: None,
}],
..Default::default()
})
.await
.unwrap();
let reason = wait_for_tun_fallback_event(&mut shared_2_events).await;
assert!(reason.contains("route prefix"));
wait_for_condition(
|| async { ping_test("net_b", "10.1.2.4", None).await },
Duration::from_secs(10),
)
.await;
wait_for_condition(
|| async { ping_test("net_b", "10.144.144.3", None).await },
Duration::from_secs(10),
)
.await;
assert_no_tun_fallback_event(&mut shared_1_events, Duration::from_secs(2)).await;
drop_insts(vec![center, shared_1, shared_2, remote]).await;
}
#[cfg(feature = "magic-dns")]
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_magic_dns_same_namespace_real_tun() {
prepare_linux_namespaces();
let center_cfg = get_inst_config("center", Some("net_a"), "10.144.144.1", "fd00::1/64");
center_cfg.set_listeners(vec![]);
let mut center = Instance::new(center_cfg);
let shared_cfg_1 = get_inst_config("shared_1", Some("net_b"), "10.144.144.2", "fd00::2/64");
shared_cfg_1.set_listeners(vec![]);
shared_cfg_1.set_socks5_portal(None);
let mut shared_flags = shared_cfg_1.get_flags();
shared_flags.dev_name = "et_shared_dns0".to_string();
shared_flags.accept_dns = true;
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_2", Some("net_b"), "10.144.144.3", "fd00::3/64");
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("remote", Some("net_c"), "10.144.144.4", "fd00::4/64");
remote_cfg.set_listeners(vec![]);
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_tun = wait_for_tun_ready_event(&mut shared_1_events).await;
let shared_2_tun = wait_for_tun_ready_event(&mut shared_2_events).await;
assert_eq!(shared_1_tun, shared_2_tun);
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 {
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_for_condition(
|| async { ping_test("net_c", "10.144.144.2", None).await },
Duration::from_secs(8),
)
.await;
wait_for_condition(
|| async { ping_test("net_c", "10.144.144.3", None).await },
Duration::from_secs(8),
)
.await;
wait_for_condition(
|| async { ping_test("net_b", "10.144.144.4", None).await },
Duration::from_secs(8),
)
.await;
let _ = MAGIC_DNS_INSTANCE_ADDR;
assert_no_tun_fallback_event(&mut shared_1_events, Duration::from_secs(2)).await;
assert_no_tun_fallback_event(&mut shared_2_events, Duration::from_secs(2)).await;
drop_insts(vec![center, shared_1, shared_2, remote]).await;
}
#[tokio::test]
#[serial_test::serial]
pub async fn shared_tun_kcp_proxy_with_source_shared_tun() {