diff --git a/.vscode/settings.json b/.vscode/settings.json index fb0827b..47fa098 100644 --- a/.vscode/settings.json +++ b/.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"] } \ No newline at end of file diff --git a/src/lib.rs b/src/lib.rs index 0ca9cb1..04d622b 100755 --- a/src/lib.rs +++ b/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()); } diff --git a/src/network/async_main.rs b/src/network/async_main.rs index ad888fa..e5329b2 100755 --- a/src/network/async_main.rs +++ b/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, cancel: CancellationToken, connecting_chan: Option>, + global_dns_rx: Receiver<(Vec, 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, 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) -> Option> { None } -async fn loop_tap(eee: &'static Node, cancel: CancellationToken) { +async fn loop_tap(eee: &'static Node, mut dns_rx: Receiver<(Vec, 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; diff --git a/src/network/ipv6.rs b/src/network/ipv6.rs index 1055945..d699750 100755 --- a/src/network/ipv6.rs +++ b/src/network/ipv6.rs @@ -69,7 +69,7 @@ pub async fn run_ipv6(edge: &'static Node, mut v6_may_change: Receiver) { } pub fn get_current_ipv6() -> Option { - return None; + // return None; let Ok(ips) = local_ip_address::list_afinet_netifas() else { error!("failed to get ip address"); return None; diff --git a/src/network/node.rs b/src/network/node.rs index 356c3b7..52500f9 100755 --- a/src/network/node.rs +++ b/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>, udpsock_for_dns: Arc, - udp_sock_for_global_dns: Arc, + 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, - pub udp_sock_for_global_dns: Arc, + pub udp_sock_for_global_dns: DynamicDNSClient, pub dns_matcher: Arc, pub server_ip: String, @@ -468,7 +469,7 @@ impl Node { connecting_chan: Option>, hostname: String, udpsock_for_dns: Arc, - udp_sock_for_global_dns: Arc, + udp_sock_for_global_dns: DynamicDNSClient, server_ip: String, install_channel: String, take_over_dns: bool, diff --git a/src/network/tun_linux.rs b/src/network/tun_linux.rs index 409211c..d3abf7f 100755 --- a/src/network/tun_linux.rs +++ b/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 diff --git a/src/network/tuntap.rs b/src/network/tuntap.rs index 8eef9ff..adaae6e 100755 --- a/src/network/tuntap.rs +++ b/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() diff --git a/src/utils/dns.rs b/src/utils/dns.rs index 1de6948..f96eda8 100644 --- a/src/utils/dns.rs +++ b/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)) } -} \ No newline at end of file +} + + + +/// 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>, + rotate_send_channel: Sender, + query_count: AtomicU8, + response_tx: Sender<(Vec, SocketAddr)>, + dns_server: SocketAddr, +} + +impl DynamicDNSClient { + pub async fn new(dns_server: SocketAddr, response_tx: Sender<(Vec, SocketAddr)>) -> Result { + 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 { + 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) { + 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, tx: Sender<(Vec, 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; + } +} +*/ \ No newline at end of file