mirror/Easytier
1
0
Fork 0
mirror of https://github.com/EasyTier/EasyTier.git synced 2026-08-07 21:19:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

upgrade hickory-dns to 0.26.0

Browse Source 
fix zone test
This commit is contained in:
Luna Yao
2026-04-17 23:10:50 +02:00
parent 6001eef736
commit 7345acfe7c
17 changed files with 362 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
easytier/Cargo.toml
+5 -7
View File
@@ -237,14 +237,12 @@ http_req = { git = "https://github.com/EasyTier/http_req.git", default-features
] }
# for dns connector
hickory-resolver = "0.25.2"
hickory-proto = "0.25.2"
hickory-proto = "0.26.0"
hickory-net = { version = "0.26.0", features = ["serde"] }
hickory-resolver = "0.26.0"
# for magic dns
hickory-client = { version = "0.25.2", optional = true }
hickory-server = { version = "0.25.2", features = [
"resolver",
], optional = true }
hickory-server = { version = "0.26.0", features = ["resolver"], optional = true }
bon = "3.9.1"
derive_builder = "0.20.2"
@@ -403,7 +401,7 @@ jemalloc-prof = [
"jemalloc-sys/stats",
]
tracing = ["tokio/tracing", "dep:console-subscriber"]
magic-dns = ["dep:hickory-client", "dep:hickory-server"]
magic-dns = ["dep:hickory-server"]
faketcp = ["dep:flume"]
zstd = ["dep:zstd"]
# For Network Extension on macOS
easytier/src/common/dns.rs
+32 -26
View File
@@ -3,10 +3,11 @@ use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use anyhow::Context;
use hickory_proto::runtime::TokioRuntimeProvider;
use hickory_proto::xfer::Protocol;
use hickory_resolver::config::{LookupIpStrategy, NameServerConfig, ResolverConfig, ResolverOpts};
use hickory_resolver::name_server::{GenericConnector, TokioConnectionProvider};
use hickory_net::runtime::TokioRuntimeProvider;
use hickory_proto::rr::RData;
use hickory_resolver::config::{
ConnectionConfig, LookupIpStrategy, NameServerConfig, ResolverConfig, ResolverOpts,
};
use hickory_resolver::system_conf::read_system_conf;
use hickory_resolver::{Resolver, TokioResolver};
use once_cell::sync::Lazy;
@@ -15,36 +16,37 @@ use tokio::net::lookup_host;
use super::error::Error;
pub fn get_default_resolver_config() -> ResolverConfig {
let mut default_resolve_config = ResolverConfig::new();
let mut default_resolve_config = ResolverConfig::default();
default_resolve_config.add_name_server(NameServerConfig::new(
"223.5.5.5:53".parse().unwrap(),
Protocol::Udp,
"223.5.5.5".parse().unwrap(),
true,
vec![ConnectionConfig::udp()],
));
default_resolve_config.add_name_server(NameServerConfig::new(
"180.184.1.1:53".parse().unwrap(),
Protocol::Udp,
"180.184.1.1".parse().unwrap(),
true,
vec![ConnectionConfig::udp()],
));
default_resolve_config
}
pub static ALLOW_USE_SYSTEM_DNS_RESOLVER: Lazy<AtomicBool> = Lazy::new(|| AtomicBool::new(true));
pub static RESOLVER: Lazy<Arc<Resolver<GenericConnector<TokioRuntimeProvider>>>> =
Lazy::new(|| {
let system_cfg = read_system_conf();
let mut cfg = get_default_resolver_config();
let mut opt = ResolverOpts::default();
if let Ok(s) = system_cfg {
for ns in s.0.name_servers() {
cfg.add_name_server(ns.clone());
}
opt = s.1;
pub static RESOLVER: Lazy<Arc<Resolver<TokioRuntimeProvider>>> = Lazy::new(|| {
let system_cfg = read_system_conf();
let mut cfg = get_default_resolver_config();
let mut opt = ResolverOpts::default();
if let Ok(s) = system_cfg {
for ns in s.0.name_servers() {
cfg.add_name_server(ns.clone());
}
opt.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
let builder = TokioResolver::builder_with_config(cfg, TokioConnectionProvider::default())
.with_options(opt);
Arc::new(builder.build())
});
opt = s.1;
}
opt.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
let builder =
TokioResolver::builder_with_config(cfg, TokioRuntimeProvider::default()).with_options(opt);
Arc::new(builder.build().unwrap())
});
pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
let r = RESOLVER.clone();
@@ -53,12 +55,16 @@ pub async fn resolve_txt_record(domain_name: &str) -> Result<String, Error> {
.await
.with_context(|| format!("txt_lookup failed, domain_name: {}", domain_name))?;
let txt_record = response
let txt_data = response
.answers()
.iter()
.filter_map(|record| match record.data {
RData::TXT(ref txt) => Some(txt),
_ => None,
})
.next()
.with_context(|| format!("no txt record found, domain_name: {}", domain_name))?;
let txt_data = String::from_utf8_lossy(&txt_record.txt_data()[0]);
tracing::info!(?txt_data, ?domain_name, "get txt record");
Ok(txt_data.to_string())
easytier/src/connector/dns_connector.rs
+14 -9
View File
@@ -13,6 +13,7 @@ use crate::{
};
use anyhow::Context;
use dashmap::DashSet;
use hickory_proto::rr::RData;
use hickory_resolver::proto::rr::rdata::SRV;
use rand::{Rng as _, seq::SliceRandom};
use strum::VariantArray;
@@ -85,24 +86,21 @@ impl DnsTunnelConnector {
fn handle_one_srv_record(record: &SRV, protocol: IpScheme) -> Result<(url::Url, u64), Error> {
// port must be non-zero
if record.port() == 0 {
if record.port == 0 {
return Err(anyhow::anyhow!("port must be non-zero").into());
}
let connector_dst = record.target().to_utf8();
let dst_url = format!("{}://{}:{}", protocol, connector_dst, record.port());
let connector_dst = record.target.to_utf8();
let dst_url = format!("{}://{}:{}", protocol, connector_dst, record.port);
Ok((
dst_url.parse().with_context(|| {
format!(
"parse dst_url failed, protocol: {}, connector_dst: {}, port: {}, dst_url: {}",
protocol,
connector_dst,
record.port(),
dst_url
protocol, connector_dst, record.port, dst_url
)
})?,
record.priority() as _,
record.priority as _,
))
}
@@ -129,7 +127,14 @@ impl DnsTunnelConnector {
format!("srv_lookup failed, srv_domain: {}", srv_domain)
})?;
tracing::info!(?response, ?srv_domain, "srv_lookup response");
for record in response.iter() {
for record in response
.answers()
.iter()
.filter_map(|record| match record.data {
RData::SRV(ref srv) => Some(srv),
_ => None,
})
{
let parsed_record = Self::handle_one_srv_record(record, **protocol);
tracing::info!(?parsed_record, ?srv_domain, "parsed_record");
if let Err(e) = &parsed_record {
easytier/src/dns/config/dns.rs
+1 -1
View File
@@ -6,8 +6,8 @@ use crate::dns::utils::addr::NameServerAddrGroup;
use crate::dns::utils::parse;
use crate::proto::dns::GetExportConfigResponse;
use derivative::Derivative;
use hickory_net::xfer::Protocol;
use hickory_proto::rr::{LowerName, Name};
use hickory_proto::xfer::Protocol;
use serde::{Deserialize, Deserializer, Serialize};
use std::collections::HashMap;
use std::iter;
easytier/src/dns/config/mod.rs
+1 -10
View File
@@ -1,6 +1,6 @@
use crate::dns::utils::addr::NameServerAddr;
use hickory_net::xfer::Protocol;
use hickory_proto::rr::LowerName;
use hickory_proto::xfer::Protocol;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::str::FromStr;
use std::sync::LazyLock;
@@ -19,15 +19,6 @@ pub const DNS_DEFAULT_ADDRESS: NameServerAddr = NameServerAddr {
addr: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(100, 100, 100, 101), 53)),
};
pub static DNS_SUPPORTED_PROTOCOLS: [Protocol; 2] = [
Protocol::Udp,
Protocol::Tcp,
// Protocol::Tls,
// Protocol::Https,
// Protocol::Quic,
// Protocol::H3,
];
pub static DNS_SERVER_RPC_ADDR: LazyLock<Url> =
LazyLock::new(|| Url::parse("tcp://127.0.0.1:49813").unwrap());
easytier/src/dns/node_mgr.rs
+17 -14
View File
@@ -8,7 +8,7 @@ use crate::proto::rpc_types;
use crate::proto::rpc_types::controller::BaseController;
use crate::proto::utils::TransientDigest;
use anyhow::Error;
use hickory_server::authority::Catalog;
use hickory_server::zone_handler::Catalog;
use itertools::Itertools;
use moka::future::Cache;
use std::collections::HashSet;
@@ -73,7 +73,7 @@ impl DnsNodeMgr {
groups
.into_iter()
.fold(Catalog::new(), |mut catalog, (origin, zones)| {
catalog.upsert(origin.clone(), zones.iter_authorities().collect());
catalog.upsert(origin.clone(), zones.iter_zone_handlers().collect());
catalog
})
}
@@ -172,7 +172,7 @@ mod tests {
};
use crate::dns::utils::response::ResponseHandle;
use hickory_proto::op::{Message, ResponseCode};
use hickory_proto::rr::{RData, RecordType, rdata};
use hickory_proto::rr::{RData, RecordType};
use std::net::Ipv4Addr;
use tokio::time::{Duration, sleep};
@@ -203,9 +203,12 @@ mod tests {
async fn lookup_a_record(mgr: &DnsNodeMgr, name: &str) -> anyhow::Result<Message> {
let request = new_request(name, RecordType::A)?;
let response = ResponseHandle::new(512);
let info = mgr.catalog().lookup(&request, None, response.clone()).await;
let info = mgr
.catalog()
.lookup(&request, None, 0, response.clone())
.await;
assert_eq!(info.response_code(), ResponseCode::NoError);
assert_eq!(info.response_code, ResponseCode::NoError);
let response = response.into_inner().expect("response should exist");
Message::from_vec(&response).map_err(Into::into)
@@ -224,10 +227,10 @@ mod tests {
let _ = send_heartbeat(&mgr, heartbeat_with_snapshot(id, snapshot)).await;
let message = lookup_a_record(&mgr, "catalog.test.").await?;
assert!(message.answers().iter().any(|record| {
assert!(message.answers.iter().any(|record| {
matches!(
record.data(),
RData::A(addr) if *addr == rdata::a::A(Ipv4Addr::new(10, 20, 30, 40))
record.data,
RData::A(addr) if *addr == Ipv4Addr::new(10, 20, 30, 40)
)
}));
@@ -255,16 +258,16 @@ mod tests {
let message_a = lookup_a_record(&mgr, "node-a.test.").await?;
let message_b = lookup_a_record(&mgr, "node-b.test.").await?;
assert!(message_a.answers().iter().any(|record| {
assert!(message_a.answers.iter().any(|record| {
matches!(
record.data(),
RData::A(addr) if *addr == rdata::a::A(Ipv4Addr::new(10, 11, 12, 13))
record.data,
RData::A(addr) if *addr == Ipv4Addr::new(10, 11, 12, 13)
)
}));
assert!(message_b.answers().iter().any(|record| {
assert!(message_b.answers.iter().any(|record| {
matches!(
record.data(),
RData::A(addr) if *addr == rdata::a::A(Ipv4Addr::new(10, 21, 22, 23))
record.data,
RData::A(addr) if *addr == Ipv4Addr::new(10, 21, 22, 23)
)
}));
easytier/src/dns/plan.md
+1 -1
View File
@@ -95,7 +95,7 @@ forwarders = [
3. - [x] (`moka::Cache`) 持续检查是否有过期（丢失心跳）的 DnsNode，需要把这些 DnsNode 提供的所有配置清除
4. - [x] 每次更新 zone 时自动添加 root zone
- [x] (`Zone::system`) 并把它的 forwarder 设置为系统 DNS
5. - [x] 使用 snapshot 更新 zone。不用合并同名 zone，直接用 Zone 结构体提供的 ChainedAuthority 按顺序插入 Catalog 就行，不过注意要先插入 MemoryAuthority，这些都是 records，后插入 ForwardAuthority，这都是 forwarders
5. - [x] 使用 snapshot 更新 zone。不用合并同名 zone，直接用 Zone 结构体提供的 ChainedZoneHandler 按顺序插入 Catalog 就行，不过注意要先插入 MemoryZoneHandler，这些都是 records，后插入 ForwardZoneHandler，这都是 forwarders
6. - [x] 更新 zone 的时候自动去掉 forwarder 中导致回环的那些，就是把 addresses 和 listeners 去掉（root zone 也需要这个逻辑）
7. - [x] 内部接口，控制 DnsServer 是否 bind 到某些 socket（也就是配置中的 listeners）
8. - [x] Listeners 绑定失败打印日志（失败一个打印一次然后就跳过），即便这时 addresses 为空也不要停机。（否则释放 socket 绑定后会有 instance 抢占 socket 试图启动 server，然后就死循环）
easytier/src/dns/readme.md
+14 -14
View File
@@ -35,7 +35,7 @@
- **服务层（数据面）**（`server.rs` + `node_mgr.rs` + `zone.rs`）
- `DnsServer`：真正处理 DNS 请求、维护监听 socket、管理 hijack addresses。
- `DnsNodeMgr`：服务端的快照管理器，接收 Node 心跳，维护节点 TTL 与 dirty 状态。
- `Zone`/`ZoneGroup`：把 records + forwarders 变成 Hickory `Authority` 并装配 `Catalog`。
- `Zone`/`ZoneGroup`：把 records + forwarders 变成 Hickory `ZoneHandler` 并装配 `Catalog`。
- **系统集成层**（`system/*`）
- 将当前 DNS 配置下发到 OS（`SystemConfigurator` 抽象）。
@@ -52,9 +52,9 @@
- **`ZoneData` / `Zone`**（`proto/dns.proto` + `zone.rs`）
- `ZoneData` 是网络传输模型（protobuf），含 `id/origin/ttl/records/forwarders`。
- `Zone` 是运行期模型：
- `records -> InMemoryAuthority`
- `forwarders -> ForwardAuthority`
- 同 origin 可链式共存（ChainedAuthority 语义）。
- `records -> InMemoryZoneHandler`
- `forwarders -> ForwardZoneHandler`
- 同 origin 可链式共存（ChainedZoneHandler 语义）。
- **`DnsSnapshot`**（`proto/dns.proto`）
- Node 发给 Server 的完整状态：`zones + addresses + listeners`。
@@ -324,7 +324,7 @@ Node 发送 heartbeat 时：
2. 追加 `Zone::system()` 作为 root zone。
3. 收集本地所有 `addresses + listeners` 形成 `local` 集合。
4. 遍历每个 zone 的 forwarders，剔除命中 `local` 的 nameserver（避免显式回环）。
5. 以 `origin -> authorities[]` 方式 `upsert` 到 Hickory `Catalog`。
5. 以 `origin -> zone_handlers[]` 方式 `upsert` 到 Hickory `Catalog`。
### 10.3 `DnsServer::run`：三路热重载
@@ -380,12 +380,12 @@ Node 发送 heartbeat 时：
2. 用 Hickory `Parser` 解析 zone 文本（origin + RR）。
3. 把 `forwarders` URL 转成 `NameServerAddr`，为空则 `forward=None`。
这确保网络收到的 `ZoneData` 能直接映射成可执行 authority。
这确保网络收到的 `ZoneData` 能直接映射成可执行 zone_handler。
### 11.3 Authority 构建策略
### 11.3 ZoneHandler 构建策略
- `create_memory_authority()`：仅当 records 非空时创建 `InMemoryAuthority`。
- `create_forward_authority()`：仅当 forward 非空时创建 `ForwardAuthority`。
- `create_memory_zone_handler()`：仅当 records 非空时创建 `InMemoryZoneHandler`。
- `create_forward_zone_handler()`：仅当 forward 非空时创建 `ForwardZoneHandler`。
因此允许 3 种 zone 形态：
@@ -396,10 +396,10 @@ Node 发送 heartbeat 时：
### 11.4 `ZoneGroup` 与同源链式行为
- `ZoneGroup::into_groups()` 按 `origin` 分组。
- `iter_authorities()` 对每个 zone 按顺序产出：先 memory，再 forward。
- `DnsNodeMgr::catalog()` 把同 origin 的多个 zone authority 以数组形式 `upsert`。
- `iter_zone_handlers()` 对每个 zone 按顺序产出：先 memory，再 forward。
- `DnsNodeMgr::catalog()` 把同 origin 的多个 zone zone_handler 以数组形式 `upsert`。
结果是同 origin 下可自然形成 ChainedAuthority，不做“硬合并单 Zone”，与 `plan.md` 一致。
结果是同 origin 下可自然形成 ChainedZoneHandler，不做“硬合并单 Zone”，与 `plan.md` 一致。
### 11.5 `Zone::system()` 的作用边界
@@ -473,7 +473,7 @@ Node 发送 heartbeat 时：
- `dns/tests.rs`：测试基建与辅助函数（构造环境、启动 `DnsNode`、DNS 查询断言工具）。
- `dns/server.rs`：数据面与 server 行为主测试集。
- `dns/node_mgr.rs`：聚合 catalog 的基本可用性测试。
- `dns/zone.rs`：配置解析、记录转换、authority 装配测试。
- `dns/zone.rs`：配置解析、记录转换、zone_handler 装配测试。
> 说明：`system/*` 也有测试，但本轮文档按约定不展开。
@@ -504,7 +504,7 @@ Node 发送 heartbeat 时：
- TOML `DnsConfig` 解析。
- `ZoneConfig -> ZoneData -> Zone` 转换链。
- record 解析/TTL 基本行为。
- memory/forward authority 构建，以及通过 server 查询验证。
- memory/forward zone_handler 构建，以及通过 server 查询验证。
该测试更多是“模型与解析正确性”，不是策略执行链路完整验证。
easytier/src/dns/server.rs
+57 -60
View File
@@ -12,13 +12,12 @@ use crate::tunnel::packet_def::ZCPacket;
use crate::tunnel::tcp::TcpTunnelListener;
use crate::utils::task::AsyncRuntime;
use derivative::Derivative;
use hickory_proto::serialize::binary::BinDecodable;
use hickory_proto::xfer::Protocol;
use hickory_server::authority::MessageRequest;
use hickory_net::runtime::{Time, TokioTime};
use hickory_net::xfer::Protocol;
use hickory_server::{
ServerFuture,
authority::Catalog,
Server,
server::{Request, RequestHandler, ResponseHandler, ResponseInfo},
zone_handler::Catalog,
};
use parking_lot::RwLock;
use pnet::packet::icmp::{IcmpTypes, MutableIcmpPacket};
@@ -55,7 +54,7 @@ impl DynamicCatalog {
#[async_trait::async_trait]
impl RequestHandler for DynamicCatalog {
async fn handle_request<R: ResponseHandler>(
async fn handle_request<R: ResponseHandler, T: Time>(
&self,
request: &Request,
response_handle: R,
@@ -63,7 +62,7 @@ impl RequestHandler for DynamicCatalog {
self.inner
.read()
.await
.handle_request(request, response_handle)
.handle_request::<_, T>(request, response_handle)
.await
}
}
@@ -87,6 +86,7 @@ pub struct DnsServer {
}
const DNS_SERVER_LISTENER_TCP_TIMEOUT: Duration = Duration::from_secs(5);
const DNS_SERVER_LISTENER_TCP_BUFFER_SIZE: usize = 32;
impl DnsServer {
pub fn new(
@@ -187,15 +187,18 @@ impl DnsServer {
let runtime = runtime.get_or_insert_default();
let mut server = ServerFuture::new(self.catalog.clone());
let mut server = Server::new(self.catalog.clone());
for listener in &listeners {
let addr = listener.addr;
tracing::info!(?addr, "binding listener");
if let Err(error) = match listener.protocol {
Protocol::Tcp => bind()
.addr(addr)
.call()
.map(|s| server.register_listener(s, DNS_SERVER_LISTENER_TCP_TIMEOUT)),
Protocol::Tcp => bind().addr(addr).call().map(|s| {
server.register_listener(
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!(),
} {
@@ -395,11 +398,12 @@ impl DnsServer {
(
src_port,
dst_port,
Request::new(
MessageRequest::from_bytes(request_payload).ok()?,
Request::from_bytes(
request_payload.to_vec(),
SocketAddr::from(SocketAddrV4::new(src_ip, src_port)),
Protocol::Udp,
),
)
.ok()?,
request_payload.len(),
)
};
@@ -412,7 +416,7 @@ impl DnsServer {
let response = ResponseHandle::new(512);
self.catalog
.handle_request(&request, response.clone())
.handle_request::<_, TokioTime>(&request, response.clone())
.await;
response.into_inner()?
@@ -480,15 +484,15 @@ mod tests {
use crate::peers::tests::create_mock_peer_manager;
use crate::proto::dns::DnsNodeMgrRpc;
use crate::proto::rpc_types::controller::BaseController;
use hickory_client::client::{Client, ClientHandle};
use hickory_net::client::{Client, ClientHandle};
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::runtime::TokioRuntimeProvider;
use hickory_proto::serialize::binary::BinEncodable;
use hickory_proto::udp::UdpClientStream;
use hickory_server::authority::Catalog;
use hickory_server::authority::ZoneType;
use hickory_server::store::in_memory::InMemoryAuthority;
use hickory_server::store::in_memory::InMemoryZoneHandler;
use hickory_server::zone_handler::ZoneType;
use hickory_server::zone_handler::{AxfrPolicy, Catalog};
use pnet::packet::icmp::{IcmpPacket, IcmpTypes, MutableIcmpPacket};
use pnet::packet::ip::IpNextHeaderProtocols;
use pnet::packet::ipv4::{Ipv4Packet, MutableIpv4Packet};
@@ -504,7 +508,11 @@ mod tests {
/// Build a `Catalog` containing a single A record: `test.example.com -> 1.2.3.4`.
fn build_test_catalog() -> Catalog {
let origin = Name::from_str("example.com.").unwrap();
let mut authority = InMemoryAuthority::empty(origin.clone(), ZoneType::Primary, false);
let mut zone_handler = InMemoryZoneHandler::<TokioRuntimeProvider>::empty(
origin.clone(),
ZoneType::Primary,
AxfrPolicy::default(),
);
let record = Record::from_rdata(
Name::from_str("test.example.com.").unwrap(),
@@ -512,16 +520,18 @@ mod tests {
RData::A(rdata::a::A(Ipv4Addr::new(1, 2, 3, 4))),
);
let rr_key =
hickory_proto::rr::RrKey::new(record.name().clone().into(), record.record_type());
hickory_proto::rr::RrKey::new(record.name.clone().into(), record.record_type());
let mut rr_set =
hickory_proto::rr::RecordSet::new(record.name().clone(), record.record_type(), 0);
hickory_proto::rr::RecordSet::new(record.name.clone(), record.record_type(), 0);
rr_set.insert(record, 0);
authority.records_get_mut().insert(rr_key, Arc::new(rr_set));
zone_handler
.records_get_mut()
.insert(rr_key, Arc::new(rr_set));
let mut catalog = Catalog::new();
catalog.upsert(
origin.into(),
vec![Arc::new(authority) as Arc<dyn hickory_server::authority::AuthorityObject>],
vec![Arc::new(zone_handler) as Arc<dyn hickory_server::zone_handler::ZoneHandler>],
);
catalog
}
@@ -578,11 +588,8 @@ mod tests {
/// Build a minimal DNS query message for `name` and encode it to bytes.
fn build_dns_query_bytes(name: &str) -> Vec<u8> {
let mut msg = Message::new();
msg.set_id(0x1234);
msg.set_message_type(MessageType::Query);
msg.set_op_code(OpCode::Query);
msg.set_recursion_desired(true);
let mut msg = Message::new(0x1234, MessageType::Query, OpCode::Query);
msg.metadata.recursion_desired = true;
let mut query = Query::new();
query.set_name(Name::from_str(name).unwrap());
query.set_query_type(RecordType::A);
@@ -796,17 +803,17 @@ mod tests {
);
let dns_reply = Message::from_vec(udp_reply.payload()).unwrap();
assert_eq!(dns_reply.id(), 0x1234);
assert_eq!(dns_reply.id, 0x1234);
assert!(
!dns_reply.answers().is_empty(),
!dns_reply.answers.is_empty(),
"DNS reply should contain answers"
);
let answer = &dns_reply.answers()[0];
if let RData::A(a) = answer.data() {
let answer = &dns_reply.answers[0];
if let RData::A(a) = answer.data {
assert_eq!(a.0, Ipv4Addr::new(1, 2, 3, 4));
} else {
panic!("expected A record in answer, got {:?}", answer.data());
panic!("expected A record in answer, got {:?}", answer.data);
}
}
@@ -814,7 +821,7 @@ mod tests {
/// send a query with a `hickory_client`, and verify the response.
#[tokio::test]
async fn should_resolve_record_via_real_udp_listener() {
use hickory_server::ServerFuture;
use hickory_server::Server;
use tokio::net::UdpSocket;
use tokio::time::timeout;
@@ -825,7 +832,7 @@ mod tests {
let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let addr = socket.local_addr().unwrap();
let mut server = ServerFuture::new(catalog);
let mut server = Server::new(catalog);
server.register_socket(socket);
let shutdown_token = server.shutdown_token().clone();
@@ -834,15 +841,10 @@ mod tests {
});
// Send a real DNS query using hickory_client.
let conn = UdpClientStream::builder(addr, TokioRuntimeProvider::default()).build();
let (mut client, bg) = timeout(Duration::from_secs(2), Client::connect(conn))
.await
.expect("client connect timeout")
.expect("client connect failed");
let stream = UdpClientStream::builder(addr, TokioRuntimeProvider::default()).build();
let (mut client, bg) = Client::<TokioRuntimeProvider>::from_sender(stream);
tokio::spawn(async move {
bg.await.ok();
});
tokio::spawn(bg);
let response = timeout(
Duration::from_secs(2),
@@ -856,12 +858,12 @@ mod tests {
.expect("query timeout")
.expect("query failed");
assert!(!response.answers().is_empty(), "should get answers");
let a_record = &response.answers()[0];
if let RData::A(a) = a_record.data() {
assert!(!response.answers.is_empty(), "should get answers");
let a_record = &response.answers[0];
if let RData::A(a) = a_record.data {
assert_eq!(a.0, Ipv4Addr::new(1, 2, 3, 4));
} else {
panic!("expected A record, got {:?}", a_record.data());
panic!("expected A record, got {:?}", a_record.data);
}
// Shutdown the server.
@@ -1028,14 +1030,9 @@ mod tests {
.await
.unwrap();
let conn = UdpClientStream::builder(good_addr, TokioRuntimeProvider::default()).build();
let (mut client, bg) = timeout(Duration::from_secs(2), Client::connect(conn))
.await
.expect("client connect timeout")
.expect("client connect failed");
tokio::spawn(async move {
let _ = bg.await;
});
let stream = UdpClientStream::builder(good_addr, TokioRuntimeProvider::default()).build();
let (mut client, bg) = Client::<TokioRuntimeProvider>::from_sender(stream);
tokio::spawn(bg);
let response = timeout(
Duration::from_secs(2),
@@ -1049,7 +1046,7 @@ mod tests {
.expect("query timeout")
.expect("query failed");
assert!(!response.answers().is_empty());
assert!(!response.answers.is_empty());
if let Some(runtime) = runtime.take() {
let _ = runtime.stop(None).await;
easytier/src/dns/tests.rs
+41 -53
View File
@@ -20,15 +20,14 @@ use crate::peers::tests::{connect_peer_manager, wait_route_appear};
use crate::proto::common::{NatType, Url};
use crate::proto::dns::{DnsSnapshot, HeartbeatRequest, ZoneData};
use cidr::Ipv4Inet;
use hickory_client::client::{Client, ClientHandle as _};
use hickory_net::client::{Client, ClientHandle};
use hickory_net::runtime::TokioRuntimeProvider;
use hickory_net::udp::UdpClientStream;
use hickory_net::xfer::Protocol;
use hickory_proto::op::{Message, MessageType, OpCode, Query};
use hickory_proto::rr;
use hickory_proto::rr::{Name, RecordType};
use hickory_proto::runtime::TokioRuntimeProvider;
use hickory_proto::serialize::binary::{BinDecodable, BinEncodable, BinEncoder};
use hickory_proto::udp::UdpClientStream;
use hickory_proto::xfer::Protocol;
use hickory_server::authority::MessageRequest;
use hickory_proto::rr::{DNSClass, Name, RData, RecordType};
use hickory_proto::serialize::binary::{BinEncodable, BinEncoder};
use hickory_server::server::Request;
use tokio::sync::Notify;
use uuid::Uuid;
@@ -159,17 +158,17 @@ pub async fn check_dns_record(fake_ip: &Ipv4Addr, domain: &str, expected_ip: &st
pub async fn check_dns_record_at(server_addr: SocketAddr, domain: &str, expected_ip: &str) {
let expected = expected_ip.parse::<Ipv4Addr>().unwrap();
let name = rr::Name::from_str(domain).unwrap();
let name = Name::from_str(domain).unwrap();
let deadline = Instant::now() + Duration::from_secs(30);
loop {
let stream = UdpClientStream::builder(server_addr, TokioRuntimeProvider::default()).build();
let (mut client, background) = Client::connect(stream).await.unwrap();
let (mut client, background) = Client::<TokioRuntimeProvider>::from_sender(stream);
let background_task = tokio::spawn(background);
let query_result = tokio::time::timeout(
Duration::from_secs(2),
client.query(name.clone(), rr::DNSClass::IN, rr::RecordType::A),
client.query(name.clone(), DNSClass::IN, RecordType::A),
)
.await;
@@ -178,13 +177,13 @@ pub async fn check_dns_record_at(server_addr: SocketAddr, domain: &str, expected
let attempt_err = match query_result {
Ok(Ok(response)) => {
if response.answers().len() == 1
&& let Some(resp) = response.answers().first()
&& resp.clone().into_parts().rdata.into_a().unwrap().0 == expected
if response.answers.len() == 1
&& let Some(resp) = response.answers.first()
&& matches!(resp.clone().data, RData::A(a) if a.0 == expected)
{
return;
}
format!("unexpected response: {:?}", response.answers())
format!("unexpected response: {:?}", response.answers)
}
Ok(Err(e)) => {
format!("DNS query failed for domain '{domain}': {e}")
@@ -213,15 +212,11 @@ pub async fn check_dns_record_missing_at(server_addr: SocketAddr, domain: &str)
loop {
let stream = UdpClientStream::builder(server_addr, TokioRuntimeProvider::default()).build();
let (mut client, background) = Client::connect(stream).await.unwrap();
let (mut client, background) = Client::<TokioRuntimeProvider>::from_sender(stream);
let background_task = tokio::spawn(background);
let query_result = tokio::time::timeout(
Duration::from_secs(2),
client.query(
rr::Name::from_str(domain).unwrap(),
rr::DNSClass::IN,
rr::RecordType::A,
),
client.query(Name::from_str(domain).unwrap(), DNSClass::IN, RecordType::A),
)
.await;
background_task.abort();
@@ -229,10 +224,10 @@ pub async fn check_dns_record_missing_at(server_addr: SocketAddr, domain: &str)
let attempt_err = match query_result {
Ok(Ok(response)) => {
if response.answers().is_empty() {
if response.answers.is_empty() {
return;
}
format!("unexpected non-empty response: {:?}", response.answers())
format!("unexpected non-empty response: {:?}", response.answers)
}
Ok(Err(e)) => {
format!("DNS query for missing record failed for domain '{domain}': {e}")
@@ -253,22 +248,19 @@ pub async fn check_dns_record_missing_at(server_addr: SocketAddr, domain: &str)
}
pub fn new_request(name: &str, rtype: RecordType) -> anyhow::Result<Request> {
let mut query = Message::new();
query.set_id(0);
query.set_message_type(MessageType::Query);
query.set_op_code(OpCode::Query);
query.set_recursion_desired(true);
query.add_query(Query::query(Name::from_ascii(name)?, rtype));
let mut msg = Message::new(0, MessageType::Query, OpCode::Query);
msg.metadata.recursion_desired = true;
msg.add_query(Query::query(Name::from_ascii(name)?, rtype));
let mut request = Vec::new();
let mut encoder = BinEncoder::new(&mut request);
query.emit(&mut encoder)?;
msg.emit(&mut encoder)?;
Ok(Request::new(
MessageRequest::from_bytes(&request)?,
Ok(Request::from_bytes(
request,
SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0).into(),
Protocol::Udp,
))
)?)
}
async fn wait_route_disappear(peer_mgr: Arc<PeerManager>, target_peer_id: u32) {
@@ -313,32 +305,28 @@ async fn check_dns_unavailable_at(server_addr: SocketAddr, domain: &str) {
loop {
let stream = UdpClientStream::builder(server_addr, TokioRuntimeProvider::default()).build();
let connect = Client::connect(stream).await;
let (mut client, background) = Client::<TokioRuntimeProvider>::from_sender(stream);
if let Ok((mut client, background)) = connect {
let background_task = tokio::spawn(background);
let query_result = tokio::time::timeout(
Duration::from_secs(1),
client.query(name.clone(), rr::DNSClass::IN, rr::RecordType::A),
)
.await;
let background_task = tokio::spawn(background);
let query_result = tokio::time::timeout(
Duration::from_secs(1),
client.query(name.clone(), DNSClass::IN, RecordType::A),
)
.await;
background_task.abort();
let _ = background_task.await;
background_task.abort();
let _ = background_task.await;
match query_result {
Ok(Ok(response)) if !response.answers().is_empty() => {
if Instant::now() >= deadline {
panic!(
"DNS endpoint {server_addr} still answered for '{domain}': {:?}",
response.answers()
);
}
match query_result {
Ok(Ok(response)) if !response.answers.is_empty() => {
if Instant::now() >= deadline {
panic!(
"DNS endpoint {server_addr} still answered for '{domain}': {:?}",
response.answers
);
}
_ => return,
}
} else {
return;
_ => return,
}
tokio::time::sleep(Duration::from_millis(200)).await;
easytier/src/dns/utils/addr.rs
+28 -43
View File
@@ -1,14 +1,15 @@
use crate::dns::config::DNS_SUPPORTED_PROTOCOLS;
use crate::proto;
use crate::proto::utils::RepeatedMessageModel;
use anyhow::{Error, anyhow};
use hickory_proto::xfer::Protocol;
use hickory_resolver::config::{NameServerConfig, NameServerConfigGroup};
use hickory_net::xfer::Protocol;
use hickory_resolver::config::{ConnectionConfig, NameServerConfig};
use serde::de::IntoDeserializer;
use serde::{Deserialize, de};
use serde_with::{DeserializeFromStr, SerializeDisplay};
use std::fmt::{Display, Formatter};
use std::net::{IpAddr, SocketAddr};
use std::str::FromStr;
use url::Url;
use url::{Host, Url};
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, SerializeDisplay, DeserializeFromStr)]
pub struct NameServerAddr {
@@ -18,15 +19,22 @@ pub struct NameServerAddr {
impl From<NameServerAddr> for NameServerConfig {
fn from(value: NameServerAddr) -> Self {
Self::new(value.addr, value.protocol)
let mut config = match value.protocol {
Protocol::Udp => ConnectionConfig::udp(),
Protocol::Tcp => ConnectionConfig::tcp(),
_ => unimplemented!(),
};
config.port = value.addr.port();
Self::new(value.addr.ip(), true, vec![config])
}
}
impl From<&NameServerConfig> for NameServerAddr {
fn from(value: &NameServerConfig) -> Self {
let connection = value.connections.first().unwrap();
Self {
protocol: value.protocol,
addr: value.socket_addr,
protocol: connection.protocol.to_protocol(),
addr: SocketAddr::new(value.ip, connection.port),
}
}
}
@@ -62,29 +70,23 @@ impl TryFrom<&Url> for NameServerAddr {
type Error = Error;
fn try_from(value: &Url) -> Result<Self, Self::Error> {
let scheme = value.scheme();
let protocol = *DNS_SUPPORTED_PROTOCOLS
.iter()
.find(|p| p.to_string() == scheme)
.ok_or(anyhow!("unsupported scheme: {}", scheme))?;
let addr = value.host_str().ok_or(anyhow!("host not found"))?;
let addr = addr
.trim_start_matches('[')
.trim_end_matches(']')
.parse::<IpAddr>()
.map_err(|e| anyhow!("invalid ip address '{}': {}", addr, e))?;
let port = if let Some(port) = value.port() {
port
} else {
match protocol {
Protocol::Udp | Protocol::Tcp => 53,
_ => return Err(anyhow!("port not found")),
let protocol = Protocol::deserialize(value.scheme().into_deserializer()).map_err(
|e: de::value::Error| anyhow!("invalid protocol '{}': {}", value.scheme(), e),
)?;
let port = value
.port()
.or_else(|| matches!(protocol, Protocol::Udp | Protocol::Tcp).then_some(53))
.ok_or_else(|| anyhow!("port not found"))?;
let ip = match value.host().ok_or(anyhow!("host not found"))? {
Host::Domain(_) => {
return Err(anyhow!("unsupported host: {}", value.host_str().unwrap()));
}
Host::Ipv4(ip) => ip.into(),
Host::Ipv6(ip) => ip.into(),
};
Ok(Self {
protocol,
addr: SocketAddr::new(addr, port),
addr: SocketAddr::new(ip, port),
})
}
}
@@ -125,20 +127,3 @@ impl Display for NameServerAddr {
}
pub type NameServerAddrGroup = RepeatedMessageModel<NameServerAddr>;
impl From<NameServerAddrGroup> for NameServerConfigGroup {
fn from(value: NameServerAddrGroup) -> Self {
value.into_iter().map(Into::into).collect::<Vec<_>>().into()
}
}
impl From<NameServerConfigGroup> for NameServerAddrGroup {
fn from(value: NameServerConfigGroup) -> Self {
value
.into_inner()
.into_iter()
.map(Into::into)
.collect::<Vec<_>>()
.into()
}
}
easytier/src/dns/utils/authority.rs
-82
View File
@@ -1,82 +0,0 @@
use crate::utils::BoxExt;
use delegate::delegate;
use derive_more::{Deref, DerefMut, From};
use hickory_proto::rr::{LowerName, RecordType};
use hickory_server::authority::{
Authority, AuthorityObject, LookupControlFlow, LookupObject, LookupOptions, MessageRequest,
UpdateResult, ZoneType,
};
use hickory_server::server::RequestInfo;
use std::sync::Arc;
pub type ArcAuthority = Arc<dyn AuthorityObject>;
#[derive(From, Deref, DerefMut)]
pub struct ChainedAuthority<A>(A)
where
A: Authority,
A::Lookup: LookupObject + 'static;
#[async_trait::async_trait]
impl<A> Authority for ChainedAuthority<A>
where
A: Authority,
A::Lookup: LookupObject + 'static,
{
type Lookup = A::Lookup;
delegate! {
to self.0 {
fn zone_type(&self) -> ZoneType;
fn is_axfr_allowed(&self) -> bool;
fn origin(&self) -> &LowerName;
}
}
#[inline]
async fn update(&self, update: &MessageRequest) -> UpdateResult<bool> {
self.0.update(update).await
}
#[inline]
async fn lookup(
&self,
name: &LowerName,
rtype: RecordType,
lookup_options: LookupOptions,
) -> LookupControlFlow<Self::Lookup> {
self.0.lookup(name, rtype, lookup_options).await
}
#[inline]
async fn consult(
&self,
name: &LowerName,
rtype: RecordType,
lookup_options: LookupOptions,
last_result: LookupControlFlow<Box<dyn LookupObject>>,
) -> LookupControlFlow<Box<dyn LookupObject>> {
if let Some(Ok(l)) = last_result.map_result() {
LookupControlFlow::Break(Ok(l))
} else {
self.0
.lookup(name, rtype, lookup_options)
.await
.map(|l| l.boxed() as _)
}
}
#[inline]
async fn search(
&self,
request_info: RequestInfo<'_>,
lookup_options: LookupOptions,
) -> LookupControlFlow<Self::Lookup> {
self.0.search(request_info, lookup_options).await
}
#[inline]
async fn get_nsec_records(
&self,
name: &LowerName,
lookup_options: LookupOptions,
) -> LookupControlFlow<Self::Lookup> {
self.0.get_nsec_records(name, lookup_options).await
}
}
easytier/src/dns/utils/mod.rs
+1 -1
View File
@@ -2,9 +2,9 @@ use hickory_proto::rr::LowerName;
use idna::AsciiDenyList;
pub mod addr;
pub mod authority;
pub mod dirty;
pub mod response;
pub mod zone_handler;
pub fn sanitize(name: &str) -> String {
let dot = name.ends_with('.');
easytier/src/dns/utils/response.rs
+6 -6
View File
@@ -1,9 +1,9 @@
use hickory_net::NetError;
use hickory_proto::rr::Record;
use hickory_proto::serialize::binary::BinEncoder;
use hickory_server::authority::MessageResponse;
use hickory_server::server::{ResponseHandler, ResponseInfo};
use hickory_server::zone_handler::MessageResponse;
use parking_lot::Mutex;
use std::io;
use std::sync::Arc;
// ResponseWrapper for serializing DNS responses into a byte buffer.
@@ -41,11 +41,11 @@ impl ResponseHandler for ResponseHandle {
impl RecordIter<'r>,
impl RecordIter<'r>,
>,
) -> io::Result<ResponseInfo> {
let max_size = if let Some(edns) = response.get_edns() {
) -> Result<ResponseInfo, NetError> {
let max_size = if let Some(edns) = response.edns() {
edns.max_payload()
} else {
hickory_proto::udp::MAX_RECEIVE_BUFFER_SIZE as u16
hickory_net::udp::MAX_RECEIVE_BUFFER_SIZE as u16
};
let mut inner = self.inner.lock();
@@ -54,6 +54,6 @@ impl ResponseHandler for ResponseHandle {
encoder.set_max_size(max_size);
response
.destructive_emit(&mut encoder)
.map_err(io::Error::other)
.map_err(NetError::Proto)
}
}
easytier/src/dns/utils/zone_handler.rs
+85
View File
@@ -0,0 +1,85 @@
use delegate::delegate;
use derive_more::{Deref, DerefMut, From};
use hickory_proto::op::ResponseCode;
use hickory_proto::rr::{LowerName, RecordType, TSigResponseContext};
use hickory_server::server::{Request, RequestInfo};
use hickory_server::zone_handler::{
AuthLookup, AxfrPolicy, LookupControlFlow, LookupOptions, ZoneHandler, ZoneType,
};
use std::sync::Arc;
pub type ArcZoneHandler = Arc<dyn ZoneHandler>;
#[derive(From, Deref, DerefMut)]
pub struct ChainedZoneHandler<H>(H)
where
H: ZoneHandler;
#[async_trait::async_trait]
impl<H> ZoneHandler for ChainedZoneHandler<H>
where
H: ZoneHandler,
{
delegate! {
to self.0 {
fn zone_type(&self) -> ZoneType;
fn axfr_policy(&self) -> AxfrPolicy;
fn origin(&self) -> &LowerName;
}
}
#[inline]
async fn update(
&self,
update: &Request,
now: u64,
) -> (Result<bool, ResponseCode>, Option<TSigResponseContext>) {
self.0.update(update, now).await
}
#[inline]
async fn lookup(
&self,
name: &LowerName,
rtype: RecordType,
request_info: Option<&RequestInfo<'_>>,
lookup_options: LookupOptions,
) -> LookupControlFlow<AuthLookup> {
self.0
.lookup(name, rtype, request_info, lookup_options)
.await
}
#[inline]
async fn consult(
&self,
name: &LowerName,
rtype: RecordType,
request_info: Option<&RequestInfo<'_>>,
lookup_options: LookupOptions,
last_result: LookupControlFlow<AuthLookup>,
) -> (LookupControlFlow<AuthLookup>, Option<TSigResponseContext>) {
let result = if let Some(Ok(l)) = last_result.map_result() {
LookupControlFlow::Break(Ok(l))
} else {
self.0
.lookup(name, rtype, request_info, lookup_options)
.await
};
(result, None)
}
#[inline]
async fn search(
&self,
request: &Request,
lookup_options: LookupOptions,
) -> (LookupControlFlow<AuthLookup>, Option<TSigResponseContext>) {
self.0.search(request, lookup_options).await
}
#[inline]
async fn nsec_records(
&self,
name: &LowerName,
lookup_options: LookupOptions,
) -> LookupControlFlow<AuthLookup> {
self.0.nsec_records(name, lookup_options).await
}
}
easytier/src/dns/zone.rs
+44 -41
View File
@@ -1,16 +1,16 @@
use crate::common::dns::get_default_resolver_config;
use crate::dns::utils::addr::NameServerAddr;
use crate::dns::utils::authority::ArcAuthority;
use crate::dns::utils::zone_handler::ArcZoneHandler;
use crate::proto;
use crate::proto::utils::RepeatedMessageModel;
use hickory_net::runtime::TokioRuntimeProvider;
use hickory_proto::rr::{LowerName, RecordSet, RrKey};
use hickory_proto::serialize::txt::Parser;
use hickory_resolver::config::ResolverOpts;
use hickory_resolver::name_server::TokioConnectionProvider;
use hickory_resolver::system_conf::read_system_conf;
use hickory_server::authority::ZoneType;
use hickory_server::store::forwarder::{ForwardAuthority, ForwardConfig};
use hickory_server::store::in_memory::InMemoryAuthority;
use hickory_server::store::forwarder::{ForwardConfig, ForwardZoneHandler};
use hickory_server::store::in_memory::InMemoryZoneHandler;
use hickory_server::zone_handler::{AxfrPolicy, ZoneType};
use indexmap::IndexMap;
use itertools::chain;
use std::collections::BTreeMap;
@@ -30,7 +30,7 @@ impl Zone {
let (config, opts) =
read_system_conf().unwrap_or((get_default_resolver_config(), ResolverOpts::default()));
let forward = ForwardConfig {
name_servers: config.name_servers().to_vec().into(),
name_servers: config.name_servers().to_vec(),
options: Some(opts),
};
let mut zone = Self::new(".".parse().unwrap());
@@ -49,10 +49,13 @@ impl Zone {
}
}
pub fn create_memory_authority(&self) -> Option<ArcAuthority> {
pub fn create_memory_zone_handler(&self) -> Option<ArcZoneHandler> {
(!self.records.is_empty()).then(|| {
let mut memory =
InMemoryAuthority::empty(self.origin.clone().into(), ZoneType::External, false);
let mut memory = InMemoryZoneHandler::<TokioRuntimeProvider>::empty(
self.origin.clone().into(),
ZoneType::External,
AxfrPolicy::default(),
);
memory.records_get_mut().extend(
self.records
@@ -61,20 +64,20 @@ impl Zone {
.map(|(k, v)| (k, Arc::new(v))),
);
Arc::new(memory) as ArcAuthority
Arc::new(memory) as ArcZoneHandler
})
}
pub fn create_forward_authority(&self) -> Option<ArcAuthority> {
pub fn create_forward_zone_handler(&self) -> Option<ArcZoneHandler> {
self.forward.as_ref().and_then(|forward| {
ForwardAuthority::builder_with_config(
ForwardZoneHandler::builder_with_config(
forward.clone(),
TokioConnectionProvider::default(),
TokioRuntimeProvider::default(),
)
.build()
.inspect_err(|e| tracing::error!("failed to create forward authority: {:?}", e))
.inspect_err(|e| tracing::error!("failed to create forward zone_handler: {:?}", e))
.ok()
.map(|f| Arc::new(f) as ArcAuthority)
.map(|f| Arc::new(f) as ArcZoneHandler)
})
}
}
@@ -92,14 +95,14 @@ impl TryFrom<&proto::dns::ZoneData> for Zone {
.parse()
.map_err(|e| anyhow::anyhow!("failed to parse zone data: {e}"))?;
let servers = value
let name_servers = value
.forwarders
.iter()
.map(TryInto::<NameServerAddr>::try_into)
.map(|a| a.map(Into::into))
.collect::<Result<Vec<_>, _>>()?;
let forward = (!servers.is_empty()).then_some(ForwardConfig {
name_servers: servers.into(),
let forward = (!name_servers.is_empty()).then_some(ForwardConfig {
name_servers,
options: None,
});
@@ -124,7 +127,7 @@ impl From<Zone> for proto::dns::ZoneData {
let forwarders = value
.forward
.into_iter()
.flat_map(|f| f.name_servers.into_inner().into_iter())
.flat_map(|f| f.name_servers.into_iter())
.map(Into::<NameServerAddr>::into)
.map(Into::into)
.collect();
@@ -149,11 +152,11 @@ impl ZoneGroup {
})
}
pub fn iter_authorities(&self) -> impl Iterator<Item = ArcAuthority> + use<'_> {
pub fn iter_zone_handlers(&self) -> impl Iterator<Item = ArcZoneHandler> + use<'_> {
self.iter().flat_map(|zone| {
chain(
zone.create_memory_authority(),
zone.create_forward_authority(),
zone.create_memory_zone_handler(),
zone.create_forward_zone_handler(),
)
})
}
@@ -167,9 +170,9 @@ mod tests {
use crate::proto::common::Url;
use crate::proto::dns::ZoneData;
use hickory_proto::op::{Message, ResponseCode};
use hickory_proto::rr::{RData, Record, RecordType, RrsetRecords, rdata};
use hickory_server::ServerFuture;
use hickory_server::authority::Catalog;
use hickory_proto::rr::{RData, Record, RecordType, RrsetRecords};
use hickory_server::Server;
use hickory_server::zone_handler::Catalog;
use std::net::{Ipv4Addr, SocketAddr};
use std::str::FromStr;
use tokio::net::UdpSocket;
@@ -209,7 +212,7 @@ mod tests {
.into_groups()
.into_iter()
.fold(Catalog::new(), |mut catalog, (origin, group)| {
catalog.upsert(origin, group.iter_authorities().collect());
catalog.upsert(origin, group.iter_zone_handlers().collect());
catalog
})
}
@@ -221,19 +224,19 @@ mod tests {
) -> anyhow::Result<(ResponseCode, Option<Message>)> {
let request = new_request(name, record_type)?;
let response = ResponseHandle::new(1024);
let info = catalog.lookup(&request, None, response.clone()).await;
let info = catalog.lookup(&request, None, 0, response.clone()).await;
let message = response
.into_inner()
.map(|raw| Message::from_vec(&raw))
.transpose()?;
Ok((info.response_code(), message))
Ok((info.response_code, message))
}
fn has_a_answer(message: &Message, expected: Ipv4Addr) -> bool {
message
.answers()
.answers
.iter()
.any(|record| matches!(record.data(), RData::A(addr) if *addr == rdata::a::A(expected)))
.any(|record| matches!(record.data, RData::A(addr) if *addr == expected))
}
async fn start_upstream_server() -> anyhow::Result<(SocketAddr, JoinHandle<()>)> {
@@ -246,13 +249,13 @@ mod tests {
let mut catalog = Catalog::new();
catalog.upsert(
upstream.origin.clone(),
vec![upstream.create_memory_authority().unwrap()],
vec![upstream.create_memory_zone_handler().unwrap()],
);
let socket = UdpSocket::bind("127.0.0.1:0").await?;
let addr = socket.local_addr()?;
let mut server = ServerFuture::new(catalog);
let mut server = Server::new(catalog);
server.register_socket(socket);
let handle = tokio::spawn(async move {
let _ = server.block_until_done().await;
@@ -292,11 +295,11 @@ mod tests {
}
#[test]
fn empty_zone_creates_no_authority() -> anyhow::Result<()> {
fn empty_zone_creates_no_zone_handler() -> anyhow::Result<()> {
let zone = Zone::try_from(&zone_data("empty.test", vec![], vec![]))?;
assert!(zone.create_memory_authority().is_none());
assert!(zone.create_forward_authority().is_none());
assert!(zone.create_memory_zone_handler().is_none());
assert!(zone.create_forward_zone_handler().is_none());
Ok(())
}
@@ -356,16 +359,16 @@ mod tests {
}
#[test]
fn zone_group_iter_authorities_returns_memory_and_forward() -> anyhow::Result<()> {
fn zone_group_iter_zone_handlers_returns_memory_and_forward() -> anyhow::Result<()> {
let zones: ZoneGroup = vec![Zone::try_from(&zone_data(
"authority.test",
"zone-handler.test",
vec!["@ IN A 10.0.0.10"],
vec!["udp://1.1.1.1:53"],
))?]
.into();
let authorities = zones.iter_authorities().collect::<Vec<_>>();
assert_eq!(authorities.len(), 2);
let zone_handlers = zones.iter_zone_handlers().collect::<Vec<_>>();
assert_eq!(zone_handlers.len(), 2);
Ok(())
}
@@ -375,11 +378,11 @@ mod tests {
let zone = Zone::system();
assert_eq!(zone.origin.to_string(), ".");
assert!(zone.forward.is_some());
assert!(zone.create_forward_authority().is_some());
assert!(zone.create_forward_zone_handler().is_some());
}
#[tokio::test]
async fn catalog_lookup_returns_a_record_from_memory_authority() -> anyhow::Result<()> {
async fn catalog_lookup_returns_a_record_from_memory_zone_handler() -> anyhow::Result<()> {
let zones: ZoneGroup = vec![Zone::try_from(&zone_data(
"memory.test",
vec!["@ IN A 10.20.30.40"],
easytier/src/peers/peer_ospf_route.rs
+15 -15
View File
@@ -3,8 +3,8 @@ use std::{
fmt::Debug,
net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::{
atomic::{AtomicBool, AtomicU32, Ordering}, Arc,
Weak,
Arc, Weak,
atomic::{AtomicBool, AtomicU32, Ordering},
},
time::{Duration, Instant, SystemTime},
};
@@ -14,12 +14,12 @@ use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr};
use crossbeam::atomic::AtomicCell;
use dashmap::DashMap;
use ordered_hash_map::OrderedHashMap;
use parking_lot::{lock_api::RwLockUpgradableReadGuard, RwLock};
use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard};
use petgraph::{
Directed,
algo::dijkstra,
graph::{Graph, NodeIndex},
visit::{EdgeRef, IntoNodeReferences},
Directed,
};
use prefix_trie::PrefixMap;
use prost::Message;
@@ -34,24 +34,24 @@ use crate::common::config::ConfigLoader;
use crate::proto::utils::TransientDigest;
use crate::{
common::{
PeerId,
config::NetworkIdentity,
constants::EASYTIER_VERSION,
global_ctx::{ArcGlobalCtx, GlobalCtxEvent},
shrink_dashmap,
stun::StunInfoCollectorTrait,
PeerId,
},
peers::route_trait::{Route, RouteInterfaceBox},
proto::{
acl::GroupIdentity,
common::{Ipv4Inet, NatType, StunInfo},
peer_rpc::{
route_foreign_network_infos, route_foreign_network_summary, sync_route_info_request::ConnInfo,
ForeignNetworkRouteInfoEntry, ForeignNetworkRouteInfoKey, OspfRouteRpc, OspfRouteRpcClientFactory,
OspfRouteRpcServer, PeerGroupInfo, PeerIdVersion, PeerIdentityType,
RouteForeignNetworkInfos, RouteForeignNetworkSummary, RoutePeerInfo, RoutePeerInfos,
SyncRouteInfoError, SyncRouteInfoRequest, SyncRouteInfoResponse,
TrustedCredentialPubkey,
ForeignNetworkRouteInfoEntry, ForeignNetworkRouteInfoKey, OspfRouteRpc,
OspfRouteRpcClientFactory, OspfRouteRpcServer, PeerGroupInfo, PeerIdVersion,
PeerIdentityType, RouteForeignNetworkInfos, RouteForeignNetworkSummary, RoutePeerInfo,
RoutePeerInfos, SyncRouteInfoError, SyncRouteInfoRequest, SyncRouteInfoResponse,
TrustedCredentialPubkey, route_foreign_network_infos, route_foreign_network_summary,
sync_route_info_request::ConnInfo,
},
rpc_types::{
self,
@@ -62,13 +62,13 @@ use crate::{
};
use super::{
PeerPacketFilter,
graph_algo::dijkstra_with_first_hop,
peer_rpc::PeerRpcManager,
route_trait::{
DefaultRouteCostCalculator, ForeignNetworkRouteInfoMap, NextHopPolicy, RouteCostCalculator,
RouteCostCalculatorInterface,
},
PeerPacketFilter,
};
use atomic_shim::AtomicU64;
@@ -3681,8 +3681,8 @@ mod tests {
use std::{
collections::{BTreeSet, HashMap},
sync::{
atomic::{AtomicU32, Ordering},
Arc,
atomic::{AtomicU32, Ordering},
},
time::{Duration, SystemTime},
};
@@ -3690,14 +3690,14 @@ mod tests {
use super::{PeerRoute, REMOVE_DEAD_PEER_INFO_AFTER};
use crate::{
common::{
global_ctx::{tests::get_mock_global_ctx, GlobalCtxEvent, TrustedKeySource},
PeerId,
global_ctx::{GlobalCtxEvent, TrustedKeySource, tests::get_mock_global_ctx},
},
connector::udp_hole_punch::tests::replace_stun_info_collector,
peers::{
create_packet_recv_chan,
peer_manager::{PeerManager, RouteAlgoType},
peer_ospf_route::{PeerIdVersion, PeerRouteServiceImpl, FORCE_USE_CONN_LIST},
peer_ospf_route::{FORCE_USE_CONN_LIST, PeerIdVersion, PeerRouteServiceImpl},
route_trait::{NextHopPolicy, Route, RouteCostCalculatorInterface, RouteInterface},
tests::{connect_peer_manager, create_mock_peer_manager, wait_route_appear},
},