#[cfg(feature = "tun")] use std::any::Any; use std::collections::HashSet; use std::net::{IpAddr, Ipv4Addr}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Weak}; #[cfg(feature = "tun")] use std::time::Duration; use anyhow::Context; use cidr::{IpCidr, Ipv4Inet}; use futures::FutureExt; use tokio::sync::{Mutex, Notify}; #[cfg(feature = "tun")] use tokio::{sync::oneshot, task::JoinSet}; #[cfg(feature = "magic-dns")] use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use crate::common::PeerId; use crate::common::acl_processor::AclRuleBuilder; use crate::common::config::ConfigLoader; use crate::common::error::Error; use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtx, GlobalCtxEvent}; use crate::connector::direct::DirectConnectorManager; use crate::connector::manual::{ConnectorManagerRpcService, ManualConnectorManager}; use crate::connector::tcp_hole_punch::TcpHolePunchConnector; use crate::connector::udp_hole_punch::UdpHolePunchConnector; use crate::gateway::icmp_proxy::IcmpProxy; #[cfg(feature = "kcp")] use crate::gateway::kcp_proxy::{KcpProxyDst, KcpProxyDstRpcService, KcpProxySrc}; #[cfg(feature = "quic")] use crate::gateway::quic_proxy::{QuicProxy, QuicProxyDstRpcService}; use crate::gateway::tcp_proxy::{NatDstTcpConnector, TcpProxy, TcpProxyRpcService}; use crate::gateway::udp_proxy::UdpProxy; use crate::peer_center::instance::{PeerCenterInstance, PeerCenterInstanceService}; use crate::peers::peer_conn::PeerConnId; use crate::peers::peer_manager::{PeerManager, RouteAlgoType}; #[cfg(feature = "tun")] use crate::peers::recv_packet_from_chan; use crate::peers::rpc_service::PeerManagerRpcService; use crate::peers::{PacketRecvChanReceiver, create_packet_recv_chan}; use crate::proto::api::config::{ ConfigPatchAction, ConfigRpc, GetConfigRequest, GetConfigResponse, PatchConfigRequest, PatchConfigResponse, PortForwardPatch, }; use crate::proto::api::instance::{ GetPrometheusStatsRequest, GetPrometheusStatsResponse, GetStatsRequest, GetStatsResponse, GetVpnPortalInfoRequest, GetVpnPortalInfoResponse, ListMappedListenerRequest, ListMappedListenerResponse, ListPortForwardRequest, ListPortForwardResponse, MappedListener, MappedListenerManageRpc, MetricSnapshot, PortForwardManageRpc, StatsRpc, VpnPortalInfo, VpnPortalRpc, }; use crate::proto::api::manage::NetworkConfig; use crate::proto::common::{PortForwardConfigPb, TunnelInfo}; use crate::proto::peer_rpc::PeerCenterRpc; use crate::proto::rpc_impl::standalone::RpcServerHook; use crate::proto::rpc_types; use crate::proto::rpc_types::controller::BaseController; use crate::rpc_service::InstanceRpcService; use crate::utils::weak_upgrade; use crate::vpn_portal::{self, VpnPortal}; #[cfg(feature = "magic-dns")] use super::dns_server::{MAGIC_DNS_FAKE_IP, runner::DnsRunner}; use super::listeners::ListenerManager; use super::public_ipv6_provider::{ reconcile_public_ipv6_provider_runtime, run_public_ipv6_provider_reconcile_task, validate_public_ipv6_config, }; #[cfg(feature = "socks5")] use crate::gateway::socks5::Socks5Server; #[derive(Clone)] struct IpProxy { tcp_proxy: Arc>, icmp_proxy: Arc, udp_proxy: Arc, global_ctx: ArcGlobalCtx, started: Arc, } impl IpProxy { fn new(global_ctx: ArcGlobalCtx, peer_manager: Arc) -> Result { let tcp_proxy = TcpProxy::new(peer_manager.clone(), NatDstTcpConnector {}); let icmp_proxy = IcmpProxy::new(global_ctx.clone(), peer_manager.clone()) .with_context(|| "create icmp proxy failed")?; let udp_proxy = UdpProxy::new(global_ctx.clone(), peer_manager) .with_context(|| "create udp proxy failed")?; Ok(IpProxy { tcp_proxy, icmp_proxy, udp_proxy, global_ctx, started: Arc::new(AtomicBool::new(false)), }) } 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. if self.global_ctx.proxy_forward_by_system() && !self.global_ctx.no_tun() { return Ok(()); } self.started.store(true, Ordering::Relaxed); self.tcp_proxy.start(true).await?; if let Err(e) = self.icmp_proxy.start().await { tracing::error!("start icmp proxy failed: {:?}", e); if cfg!(not(any( target_os = "android", any( target_os = "ios", all(target_os = "macos", feature = "macos-ne") ), target_env = "ohos" ))) { // android, ios and ohos not support icmp proxy return Err(e); } } self.udp_proxy.start().await?; Ok(()) } } #[cfg(feature = "tun")] type NicCtx = super::virtual_nic::NicCtx; #[cfg(feature = "magic-dns")] struct MagicDnsContainer { dns_runner_task: AbortOnDropHandle<()>, dns_runner_cancel_token: CancellationToken, } // nic container will be cleared when dhcp ip changed #[cfg(feature = "tun")] pub struct NicCtxContainer { nic_ctx: Option>, #[cfg(feature = "magic-dns")] magic_dns: Option, } #[cfg(feature = "tun")] impl NicCtxContainer { #[cfg(not(feature = "magic-dns"))] fn new(nic_ctx: NicCtx) -> Self { Self { nic_ctx: Some(Box::new(nic_ctx)), } } #[cfg(feature = "magic-dns")] fn new(nic_ctx: NicCtx, dns_runner: Option) -> Self { if let Some(mut dns_runner) = dns_runner { let token = CancellationToken::new(); let token_clone = token.clone(); let task = tokio::spawn(async move { let _ = dns_runner.run(token_clone).await; }); Self { nic_ctx: Some(Box::new(nic_ctx)), magic_dns: Some(MagicDnsContainer { dns_runner_task: AbortOnDropHandle::new(task), dns_runner_cancel_token: token, }), } } else { Self { nic_ctx: Some(Box::new(nic_ctx)), magic_dns: None, } } } fn new_with_any(ctx: T) -> Self { Self { nic_ctx: Some(Box::new(ctx)), #[cfg(feature = "magic-dns")] magic_dns: None, } } } #[cfg(feature = "tun")] type ArcNicCtx = Arc>>; pub struct InstanceRpcServerHook { rpc_portal_whitelist: Vec, } impl InstanceRpcServerHook { pub fn new(rpc_portal_whitelist: Option>) -> Self { let rpc_portal_whitelist = rpc_portal_whitelist .unwrap_or_else(|| vec!["127.0.0.0/8".parse().unwrap(), "::1/128".parse().unwrap()]); InstanceRpcServerHook { rpc_portal_whitelist, } } } #[async_trait::async_trait] impl RpcServerHook for InstanceRpcServerHook { async fn on_new_client( &self, tunnel_info: Option, ) -> Result, anyhow::Error> { let tunnel_info = tunnel_info.ok_or_else(|| anyhow::anyhow!("tunnel info is None"))?; let remote_url = tunnel_info .remote_addr .clone() .ok_or_else(|| anyhow::anyhow!("remote_addr is None"))?; let url_str = &remote_url.url; let url = url::Url::parse(url_str) .map_err(|e| anyhow::anyhow!("Failed to parse remote URL '{}': {}", url_str, e))?; let host = url .host_str() .ok_or_else(|| anyhow::anyhow!("No host found in remote URL '{}'", url_str))?; let ip_addr: IpAddr = host .parse() .map_err(|e| anyhow::anyhow!("Failed to parse IP address '{}': {}", host, e))?; for cidr in &self.rpc_portal_whitelist { if cidr.contains(&ip_addr) { return Ok(Some(tunnel_info)); } } return Err(anyhow::anyhow!( "Rpc portal client IP {} not in whitelist: {:?}, ignoring client.", ip_addr, self.rpc_portal_whitelist )); } } #[derive(Clone)] pub struct InstanceConfigPatcher { global_ctx: Weak, #[cfg(feature = "socks5")] socks5_server: Weak, peer_manager: Weak, conn_manager: Weak, } impl InstanceConfigPatcher { pub async fn apply_patch( &self, patch: crate::proto::api::config::InstanceConfigPatch, ) -> Result<(), anyhow::Error> { let patch_for_event = patch.clone(); self.patch_port_forwards(patch.port_forwards).await?; self.patch_acl(patch.acl).await?; self.patch_proxy_networks(patch.proxy_networks).await?; self.patch_routes(patch.routes).await?; self.patch_exit_nodes(patch.exit_nodes).await?; self.patch_mapped_listeners(patch.mapped_listeners).await?; self.patch_connector(patch.connectors).await?; let global_ctx = weak_upgrade(&self.global_ctx)?; let provider_reconcile_was_running = global_ctx.config.get_ipv6_public_addr_provider() && global_ctx.config.get_ipv6_public_addr_prefix().is_none(); let mut provider_config_changed = false; if let Some(hostname) = patch.hostname { global_ctx.set_hostname(hostname.clone()); global_ctx.config.set_hostname(Some(hostname)); } if let Some(ipv4) = patch.ipv4 && !global_ctx.config.get_dhcp() { global_ctx.set_ipv4(Some(ipv4.into())); global_ctx.config.set_ipv4(Some(ipv4.into())); } if let Some(ipv6) = patch.ipv6 { global_ctx.set_ipv6(Some(ipv6.into())); global_ctx.config.set_ipv6(Some(ipv6.into())); } if let Some(enabled) = patch.ipv6_public_addr_provider { global_ctx.config.set_ipv6_public_addr_provider(enabled); provider_config_changed = true; } if let Some(enabled) = patch.ipv6_public_addr_auto { global_ctx.config.set_ipv6_public_addr_auto(enabled); } if let Some(prefix) = patch.ipv6_public_addr_prefix { let prefix = prefix.trim(); let parsed = if prefix.is_empty() { None } else { Some(prefix.parse().with_context(|| { format!("failed to parse ipv6 public address prefix: {prefix}") })?) }; global_ctx.config.set_ipv6_public_addr_prefix(parsed); provider_config_changed = true; } global_ctx.issue_event(GlobalCtxEvent::ConfigPatched(patch_for_event)); if provider_config_changed { reconcile_public_ipv6_provider_runtime(&global_ctx).await; let provider_reconcile_should_run = global_ctx.config.get_ipv6_public_addr_provider() && global_ctx.config.get_ipv6_public_addr_prefix().is_none(); if !provider_reconcile_was_running && provider_reconcile_should_run { run_public_ipv6_provider_reconcile_task(&global_ctx); } } Ok(()) } fn trace_patchables( patches: &Vec>, ) { for patch in patches { match patch.action { Some(ConfigPatchAction::Add) | Some(ConfigPatchAction::Remove) => { if let Some(value) = &patch.value { tracing::info!("{:?} {:?}", patch.action, value); } else { tracing::warn!( "Ignored {:?} patch with no value for type '{}'. Please ensure the patch value is provided.", patch.action, std::any::type_name::() ); } } Some(ConfigPatchAction::Clear) => { tracing::info!("Clear all for type '{}'", std::any::type_name::()); } None => { tracing::warn!( "Invalid patch action for type '{}'", std::any::type_name::() ); } } } } async fn patch_port_forwards( &self, port_forwards: Vec, ) -> Result<(), anyhow::Error> { if port_forwards.is_empty() { return Ok(()); } #[cfg(feature = "socks5")] let Some(socks5_server) = self.socks5_server.upgrade() else { return Err(anyhow::anyhow!("socks5 server not available")); }; let global_ctx = weak_upgrade(&self.global_ctx)?; let mut current_forwards = global_ctx.config.get_port_forwards(); let patches = port_forwards.into_iter().map(Into::into).collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_forwards, patches); global_ctx .config .set_port_forwards(current_forwards.clone()); #[cfg(feature = "socks5")] socks5_server .reload_port_forwards(¤t_forwards) .await .with_context(|| "Failed to reload port forwards")?; Ok(()) } async fn patch_acl( &self, acl_patch: Option, ) -> Result<(), anyhow::Error> { let Some(acl_patch) = acl_patch else { return Ok(()); }; let global_ctx = weak_upgrade(&self.global_ctx)?; if let Some(acl) = acl_patch.acl { global_ctx.config.set_acl(Some(acl)); } if !acl_patch.tcp_whitelist.is_empty() { let mut current_whitelist = global_ctx.config.get_tcp_whitelist(); let patches = acl_patch .tcp_whitelist .into_iter() .map(Into::into) .collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_whitelist, patches); global_ctx.config.set_tcp_whitelist(current_whitelist); } if !acl_patch.udp_whitelist.is_empty() { let mut current_whitelist = global_ctx.config.get_udp_whitelist(); let patches = acl_patch .udp_whitelist .into_iter() .map(Into::into) .collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_whitelist, patches); global_ctx.config.set_udp_whitelist(current_whitelist); } global_ctx .get_acl_filter() .reload_rules(AclRuleBuilder::build(&global_ctx)?.as_ref()); weak_upgrade(&self.peer_manager)? .get_route() .refresh_acl_groups() .await; Ok(()) } async fn patch_proxy_networks( &self, proxy_networks: Vec, ) -> Result<(), anyhow::Error> { if proxy_networks.is_empty() { return Ok(()); } let global_ctx = weak_upgrade(&self.global_ctx)?; for proxy_network_patch in proxy_networks { let Some(cidr) = proxy_network_patch.cidr.map(|c| c.into()) else { tracing::warn!("Proxy network cidr is None, skipping."); continue; }; let mapped_cidr: Option = proxy_network_patch.mapped_cidr.map(|s| s.into()); match ConfigPatchAction::try_from(proxy_network_patch.action) { Ok(ConfigPatchAction::Add) => { tracing::info!("Proxy network added: {}", cidr); global_ctx.config.add_proxy_cidr(cidr, mapped_cidr)?; } Ok(ConfigPatchAction::Remove) => { tracing::info!("Proxy network removed: {}", cidr); global_ctx.config.remove_proxy_cidr(cidr); } Ok(ConfigPatchAction::Clear) => { tracing::info!("Proxy networks cleared."); global_ctx.config.clear_proxy_cidrs(); } Err(_) => { tracing::warn!( "Invalid proxy network action: {}", proxy_network_patch.action ); } } } Ok(()) } async fn patch_routes( &self, routes: Vec, ) -> Result<(), anyhow::Error> { if routes.is_empty() { return Ok(()); } let global_ctx = weak_upgrade(&self.global_ctx)?; let mut current_routes = global_ctx.config.get_routes().unwrap_or_default(); let patches = routes.into_iter().map(Into::into).collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_routes, patches); if current_routes.is_empty() { global_ctx.config.set_routes(None); } else { global_ctx.config.set_routes(Some(current_routes)); } Ok(()) } async fn patch_exit_nodes( &self, exit_nodes: Vec, ) -> Result<(), anyhow::Error> { if exit_nodes.is_empty() { return Ok(()); } let global_ctx = weak_upgrade(&self.global_ctx)?; let peer_manager = weak_upgrade(&self.peer_manager)?; let mut current_exit_nodes = global_ctx.config.get_exit_nodes(); let patches = exit_nodes.into_iter().map(Into::into).collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_exit_nodes, patches); global_ctx.config.set_exit_nodes(current_exit_nodes); peer_manager.update_exit_nodes().await; Ok(()) } async fn patch_mapped_listeners( &self, mapped_listeners: Vec, ) -> Result<(), anyhow::Error> { if mapped_listeners.is_empty() { return Ok(()); } let global_ctx = weak_upgrade(&self.global_ctx)?; let mut current_mapped_listeners = global_ctx.config.get_mapped_listeners(); let patches = mapped_listeners.into_iter().map(Into::into).collect(); InstanceConfigPatcher::trace_patchables(&patches); crate::proto::api::config::patch_vec(&mut current_mapped_listeners, patches); if current_mapped_listeners.is_empty() { global_ctx.config.set_mapped_listeners(None); } else { global_ctx .config .set_mapped_listeners(Some(current_mapped_listeners)); } Ok(()) } async fn patch_connector( &self, connectors: Vec, ) -> Result<(), anyhow::Error> { if connectors.is_empty() { return Ok(()); } let conn_manager = weak_upgrade(&self.conn_manager)?; for connector in connectors { let Some(url) = connector.url.map(Into::::into) else { tracing::warn!("Connector url is None, skipping."); return Ok(()); }; match ConfigPatchAction::try_from(connector.action) { Ok(ConfigPatchAction::Add) => { tracing::info!("Connector added: {}", url); conn_manager.add_connector_by_url(url).await?; } Ok(ConfigPatchAction::Remove) => { tracing::info!("Connector removed: {}", url); conn_manager.remove_connector(url).await?; } Ok(ConfigPatchAction::Clear) => { tracing::info!("Connectors cleared."); conn_manager.clear_connectors().await; } Err(_) => { tracing::warn!("Invalid connector action: {}", connector.action); } } } Ok(()) } } pub struct Instance { inst_name: String, id: uuid::Uuid, #[cfg(feature = "tun")] nic_ctx: ArcNicCtx, peer_packet_receiver: Arc>, peer_manager: Arc, listener_manager: Arc>>, conn_manager: Arc, direct_conn_manager: Arc, udp_hole_puncher: Arc>, tcp_hole_puncher: Arc>, ip_proxy: Option, #[cfg(feature = "kcp")] kcp_proxy_src: Option, #[cfg(feature = "kcp")] kcp_proxy_dst: Option, #[cfg(feature = "quic")] quic_proxy: Option, peer_center: Arc, vpn_portal: Arc>>, #[cfg(feature = "socks5")] socks5_server: Arc, proxy_cidrs_monitor: Option>, global_ctx: ArcGlobalCtx, } impl Instance { pub fn new(config: impl ConfigLoader + 'static) -> Self { let global_ctx = Arc::new(GlobalCtx::new(config)); tracing::info!( "[INIT] instance creating. config: {}", global_ctx.config.dump() ); let (peer_packet_sender, peer_packet_receiver) = create_packet_recv_chan(); let id = global_ctx.get_id(); let peer_manager = Arc::new(PeerManager::new( RouteAlgoType::Ospf, global_ctx.clone(), peer_packet_sender, )); peer_manager.set_allow_loopback_tunnel(false); let listener_manager = Arc::new(Mutex::new(ListenerManager::new( global_ctx.clone(), peer_manager.clone(), ))); let conn_manager = Arc::new(ManualConnectorManager::new( global_ctx.clone(), peer_manager.clone(), )); let mut direct_conn_manager = DirectConnectorManager::new(global_ctx.clone(), peer_manager.clone()); direct_conn_manager.run(); let direct_conn_manager = Arc::new(direct_conn_manager); let udp_hole_puncher = Arc::new(Mutex::new(UdpHolePunchConnector::new(peer_manager.clone()))); let tcp_hole_puncher = Arc::new(Mutex::new(TcpHolePunchConnector::new(peer_manager.clone()))); let peer_center = Arc::new(PeerCenterInstance::new(peer_manager.clone())); #[cfg(feature = "wireguard")] let vpn_portal_inst = vpn_portal::wireguard::WireGuard::default(); #[cfg(not(feature = "wireguard"))] let vpn_portal_inst = vpn_portal::NullVpnPortal; #[cfg(feature = "socks5")] let socks5_server = Socks5Server::new(global_ctx.clone(), peer_manager.clone(), None); Instance { inst_name: global_ctx.inst_name.clone(), id, peer_packet_receiver: Arc::new(Mutex::new(peer_packet_receiver)), #[cfg(feature = "tun")] nic_ctx: Arc::new(Mutex::new(None)), peer_manager, listener_manager, conn_manager, direct_conn_manager, udp_hole_puncher, tcp_hole_puncher, ip_proxy: None, #[cfg(feature = "kcp")] kcp_proxy_src: None, #[cfg(feature = "kcp")] kcp_proxy_dst: None, #[cfg(feature = "quic")] quic_proxy: None, peer_center, vpn_portal: Arc::new(Mutex::new(Box::new(vpn_portal_inst))), #[cfg(feature = "socks5")] socks5_server, proxy_cidrs_monitor: None, global_ctx, } } pub fn get_conn_manager(&self) -> Arc { self.conn_manager.clone() } async fn add_initial_peers(&self) -> Result<(), Error> { for peer in self.global_ctx.config.get_peers().iter() { self.get_conn_manager() .add_connector_by_url(peer.uri.clone()) .await?; } Ok(()) } async fn prepare_public_ipv6_config(&self) -> Result<(), Error> { validate_public_ipv6_config(&self.global_ctx)?; reconcile_public_ipv6_provider_runtime(&self.global_ctx).await; Ok(()) } // use a mock nic ctx to consume packets. #[cfg(feature = "tun")] async fn clear_nic_ctx( arc_nic_ctx: ArcNicCtx, packet_recv: Arc>, ) { #[cfg(feature = "magic-dns")] if let Some(old_ctx) = arc_nic_ctx.lock().await.take() && let Some(dns_runner) = old_ctx.magic_dns { dns_runner.dns_runner_cancel_token.cancel(); tracing::debug!("cancelling dns runner task"); let ret = dns_runner.dns_runner_task.await; tracing::debug!("dns runner task cancelled, ret: {:?}", ret); }; let mut tasks = JoinSet::new(); tasks.spawn(async move { let mut packet_recv = packet_recv.lock().await; while let Ok(packet) = recv_packet_from_chan(&mut packet_recv).await { tracing::trace!("packet consumed by mock nic ctx: {:?}", packet); } }); arc_nic_ctx .lock() .await .replace(NicCtxContainer::new_with_any(tasks)); tracing::debug!("nic ctx cleared."); } #[cfg(feature = "magic-dns")] fn create_magic_dns_runner( peer_mgr: Arc, tun_dev: Option, tun_ip: Ipv4Inet, ) -> Option { let ctx = peer_mgr.get_global_ctx(); if !ctx.config.get_flags().accept_dns { return None; } let runner = DnsRunner::new( peer_mgr, tun_dev, tun_ip, MAGIC_DNS_FAKE_IP.parse().unwrap(), ); Some(runner) } #[cfg(feature = "tun")] async fn use_new_nic_ctx( arc_nic_ctx: ArcNicCtx, nic_ctx: NicCtx, #[cfg(feature = "magic-dns")] magic_dns: Option, ) { let mut g = arc_nic_ctx.lock().await; *g = Some(NicCtxContainer::new( nic_ctx, #[cfg(feature = "magic-dns")] magic_dns, )); tracing::debug!("nic ctx updated."); } // Warning, if there is an IP conflict in the network when using DHCP, the IP will be automatically changed. fn check_dhcp_ip_conflict(&self) { use rand::Rng; let peer_manager_c = Arc::downgrade(&self.peer_manager.clone()); let global_ctx_c = self.get_global_ctx(); #[cfg(feature = "tun")] let nic_ctx = self.nic_ctx.clone(); let _peer_packet_receiver = self.peer_packet_receiver.clone(); tokio::spawn(async move { let default_ipv4_addr = Ipv4Inet::new(Ipv4Addr::new(10, 126, 126, 0), 24).unwrap(); let mut current_dhcp_ip: Option = None; let mut next_sleep_time = 0; let nic_closed_notifier = Arc::new(Notify::new()); loop { tokio::time::sleep(std::time::Duration::from_secs(next_sleep_time)).await; let Some(peer_manager_c) = peer_manager_c.upgrade() else { tracing::warn!("peer manager is dropped, stop dhcp check."); return; }; if nic_closed_notifier.notified().now_or_never().is_some() { tracing::debug!("nic ctx is closed, try recreate it"); current_dhcp_ip = None; } // do not allocate ip if no peer connected let routes = peer_manager_c.list_routes().await; if routes.is_empty() { next_sleep_time = 1; continue; } else { next_sleep_time = rand::thread_rng().gen_range(5..10); } let mut used_ipv4 = HashSet::new(); for route in routes { let Some(peer_ipv4_addr) = route.ipv4_addr else { continue; }; used_ipv4.insert(peer_ipv4_addr.into()); } let dhcp_inet = used_ipv4.iter().next().unwrap_or(&default_ipv4_addr); // if old ip is already in this subnet and not conflicted, use it if let Some(ip) = current_dhcp_ip && ip.network() == dhcp_inet.network() && !used_ipv4.contains(&ip) { continue; } // find an available ip in the subnet let candidate_ipv4_addr = dhcp_inet.network().iter().find(|ip| { ip.address() != dhcp_inet.first_address() && ip.address() != dhcp_inet.last_address() && !used_ipv4.contains(ip) }); if current_dhcp_ip == candidate_ipv4_addr { continue; } let last_ip = current_dhcp_ip; tracing::debug!( ?current_dhcp_ip, ?candidate_ipv4_addr, "dhcp start changing ip" ); #[cfg(feature = "tun")] Self::clear_nic_ctx(nic_ctx.clone(), _peer_packet_receiver.clone()).await; if let Some(ip) = candidate_ipv4_addr { if global_ctx_c.no_tun() { current_dhcp_ip = Some(ip); global_ctx_c.set_ipv4(Some(ip)); global_ctx_c .issue_event(GlobalCtxEvent::DhcpIpv4Changed(last_ip, Some(ip))); continue; } #[cfg(all(not(mobile), feature = "tun"))] { let mut new_nic_ctx = NicCtx::new( global_ctx_c.clone(), &peer_manager_c, _peer_packet_receiver.clone(), nic_closed_notifier.clone(), ); if let Err(e) = new_nic_ctx.run(Some(ip), global_ctx_c.get_ipv6()).await { tracing::error!( ?current_dhcp_ip, ?candidate_ipv4_addr, ?e, "add ip failed" ); global_ctx_c.set_ipv4(None); continue; } #[cfg(feature = "magic-dns")] let ifname = new_nic_ctx.ifname().await; Self::use_new_nic_ctx( nic_ctx.clone(), new_nic_ctx, #[cfg(feature = "magic-dns")] Self::create_magic_dns_runner(peer_manager_c.clone(), ifname, ip), ) .await; } current_dhcp_ip = Some(ip); global_ctx_c.set_ipv4(Some(ip)); global_ctx_c.issue_event(GlobalCtxEvent::DhcpIpv4Changed(last_ip, Some(ip))); } else { current_dhcp_ip = None; global_ctx_c.set_ipv4(None); global_ctx_c.issue_event(GlobalCtxEvent::DhcpIpv4Conflicted(last_ip)); } } }); } #[cfg(all(not(mobile), feature = "tun"))] fn check_for_static_ip(&self, first_round_output: oneshot::Sender>) { let ipv4_addr = self.global_ctx.get_ipv4(); let ipv6_addr = self.global_ctx.get_ipv6(); // Only run if we have at least one IP address (IPv4 or IPv6) if ipv4_addr.is_none() && ipv6_addr.is_none() { let _ = first_round_output.send(Ok(())); return; } let nic_ctx = self.nic_ctx.clone(); let peer_mgr = Arc::downgrade(&self.peer_manager); let peer_packet_receiver = self.peer_packet_receiver.clone(); tokio::spawn(async move { let mut output_tx = Some(first_round_output); loop { let close_notifier = Arc::new(Notify::new()); { let Some(peer_mgr) = peer_mgr.upgrade() else { tracing::warn!("peer manager is dropped, stop static ip check."); if let Some(output_tx) = output_tx.take() { let _ = output_tx.send(Err(Error::Unknown)); return; } return; }; let mut new_nic_ctx = NicCtx::new( peer_mgr.get_global_ctx(), &peer_mgr, peer_packet_receiver.clone(), close_notifier.clone(), ); if let Err(e) = new_nic_ctx.run(ipv4_addr, ipv6_addr).await { if let Some(output_tx) = output_tx.take() { let _ = output_tx.send(Err(e)); return; } tracing::error!("failed to create new nic ctx, err: {:?}", e); tokio::time::sleep(Duration::from_secs(1)).await; continue; } // Create Magic DNS runner only if we have IPv4 #[cfg(feature = "magic-dns")] { let ifname = new_nic_ctx.ifname().await; let dns_runner = if let Some(ipv4) = ipv4_addr { Self::create_magic_dns_runner(peer_mgr, ifname, ipv4) } else { None }; Self::use_new_nic_ctx(nic_ctx.clone(), new_nic_ctx, dns_runner).await; } #[cfg(not(feature = "magic-dns"))] Self::use_new_nic_ctx(nic_ctx.clone(), new_nic_ctx).await; } if let Some(output_tx) = output_tx.take() { let _ = output_tx.send(Ok(())); } // NOTICE: make sure we do not hold the peer manager here, while close_notifier.notified().now_or_never().is_none() { tokio::time::sleep(Duration::from_secs(1)).await; if peer_mgr.strong_count() == 0 { tracing::warn!("peer manager is dropped, stop static ip check."); return; } } } }); } pub async fn run(&mut self) -> Result<(), Error> { self.prepare_public_ipv6_config().await?; self.listener_manager .lock() .await .prepare_listeners() .await?; self.listener_manager.lock().await.run().await?; self.peer_manager.run().await?; run_public_ipv6_provider_reconcile_task(&self.global_ctx); #[cfg(feature = "tun")] { Self::clear_nic_ctx(self.nic_ctx.clone(), self.peer_packet_receiver.clone()).await; #[cfg(not(mobile))] if !self.global_ctx.config.get_flags().no_tun { let (output_tx, output_rx) = oneshot::channel(); self.check_for_static_ip(output_tx); output_rx.await.unwrap()?; } } if self.global_ctx.config.get_dhcp() { self.check_dhcp_ip_conflict(); } #[cfg(feature = "kcp")] if self.global_ctx.get_flags().enable_kcp_proxy { let src_proxy = KcpProxySrc::new(self.get_peer_manager()).await; src_proxy.start().await; self.kcp_proxy_src = Some(src_proxy); } #[cfg(feature = "kcp")] if !self.global_ctx.get_flags().disable_kcp_input { let mut dst_proxy = KcpProxyDst::new(self.get_peer_manager()).await; dst_proxy.start().await; self.kcp_proxy_dst = Some(dst_proxy); } #[cfg(feature = "quic")] { let quic_src = self.global_ctx.get_flags().enable_quic_proxy; let quic_dst = !self.global_ctx.get_flags().disable_quic_input; if quic_src || quic_dst { let mut quic_proxy = QuicProxy::new(self.get_peer_manager()); quic_proxy.run(quic_src, quic_dst).await; self.quic_proxy = Some(quic_proxy); } } self.global_ctx .get_acl_filter() .reload_rules(AclRuleBuilder::build(&self.global_ctx)?.as_ref()); // run after tun device created, so listener can bind to tun device, which may be required by win 10 self.ip_proxy = Some(IpProxy::new( self.get_global_ctx(), self.get_peer_manager(), )?); self.run_ip_proxy().await?; self.udp_hole_puncher.lock().await.run().await?; self.tcp_hole_puncher.lock().await.run().await?; self.peer_center.init().await; let route_calc = self.peer_center.get_cost_calculator(); self.peer_manager .get_route() .set_route_cost_fn(route_calc) .await; self.add_initial_peers().await?; let monitor = super::proxy_cidrs_monitor::ProxyCidrsMonitor::new( self.peer_manager.clone(), self.global_ctx.clone(), ); self.proxy_cidrs_monitor = Some(monitor.start()); if self.global_ctx.get_vpn_portal_cidr().is_some() { self.run_vpn_portal().await?; } #[cfg(feature = "socks5")] self.socks5_server .run( #[cfg(feature = "kcp")] self.kcp_proxy_src .as_ref() .map(|x| Arc::downgrade(&x.get_kcp_endpoint())), ) .await?; Ok(()) } pub async fn run_ip_proxy(&mut self) -> Result<(), Error> { if self.ip_proxy.is_none() { return Err(anyhow::anyhow!("ip proxy not enabled.").into()); } self.ip_proxy.as_ref().unwrap().start().await?; Ok(()) } pub async fn run_vpn_portal(&mut self) -> Result<(), Error> { if self.global_ctx.get_vpn_portal_cidr().is_none() { return Err(anyhow::anyhow!("vpn portal cidr not set.").into()); } self.vpn_portal .lock() .await .start(self.get_global_ctx(), self.get_peer_manager()) .await?; Ok(()) } pub fn get_peer_manager(&self) -> Arc { self.peer_manager.clone() } pub async fn close_peer_conn( &mut self, peer_id: PeerId, conn_id: &PeerConnId, ) -> Result<(), Error> { self.peer_manager .get_peer_map() .close_peer_conn(peer_id, conn_id) .await?; Ok(()) } pub async fn wait(&self) { self.peer_manager.wait().await; } pub fn id(&self) -> uuid::Uuid { self.id } pub fn peer_id(&self) -> PeerId { self.peer_manager.my_peer_id() } fn get_vpn_portal_rpc_service( &self, ) -> impl VpnPortalRpc + Clone + use<> { #[derive(Clone)] struct VpnPortalRpcService { peer_mgr: Weak, vpn_portal: Weak>>, } #[async_trait::async_trait] impl VpnPortalRpc for VpnPortalRpcService { type Controller = BaseController; async fn get_vpn_portal_info( &self, _: BaseController, _request: GetVpnPortalInfoRequest, ) -> Result { let Some(vpn_portal) = self.vpn_portal.upgrade() else { return Err(anyhow::anyhow!("vpn portal not available").into()); }; let Some(peer_mgr) = self.peer_mgr.upgrade() else { return Err(anyhow::anyhow!("peer manager not available").into()); }; let vpn_portal = vpn_portal.lock().await; let ret = GetVpnPortalInfoResponse { vpn_portal_info: Some(VpnPortalInfo { vpn_type: vpn_portal.name(), client_config: vpn_portal.dump_client_config(peer_mgr).await, connected_clients: vpn_portal.list_clients().await, }), }; Ok(ret) } } VpnPortalRpcService { peer_mgr: Arc::downgrade(&self.peer_manager), vpn_portal: Arc::downgrade(&self.vpn_portal), } } fn get_mapped_listener_manager_rpc_service( &self, ) -> impl MappedListenerManageRpc + Clone + use<> { #[derive(Clone)] pub struct MappedListenerManagerRpcService(Weak); #[async_trait::async_trait] impl MappedListenerManageRpc for MappedListenerManagerRpcService { type Controller = BaseController; async fn list_mapped_listener( &self, _: BaseController, _request: ListMappedListenerRequest, ) -> Result { let mut ret = ListMappedListenerResponse::default(); let urls = weak_upgrade(&self.0)?.config.get_mapped_listeners(); let mapped_listeners: Vec = urls .into_iter() .map(|u| MappedListener { url: Some(u.into()), }) .collect(); ret.mappedlisteners = mapped_listeners; Ok(ret) } } MappedListenerManagerRpcService(Arc::downgrade(&self.global_ctx)) } fn get_port_forward_manager_rpc_service( &self, ) -> impl PortForwardManageRpc + Clone + use<> { #[derive(Clone)] pub struct PortForwardManagerRpcService { global_ctx: Weak, #[cfg(feature = "socks5")] socks5_server: Weak, } #[async_trait::async_trait] impl PortForwardManageRpc for PortForwardManagerRpcService { type Controller = BaseController; async fn list_port_forward( &self, _: BaseController, _request: ListPortForwardRequest, ) -> Result { let forwards = weak_upgrade(&self.global_ctx)?.config.get_port_forwards(); let cfgs: Vec = forwards.into_iter().map(Into::into).collect(); Ok(ListPortForwardResponse { cfgs }) } } PortForwardManagerRpcService { global_ctx: Arc::downgrade(&self.global_ctx), #[cfg(feature = "socks5")] socks5_server: Arc::downgrade(&self.socks5_server), } } fn get_stats_rpc_service(&self) -> impl StatsRpc + Clone + use<> { #[derive(Clone)] pub struct StatsRpcService { global_ctx: Weak, } #[async_trait::async_trait] impl StatsRpc for StatsRpcService { type Controller = BaseController; async fn get_stats( &self, _: BaseController, _request: GetStatsRequest, ) -> Result { let snapshots = weak_upgrade(&self.global_ctx)? .stats_manager() .get_all_metrics(); let metrics = snapshots .into_iter() .map(|snapshot| { let mut labels = std::collections::BTreeMap::new(); for label in snapshot.labels.labels() { labels.insert(label.key.clone(), label.value.clone()); } MetricSnapshot { name: snapshot.name_str(), value: snapshot.value, labels, } }) .collect(); Ok(GetStatsResponse { metrics }) } async fn get_prometheus_stats( &self, _: BaseController, _request: GetPrometheusStatsRequest, ) -> Result { let prometheus_text = weak_upgrade(&self.global_ctx)? .stats_manager() .export_prometheus(); Ok(GetPrometheusStatsResponse { prometheus_text }) } } StatsRpcService { global_ctx: Arc::downgrade(&self.global_ctx), } } pub fn get_config_patcher(&self) -> InstanceConfigPatcher { InstanceConfigPatcher { global_ctx: Arc::downgrade(&self.global_ctx), #[cfg(feature = "socks5")] socks5_server: Arc::downgrade(&self.socks5_server), peer_manager: Arc::downgrade(&self.peer_manager), conn_manager: Arc::downgrade(&self.conn_manager), } } fn get_config_service(&self) -> impl ConfigRpc + Clone + use<> { #[derive(Clone)] pub struct ConfigRpcService { patcher: InstanceConfigPatcher, global_ctx: Weak, } #[async_trait::async_trait] impl ConfigRpc for ConfigRpcService { type Controller = BaseController; async fn patch_config( &self, _: Self::Controller, request: PatchConfigRequest, ) -> crate::proto::rpc_types::error::Result { let Some(patch) = request.patch else { return Ok(PatchConfigResponse::default()); }; self.patcher.apply_patch(patch).await?; Ok(PatchConfigResponse::default()) } async fn get_config( &self, _: Self::Controller, _request: GetConfigRequest, ) -> crate::proto::rpc_types::error::Result { let global_ctx = weak_upgrade(&self.global_ctx)?; let config = NetworkConfig::new_from_config(&global_ctx.config)?; Ok(GetConfigResponse { config: Some(config), }) } } ConfigRpcService { patcher: self.get_config_patcher(), global_ctx: Arc::downgrade(&self.global_ctx), } } pub fn get_api_rpc_service(&self) -> impl InstanceRpcService + use<> { use crate::proto::api::instance::*; #[derive(Clone)] struct ApiRpcServiceImpl { peer_mgr_rpc_service: A, connector_mgr_rpc_service: B, mapped_listener_mgr_rpc_service: C, vpn_portal_rpc_service: D, tcp_proxy_rpc_services: dashmap::DashMap< String, Arc + Send + Sync>, >, acl_manage_rpc_service: E, port_forward_manage_rpc_service: F, stats_rpc_service: G, config_rpc_service: H, peer_center_rpc_service: Arc, credential_manage_rpc_service: PeerManagerRpcService, } #[async_trait::async_trait] impl< A: PeerManageRpc + Send + Sync, B: ConnectorManageRpc + Send + Sync, C: MappedListenerManageRpc + Send + Sync, D: VpnPortalRpc + Send + Sync, E: AclManageRpc + Send + Sync, F: PortForwardManageRpc + Send + Sync, G: StatsRpc + Send + Sync, H: ConfigRpc + Send + Sync, > InstanceRpcService for ApiRpcServiceImpl { fn get_peer_manage_service(&self) -> &dyn PeerManageRpc { &self.peer_mgr_rpc_service } fn get_connector_manage_service( &self, ) -> &dyn ConnectorManageRpc { &self.connector_mgr_rpc_service } fn get_mapped_listener_manage_service( &self, ) -> &dyn MappedListenerManageRpc { &self.mapped_listener_mgr_rpc_service } fn get_vpn_portal_service(&self) -> &dyn VpnPortalRpc { &self.vpn_portal_rpc_service } fn get_proxy_service( &self, client_type: &str, ) -> Option + Send + Sync>> { self.tcp_proxy_rpc_services .get(client_type) .map(|e| e.clone()) } fn get_acl_manage_service(&self) -> &dyn AclManageRpc { &self.acl_manage_rpc_service } fn get_port_forward_manage_service( &self, ) -> &dyn PortForwardManageRpc { &self.port_forward_manage_rpc_service } fn get_stats_service(&self) -> &dyn StatsRpc { &self.stats_rpc_service } fn get_config_service(&self) -> &dyn ConfigRpc { &self.config_rpc_service } fn get_peer_center_service( &self, ) -> Arc + Send + Sync> { self.peer_center_rpc_service.clone() } fn get_credential_manage_service( &self, ) -> &dyn CredentialManageRpc { &self.credential_manage_rpc_service } } ApiRpcServiceImpl { peer_mgr_rpc_service: PeerManagerRpcService::new(self.peer_manager.clone()), connector_mgr_rpc_service: ConnectorManagerRpcService(Arc::downgrade( &self.conn_manager, )), mapped_listener_mgr_rpc_service: self.get_mapped_listener_manager_rpc_service(), vpn_portal_rpc_service: self.get_vpn_portal_rpc_service(), tcp_proxy_rpc_services: { let tcp_proxy_rpc_services: dashmap::DashMap< String, Arc + Send + Sync>, > = dashmap::DashMap::new(); if let Some(ip_proxy) = self.ip_proxy.as_ref() { tcp_proxy_rpc_services.insert( "tcp".to_string(), Arc::new(TcpProxyRpcService::new(ip_proxy.tcp_proxy.clone())), ); } #[cfg(feature = "kcp")] if let Some(kcp_proxy) = self.kcp_proxy_src.as_ref() { tcp_proxy_rpc_services.insert( "kcp_src".to_string(), Arc::new(TcpProxyRpcService::new(kcp_proxy.get_tcp_proxy())), ); } #[cfg(feature = "kcp")] if let Some(kcp_proxy) = self.kcp_proxy_dst.as_ref() { tcp_proxy_rpc_services.insert( "kcp_dst".to_string(), Arc::new(KcpProxyDstRpcService::new(kcp_proxy)), ); } #[cfg(feature = "quic")] if let Some(quic_proxy) = self.quic_proxy.as_ref() { if let Some(quic_src) = quic_proxy.src() { tcp_proxy_rpc_services.insert( "quic_src".to_string(), Arc::new(TcpProxyRpcService::new(quic_src.get_tcp_proxy())), ); } if let Some(quic_dst) = quic_proxy.dst() { tcp_proxy_rpc_services.insert( "quic_dst".to_string(), Arc::new(QuicProxyDstRpcService::new(quic_dst)), ); } } tcp_proxy_rpc_services }, acl_manage_rpc_service: PeerManagerRpcService::new(self.peer_manager.clone()), port_forward_manage_rpc_service: self.get_port_forward_manager_rpc_service(), stats_rpc_service: self.get_stats_rpc_service(), config_rpc_service: self.get_config_service(), peer_center_rpc_service: Arc::new(self.peer_center.get_rpc_service()), credential_manage_rpc_service: PeerManagerRpcService::new(self.peer_manager.clone()), } } pub fn get_global_ctx(&self) -> ArcGlobalCtx { self.global_ctx.clone() } pub fn get_vpn_portal_inst(&self) -> Arc>> { self.vpn_portal.clone() } #[cfg(feature = "tun")] pub fn get_nic_ctx(&self) -> ArcNicCtx { self.nic_ctx.clone() } pub fn get_peer_packet_receiver(&self) -> Arc> { self.peer_packet_receiver.clone() } #[cfg(mobile)] pub async fn setup_nic_ctx_for_mobile( nic_ctx: ArcNicCtx, global_ctx: ArcGlobalCtx, peer_manager: Arc, peer_packet_receiver: Arc>, fd: i32, ) -> Result<(), anyhow::Error> { tracing::info!("setup_nic_ctx_for_mobile, fd: {}", fd); Self::clear_nic_ctx(nic_ctx.clone(), peer_packet_receiver.clone()).await; if fd <= 0 { return Ok(()); } let close_notifier = Arc::new(Notify::new()); let mut new_nic_ctx = NicCtx::new( global_ctx.clone(), &peer_manager, peer_packet_receiver.clone(), close_notifier.clone(), ); new_nic_ctx .run_for_mobile(fd) .await .with_context(|| "add ip failed")?; let magic_dns_runner = if let Some(ipv4) = global_ctx.get_ipv4() { Self::create_magic_dns_runner(peer_manager.clone(), None, ipv4) } else { None }; Self::use_new_nic_ctx(nic_ctx.clone(), new_nic_ctx, magic_dns_runner).await; Ok(()) } pub async fn clear_resources(&mut self) { self.peer_manager.clear_resources().await; #[cfg(feature = "tun")] let _ = self.nic_ctx.lock().await.take(); } } impl Drop for Instance { fn drop(&mut self) { let my_peer_id = self.peer_manager.my_peer_id(); let pm = Arc::downgrade(&self.peer_manager); #[cfg(feature = "tun")] let nic_ctx = self.nic_ctx.clone(); tokio::spawn(async move { #[cfg(feature = "tun")] nic_ctx.lock().await.take(); if let Some(pm) = pm.upgrade() { pm.clear_resources().await; }; let now = std::time::Instant::now(); while now.elapsed().as_secs() < 10 { tokio::time::sleep(std::time::Duration::from_millis(50)).await; if pm.strong_count() == 0 { tracing::info!( "Instance for peer {} dropped, all resources cleared.", my_peer_id ); return; } } debug_assert!( false, "Instance for peer {} dropped, but resources not cleared in 1 seconds.", my_peer_id ); }); } } #[cfg(test)] mod tests { use crate::{ instance::instance::InstanceRpcServerHook, proto::rpc_impl::standalone::RpcServerHook, }; #[tokio::test] async fn test_rpc_portal_whitelist() { use cidr::IpCidr; struct TestCase { remote_url: String, whitelist: Option>, expected_result: bool, } let test_cases: Vec = vec![ // Test default whitelist (127.0.0.0/8, ::1/128) TestCase { remote_url: "tcp://127.0.0.1:15888".to_string(), whitelist: None, expected_result: true, }, TestCase { remote_url: "tcp://127.1.2.3:15888".to_string(), whitelist: None, expected_result: true, }, TestCase { remote_url: "tcp://192.168.1.1:15888".to_string(), whitelist: None, expected_result: false, }, // Test custom whitelist TestCase { remote_url: "tcp://192.168.1.10:15888".to_string(), whitelist: Some(vec![ "192.168.1.0/24".parse().unwrap(), "10.0.0.0/8".parse().unwrap(), ]), expected_result: true, }, TestCase { remote_url: "tcp://10.1.2.3:15888".to_string(), whitelist: Some(vec![ "192.168.1.0/24".parse().unwrap(), "10.0.0.0/8".parse().unwrap(), ]), expected_result: true, }, TestCase { remote_url: "tcp://172.16.0.1:15888".to_string(), whitelist: Some(vec![ "192.168.1.0/24".parse().unwrap(), "10.0.0.0/8".parse().unwrap(), ]), expected_result: false, }, // Test empty whitelist (should reject all connections) TestCase { remote_url: "tcp://127.0.0.1:15888".to_string(), whitelist: Some(vec![]), expected_result: false, }, // Test broad whitelist (0.0.0.0/0 and ::/0 accept all IP addresses) TestCase { remote_url: "tcp://8.8.8.8:15888".to_string(), whitelist: Some(vec!["0.0.0.0/0".parse().unwrap()]), expected_result: true, }, // Test edge case: specific IP whitelist TestCase { remote_url: "tcp://192.168.1.5:15888".to_string(), whitelist: Some(vec!["192.168.1.5/32".parse().unwrap()]), expected_result: true, }, TestCase { remote_url: "tcp://192.168.1.6:15888".to_string(), whitelist: Some(vec!["192.168.1.5/32".parse().unwrap()]), expected_result: false, }, // Test invalid URL (this case will fail during URL parsing) TestCase { remote_url: "invalid-url".to_string(), whitelist: None, expected_result: false, }, // Test URL without IP address (this case will fail during IP parsing) TestCase { remote_url: "tcp://localhost:15888".to_string(), whitelist: None, expected_result: false, }, ]; for case in test_cases { let hook = InstanceRpcServerHook::new(case.whitelist.clone()); let tunnel_info = Some(crate::proto::common::TunnelInfo { remote_addr: Some(crate::proto::common::Url { url: case.remote_url.clone(), }), ..Default::default() }); let result = hook.on_new_client(tunnel_info).await; if case.expected_result { assert!( result.is_ok(), "Expected success for remote_url:{},whitelist:{:?},but got: {:?}", case.remote_url, case.whitelist, result ); } else { assert!( result.is_err(), "Expected failure for remote_url:{},whitelist:{:?},but got: {:?}", case.remote_url, case.whitelist, result ); } } } }