server: add addresses to nic and bind to them

fmt

add address

save bindings

easytier/src/dns/node.rs

@@ -1,7 +1,5 @@
 use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent};
-use crate::dns::config::{
+use crate::dns::config::{DNS_NODE_RR_INTERVAL, DNS_SERVER_ELECTION_INTERVAL, DNS_SERVER_RPC_ADDR};
-    DNS_NODE_RR_INTERVAL, DNS_SERVER_ELECTION_INTERVAL, DNS_SERVER_RPC_ADDR,
-};
 use crate::dns::peer_mgr::DnsPeerMgr;
 use crate::dns::server::DnsServer;
 #[cfg(feature = "tun")]
@@ -224,7 +222,7 @@ impl DnsNode {
             let runtime = runtime.clone();
             CancellableTask::spawn(|token| async move {
                 runtime.elect.notify_one();
-                tokio::join!(runtime.run_election(token.clone()), runtime.run(token),);
+                tokio::join!(runtime.run_election(token.clone()), runtime.run(token));
             })
         };
 

easytier/src/dns/server.rs

@@ -2,10 +2,15 @@ use crate::common::global_ctx::ArcGlobalCtx;
 use crate::dns::node_mgr::DnsNodeMgr;
 use crate::dns::system;
 use crate::dns::utils::addr::NameServerAddr;
+#[cfg(feature = "tun")]
+use crate::instance::instance::{ArcNicCtx, NicCtx};
 use crate::peers::peer_manager::PeerManager;
 use crate::proto::dns::DnsNodeMgrRpcServer;
 use crate::proto::rpc_impl::standalone::StandAloneServer;
+use crate::tunnel::common::bind;
 use crate::tunnel::tcp::TcpTunnelListener;
+use crate::utils::task::CancellableTask;
+use anyhow::Context;
 use derivative::Derivative;
 use guarden::guarded;
 use hickory_net::runtime::Time;
@@ -15,18 +20,15 @@ use hickory_server::{
     server::{Request, RequestHandler, ResponseHandler, ResponseInfo},
     zone_handler::Catalog,
 };
+use itertools::chain;
 use parking_lot::RwLock;
 use std::collections::HashSet;
-use std::net::SocketAddr;
+use std::net::IpAddr;
 use std::{sync::Arc, time::Duration};
+use tokio::sync::Mutex;
 use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, instrument};
 
