use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
use log::{debug, warn};
use tokio::net::UdpSocket;
use super::cache;
use super::chnroute::ChnRoute;
use super::dns;
use super::gfwlist::GfwList;
use super::Mode;
const MAX_DNS_MSG: usize = 4096;
pub trait IpSink: Send + Sync {
fn add(&self, ip: Ipv4Addr);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Decision {
Direct,
Tunnel,
}
pub fn chinadns_decision(local_ips: &[Ipv4Addr], chnroute: &ChnRoute) -> Decision {
if local_ips.iter().any(|ip| chnroute.contains(*ip)) {
Decision::Direct
} else {
Decision::Tunnel
}
}
struct Decided {
response: Vec<u8>,
tunnel_ips: Vec<Ipv4Addr>,
}
pub struct Resolver {
mode: Mode,
gfwlist: GfwList,
chnroute: ChnRoute,
local: SocketAddr,
remote: SocketAddr,
timeout: Duration,
sink: Arc<dyn IpSink>,
cache: Arc<cache::DnsCache>,
local_bind: IpAddr,
remote_bind: IpAddr,
}
impl Resolver {
#[allow(clippy::too_many_arguments)]
pub fn new(
mode: Mode,
gfwlist: GfwList,
chnroute: ChnRoute,
local: SocketAddr,
remote: SocketAddr,
timeout: Duration,
sink: Arc<dyn IpSink>,
cache: Arc<cache::DnsCache>,
) -> Self {
Self {
mode,
gfwlist,
chnroute,
local,
remote,
timeout,
sink,
cache,
local_bind: IpAddr::V4(Ipv4Addr::UNSPECIFIED),
remote_bind: IpAddr::V4(Ipv4Addr::UNSPECIFIED),
}
}
pub fn with_bind_sources(mut self, local_bind: IpAddr, remote_bind: IpAddr) -> Self {
self.local_bind = local_bind;
self.remote_bind = remote_bind;
self
}
pub async fn resolve(&self, query: &[u8]) -> Option<Vec<u8>> {
if matches!(self.mode, Mode::Full) {
return None; }
if let Some((response, tunnel_ips)) = self.cache.get(query) {
self.tunnel(&tunnel_ips);
return Some(response);
}
let decided = match self.mode {
Mode::Full => return None,
Mode::GfwList => {
let name = dns::question_name(query);
self.decide_gfwlist(query, name.as_deref()).await?
}
Mode::ChinaDns => {
let name = dns::question_name(query);
self.decide_chinadns(query, name.as_deref()).await?
}
};
self.tunnel(&decided.tunnel_ips);
self.cache
.put(query, &decided.response, &decided.tunnel_ips);
Some(decided.response)
}
async fn decide_gfwlist(&self, query: &[u8], name: Option<&str>) -> Option<Decided> {
let tunnel = name.map(|n| self.gfwlist.matches(n)).unwrap_or(false);
let upstream = if tunnel { self.remote } else { self.local };
let bind = if tunnel {
self.remote_bind
} else {
self.local_bind
};
let response = match query_upstream(upstream, query, self.timeout, bind).await {
Ok(r) => r,
Err(e) => {
debug!("gfwlist: upstream {upstream} failed for {name:?}: {e}");
return None;
}
};
let tunnel_ips = if tunnel {
dns::a_records(&response)
} else {
Vec::new()
};
Some(Decided {
response,
tunnel_ips,
})
}
async fn decide_chinadns(&self, query: &[u8], name: Option<&str>) -> Option<Decided> {
if name.map(|n| self.gfwlist.matches(n)).unwrap_or(false) {
let response =
match query_upstream(self.remote, query, self.timeout, self.remote_bind).await {
Ok(r) => r,
Err(e) => {
debug!(
"chinadns: forced-tunnel upstream {} failed for {name:?}: {e}",
self.remote
);
return None;
}
};
let tunnel_ips = dns::a_records(&response);
return Some(Decided {
response,
tunnel_ips,
});
}
let remote_query = query.to_vec();
let remote = self.remote;
let remote_bind = self.remote_bind;
let timeout = self.timeout;
let remote_task = tokio::spawn(async move {
query_upstream(remote, &remote_query, timeout, remote_bind)
.await
.ok()
});
let local_res = query_upstream(self.local, query, self.timeout, self.local_bind)
.await
.ok();
let local_ips = local_res.as_deref().map(dns::a_records).unwrap_or_default();
if local_res.is_some() && chinadns_decision(&local_ips, &self.chnroute) == Decision::Direct
{
remote_task.abort();
return local_res.map(|response| Decided {
response,
tunnel_ips: Vec::new(),
});
}
let remote_res = remote_task.await.ok().flatten();
let response = remote_res.or(local_res)?;
let tunnel_ips = dns::a_records(&response);
Some(Decided {
response,
tunnel_ips,
})
}
fn tunnel(&self, ips: &[Ipv4Addr]) {
for ip in ips {
self.sink.add(*ip);
}
}
}
async fn query_upstream(
server: SocketAddr,
query: &[u8],
timeout: Duration,
bind_src: IpAddr,
) -> Result<Vec<u8>> {
let bind: SocketAddr = (bind_src, 0).into();
let sock = UdpSocket::bind(bind)
.await
.context("bind upstream DNS socket")?;
sock.connect(server)
.await
.with_context(|| format!("connect to DNS upstream {server}"))?;
sock.send(query).await.context("send DNS query")?;
let mut buf = vec![0u8; MAX_DNS_MSG];
let n = tokio::time::timeout(timeout, sock.recv(&mut buf))
.await
.with_context(|| format!("DNS upstream {server} timed out"))?
.context("recv DNS response")?;
buf.truncate(n);
Ok(buf)
}
#[cfg(windows)]
fn disable_udp_conn_reset(sock: &UdpSocket) {
use std::os::windows::io::AsRawSocket;
use windows_sys::Win32::Networking::WinSock::WSAIoctl;
const SIO_UDP_CONNRESET: u32 = 0x9800_000C;
let disable: u32 = 0;
let mut returned: u32 = 0;
let ret = unsafe {
WSAIoctl(
sock.as_raw_socket() as _,
SIO_UDP_CONNRESET,
&disable as *const u32 as *const core::ffi::c_void,
std::mem::size_of::<u32>() as u32,
std::ptr::null_mut(),
0,
&mut returned,
std::ptr::null_mut(),
None,
)
};
if ret != 0 {
debug!("WSAIoctl(SIO_UDP_CONNRESET) failed; relying on recv-loop fallback");
}
}
pub async fn serve(listener: UdpSocket, resolver: Arc<Resolver>) -> Result<()> {
let listener = Arc::new(listener);
#[cfg(windows)]
disable_udp_conn_reset(&listener);
let mut buf = vec![0u8; MAX_DNS_MSG];
loop {
let (n, client) = match listener.recv_from(&mut buf).await {
Ok(v) => v,
Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => {
debug!("DNS proxy: ignoring spurious connection-reset on recv");
continue;
}
Err(e) => return Err(e).context("DNS proxy recv_from failed"),
};
let query = buf[..n].to_vec();
let resolver = Arc::clone(&resolver);
let listener = Arc::clone(&listener);
tokio::spawn(async move {
match resolver.resolve(&query).await {
Some(resp) => {
if let Err(e) = listener.send_to(&resp, client).await {
warn!("DNS proxy failed to reply to {client}: {e}");
}
}
None => debug!("DNS proxy: no answer for query from {client}"),
}
});
}
}
pub async fn prewarm(resolver: Arc<Resolver>, domains: Vec<String>) {
let total = domains.len();
for batch in domains.chunks(8) {
let mut handles = Vec::with_capacity(batch.len());
for (i, domain) in batch.iter().enumerate() {
let resolver = Arc::clone(&resolver);
let query = dns::build_query(i as u16 + 1, domain);
handles.push(tokio::spawn(async move { resolver.resolve(&query).await }));
}
for h in handles {
let _ = h.await;
}
}
debug!("pre-warmed {total} domains into the DNS cache");
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
#[derive(Default)]
struct VecSink(Mutex<Vec<Ipv4Addr>>);
impl IpSink for VecSink {
fn add(&self, ip: Ipv4Addr) {
self.0.lock().unwrap().push(ip);
}
}
impl VecSink {
fn ips(&self) -> Vec<Ipv4Addr> {
self.0.lock().unwrap().clone()
}
}
fn query(name: &str) -> Vec<u8> {
let mut m = vec![0x12, 0x34, 0x01, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0];
for label in name.split('.') {
m.push(label.len() as u8);
m.extend_from_slice(label.as_bytes());
}
m.push(0);
m.extend_from_slice(&[0, 1, 0, 1]); m
}
fn response(query: &[u8], ips: &[Ipv4Addr]) -> Vec<u8> {
let mut m = query.to_vec();
m[2] = 0x81;
m[3] = 0x80;
m[6] = (ips.len() >> 8) as u8;
m[7] = ips.len() as u8;
for ip in ips {
m.extend_from_slice(&[0xC0, 0x0C]); m.extend_from_slice(&[0, 1, 0, 1]); m.extend_from_slice(&300u32.to_be_bytes());
m.extend_from_slice(&4u16.to_be_bytes());
m.extend_from_slice(&ip.octets());
}
m
}
async fn mock_upstream(ips: Vec<Ipv4Addr>) -> SocketAddr {
let sock = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
let addr = sock.local_addr().unwrap();
tokio::spawn(async move {
let mut buf = vec![0u8; MAX_DNS_MSG];
loop {
let (n, from) = sock.recv_from(&mut buf).await.unwrap();
let resp = response(&buf[..n], &ips);
sock.send_to(&resp, from).await.unwrap();
}
});
addr
}
#[tokio::test]
async fn gfwlist_tunnels_only_matched_domains() {
let local = mock_upstream(vec![Ipv4Addr::new(10, 0, 0, 1)]).await;
let remote = mock_upstream(vec![Ipv4Addr::new(93, 184, 216, 34)]).await;
let sink = Arc::new(VecSink::default());
let r = Resolver::new(
Mode::GfwList,
GfwList::from_lines(["blocked.com"]),
ChnRoute::default(),
local,
remote,
Duration::from_secs(2),
sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let resp = r.resolve(&query("www.blocked.com")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![Ipv4Addr::new(93, 184, 216, 34)]);
let resp = r.resolve(&query("safe.cn")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![Ipv4Addr::new(10, 0, 0, 1)]);
assert_eq!(sink.ips(), vec![Ipv4Addr::new(93, 184, 216, 34)]);
}
#[tokio::test]
async fn chinadns_routes_by_china_membership() {
let china_ip = Ipv4Addr::new(114, 114, 114, 114);
let poison_ip = Ipv4Addr::new(8, 7, 6, 5);
let real_ip = Ipv4Addr::new(93, 184, 216, 34);
let chnroute = ChnRoute::from_lines(["114.114.114.0/24"]);
let sink = Arc::new(VecSink::default());
let r_dom = Resolver::new(
Mode::ChinaDns,
GfwList::default(),
chnroute.clone(),
mock_upstream(vec![china_ip]).await,
mock_upstream(vec![real_ip]).await,
Duration::from_secs(2),
sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let resp = r_dom.resolve(&query("baidu.cn")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![china_ip]);
assert!(sink.ips().is_empty());
let r_blk = Resolver::new(
Mode::ChinaDns,
GfwList::default(),
chnroute,
mock_upstream(vec![poison_ip]).await,
mock_upstream(vec![real_ip]).await,
Duration::from_secs(2),
sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let resp = r_blk.resolve(&query("blocked.com")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![real_ip]);
assert_eq!(sink.ips(), vec![real_ip]);
}
#[tokio::test]
async fn chinadns_gfwlist_forces_tunnel_over_china_answer() {
let china_ip = Ipv4Addr::new(114, 114, 114, 114);
let real_ip = Ipv4Addr::new(93, 184, 216, 34);
let chnroute = ChnRoute::from_lines(["114.114.114.0/24"]);
let sink = Arc::new(VecSink::default());
let r = Resolver::new(
Mode::ChinaDns,
GfwList::from_lines(["blocked.com"]),
chnroute,
mock_upstream(vec![china_ip]).await, mock_upstream(vec![real_ip]).await, Duration::from_secs(2),
sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let resp = r.resolve(&query("www.blocked.com")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![real_ip]);
assert_eq!(sink.ips(), vec![real_ip]);
let resp = r.resolve(&query("safe.cn")).await.unwrap();
assert_eq!(dns::a_records(&resp), vec![china_ip]);
assert_eq!(sink.ips(), vec![real_ip]); }
async fn mock_counting() -> SocketAddr {
let sock = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
let addr = sock.local_addr().unwrap();
tokio::spawn(async move {
let mut buf = vec![0u8; MAX_DNS_MSG];
let mut n: u8 = 0;
loop {
let (len, from) = sock.recv_from(&mut buf).await.unwrap();
n += 1;
let resp = response(&buf[..len], &[Ipv4Addr::new(10, 0, 0, n)]);
sock.send_to(&resp, from).await.unwrap();
}
});
addr
}
#[tokio::test]
async fn second_lookup_is_served_from_cache() {
let remote = mock_counting().await;
let sink = Arc::new(VecSink::default());
let r = Resolver::new(
Mode::GfwList,
GfwList::from_lines(["blocked.com"]),
ChnRoute::default(),
mock_upstream(vec![Ipv4Addr::new(10, 0, 0, 1)]).await,
remote,
Duration::from_secs(2),
sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let first = r.resolve(&query("www.blocked.com")).await.unwrap();
assert_eq!(dns::a_records(&first), vec![Ipv4Addr::new(10, 0, 0, 1)]);
let second = r.resolve(&query("www.blocked.com")).await.unwrap();
assert_eq!(dns::a_records(&second), vec![Ipv4Addr::new(10, 0, 0, 1)]);
}
#[tokio::test]
async fn chinadns_does_not_wait_for_remote_on_china_domain() {
let china = Ipv4Addr::new(114, 114, 114, 114);
let local = mock_upstream(vec![china]).await;
let dead: SocketAddr = "192.0.2.1:53".parse().unwrap();
let sink = Arc::new(VecSink::default());
let r = Resolver::new(
Mode::ChinaDns,
GfwList::default(),
ChnRoute::from_lines(["114.114.114.0/24"]),
local,
dead,
Duration::from_secs(30), sink.clone(),
Arc::new(cache::DnsCache::new()),
);
let resp = tokio::time::timeout(Duration::from_secs(2), r.resolve(&query("baidu.cn")))
.await
.expect("resolved without waiting for the remote")
.expect("got an answer");
assert_eq!(dns::a_records(&resp), vec![china]);
assert!(sink.ips().is_empty());
}
#[test]
fn decision_is_china_aware() {
let chn = ChnRoute::from_lines(["1.2.3.0/24"]);
assert_eq!(
chinadns_decision(&[Ipv4Addr::new(1, 2, 3, 4)], &chn),
Decision::Direct
);
assert_eq!(
chinadns_decision(&[Ipv4Addr::new(8, 8, 8, 8)], &chn),
Decision::Tunnel
);
assert_eq!(chinadns_decision(&[], &chn), Decision::Tunnel);
}
}