mirror/Easytier
1
0
Fork 0
mirror of https://github.com/EasyTier/EasyTier.git synced 2026-08-06 12:39:51 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
cf3fcb75b5a249de4538f0ab5df6628908222705
Easytier/easytier/src/instance/instance.rs
T
sijie.sun cf3fcb75b5 test: cover shared tun magic dns in netns 
Add an end-to-end shared tun Magic DNS test with two members
sharing one device in the same namespace. The test checks real
netns ping paths and queries the Magic DNS fake IP through dig.

Teach DnsRunner to pass the instance netns into
MagicDnsServerInstance so fake DNS route add/remove operations run
where the tun device exists. Keep the existing constructors as
wrappers so current callers stay source-compatible.
2026-06-14 01:26:19 +08:00

1993 lines
71 KiB
Rust
Raw Blame History

#[cfg(feature = "tun")]
use std::any::Any;
use std::collections::HashSet;
use std::net::{IpAddr, Ipv4Addr};
#[cfg(feature = "tun")]
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Weak};
#[cfg(all(feature = "tun", not(mobile)))]
use std::time::Duration;
use anyhow::Context;
use cidr::{IpCidr, Ipv4Inet};
use futures::FutureExt;
#[cfg(all(feature = "tun", not(mobile)))]
use tokio::sync::oneshot;
use tokio::sync::{Mutex, Notify};
#[cfg(feature = "tun")]
use tokio::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,
should_run_public_ipv6_provider_reconcile, validate_public_ipv6_config,
validate_public_ipv6_config_values,
};
#[cfg(feature = "tun")]
use super::shared_virtual_nic::SharedVirtualNicRegistry;
#[cfg(feature = "socks5")]
use crate::gateway::socks5::Socks5Server;
#[derive(Clone)]
struct IpProxy {
tcp_proxy: Arc<TcpProxy<NatDstTcpConnector>>,
icmp_proxy: Arc<IcmpProxy>,
udp_proxy: Arc<UdpProxy>,
global_ctx: ArcGlobalCtx,
started: Arc<AtomicBool>,
}
impl IpProxy {
fn new(global_ctx: ArcGlobalCtx, peer_manager: Arc<PeerManager>) -> Result<Self, Error> {
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<Box<dyn Any + 'static + Send>>,
#[cfg(feature = "magic-dns")]
magic_dns: Option<MagicDnsContainer>,
}
#[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<DnsRunner>) -> 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<T: 'static + Send>(ctx: T) -> Self {
Self {
nic_ctx: Some(Box::new(ctx)),
#[cfg(feature = "magic-dns")]
magic_dns: None,
}
}
}
#[cfg(feature = "tun")]
type ArcNicCtx = Arc<Mutex<Option<NicCtxContainer>>>;
#[cfg(feature = "tun")]
type ArcSharedVirtualNicRegistry = Arc<Mutex<SharedVirtualNicRegistry>>;
#[cfg(feature = "tun")]
fn default_shared_virtual_nic_registry() -> ArcSharedVirtualNicRegistry {
static REGISTRY: OnceLock<ArcSharedVirtualNicRegistry> = OnceLock::new();
REGISTRY
.get_or_init(|| Arc::new(Mutex::new(SharedVirtualNicRegistry::new())))
.clone()
}
pub struct InstanceRpcServerHook {
rpc_portal_whitelist: Vec<IpCidr>,
}
impl InstanceRpcServerHook {
pub fn new(rpc_portal_whitelist: Option<Vec<IpCidr>>) -> 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<TunnelInfo>,
) -> Result<Option<TunnelInfo>, 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<GlobalCtx>,
#[cfg(feature = "socks5")]
socks5_server: Weak<Socks5Server>,
peer_manager: Weak<PeerManager>,
conn_manager: Weak<ManualConnectorManager>,
}
impl InstanceConfigPatcher {
fn parse_ipv6_public_addr_prefix_patch(
prefix: Option<&str>,
) -> Result<Option<Option<cidr::Ipv6Cidr>>, anyhow::Error> {
let Some(prefix) = prefix else {
return Ok(None);
};
let prefix = prefix.trim();
if prefix.is_empty() {
return Ok(Some(None));
}
let parsed = prefix
.parse()
.with_context(|| format!("failed to parse ipv6 public address prefix: {prefix}"))?;
Ok(Some(Some(parsed)))
}
fn effective_ipv6_for_public_ipv6_validation(
global_ctx: &ArcGlobalCtx,
patch: &crate::proto::api::config::InstanceConfigPatch,
_auto_enabled: bool,
) -> Option<cidr::Ipv6Inet> {
if let Some(ipv6) = patch.ipv6 {
return Some(ipv6.into());
}
global_ctx.get_ipv6()
}
fn validate_public_ipv6_patch(
global_ctx: &ArcGlobalCtx,
patch: &crate::proto::api::config::InstanceConfigPatch,
) -> Result<Option<Option<cidr::Ipv6Cidr>>, anyhow::Error> {
let parsed_prefix =
Self::parse_ipv6_public_addr_prefix_patch(patch.ipv6_public_addr_prefix.as_deref())?;
let auto_enabled = patch
.ipv6_public_addr_auto
.unwrap_or(global_ctx.config.get_ipv6_public_addr_auto());
let provider_enabled = patch
.ipv6_public_addr_provider
.unwrap_or(global_ctx.config.get_ipv6_public_addr_provider());
let prefix =
parsed_prefix.unwrap_or_else(|| global_ctx.config.get_ipv6_public_addr_prefix());
let ipv6 = Self::effective_ipv6_for_public_ipv6_validation(global_ctx, patch, auto_enabled);
validate_public_ipv6_config_values(ipv6, provider_enabled, auto_enabled, prefix)?;
Ok(parsed_prefix)
}
pub async fn apply_patch(
&self,
patch: crate::proto::api::config::InstanceConfigPatch,
) -> Result<(), anyhow::Error> {
let patch_for_event = patch.clone();
let global_ctx = weak_upgrade(&self.global_ctx)?;
let parsed_ipv6_public_addr_prefix = Self::validate_public_ipv6_patch(&global_ctx, &patch)?;
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 provider_reconcile_was_running = should_run_public_ipv6_provider_reconcile(&global_ctx);
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(disable_relay_data) = patch.disable_relay_data {
let mut flags = global_ctx.get_flags();
flags.disable_relay_data = disable_relay_data;
global_ctx.set_flags(flags);
}
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) = parsed_ipv6_public_addr_prefix {
global_ctx.config.set_ipv6_public_addr_prefix(prefix);
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 =
should_run_public_ipv6_provider_reconcile(&global_ctx);
if !provider_reconcile_was_running && provider_reconcile_should_run {
run_public_ipv6_provider_reconcile_task(&global_ctx);
}
}
Ok(())
}
fn trace_patchables<T: std::fmt::Debug>(
patches: &Vec<crate::proto::api::config::Patchable<T>>,
) {
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::<T>()
);
}
}
Some(ConfigPatchAction::Clear) => {
tracing::info!("Clear all for type '{}'", std::any::type_name::<T>());
}
None => {
tracing::warn!(
"Invalid patch action for type '{}'",
std::any::type_name::<T>()
);
}
}
}
}
async fn patch_port_forwards(
&self,
port_forwards: Vec<PortForwardPatch>,
) -> 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(&current_forwards)
.await
.with_context(|| "Failed to reload port forwards")?;
Ok(())
}
async fn patch_acl(
&self,
acl_patch: Option<crate::proto::api::config::AclPatch>,
) -> 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<crate::proto::api::config::ProxyNetworkPatch>,
) -> 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<cidr::Ipv4Cidr> =
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<crate::proto::api::config::RoutePatch>,
) -> 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<crate::proto::api::config::ExitNodePatch>,
) -> 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<crate::proto::api::config::UrlPatch>,
) -> 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<crate::proto::api::config::UrlPatch>,
) -> 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::<url::Url>::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,
#[cfg(feature = "tun")]
shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
peer_packet_receiver: Arc<Mutex<PacketRecvChanReceiver>>,
peer_manager: Arc<PeerManager>,
listener_manager: Arc<Mutex<ListenerManager<PeerManager>>>,
conn_manager: Arc<ManualConnectorManager>,
direct_conn_manager: Arc<DirectConnectorManager>,
udp_hole_puncher: Arc<Mutex<UdpHolePunchConnector>>,
tcp_hole_puncher: Arc<Mutex<TcpHolePunchConnector>>,
ip_proxy: Option<IpProxy>,
#[cfg(feature = "kcp")]
kcp_proxy_src: Option<KcpProxySrc>,
#[cfg(feature = "kcp")]
kcp_proxy_dst: Option<KcpProxyDst>,
#[cfg(feature = "quic")]
quic_proxy: Option<QuicProxy>,
peer_center: Arc<PeerCenterInstance>,
vpn_portal: Arc<Mutex<Box<dyn VpnPortal>>>,
#[cfg(feature = "socks5")]
socks5_server: Arc<Socks5Server>,
proxy_cidrs_monitor: Option<AbortOnDropHandle<()>>,
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)),
#[cfg(feature = "tun")]
shared_virtual_nic_registry: default_shared_virtual_nic_registry(),
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<ManualConnectorManager> {
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<Mutex<PacketRecvChanReceiver>>,
) {
#[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<PeerManager>,
tun_dev: Option<String>,
tun_ip: Ipv4Inet,
) -> Option<DnsRunner> {
let ctx = peer_mgr.get_global_ctx();
if !ctx.config.get_flags().accept_dns {
return None;
}
let runner = DnsRunner::new_with_netns(
peer_mgr,
tun_dev,
tun_ip,
MAGIC_DNS_FAKE_IP.parse().unwrap(),
ctx.net_ns.name(),
);
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<DnsRunner>,
) {
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.");
}
#[cfg(feature = "tun")]
async fn new_nic_ctx(
global_ctx: ArcGlobalCtx,
peer_manager: &Arc<PeerManager>,
peer_packet_receiver: Arc<Mutex<PacketRecvChanReceiver>>,
close_notifier: Arc<Notify>,
shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
) -> Result<NicCtx, Error> {
if global_ctx.get_flags().dev_name.is_empty() {
return Ok(NicCtx::new(
global_ctx,
peer_manager,
peer_packet_receiver,
close_notifier,
));
}
NicCtx::new_shared(
global_ctx,
peer_manager,
peer_packet_receiver,
close_notifier,
shared_virtual_nic_registry,
uuid::Uuid::new_v4(),
)
.await
}
// 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();
#[cfg(all(not(mobile), feature = "tun"))]
let shared_virtual_nic_registry = self.shared_virtual_nic_registry.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<Ipv4Inet> = 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 = match Self::new_nic_ctx(
global_ctx_c.clone(),
&peer_manager_c,
_peer_packet_receiver.clone(),
nic_closed_notifier.clone(),
shared_virtual_nic_registry.clone(),
)
.await
{
Ok(nic_ctx) => nic_ctx,
Err(e) => {
tracing::error!(
?current_dhcp_ip,
?candidate_ipv4_addr,
?e,
"create nic ctx failed"
);
global_ctx_c.set_ipv4(None);
continue;
}
};
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<Result<(), Error>>) {
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();
let shared_virtual_nic_registry = self.shared_virtual_nic_registry.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 = match Self::new_nic_ctx(
peer_mgr.get_global_ctx(),
&peer_mgr,
peer_packet_receiver.clone(),
close_notifier.clone(),
shared_virtual_nic_registry.clone(),
)
.await
{
Ok(nic_ctx) => nic_ctx,
Err(e) => {
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;
}
};
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<PeerManager> {
self.peer_manager.clone()
}
#[cfg(feature = "ffi-dataplane")]
pub fn get_socks5_server(&self) -> Arc<Socks5Server> {
self.socks5_server.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<Controller = BaseController> + Clone + use<> {
#[derive(Clone)]
struct VpnPortalRpcService {
peer_mgr: Weak<PeerManager>,
vpn_portal: Weak<Mutex<Box<dyn VpnPortal>>>,
}
#[async_trait::async_trait]
impl VpnPortalRpc for VpnPortalRpcService {
type Controller = BaseController;
async fn get_vpn_portal_info(
&self,
_: BaseController,
_request: GetVpnPortalInfoRequest,
) -> Result<GetVpnPortalInfoResponse, rpc_types::error::Error> {
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<Controller = BaseController> + Clone + use<> {
#[derive(Clone)]
pub struct MappedListenerManagerRpcService(Weak<GlobalCtx>);
#[async_trait::async_trait]
impl MappedListenerManageRpc for MappedListenerManagerRpcService {
type Controller = BaseController;
async fn list_mapped_listener(
&self,
_: BaseController,
_request: ListMappedListenerRequest,
) -> Result<ListMappedListenerResponse, rpc_types::error::Error> {
let mut ret = ListMappedListenerResponse::default();
let urls = weak_upgrade(&self.0)?.config.get_mapped_listeners();
let mapped_listeners: Vec<MappedListener> = 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<Controller = BaseController> + Clone + use<> {
#[derive(Clone)]
pub struct PortForwardManagerRpcService {
global_ctx: Weak<GlobalCtx>,
#[cfg(feature = "socks5")]
socks5_server: Weak<Socks5Server>,
}
#[async_trait::async_trait]
impl PortForwardManageRpc for PortForwardManagerRpcService {
type Controller = BaseController;
async fn list_port_forward(
&self,
_: BaseController,
_request: ListPortForwardRequest,
) -> Result<ListPortForwardResponse, rpc_types::error::Error> {
let forwards = weak_upgrade(&self.global_ctx)?.config.get_port_forwards();
let cfgs: Vec<PortForwardConfigPb> = 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<Controller = BaseController> + Clone + use<> {
#[derive(Clone)]
pub struct StatsRpcService {
global_ctx: Weak<GlobalCtx>,
}
#[async_trait::async_trait]
impl StatsRpc for StatsRpcService {
type Controller = BaseController;
async fn get_stats(
&self,
_: BaseController,
_request: GetStatsRequest,
) -> Result<GetStatsResponse, rpc_types::error::Error> {
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<GetPrometheusStatsResponse, rpc_types::error::Error> {
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<Controller = BaseController> + Clone + use<> {
#[derive(Clone)]
pub struct ConfigRpcService {
patcher: InstanceConfigPatcher,
global_ctx: Weak<GlobalCtx>,
}
#[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<PatchConfigResponse> {
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<GetConfigResponse> {
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<A, B, C, D, E, F, G, H> {
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<dyn TcpProxyRpc<Controller = BaseController> + 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<PeerCenterInstanceService>,
credential_manage_rpc_service: PeerManagerRpcService,
}
#[async_trait::async_trait]
impl<
A: PeerManageRpc<Controller = BaseController> + Send + Sync,
B: ConnectorManageRpc<Controller = BaseController> + Send + Sync,
C: MappedListenerManageRpc<Controller = BaseController> + Send + Sync,
D: VpnPortalRpc<Controller = BaseController> + Send + Sync,
E: AclManageRpc<Controller = BaseController> + Send + Sync,
F: PortForwardManageRpc<Controller = BaseController> + Send + Sync,
G: StatsRpc<Controller = BaseController> + Send + Sync,
H: ConfigRpc<Controller = BaseController> + Send + Sync,
> InstanceRpcService for ApiRpcServiceImpl<A, B, C, D, E, F, G, H>
{
fn get_peer_manage_service(&self) -> &dyn PeerManageRpc<Controller = BaseController> {
&self.peer_mgr_rpc_service
}
fn get_connector_manage_service(
&self,
) -> &dyn ConnectorManageRpc<Controller = BaseController> {
&self.connector_mgr_rpc_service
}
fn get_mapped_listener_manage_service(
&self,
) -> &dyn MappedListenerManageRpc<Controller = BaseController> {
&self.mapped_listener_mgr_rpc_service
}
fn get_vpn_portal_service(&self) -> &dyn VpnPortalRpc<Controller = BaseController> {
&self.vpn_portal_rpc_service
}
fn get_proxy_service(
&self,
client_type: &str,
) -> Option<Arc<dyn TcpProxyRpc<Controller = BaseController> + Send + Sync>>
{
self.tcp_proxy_rpc_services
.get(client_type)
.map(|e| e.clone())
}
fn get_acl_manage_service(&self) -> &dyn AclManageRpc<Controller = BaseController> {
&self.acl_manage_rpc_service
}
fn get_port_forward_manage_service(
&self,
) -> &dyn PortForwardManageRpc<Controller = BaseController> {
&self.port_forward_manage_rpc_service
}
fn get_stats_service(&self) -> &dyn StatsRpc<Controller = BaseController> {
&self.stats_rpc_service
}
fn get_config_service(&self) -> &dyn ConfigRpc<Controller = BaseController> {
&self.config_rpc_service
}
fn get_peer_center_service(
&self,
) -> Arc<dyn PeerCenterRpc<Controller = BaseController> + Send + Sync> {
self.peer_center_rpc_service.clone()
}
fn get_credential_manage_service(
&self,
) -> &dyn CredentialManageRpc<Controller = BaseController> {
&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<dyn TcpProxyRpc<Controller = BaseController> + 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<Mutex<Box<dyn VpnPortal>>> {
self.vpn_portal.clone()
}
#[cfg(feature = "tun")]
pub fn get_nic_ctx(&self) -> ArcNicCtx {
self.nic_ctx.clone()
}
#[cfg(feature = "tun")]
pub fn get_shared_virtual_nic_registry(&self) -> ArcSharedVirtualNicRegistry {
self.shared_virtual_nic_registry.clone()
}
pub fn get_peer_packet_receiver(&self) -> Arc<Mutex<PacketRecvChanReceiver>> {
self.peer_packet_receiver.clone()
}
#[cfg(mobile)]
pub async fn setup_nic_ctx_for_mobile(
nic_ctx: ArcNicCtx,
global_ctx: ArcGlobalCtx,
peer_manager: Arc<PeerManager>,
peer_packet_receiver: Arc<Mutex<PacketRecvChanReceiver>>,
shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
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 = Self::new_nic_ctx(
global_ctx.clone(),
&peer_manager,
peer_packet_receiver.clone(),
close_notifier.clone(),
shared_virtual_nic_registry,
)
.await
.with_context(|| "create nic ctx failed")?;
new_nic_ctx
.run_for_mobile(fd)
.await
.with_context(|| "add ip failed")?;
#[cfg(feature = "magic-dns")]
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,
#[cfg(feature = "magic-dns")]
magic_dns_runner,
)
.await;
Ok(())
}
pub async fn clear_resources(&mut self) {
#[cfg(feature = "tun")]
Self::clear_nic_ctx(self.nic_ctx.clone(), self.peer_packet_receiver.clone()).await;
self.peer_manager.clear_resources().await;
}
}
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();
#[cfg(feature = "tun")]
let peer_packet_receiver = self.peer_packet_receiver.clone();
tokio::spawn(async move {
#[cfg(feature = "tun")]
Self::clear_nic_ctx(nic_ctx, peer_packet_receiver).await;
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 {
#[cfg(feature = "tun")]
use std::sync::Arc;
#[cfg(feature = "tun")]
use tokio::sync::{Mutex, Notify};
use crate::{
common::global_ctx::tests::get_mock_global_ctx,
instance::instance::{InstanceConfigPatcher, InstanceRpcServerHook},
proto::{api::config::InstanceConfigPatch, rpc_impl::standalone::RpcServerHook},
};
#[cfg(feature = "tun")]
use crate::{
instance::shared_virtual_nic::SharedVirtualNicRegistry,
peers::{
create_packet_recv_chan,
peer_manager::{PeerManager, RouteAlgoType},
},
};
#[cfg(feature = "tun")]
async fn new_test_nic_ctx(
dev_name: &str,
registry: super::ArcSharedVirtualNicRegistry,
) -> super::NicCtx {
let global_ctx = get_mock_global_ctx();
set_dev_name(&global_ctx, dev_name);
let (packet_sender, packet_receiver) = create_packet_recv_chan();
let peer_manager = Arc::new(PeerManager::new(
RouteAlgoType::Ospf,
global_ctx.clone(),
packet_sender,
));
super::Instance::new_nic_ctx(
global_ctx,
&peer_manager,
Arc::new(Mutex::new(packet_receiver)),
Arc::new(Notify::new()),
registry,
)
.await
.unwrap()
}
#[cfg(feature = "tun")]
fn set_dev_name(global_ctx: &crate::common::global_ctx::ArcGlobalCtx, dev_name: &str) {
let mut flags = global_ctx.get_flags();
flags.dev_name = dev_name.to_string();
global_ctx.set_flags(flags);
}
#[cfg(feature = "tun")]
#[tokio::test]
async fn new_nic_ctx_keeps_empty_dev_name_dedicated() {
let registry = Arc::new(Mutex::new(SharedVirtualNicRegistry::new()));
let nic_ctx = new_test_nic_ctx("", registry).await;
assert!(nic_ctx.is_dedicated_backend_for_test());
assert!(nic_ctx.shared_member_id_for_test().is_none());
assert!(nic_ctx.shared_nic_for_test().is_none());
}
#[cfg(feature = "tun")]
#[tokio::test]
async fn new_nic_ctx_shares_dev_name_with_fresh_members() {
let registry = Arc::new(Mutex::new(SharedVirtualNicRegistry::new()));
let first = new_test_nic_ctx("et-shared", registry.clone()).await;
let second = new_test_nic_ctx("et-shared", registry.clone()).await;
let first_member = first.shared_member_id_for_test().unwrap();
let second_member = second.shared_member_id_for_test().unwrap();
assert_ne!(first_member, second_member);
let first_shared_nic = first.shared_nic_for_test().unwrap();
let second_shared_nic = second.shared_nic_for_test().unwrap();
assert!(Arc::ptr_eq(&first_shared_nic, &second_shared_nic));
let registered_nic = registry.lock().await.get("et-shared").unwrap();
assert!(Arc::ptr_eq(&registered_nic, &first_shared_nic));
}
#[tokio::test]
async fn test_rpc_portal_whitelist() {
use cidr::IpCidr;
struct TestCase {
remote_url: String,
whitelist: Option<Vec<IpCidr>>,
expected_result: bool,
}
let test_cases: Vec<TestCase> = 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
);
}
}
}
#[tokio::test]
async fn validate_public_ipv6_patch_rejects_non_global_prefix() {
let global_ctx = get_mock_global_ctx();
let patch = InstanceConfigPatch {
ipv6_public_addr_provider: Some(true),
ipv6_public_addr_prefix: Some("fd00::/64".to_string()),
..Default::default()
};
let err =
InstanceConfigPatcher::validate_public_ipv6_patch(&global_ctx, &patch).unwrap_err();
assert!(
err.to_string()
.contains("not a valid global unicast IPv6 prefix")
);
}
#[tokio::test]
async fn validate_public_ipv6_patch_allows_enabling_auto_with_manual_ipv6() {
let global_ctx = get_mock_global_ctx();
global_ctx.set_ipv6(Some("fd00::1/64".parse().unwrap()));
let patch = InstanceConfigPatch {
ipv6_public_addr_auto: Some(true),
..Default::default()
};
assert!(InstanceConfigPatcher::validate_public_ipv6_patch(&global_ctx, &patch).is_ok());
}
#[tokio::test]
async fn validate_public_ipv6_patch_ignores_runtime_auto_ipv6_cache() {
let global_ctx = get_mock_global_ctx();
global_ctx.config.set_ipv6_public_addr_auto(true);
global_ctx.set_ipv6(Some("2001:db8::10/64".parse().unwrap()));
let patch = InstanceConfigPatch {
ipv6_public_addr_provider: Some(true),
ipv6_public_addr_prefix: Some("2001:db8:100::/64".to_string()),
..Default::default()
};
assert!(InstanceConfigPatcher::validate_public_ipv6_patch(&global_ctx, &patch).is_ok());
}
}
Reference in New Issue View Git Blame Copy Permalink