dns client, rotate every 60 seconds or 100 queries

.vscode/settings.json

@@ -1,4 +1,6 @@
 {
     // "rust-analyzer.cargo.target": "x86_64-pc-windows-gnu",
+    // "rust-analyzer.cargo.target": "x86_64-unknown-linux-gnu",
+
     // "rust-analyzer.cargo.features": ["tun"]
 }

src/lib.rs

@@ -30,6 +30,8 @@ use sdlan_sn_rs::{
     utils::{create_or_load_uuid, get_sdlan_sock_from_socketaddr, Result, SDLanError},
 };
 
+use crate::utils::DynamicDNSClient;
+
 #[cfg(all(feature = "tun", target_os = "windows"))]
 compile_error!("tun feature is not supported on windows");
 
@@ -91,6 +93,9 @@ pub async fn run_sdlan(
         }
     };
 
+    let (global_dns_tx, global_dns_rx) = channel(10);
+    let udp_sock_for_global_dns = DynamicDNSClient::new("223.5.5.5:53".parse().unwrap(), global_dns_tx).await.unwrap();
+
     init_edge(
         // &args.token,
         // &args.network_code,
@@ -118,7 +123,7 @@ pub async fn run_sdlan(
     // let install_chan = install_channel.to_owned();
     tokio::spawn(async move {
         if let Err(e) =
-            async_main(args, start_stop_chan, cancel, connecting_chan).await
+            async_main(args, start_stop_chan, cancel, connecting_chan, global_dns_rx).await
         {
             error!("failed to run async main: {}", e.as_str());
         }

src/network/async_main.rs

@@ -1,3 +1,4 @@
+use std::net::SocketAddr;
 use std::sync::atomic::{Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -29,6 +30,7 @@ pub async fn async_main(
     start_stop_chan: Receiver<StartStopInfo>,
     cancel: CancellationToken,
     connecting_chan: Option<Sender<ConnectionInfo>>,
+    global_dns_rx: Receiver<(Vec<u8>, SocketAddr)>,
 ) -> Result<()> {
     // let _ = PidRecorder::new(".pid");
     let edge = get_edge();
@@ -74,7 +76,7 @@ pub async fn async_main(
     {
         let cancel = cancel.clone();
         tokio::spawn(async move {
-            run_edge_loop(edge, cancel).await;
+            run_edge_loop(edge, global_dns_rx, cancel).await;
         });
     }
 
@@ -118,7 +120,7 @@ pub async fn async_main(
     Ok(())
 }
 
-async fn run_edge_loop(eee: &'static Node, cancel: CancellationToken) {
+async fn run_edge_loop(eee: &'static Node, global_dns_rx: Receiver<(Vec<u8>, SocketAddr)>, cancel: CancellationToken) {
     ping_to_sn().await;
     {
         let cancel2 = cancel.clone();
@@ -133,7 +135,7 @@ async fn run_edge_loop(eee: &'static Node, cancel: CancellationToken) {
 
     {
         tokio::spawn(async move {
-            loop_tap(eee, cancel).await;
+            loop_tap(eee, global_dns_rx, cancel).await;
         });
     }
 }
@@ -212,7 +214,7 @@ async fn receive_dns_reply(sock: &Arc<UdpSocket>) -> Option<Vec<u8>> {
     None
 }
 
-async fn loop_tap(eee: &'static Node, cancel: CancellationToken) {
+async fn loop_tap(eee: &'static Node, mut dns_rx: Receiver<(Vec<u8>, SocketAddr)>, cancel: CancellationToken) {
     debug!("loop tap");
     let (tx, mut rx) = channel(10);
     tokio::spawn(async {
@@ -225,15 +227,16 @@ async fn loop_tap(eee: &'static Node, cancel: CancellationToken) {
                 drop(rx);
                 break;
             }
-            global_reply_global = receive_dns_reply(&eee.udp_sock_for_global_dns)  => {
+            global_reply_global = dns_rx.recv() => {
                 // global reply, only dns payload
                 if let Some(data) = global_reply_global {
-                    if let Ok(mut dns) = simple_dns::Packet::parse(&data) {
+                    if let Ok(mut dns) = simple_dns::Packet::parse(&data.0) {
                         let transaction_id = dns.id();
                         if let Some((ip, port, origin_transaction_id)) = eee.dns_matcher.get_client_info(transaction_id) {
                             warn!("got dns reply from global 223.5.5.5, will send to {}:{}", 
                                 ip_to_string(&ip), port);
 
+                            
                             let dstmac = eee.device_config.get_mac();
                             let srcmac = eee.device_config.dns_mac;
 

src/network/ipv6.rs

@@ -69,7 +69,7 @@ pub async fn run_ipv6(edge: &'static Node, mut v6_may_change: Receiver<bool>) {
 }
 
 pub fn get_current_ipv6() -> Option<Ipv6Addr> {
-    return None;
+    // return None;
     let Ok(ips) = local_ip_address::list_afinet_netifas() else {
         error!("failed to get ip address");
         return None;

src/network/node.rs

@@ -17,6 +17,7 @@ use tokio::sync::mpsc::Sender;
 use tracing::{debug, error, warn};
 
 use crate::network::{ArpTable, RouteTable2};
+use crate::utils::DynamicDNSClient;
 use crate::pb::{
     encode_to_tcp_message, encode_to_udp_message, SdlArpRequest, SdlEmpty, SdlStunProbe,
     SdlStunProbeReply,
@@ -52,7 +53,7 @@ pub async fn init_edge(
     // mtu: u32,
     connecting_chan: Option<Sender<ConnectionInfo>>,
     udpsock_for_dns: Arc<UdpSocket>,
-    udp_sock_for_global_dns: Arc<UdpSocket>,
+    udp_sock_for_global_dns: DynamicDNSClient,
     hostname: String,
     server_ip: String,
     install_channel: String,
@@ -255,7 +256,7 @@ pub struct Node {
     pub quic_endpoint: Endpoint,
 
     pub udp_sock_for_dns: Arc<UdpSocket>,
-    pub udp_sock_for_global_dns: Arc<UdpSocket>,
+    pub udp_sock_for_global_dns: DynamicDNSClient,
     pub dns_matcher: Arc<DNSMatcher>,
 
     pub server_ip: String,
@@ -468,7 +469,7 @@ impl Node {
         connecting_chan: Option<Sender<ConnectionInfo>>,
         hostname: String,
         udpsock_for_dns: Arc<UdpSocket>,
-        udp_sock_for_global_dns: Arc<UdpSocket>,
+        udp_sock_for_global_dns: DynamicDNSClient,
         server_ip: String,
         install_channel: String,
         take_over_dns: bool,

src/network/tun_linux.rs

@@ -877,7 +877,7 @@ pub fn get_install_channel() -> String {
 }
 
 fn check_has_resolvectl() -> bool {
-    // return false;
+    return false;
     let res = Command::new("resolvectl").arg("status").output();
     if let Ok(_) = res {
         true

src/network/tuntap.rs

@@ -211,10 +211,11 @@ pub async fn parse_dns_payload(
                     if let Ok(res) = dns.build_bytes_vec() {
                         if let Err(e) = edge
                             .udp_sock_for_global_dns
-                            .send_to(&res, "223.5.5.5:53")
+                            .send_query(&res)
+                            // .send_to(&res, "223.5.5.5:53")
                             .await
                         {
-                            error!("failed to query for global dns: {}", e);
+                            error!("failed to query for global dns: {}", e.as_str());
                         }
                     }
                     // edge.udp_sock_for_global_dns.send_to()

src/utils/dns.rs

@@ -1,4 +1,13 @@
+use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::{Arc, atomic::AtomicU16};
+use std::{net::{Ipv4Addr, SocketAddr}, time::Duration};
+
+use arc_swap::ArcSwap;
+use sdlan_sn_rs::utils::Result;
+use simple_dns::{Name, QCLASS, QTYPE, Question, rdata::RData};
+use tokio::signal;
+use tokio::sync::mpsc::{channel, Sender, Receiver};
+use tokio::{net::UdpSocket, time};
 
 use dashmap::DashMap;
 use sdlan_sn_rs::utils::get_current_timestamp;
@@ -44,4 +53,152 @@ impl DNSMatcher {
         let res = self.matcher.remove(&transaction_id)?;
         Some((res.1.src_ip, res.1.src_port, res.1.origin_transaction_id))
     }
-}
+}
+
+
+
+/// usage
+/// ```rust
+/// let (tx, rx) = tokio::sync::mpsc::channel(10);
+/// let client = DynamicDNSClient::new("223.5.5.5:53".parse().unwrap(), tx);
+/// 
+/// ```
+pub struct DynamicDNSClient {
+    current_socket: Arc<ArcSwap<UdpSocket>>,
+    rotate_send_channel: Sender<bool>,
+    query_count: AtomicU8,
+    response_tx: Sender<(Vec<u8>, SocketAddr)>,
+    dns_server: SocketAddr,
+}
+
+impl DynamicDNSClient {
+    pub async fn new(dns_server: SocketAddr, response_tx: Sender<(Vec<u8>, SocketAddr)>) -> Result<Self> {
+        let initial_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await?);
+        let current_socket = Arc::new(ArcSwap::from(initial_socket.clone()));
+
+        Self::spawn_receiver(initial_socket, response_tx.clone());
+
+        let (tx, rx) = channel(5);
+
+        let result = Self {
+            current_socket,
+            dns_server,
+            response_tx,
+            rotate_send_channel: tx,
+            query_count: AtomicU8::new(0),
+        };
+        result.start_refresher(rx);
+        Ok(result)
+    }
+
+    pub async fn send_query(&self, packet: &[u8]) -> Result<usize> {
+        let sock_guard = self.current_socket.load();
+        let size = sock_guard.send_to(packet, self.dns_server).await?;
+        let count = self.query_count.fetch_add(1, Ordering::Release);
+        if count >= 100 {
+            self.rotate_send_channel.send(true).await;
+        }
+        Ok(size)
+    }
+
+    fn start_refresher(&self, mut rx: Receiver<bool>) {
+        let socket_swap = self.current_socket.clone();
+        let response_tx = self.response_tx.clone();
+
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = tokio::time::sleep(Duration::from_secs(60)) => {
+
+                    }
+                    data = rx.recv() => {
+                        if data.is_none() {
+                            panic!("global dns rx None");
+                        }
+                    }
+                };
+                // tokio::time::sleep(Duration::from_secs(60)).await;
+                match UdpSocket::bind("0.0.0.0:0").await {
+                    Ok(new_socket) => {
+                        let new_socket = Arc::new(new_socket);
+                        // let port = new_socket.local_addr().unwrap().port();
+                        Self::spawn_receiver(new_socket.clone(), response_tx.clone());
+                        socket_swap.store(new_socket);
+                    }
+                    Err(e) => {
+                        eprintln!("failed to refresh: {}", e);
+                    }
+                }
+            }
+        });
+    }
+
+    fn spawn_receiver(socket: Arc<UdpSocket>, tx: Sender<(Vec<u8>, SocketAddr)>) {
+        tokio::spawn(async move {
+            let mut buf = vec![0u8; 1024];
+            let port = socket.local_addr().unwrap().port();
+
+            loop {
+                match time::timeout(Duration::from_secs(65), socket.recv_from(&mut buf)).await {
+                    Ok(Ok((len, from))) => {
+                        let data = buf[..len].to_vec();
+                        if tx.send((data, from)).await.is_err() {
+                            break;
+                        }
+                    }
+                    Ok(Err(_)) => break,
+                    Err(_) => {
+                        // timeout occured
+                        break;
+                    }
+                }
+            }
+            eprintln!("port {} has been closed", socket.local_addr().unwrap().port());
+        });
+    }
+}
+
+/*
+async fn test_simple_dns() {
+    let mut id = 0;
+    let (tx, mut rx) = tokio::sync::mpsc::channel(10);
+    let client = DynamicDNSClient::new("223.5.5.5:53".parse().unwrap(), tx).await.unwrap();
+
+    tokio::spawn(async move {
+        loop {
+            let data = rx.recv().await;
+            if let Some(data) = data {
+                let packet = simple_dns::Packet::parse(&data.0).unwrap();
+                println!("got response: id = {}", packet.id());
+                for answer in packet.answers {
+                    println!("  domain: {}", answer.name);
+                    match answer.rdata {
+                        RData::A(a_record) => {
+                            println!("    {}", Ipv4Addr::from_bits(a_record.address).to_string())
+                        }
+                        _other => {
+                            println!("    other response type")
+                        }
+                    }
+                }
+            }
+        }
+    });
+
+    loop {
+        let mut packet = simple_dns::Packet::new_query(id);
+        id += 1;
+        let name = Name::new("www.baidu.com").unwrap();
+        let question = Question {
+            qname: name,
+            qtype: QTYPE::TYPE(simple_dns::TYPE::A),
+            qclass: QCLASS::CLASS(simple_dns::CLASS::IN),
+            unicast_response: false,
+        };
+        packet.questions.push(question);
+        let question = packet.build_bytes_vec().unwrap();
+        client.send_query(&question).await;
+        tokio::time::sleep(Duration::from_secs(5)).await;
+    }
+}
+*/