-#[cfg(feature = "tun")]
-use crate::instance::instance::{ArcNicCtx, NicCtx};
-use crate::tunnel::common::bind;
-use crate::utils::task::CancellableTask;
-
 #[derive(Clone)]
 struct DynamicCatalog {
     inner: Arc<tokio::sync::RwLock<Catalog>>,
@@ -72,13 +74,15 @@ pub struct DnsServer {
 
     #[derivative(Debug = "ignore")]
     catalog: DynamicCatalog,
+    runtime: Mutex<Option<CancellableTask<()>>>,
+    bindings: RwLock<HashSet<NameServerAddr>>,
 
-    listeners: RwLock<HashSet<NameServerAddr>>,
     addresses: RwLock<HashSet<NameServerAddr>>,
+    listeners: RwLock<HashSet<NameServerAddr>>,
 }
 
-const DNS_SERVER_LISTENER_TCP_TIMEOUT: Duration = Duration::from_secs(5);
+const DNS_SERVER_TCP_TIMEOUT: Duration = Duration::from_secs(5);
-const DNS_SERVER_LISTENER_TCP_BUFFER_SIZE: usize = 32;
+const DNS_SERVER_TCP_BUFFER_SIZE: usize = 32;
 
 impl DnsServer {
     pub fn new(
@@ -93,6 +97,8 @@ impl DnsServer {
             peer_mgr,
             global_ctx,
             catalog: DynamicCatalog::new(),
+            runtime: Default::default(),
+            bindings: Default::default(),
             listeners: Default::default(),
             addresses: Default::default(),
         }
@@ -103,98 +109,76 @@ impl DnsServer {
             .register(DnsNodeMgrRpcServer::new_arc(self.mgr.clone()), "");
     }
 
-    pub fn addresses(&self) -> HashSet<SocketAddr> {
+    #[cfg(feature = "tun")]
-        self.addresses.read().iter().map(|a| a.addr).collect()
+    async fn update_system(&self, nameservers: &HashSet<NameServerAddr>) -> anyhow::Result<()> {
-    }
+        let nic_ctx = self.nic_ctx.lock().await;
-
+        let nic_ctx = nic_ctx
-    #[instrument(skip_all)]
+            .as_ref()
-    async fn reload_addresses(
+            .and_then(|nic_ctx| nic_ctx.downcast_ref::<NicCtx>())
-        &self,
+            .with_context(|| "failed to get NicCtx")?;
-        addresses: impl IntoIterator<Item = NameServerAddr>,
+        let ifname = nic_ctx
-    ) -> anyhow::Result<()> {
+            .ifname()
-        let addresses = addresses.into_iter().collect();
+            .await
-
+            .with_context(|| "failed to get interface name from NicCtx")?;
-        if *self.addresses.read() == addresses {
+        let system = system::get(&ifname)?.with_context(|| "failed to get system configurator")?;
-            tracing::info!("addresses unchanged, no need to reload");
+        let config = self.global_ctx.config.get_dns();
-            return Ok(());
+        let domain = vec![config.domain.to_string()];
-        }
+        system.set_dns(&system::SystemConfig {
-        tracing::info!(?addresses, "reloading");
+            nameservers: nameservers
-
+                .iter()
-        #[cfg(feature = "tun")]
+                .filter_map(|a| {
-        {
+                    (a.protocol == Protocol::Udp && a.addr.port() == 53)
-            let nic_ctx = self.nic_ctx.lock().await;
+                        .then_some(a.addr.ip().to_string())
-            if let Some(nic_ctx) = nic_ctx
+                })
-                .as_ref()
+                .collect(),
-                .and_then(|nic_ctx| nic_ctx.downcast_ref::<NicCtx>())
+            search_domains: domain.clone(),
-                && let Some(system) = nic_ctx
+            match_domains: domain
-                    .ifname()
+                .into_iter()
-                    .await
+                .chain(config.zones.iter().map(|z| z.origin.to_string()))
-                    .map(|ifname| system::get(&ifname))
+                .collect(),
-                    .transpose()?
+        })?;
-                    .flatten()
-            {
-                let config = self.global_ctx.config.get_dns();
-                let domain = vec![config.domain.to_string()];
-                system.set_dns(&system::SystemConfig {
-                    nameservers: addresses
-                        .iter()
-                        .filter_map(|a| {
-                            (a.protocol == Protocol::Udp && a.addr.port() == 53)
-                                .then_some(a.addr.ip().to_string())
-                        })
-                        .collect(),
-                    search_domains: domain.clone(),
-                    match_domains: domain
-                        .into_iter()
-                        .chain(config.zones.iter().map(|z| z.origin.to_string()))
-                        .collect(),
-                })?;
-            }
-        }
-
-        *self.addresses.write() = addresses;
-
         Ok(())
     }
 
     #[instrument(skip_all)]
-    async fn reload_listeners(
+    async fn rebind(&self) -> anyhow::Result<bool> {
-        &self,
+        let Ok(mut runtime) = self.runtime.try_lock() else {
-        listeners: impl IntoIterator<Item = NameServerAddr>,
+            return Ok(false);
-        runtime: &mut Option<CancellableTask<()>>,
+        };
-    ) -> anyhow::Result<()> {
-        let listeners = listeners.into_iter().collect();
 
-        if *self.listeners.read() == listeners {
+        let mut bindings = {
-            tracing::info!("listeners unchanged, no need to reload");
+            let current = self.bindings.read();
-            return Ok(());
+            let bindings = chain(
-        }
+                self.addresses.read().iter().cloned(),
-        tracing::info!(?listeners, "reloading");
+                self.listeners.read().iter().cloned(),
+            )
+            .collect();
+            if *current == bindings {
+                tracing::info!("bindings unchanged, no need to rebind");
+                return Ok(false);
+            }
+            bindings
+        };
 
-        if let Some(runtime) = runtime.take()
+        if let Some(runtime) = runtime.take() {
-            && let Err(error) = runtime.stop(None).await
+            runtime.stop(None).await?;
-        {
-            tracing::error!(?error, "failed to stop old DNS server runtime");
         }
 
         let mut server = Server::new(self.catalog.clone());
-        for listener in &listeners {
+
-            let addr = listener.addr;
+        bindings.retain(|binding| {
-            tracing::info!(?addr, "binding listener");
+            let addr = binding.addr;
-            if let Err(error) = match listener.protocol {
+            tracing::info!(?addr, "binding");
+            match binding.protocol {
                 Protocol::Tcp => bind().addr(addr).call().map(|s| {
-                    server.register_listener(
+                    server.register_listener(s, DNS_SERVER_TCP_TIMEOUT, DNS_SERVER_TCP_BUFFER_SIZE)
-                        s,
-                        DNS_SERVER_LISTENER_TCP_TIMEOUT,
-                        DNS_SERVER_LISTENER_TCP_BUFFER_SIZE,
-                    )
                 }),
                 Protocol::Udp => bind().addr(addr).call().map(|s| server.register_socket(s)),
                 _ => unimplemented!(),
-            } {
-                tracing::error!(?addr, ?error, "failed to bind listener");
             }
-        }
+            .inspect_err(|error| tracing::error!(?addr, ?error, "failed to bind"))
+            .is_ok()
+        });
 
         let token = server.shutdown_token().clone();
         let handle = tokio::spawn(
@@ -209,8 +193,83 @@ impl DnsServer {
 
         *runtime = Some(CancellableTask::with_handle(token, handle));
 
+        #[cfg(feature = "tun")]
+        if let Err(error) = self.update_system(&bindings).await {
+            tracing::error!(?error, "failed to update system DNS settings");
+        }
+
+        *self.bindings.write() = bindings;
+
+        Ok(true)
+    }
+
+    #[instrument(skip_all)]
+    async fn reload_addresses(&self) -> anyhow::Result<()> {
+        let addresses = self.mgr.iter_addresses().collect();
+
+        let removed = {
+            let current = self.addresses.read();
+            if *current == addresses {
+                tracing::info!("addresses unchanged, no need to reload");
+                return Ok(());
+            }
+            current
+                .difference(&addresses)
+                .cloned()
+                .collect::<HashSet<_>>()
+        };
+        tracing::info!(?addresses, "reloading");
+
+        #[cfg(feature = "tun")]
+        {
+            let nic_ctx = self.nic_ctx.lock().await;
+            if let Some(nic_ctx) = nic_ctx
+                .as_ref()
+                .and_then(|nic_ctx| nic_ctx.downcast_ref::<NicCtx>())
+            {
+                for addr in &addresses {
+                    let ip = addr.addr.ip();
+                    if let Err(error) = match ip {
+                        IpAddr::V4(ipv4) => nic_ctx.add_ipv4_to_tun_device(ipv4.into()).await,
+                        IpAddr::V6(ipv6) => nic_ctx.add_ipv6_to_tun_device(ipv6.into()).await,
+                    } {
+                        tracing::error!(?addr, ?error, "failed to add address to tun device");
+                    }
+                }
+
+                for addr in removed {
+                    let ip = addr.addr.ip();
+                    if let Err(error) = match ip {
+                        IpAddr::V4(ipv4) => nic_ctx.remove_ipv4_from_tun_device(ipv4.into()).await,
+                        IpAddr::V6(ipv6) => nic_ctx.remove_ipv6_from_tun_device(ipv6.into()).await,
+                    } {
+                        tracing::error!(?addr, ?error, "failed to remove address from tun device");
+                    }
+                }
+            }
+        }
+
+        *self.addresses.write() = addresses;
+
+        self.rebind().await?;
+
+        Ok(())
+    }
+
+    #[instrument(skip_all)]
+    async fn reload_listeners(&self) -> anyhow::Result<()> {
+        let listeners = self.mgr.iter_listeners().collect();
+
+        if *self.listeners.read() == listeners {
+            tracing::info!("listeners unchanged, no need to reload");
+            return Ok(());
+        }
+        tracing::info!(?listeners, "reloading");
+
         *self.listeners.write() = listeners;
 
+        self.rebind().await?;
+
         Ok(())
     }
 
@@ -266,7 +325,7 @@ impl DnsServer {
             loop {
                 dirty.addresses.wait().await;
                 if dirty.addresses.reset()
-                    && let Err(error) = self.reload_addresses(self.mgr.iter_addresses()).await
+                    && let Err(error) = self.reload_addresses().await
                 {
                     tracing::error!(?error, "failed to reload addresses");
                     dirty.addresses.mark();
@@ -279,9 +338,7 @@ impl DnsServer {
             loop {
                 dirty.listeners.wait().await;
                 if dirty.listeners.reset()
-                    && let Err(error) = self
+                    && let Err(error) = self.reload_listeners().await
-                        .reload_listeners(self.mgr.iter_listeners(), runtime)
-                        .await
                 {
                     tracing::error!(?error, "failed to reload listeners");
                     dirty.listeners.mark();
@@ -314,7 +371,7 @@ mod tests {
     use hickory_net::runtime::TokioRuntimeProvider;
     use hickory_net::udp::UdpClientStream;
     use hickory_proto::op::{Message, MessageType, OpCode, Query};
-    use hickory_proto::rr::{DNSClass, Name, RData, Record, RecordType, rdata};
+    use hickory_proto::rr::{rdata, DNSClass, Name, RData, Record, RecordType};
     use hickory_proto::serialize::binary::BinEncodable;
     use hickory_server::store::in_memory::InMemoryZoneHandler;
     use hickory_server::zone_handler::ZoneType;
@@ -322,7 +379,7 @@ mod tests {
     use pnet::packet::icmp::{IcmpTypes, MutableIcmpPacket};
     use pnet::packet::ipv4::MutableIpv4Packet;
     use pnet::packet::udp::MutableUdpPacket;
-    use pnet::packet::{MutablePacket, icmp, ipv4, udp};
+    use pnet::packet::{icmp, ipv4, udp, MutablePacket};
     use std::net::Ipv4Addr;
     use std::str::FromStr;
     use std::time::Duration;
@@ -520,16 +577,21 @@ mod tests {
             "udp://10.10.10.53:53".parse::<NameServerAddr>().unwrap(),
             "tcp://10.10.10.54:5353".parse::<NameServerAddr>().unwrap(),
         ];
-        server.reload_addresses(addrs.clone()).await.unwrap();
+        server.reload_addresses().await.unwrap();
 
-        let as_socket = server.addresses();
+        let as_socket = server
+            .addresses
+            .read()
+            .iter()
+            .map(|a| a.addr)
+            .collect::<HashSet<_>>();
         assert_eq!(as_socket.len(), 2);
         assert!(as_socket.contains(&addrs[0].addr));
         assert!(as_socket.contains(&addrs[1].addr));
 
         // No-op reload should keep the same content.
-        server.reload_addresses(addrs).await.unwrap();
+        server.reload_addresses().await.unwrap();
-        assert_eq!(server.addresses().len(), 2);
+        assert_eq!(server.addresses.read().len(), 2);
     }
 
     #[tokio::test]
@@ -555,11 +617,7 @@ mod tests {
             },
         ];
 
-        let mut runtime = None;
+        server.reload_listeners().await.unwrap();
-        server
-            .reload_listeners(listeners, &mut runtime)
-            .await
-            .unwrap();
 
         let stream = UdpClientStream::builder(good_addr, TokioRuntimeProvider::default()).build();
         let (mut client, bg) = Client::<TokioRuntimeProvider>::from_sender(stream);
@@ -578,9 +636,5 @@ mod tests {
         .expect("query failed");
 
         assert!(!response.answers.is_empty());
-
-        if let Some(runtime) = runtime.take() {
-            let _ = runtime.stop(None).await;
-        }
     }
 }

easytier/src/instance/virtual_nic.rs

@@ -827,6 +827,34 @@ impl NicCtx {
         nic.ifname.as_ref().map(|s| s.to_owned())
     }
 
+    pub async fn add_ipv4_to_tun_device(&self, ipv4_addr: cidr::Ipv4Inet) -> Result<(), Error> {
+        let nic = self.nic.lock().await;
+        nic.add_ip(ipv4_addr.address(), ipv4_addr.network_length() as i32)
+            .await
+    }
+
+    pub async fn add_ipv6_to_tun_device(&self, ipv6_addr: cidr::Ipv6Inet) -> Result<(), Error> {
+        let nic = self.nic.lock().await;
+        nic.add_ipv6(ipv6_addr.address(), ipv6_addr.network_length() as i32)
+            .await
+    }
+
+    pub async fn remove_ipv4_from_tun_device(
+        &self,
+        ipv4_addr: cidr::Ipv4Inet,
+    ) -> Result<(), Error> {
+        let nic = self.nic.lock().await;
+        nic.remove_ip(Some(ipv4_addr)).await
+    }
+
+    pub async fn remove_ipv6_from_tun_device(
+        &self,
+        ipv6_addr: cidr::Ipv6Inet,
+    ) -> Result<(), Error> {
+        let nic = self.nic.lock().await;
+        nic.remove_ipv6(Some(ipv6_addr)).await
+    }
+
     pub async fn assign_ipv4_to_tun_device(&self, ipv4_addr: cidr::Ipv4Inet) -> Result<(), Error> {
         let nic = self.nic.lock().await;
         nic.link_up().await?;