yeyinlaoying/vnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add TAP device mode and Ethernet interoperability

Browse Source
This commit is contained in:
lbl
2026-08-22 21:47:49 +08:00
parent 26a44e64bd
commit de0183139d
26 changed files with 1073 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt-core/src/compression/lz4_compression.rs
+7 -2
View File
@@ -89,11 +89,13 @@ mod tests {
// --- 构造原始包 ---
let payload = vec![1u8; 200];
let original = make_packet(&payload);
let mut original = make_packet(&payload);
original.set_ethernet_flag(true);
// --- 压缩 ---
let compressed = lz.compress(original, 0).unwrap();
assert!(compressed.is_compressed());
assert!(compressed.is_ethernet());
// 压缩后的 payload 应变小
assert!(
@@ -106,11 +108,14 @@ mod tests {
// 标志应清除
assert!(!decompressed.is_compressed());
assert!(decompressed.is_ethernet());
// HEAD 不变
let mut expected_head = [0u8; HEAD_LENGTH];
expected_head[2] = 0x10;
assert_eq!(
&decompressed.buffer()[..HEAD_LENGTH],
&[0u8; HEAD_LENGTH][..],
&expected_head,
"HEAD 必须保持不变"
);
vnt-core/src/context/config.rs
+64 -1
View File
@@ -6,7 +6,9 @@ use crate::tunnel_core::server::transport::config::{ConnectRegConfig, ProtocolAd
use anyhow::bail;
use ipnet::Ipv4Net;
use std::collections::HashSet;
use std::fmt::{Display, Formatter};
use std::net::Ipv4Addr;
use std::str::FromStr;
pub const MAX_NETWORK_CODE_LEN: usize = 32;
pub const MAX_DEVICE_ID_LEN: usize = 64;
@@ -14,6 +16,44 @@ pub const MAX_NAME_LEN: usize = 128;
pub const MAX_VERSION_LEN: usize = 32;
pub const MAX_MTU: u16 = 1500;
#[derive(Debug, Copy, Clone, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "lowercase")]
pub enum DeviceMode {
No,
#[default]
Tun,
Tap,
}
impl DeviceMode {
pub fn has_device(self) -> bool {
!matches!(self, Self::No)
}
}
impl Display for DeviceMode {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
Self::No => "no",
Self::Tun => "tun",
Self::Tap => "tap",
})
}
}
impl FromStr for DeviceMode {
type Err = anyhow::Error;
fn from_str(value: &str) -> Result<Self, Self::Err> {
match value.to_ascii_lowercase().as_str() {
"no" => Ok(Self::No),
"tun" => Ok(Self::Tun),
"tap" => Ok(Self::Tap),
_ => bail!("invalid device_mode '{value}', expected one of: no, tun, tap"),
}
}
}
#[derive(Debug, Clone, Default)]
pub struct Config {
pub server_addr: Vec<ProtocolAddress>,
@@ -33,7 +73,7 @@ pub struct Config {
pub input: Vec<NetInput>,
pub output: Vec<Ipv4Net>,
pub no_nat: bool,
pub no_tun: bool,
pub device_mode: DeviceMode,
pub mtu: Option<u16>,
pub port_mapping: Vec<PortMapping>,
pub allow_port_mapping: bool,
@@ -43,6 +83,10 @@ pub struct Config {
}
impl Config {
pub fn check(&self) -> anyhow::Result<()> {
#[cfg(any(target_os = "android", target_os = "ios", target_os = "tvos"))]
if self.device_mode == DeviceMode::Tap {
bail!("TAP mode is not supported on mobile VPN interfaces");
}
if self.server_addr.is_empty() {
bail!("服务器地址不能为空");
}
@@ -107,3 +151,22 @@ impl Config {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn device_mode_parse_and_display() {
for (text, mode) in [
("no", DeviceMode::No),
("tun", DeviceMode::Tun),
("tap", DeviceMode::Tap),
] {
assert_eq!(text.parse::<DeviceMode>().unwrap(), mode);
assert_eq!(mode.to_string(), text);
}
assert!("bridge".parse::<DeviceMode>().is_err());
assert_eq!(DeviceMode::default(), DeviceMode::Tun);
}
}
vnt-core/src/core/mod.rs
+49 -24
View File
@@ -1,6 +1,6 @@
use crate::api::VntApi;
use crate::compression::PacketCompression;
use crate::context::config::Config;
use crate::context::config::{Config, DeviceMode};
use crate::context::{AppState, NetworkAddr, NetworkRoute};
use crate::crypto::PacketCrypto;
use crate::enhanced_tunnel::enhanced_ipv4_tunnel;
@@ -21,7 +21,7 @@ use crate::tunnel_core::server::connection_manager::{
};
use crate::tunnel_core::server::rpc::ServerRPC;
use crate::utils::task_control::TaskGroup;
use anyhow::bail;
use anyhow::{Context, bail};
use ipnet::Ipv4Net;
#[cfg(not(target_os = "android"))]
use std::net::Ipv4Addr;
@@ -137,12 +137,12 @@ impl NetworkManager {
fec_encoder,
);
let port_mapping_manager = PortMappingManager::new(
config.no_tun,
config.device_mode == DeviceMode::No,
config.allow_port_mapping,
app_state.network.clone(),
default_interface.clone(),
);
let internal_nat_inbound = if config.no_nat && !config.no_tun {
let internal_nat_inbound = if config.no_nat && config.device_mode != DeviceMode::No {
None
} else {
let nat_inbound = InternalNatInbound::create(
@@ -151,26 +151,32 @@ impl NetworkManager {
hybrid_outbound.clone(),
allow_subnet.clone(),
app_state.network.clone(),
config.no_tun,
config.device_mode == DeviceMode::No,
default_interface.clone(),
)
.await?;
Some(nat_inbound)
};
let (enhanced_tun_inbound, tun_receiver) = if config.no_tun {
(
let (enhanced_tun_inbound, tun_receiver) = match config.device_mode {
DeviceMode::No => (
EnhancedTunInbound::Nat(
internal_nat_inbound
.clone()
.expect("internal_nat_inbound must be Some when no_tun is true"),
.expect("internal_nat_inbound must be Some in no-device mode"),
),
None,
)
} else {
let (tun_inbound, tun_receiver) = tun_channel();
let tun_data_sender = TunDataInbound::new(tun_inbound, allow_subnet.clone());
(EnhancedTunInbound::Tun(tun_data_sender), Some(tun_receiver))
),
mode @ (DeviceMode::Tun | DeviceMode::Tap) => {
let (tun_inbound, tun_receiver) = tun_channel();
let tun_data_sender = TunDataInbound::new(tun_inbound, allow_subnet.clone(), mode);
let inbound = if mode == DeviceMode::Tap {
EnhancedTunInbound::Tap(tun_data_sender)
} else {
EnhancedTunInbound::Tun(tun_data_sender)
};
(inbound, Some(tun_receiver))
}
};
let (enhanced_inbound, enhanced_outbound) = enhanced_ipv4_tunnel(
@@ -182,6 +188,7 @@ impl NetworkManager {
password: config.password.clone(),
open_quic_client: config.rtx,
port_mapping: config.port_mapping.clone(),
device_mode: config.device_mode,
},
crate::enhanced_tunnel::TunnelComponents {
hybrid_outbound: hybrid_outbound.clone(),
@@ -325,16 +332,25 @@ impl NetworkManager {
Ok(RegisterResponse::Success(network_addr))
}
pub fn is_no_tun(&self) -> bool {
self.config.no_tun
pub fn device_mode(&self) -> DeviceMode {
self.config.device_mode
}
pub async fn start_tun(&mut self) -> anyhow::Result<()> {
pub async fn start_device(&mut self) -> anyhow::Result<()> {
if self.tun_receiver.is_none() || self.enhanced_outbound.is_none() {
bail!("start_tun can only be called once");
bail!("start_device requires tun/tap mode and can only be called once");
}
let mut config = DeviceConfig::default();
config = config.set_mtu(self.config.mtu.unwrap_or(DEFAULT_MTU));
config = config
.set_device_mode(self.config.device_mode)
.set_mtu(self.config.mtu.unwrap_or(DEFAULT_MTU));
if self.config.device_mode == DeviceMode::Tap {
let net = self
.app_state
.get_network()
.context("network is not registered")?;
config = config.set_mac_addr(crate::ethernet::mac_from_ip(net.ip));
}
if let Some(tun_name) = self.config.tun_name.clone() {
config = config.set_tun_name(tun_name);
}
@@ -344,11 +360,20 @@ impl NetworkManager {
.await
}
#[cfg(unix)]
pub async fn start_tun_fd(&mut self, tun_fd: Option<i32>) -> anyhow::Result<()> {
pub async fn start_device_fd(&mut self, tun_fd: Option<i32>) -> anyhow::Result<()> {
if self.tun_receiver.is_none() || self.enhanced_outbound.is_none() {
bail!("start_tun_fd can only be called once");
bail!("start_device_fd requires tun/tap mode and can only be called once");
}
let mut config = DeviceConfig::default()
.set_device_mode(self.config.device_mode)
.set_mtu(self.config.mtu.unwrap_or(DEFAULT_MTU));
if self.config.device_mode == DeviceMode::Tap {
let net = self
.app_state
.get_network()
.context("network is not registered")?;
config = config.set_mac_addr(crate::ethernet::mac_from_ip(net.ip));
}
let mut config = DeviceConfig::default();
if let Some(tun_fd) = tun_fd {
config = config.set_tun_fd(tun_fd);
}
@@ -360,7 +385,7 @@ impl NetworkManager {
.await
}
#[cfg(not(target_os = "android"))]
pub async fn set_tun_network_ip(&self, ip: Ipv4Addr, prefix_len: u8) -> anyhow::Result<()> {
pub async fn set_device_network_ip(&self, ip: Ipv4Addr, prefix_len: u8) -> anyhow::Result<()> {
self.device_io_manager.set_network(ip, prefix_len).await?;
Ok(())
}
@@ -370,8 +395,8 @@ impl NetworkManager {
self.app_state.stop_network();
}
#[cfg(not(target_os = "android"))]
pub async fn tun_if_index(&self) -> anyhow::Result<u32> {
self.device_io_manager.tun_if_index().await
pub async fn device_if_index(&self) -> anyhow::Result<u32> {
self.device_io_manager.device_if_index().await
}
pub async fn wait_all_stopped(&mut self) {
self.task_group.wait_all_stopped().await;
vnt-core/src/crypto/chacha20_poly1305.rs
+1 -1
View File
@@ -74,7 +74,7 @@ impl PacketCrypto {
/// AAD 承担"认证"职责：覆盖传输中不变、但不参与 nonce 的头部字节
/// byte0(msg_type)/byte2(flags)/byte3(reserved)。
/// msg_type 与 flags（COMPRESSED/FEC/GATEWAY）只由发送方设置、
/// msg_type 与 flags（COMPRESSED/FEC/GATEWAY/ETHERNET）只由发送方设置、
/// 传输中不会被修改，必须纳入认证，否则中间人可翻转造成不可检测的
/// 丢包/语义篡改；ttl(byte1) 在中继转发时会递减，不能纳入 AAD。
fn make_aad<B: AsRef<[u8]>>(pkt: &NetPacket<B>) -> [u8; 3] {
vnt-core/src/enhanced_tunnel/inbound.rs
+52 -4
View File
@@ -1,9 +1,12 @@
use crate::context::config::DeviceMode;
use crate::context::{NetworkAddr, TrafficStats};
use crate::enhanced_tunnel::quic_over::quic_inbound::EnhancedQuicInbound;
use crate::ethernet::{ETHERTYPE_IPV4, build_arp_reply, parse_arp_ipv4, parse_frame};
use crate::nat::internal_nat::InternalNatInbound;
use crate::protocol::ip_packet_protocol::{HEAD_LENGTH, MsgType, NetPacket};
use crate::protocol::transmission::TransmissionBytes;
use crate::tun::enhanced_tun::EnhancedTunInbound;
use crate::tunnel_core::outbound::HybridOutbound;
use anyhow::{Context, bail};
use pnet_packet::ipv4::Ipv4Packet;
use std::net::Ipv4Addr;
@@ -14,6 +17,8 @@ pub(crate) struct EnhancedInbound {
quic_inbound: EnhancedQuicInbound,
internal_nat_inbound: Option<InternalNatInbound>,
traffic_stats: TrafficStats,
device_mode: DeviceMode,
hybrid_outbound: HybridOutbound,
}
impl EnhancedInbound {
@@ -22,12 +27,16 @@ impl EnhancedInbound {
quic_inbound: EnhancedQuicInbound,
internal_nat_inbound: Option<InternalNatInbound>,
traffic_stats: TrafficStats,
device_mode: DeviceMode,
hybrid_outbound: HybridOutbound,
) -> Self {
Self {
tun_data_inbound,
quic_inbound,
internal_nat_inbound,
traffic_stats,
device_mode,
hybrid_outbound,
}
}
pub async fn inbound(
@@ -37,26 +46,65 @@ impl EnhancedInbound {
src: Ipv4Addr,
packet: NetPacket<TransmissionBytes>,
) -> anyhow::Result<()> {
let ethernet = packet.is_ethernet();
let mut buf = packet.into_buffer();
self.traffic_stats.record_rx(src, buf.len() as u64);
buf.advance_head(HEAD_LENGTH)?;
if ethernet && parse_frame(buf.as_ref()).is_none() {
return Ok(());
}
if ethernet && self.device_mode != DeviceMode::Tap {
if let Some(arp) = parse_arp_ipv4(buf.as_ref())
&& arp.operation == 1
&& arp.target_ip == network_addr.ip
{
if let Some(reply) = build_arp_reply(buf.as_ref(), network_addr.ip) {
self.hybrid_outbound
.ethernet_unicast_outbound(*network_addr, src, reply)
.await?;
}
return Ok(());
}
if parse_frame(buf.as_ref()).is_none_or(|frame| frame.ethertype != ETHERTYPE_IPV4) {
return Ok(());
}
}
match msg_type {
MsgType::Turn => {
if let Some(internal_nat_inbound) = self.internal_nat_inbound.as_ref() {
let Some(ipv4) = Ipv4Packet::new(&buf) else {
let ip_data = if ethernet {
let Some(frame) = parse_frame(buf.as_ref()) else {
return Ok(());
};
if frame.ethertype != ETHERTYPE_IPV4 {
self.tun_data_inbound
.inbound(buf, network_addr, src, true)
.await?;
return Ok(());
}
&buf[frame.payload_offset..]
} else {
buf.as_ref()
};
let Some(ipv4) = Ipv4Packet::new(ip_data) else {
bail!("EnhancedInbound not ipv4")
};
let dest = ipv4.get_destination();
if dest != network_addr.ip && !network_addr.network().contains(&dest) {
internal_nat_inbound.send(&buf, network_addr).await?;
internal_nat_inbound.send(ip_data, network_addr).await?;
return Ok(());
}
}
self.tun_data_inbound.inbound(buf, network_addr).await?;
self.tun_data_inbound
.inbound(buf, network_addr, src, ethernet)
.await?;
}
MsgType::Broadcast | MsgType::ExcludeBroadcast => {
self.tun_data_inbound.inbound(buf, network_addr).await?;
self.tun_data_inbound
.inbound(buf, network_addr, src, ethernet)
.await?;
}
MsgType::Quic => {
let payload = buf.into_bytes().freeze();
vnt-core/src/enhanced_tunnel/mod.rs
+5 -1
View File
@@ -1,4 +1,5 @@
use crate::context::AppState;
use crate::context::config::DeviceMode;
use crate::enhanced_tunnel::inbound::EnhancedInbound;
use crate::enhanced_tunnel::outbound::EnhancedOutbound;
use crate::nat::SubnetExternalRoute;
@@ -18,6 +19,7 @@ pub(crate) struct TunnelConfig {
pub password: Option<String>,
pub open_quic_client: bool,
pub port_mapping: Vec<PortMapping>,
pub device_mode: DeviceMode,
}
pub(crate) struct TunnelComponents {
@@ -36,7 +38,7 @@ pub(crate) async fn enhanced_ipv4_tunnel(
) -> anyhow::Result<(EnhancedInbound, Option<EnhancedOutbound>)> {
let password = config.password.unwrap_or_else(|| "password".to_string());
let tun = match &tun_data_sender {
EnhancedTunInbound::Tun(tun) => Some(tun.clone()),
EnhancedTunInbound::Tun(tun) | EnhancedTunInbound::Tap(tun) => Some(tun.clone()),
EnhancedTunInbound::Nat(_) => None,
};
let (inbound, outbound) = quic_over::boot::quic_tunnel_start(
@@ -62,6 +64,8 @@ pub(crate) async fn enhanced_ipv4_tunnel(
inbound,
components.internal_nat_inbound,
app_state.traffic_stats.clone(),
config.device_mode,
components.hybrid_outbound.clone(),
);
let enhanced_outbound = outbound.map(|outbound| {
vnt-core/src/enhanced_tunnel/outbound.rs
+72
View File
@@ -1,5 +1,9 @@
use crate::context::SharedNetworkAddr;
use crate::enhanced_tunnel::quic_over::quic_outbound::EnhancedQuicOutbound;
use crate::ethernet::{
ETHERTYPE_ARP, ETHERTYPE_IPV4, build_arp_reply, ip_from_mac, is_broadcast_or_multicast,
parse_arp_ipv4, parse_frame,
};
use crate::protocol::transmission::TransmissionBytes;
use crate::tunnel_core::outbound::HybridOutbound;
use pnet_packet::ipv4::Ipv4Packet;
@@ -30,6 +34,74 @@ impl EnhancedOutbound {
log::warn!("EnhancedOutbound error: {:?}", e);
}
}
pub async fn ethernet_outbound(&self, data: TransmissionBytes) -> Option<TransmissionBytes> {
match self.ethernet_outbound_impl(data).await {
Ok(reply) => reply,
Err(e) => {
log::warn!("EnhancedOutbound Ethernet error: {e:?}");
None
}
}
}
async fn ethernet_outbound_impl(
&self,
data: TransmissionBytes,
) -> anyhow::Result<Option<TransmissionBytes>> {
let Some(frame) = parse_frame(data.as_ref()) else {
return Ok(None);
};
let Some(net) = self.network.get() else {
return Ok(None);
};
match frame.ethertype {
ETHERTYPE_IPV4 => {
let Some(ipv4) = Ipv4Packet::new(&data[frame.payload_offset..]) else {
return Ok(None);
};
let src = ipv4.get_source();
let dest = ipv4.get_destination();
if dest == src || dest.is_unspecified() {
return Ok(None);
}
self.hybrid_outbound
.ethernet_ipv4_outbound(net, data, dest)
.await?;
}
ETHERTYPE_ARP => {
let Some(arp) = parse_arp_ipv4(data.as_ref()) else {
return Ok(None);
};
if arp.operation == 1 && arp.target_ip == net.gateway {
return Ok(build_arp_reply(data.as_ref(), net.gateway));
}
if arp.operation == 2 {
let dest = ip_from_mac(frame.destination).unwrap_or(arp.target_ip);
self.hybrid_outbound
.ethernet_unicast_outbound(net, dest, data)
.await?;
} else {
self.hybrid_outbound
.ethernet_broadcast_outbound(net, data)
.await?;
}
}
_ => {
if !is_broadcast_or_multicast(frame.destination)
&& let Some(dest) = ip_from_mac(frame.destination)
&& net.network().contains(&dest)
{
self.hybrid_outbound
.ethernet_unicast_outbound(net, dest, data)
.await?;
} else {
self.hybrid_outbound
.ethernet_broadcast_outbound(net, data)
.await?;
}
}
}
Ok(None)
}
async fn ipv4_outbound_impl(&self, data: TransmissionBytes) -> anyhow::Result<()> {
let Some(ipv4) = Ipv4Packet::new(data.as_ref()) else {
return Ok(());
vnt-core/src/enhanced_tunnel/quic_over/boot.rs
+8 -1
View File
@@ -16,6 +16,7 @@ use crate::tun::TunDataInbound;
use crate::tunnel_core::outbound::HybridOutbound;
use crate::utils::task_control::TaskGroup;
use anyhow::Context;
use pnet_packet::ipv4::Ipv4Packet;
use quinn::congestion::BbrConfig;
use quinn::crypto::rustls::QuicServerConfig;
use quinn::{ClientConfig, Endpoint, EndpointConfig, TransportConfig, default_runtime};
@@ -186,7 +187,13 @@ async fn ip_stack_recv_task(
log::error!("not network");
break;
};
match tun_data_sender.send((&buf[..len]).into(), &net).await {
let Some(ipv4) = Ipv4Packet::new(&buf[..len]) else {
continue;
};
match tun_data_sender
.send_ip((&buf[..len]).into(), &net, ipv4.get_source())
.await
{
Ok(_) => {}
Err(e) => {
log::error!("IP stack send error: {:?}", e);
vnt-core/src/ethernet.rs
+231
View File
@@ -0,0 +1,231 @@
use crate::context::NetworkAddr;
use crate::protocol::ip_packet_protocol::HEAD_LENGTH;
use crate::protocol::transmission::TransmissionBytes;
use std::net::Ipv4Addr;
pub const ETHERNET_HEADER_LEN: usize = 14;
pub const ETHERTYPE_IPV4: u16 = 0x0800;
pub const ETHERTYPE_ARP: u16 = 0x0806;
const ETHERTYPE_VLAN: u16 = 0x8100;
const ETHERTYPE_QINQ: u16 = 0x88a8;
const ETHERTYPE_VLAN_9100: u16 = 0x9100;
const ARP_IPV4_LEN: usize = 28;
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub struct FrameInfo {
pub destination: [u8; 6],
pub source: [u8; 6],
pub ethertype: u16,
pub payload_offset: usize,
}
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub struct ArpIpv4 {
pub operation: u16,
pub sender_mac: [u8; 6],
pub sender_ip: Ipv4Addr,
pub target_mac: [u8; 6],
pub target_ip: Ipv4Addr,
}
pub fn mac_from_ip(ip: Ipv4Addr) -> [u8; 6] {
let octets = ip.octets();
[0x02, 0x00, octets[0], octets[1], octets[2], octets[3]]
}
pub fn ip_from_mac(mac: [u8; 6]) -> Option<Ipv4Addr> {
(mac[0] == 0x02 && mac[1] == 0x00).then(|| Ipv4Addr::new(mac[2], mac[3], mac[4], mac[5]))
}
pub fn parse_frame(frame: &[u8]) -> Option<FrameInfo> {
if frame.len() < ETHERNET_HEADER_LEN {
return None;
}
let destination = frame[0..6].try_into().ok()?;
let source = frame[6..12].try_into().ok()?;
let mut ethertype = u16::from_be_bytes(frame[12..14].try_into().ok()?);
let mut payload_offset = ETHERNET_HEADER_LEN;
// Support stacked VLAN tags. Four tags is already beyond normal Q-in-Q usage
// and bounds work performed for an untrusted frame.
for _ in 0..4 {
if !matches!(
ethertype,
ETHERTYPE_VLAN | ETHERTYPE_QINQ | ETHERTYPE_VLAN_9100
) {
break;
}
if frame.len() < payload_offset + 4 {
return None;
}
ethertype = u16::from_be_bytes(
frame[payload_offset + 2..payload_offset + 4]
.try_into()
.ok()?,
);
payload_offset += 4;
}
(frame.len() >= payload_offset).then_some(FrameInfo {
destination,
source,
ethertype,
payload_offset,
})
}
pub fn parse_arp_ipv4(frame: &[u8]) -> Option<ArpIpv4> {
let info = parse_frame(frame)?;
if info.ethertype != ETHERTYPE_ARP || frame.len() < info.payload_offset + ARP_IPV4_LEN {
return None;
}
let arp = &frame[info.payload_offset..];
if u16::from_be_bytes(arp[0..2].try_into().ok()?) != 1
|| u16::from_be_bytes(arp[2..4].try_into().ok()?) != ETHERTYPE_IPV4
|| arp[4] != 6
|| arp[5] != 4
{
return None;
}
Some(ArpIpv4 {
operation: u16::from_be_bytes(arp[6..8].try_into().ok()?),
sender_mac: arp[8..14].try_into().ok()?,
sender_ip: Ipv4Addr::from(<[u8; 4]>::try_from(&arp[14..18]).ok()?),
target_mac: arp[18..24].try_into().ok()?,
target_ip: Ipv4Addr::from(<[u8; 4]>::try_from(&arp[24..28]).ok()?),
})
}
pub fn build_arp_reply(request: &[u8], own_ip: Ipv4Addr) -> Option<TransmissionBytes> {
let info = parse_frame(request)?;
let arp = parse_arp_ipv4(request)?;
if arp.operation != 1 || arp.target_ip != own_ip {
return None;
}
let frame_len = request.len().max(info.payload_offset + ARP_IPV4_LEN);
let mut bytes = TransmissionBytes::with_capacity(HEAD_LENGTH, HEAD_LENGTH + frame_len);
bytes.put(request).ok()?;
if bytes.len() < frame_len {
bytes.extend_end(frame_len - bytes.len());
}
let own_mac = mac_from_ip(own_ip);
bytes[0..6].copy_from_slice(&arp.sender_mac);
bytes[6..12].copy_from_slice(&own_mac);
let payload = info.payload_offset;
bytes[payload + 6..payload + 8].copy_from_slice(&2u16.to_be_bytes());
bytes[payload + 8..payload + 14].copy_from_slice(&own_mac);
bytes[payload + 14..payload + 18].copy_from_slice(&own_ip.octets());
bytes[payload + 18..payload + 24].copy_from_slice(&arp.sender_mac);
bytes[payload + 24..payload + 28].copy_from_slice(&arp.sender_ip.octets());
Some(bytes)
}
pub fn strip_ipv4(mut frame: TransmissionBytes) -> Option<TransmissionBytes> {
let info = parse_frame(frame.as_ref())?;
if info.ethertype != ETHERTYPE_IPV4 {
return None;
}
frame.advance_head(info.payload_offset).ok()?;
Some(frame)
}
pub fn wrap_ipv4(
mut packet: TransmissionBytes,
src_node: Ipv4Addr,
net: &NetworkAddr,
) -> Option<TransmissionBytes> {
let ipv4 = pnet_packet::ipv4::Ipv4Packet::new(packet.as_ref())?;
let destination_ip = ipv4.get_destination();
let destination_mac = if destination_ip.is_broadcast() || destination_ip == net.broadcast {
[0xff; 6]
} else if destination_ip.is_multicast() {
let octets = destination_ip.octets();
[0x01, 0x00, 0x5e, octets[1] & 0x7f, octets[2], octets[3]]
} else {
mac_from_ip(net.ip)
};
packet.retreat_head(ETHERNET_HEADER_LEN).ok()?;
packet[0..6].copy_from_slice(&destination_mac);
packet[6..12].copy_from_slice(&mac_from_ip(src_node));
packet[12..14].copy_from_slice(&ETHERTYPE_IPV4.to_be_bytes());
Some(packet)
}
pub fn is_broadcast_or_multicast(mac: [u8; 6]) -> bool {
mac == [0xff; 6] || mac[0] & 1 != 0
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn node_mac_round_trip() {
let ip = Ipv4Addr::new(10, 26, 1, 9);
assert_eq!(mac_from_ip(ip), [2, 0, 10, 26, 1, 9]);
assert_eq!(ip_from_mac(mac_from_ip(ip)), Some(ip));
assert_eq!(ip_from_mac([0; 6]), None);
}
#[test]
fn parses_vlan_ipv4() {
let mut frame = vec![0u8; 18 + 20];
frame[12..14].copy_from_slice(&ETHERTYPE_VLAN.to_be_bytes());
frame[16..18].copy_from_slice(&ETHERTYPE_IPV4.to_be_bytes());
let info = parse_frame(&frame).unwrap();
assert_eq!(info.ethertype, ETHERTYPE_IPV4);
assert_eq!(info.payload_offset, 18);
assert!(parse_frame(&[0u8; 13]).is_none());
let mut truncated_vlan = vec![0u8; 16];
truncated_vlan[12..14].copy_from_slice(&ETHERTYPE_VLAN.to_be_bytes());
assert!(parse_frame(&truncated_vlan).is_none());
}
#[test]
fn builds_arp_reply_for_node_ip() {
let sender_ip = Ipv4Addr::new(10, 26, 0, 8);
let target_ip = Ipv4Addr::new(10, 26, 0, 9);
let sender_mac = mac_from_ip(sender_ip);
let mut request = vec![0u8; 42];
request[0..6].copy_from_slice(&[0xff; 6]);
request[6..12].copy_from_slice(&sender_mac);
request[12..14].copy_from_slice(&ETHERTYPE_ARP.to_be_bytes());
request[14..16].copy_from_slice(&1u16.to_be_bytes());
request[16..18].copy_from_slice(&ETHERTYPE_IPV4.to_be_bytes());
request[18] = 6;
request[19] = 4;
request[20..22].copy_from_slice(&1u16.to_be_bytes());
request[22..28].copy_from_slice(&sender_mac);
request[28..32].copy_from_slice(&sender_ip.octets());
request[38..42].copy_from_slice(&target_ip.octets());
let reply = build_arp_reply(&request, target_ip).unwrap();
let arp = parse_arp_ipv4(reply.as_ref()).unwrap();
assert_eq!(arp.operation, 2);
assert_eq!(arp.sender_ip, target_ip);
assert_eq!(arp.sender_mac, mac_from_ip(target_ip));
assert_eq!(arp.target_ip, sender_ip);
assert_eq!(arp.target_mac, sender_mac);
}
#[test]
fn wraps_ipv4_for_tap_and_strips_it_again() {
let net = NetworkAddr {
gateway: Ipv4Addr::new(10, 26, 0, 1),
broadcast: Ipv4Addr::new(10, 26, 0, 255),
ip: Ipv4Addr::new(10, 26, 0, 9),
prefix_len: 24,
};
let src = Ipv4Addr::new(10, 26, 0, 8);
let mut packet = TransmissionBytes::with_capacity(HEAD_LENGTH, HEAD_LENGTH + 20);
packet.put(&[0u8; 20]).unwrap();
packet[0] = 0x45;
packet[12..16].copy_from_slice(&src.octets());
packet[16..20].copy_from_slice(&net.ip.octets());
let original = packet.as_ref().to_vec();
let frame = wrap_ipv4(packet, src, &net).unwrap();
let info = parse_frame(frame.as_ref()).unwrap();
assert_eq!(info.source, mac_from_ip(src));
assert_eq!(info.destination, mac_from_ip(net.ip));
assert_eq!(strip_ipv4(frame).unwrap().as_ref(), original);
}
}
vnt-core/src/fec/decoder.rs
+12
View File
@@ -436,6 +436,18 @@ mod tests {
shard
}
#[test]
fn test_original_fec_packet_preserves_ethernet_flag() {
let decoder = FecDecoder::new();
let packets = decoder
.receive(build_data_packet(99, 0, 0x81, 0x10, &[1, 2, 3]))
.unwrap()
.unwrap();
assert_eq!(packets.len(), 1);
assert!(packets[0].is_ethernet());
assert!(!packets[0].is_fec());
}
/// 变长批次：丢一个数据包，靠校验包必须能恢复（修复前必报 IncorrectShardSize）
#[test]
fn test_reconstruct_variable_length_batch() {
vnt-core/src/lib.rs
+1
View File
@@ -2,6 +2,7 @@ pub(crate) mod compression;
pub mod context;
pub mod core;
pub mod crypto;
pub(crate) mod ethernet;
pub(crate) mod fec;
pub mod nat;
pub mod protocol;
vnt-core/src/protocol/ip_packet_protocol.rs
+22 -2
View File
@@ -2,7 +2,7 @@
0 15 31
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| 1 | msg_type(7) |max ttl(4) |curr ttl(4)| C | G | R | reserve(13) |
| 1 | msg_type(7) |max ttl(4) |curr ttl(4)| C | G | F | E | reserve(12) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| seq(32) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
@@ -27,7 +27,7 @@ pub struct NetHeader {
pub type_byte: u8,
/// Byte 1: high 4 = max ttl, low 4 = curr ttl
pub ttl_byte: u8,
/// Byte 2: C(0x80) | G(0x40) | reserve
/// Byte 2: C(0x80) | G(0x40) | F(0x20) | ETHERNET(0x10) | reserve
pub flags_byte: u8,
/// Byte 3: reserve
pub _reserved: u8,
@@ -39,6 +39,7 @@ pub struct NetHeader {
const COMPRESSED: u8 = 0x80;
const GATEWAY: u8 = 0x40;
const FEC: u8 = 0x20;
const ETHERNET: u8 = 0x10;
impl NetHeader {
#[inline]
pub fn msg_type(&self) -> u8 {
@@ -210,6 +211,9 @@ impl<B: AsRef<[u8]>> NetPacket<B> {
pub fn is_fec(&self) -> bool {
(self.header().flags_byte & FEC) != 0
}
pub fn is_ethernet(&self) -> bool {
(self.header().flags_byte & ETHERNET) != 0
}
pub fn head(&self) -> &[u8] {
&self.buffer.as_ref()[..HEAD_LENGTH]
}
@@ -258,6 +262,9 @@ impl<B: AsRef<[u8]> + AsMut<[u8]>> NetPacket<B> {
pub fn set_fec_flag(&mut self, fec: bool) {
self.header_mut().set_flag(FEC, fec);
}
pub fn set_ethernet_flag(&mut self, ethernet: bool) {
self.header_mut().set_flag(ETHERNET, ethernet);
}
pub fn set_payload(&mut self, data: &[u8]) -> io::Result<()> {
let buf = self.buffer.as_mut();
@@ -345,6 +352,19 @@ mod tests {
assert!(MsgType::try_from(20u8).is_err());
}
#[test]
fn ethernet_flag_round_trip() {
let mut packet = NetPacket::new(BytesMut::from(&[0u8; HEAD_LENGTH][..])).unwrap();
assert!(!packet.is_ethernet());
packet.set_ethernet_flag(true);
assert!(packet.is_ethernet());
packet.set_fec_flag(true);
assert!(packet.is_ethernet());
packet.set_ethernet_flag(false);
assert!(!packet.is_ethernet());
assert!(packet.is_fec());
}
/// 中继转发语义：包每经过一跳 curr_ttl 减 1，curr_ttl >= 1 时才继续转发，
/// 接收方以 metric = max_ttl - curr_ttl 计算路由距离。
#[test]
vnt-core/src/tun/enhanced_tun.rs
+119 -3
View File
@@ -1,18 +1,134 @@
use crate::context::NetworkAddr;
use crate::ethernet::strip_ipv4;
use crate::nat::internal_nat::InternalNatInbound;
use crate::protocol::transmission::TransmissionBytes;
use crate::tun::TunDataInbound;
use std::net::Ipv4Addr;
#[derive(Clone)]
pub enum EnhancedTunInbound {
Tun(TunDataInbound),
Tap(TunDataInbound),
Nat(InternalNatInbound),
}
impl EnhancedTunInbound {
pub async fn inbound(&self, data: TransmissionBytes, net: &NetworkAddr) -> anyhow::Result<()> {
pub async fn inbound(
&self,
data: TransmissionBytes,
net: &NetworkAddr,
src_node: Ipv4Addr,
ethernet: bool,
) -> anyhow::Result<()> {
match self {
EnhancedTunInbound::Tun(tun) => tun.send(data, net).await,
EnhancedTunInbound::Nat(nat) => nat.send(&data, net).await,
EnhancedTunInbound::Tun(tun) => {
let data = if ethernet {
let Some(ip) = strip_ipv4(data) else {
return Ok(());
};
ip
} else {
data
};
tun.send_ip(data, net, src_node).await
}
EnhancedTunInbound::Tap(tap) => {
if ethernet {
tap.send_frame(data).await
} else {
tap.send_ip(data, net, src_node).await
}
}
EnhancedTunInbound::Nat(nat) => {
let data = if ethernet {
let Some(ip) = strip_ipv4(data) else {
return Ok(());
};
ip
} else {
data
};
nat.send(&data, net).await
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::context::config::DeviceMode;
use crate::ethernet::{ETHERTYPE_IPV4, parse_frame, wrap_ipv4};
use crate::nat::AllowSubnetExternalRoute;
use crate::protocol::ip_packet_protocol::HEAD_LENGTH;
use crate::tun::{TunDataInbound, tun_channel};
fn network() -> NetworkAddr {
NetworkAddr {
gateway: Ipv4Addr::new(10, 26, 0, 1),
broadcast: Ipv4Addr::new(10, 26, 0, 255),
ip: Ipv4Addr::new(10, 26, 0, 9),
prefix_len: 24,
}
}
fn ipv4(src: Ipv4Addr, dest: Ipv4Addr) -> TransmissionBytes {
let mut packet = TransmissionBytes::with_capacity(HEAD_LENGTH, HEAD_LENGTH + 20);
packet.put(&[0u8; 20]).unwrap();
packet[0] = 0x45;
packet[12..16].copy_from_slice(&src.octets());
packet[16..20].copy_from_slice(&dest.octets());
packet
}
#[tokio::test]
async fn tap_wraps_ip_and_tun_strips_ethernet() {
let net = network();
let src = Ipv4Addr::new(10, 26, 0, 8);
let (tap_tx, mut tap_rx) = tun_channel();
let tap = EnhancedTunInbound::Tap(TunDataInbound::new(
tap_tx,
AllowSubnetExternalRoute::new(vec![]),
DeviceMode::Tap,
));
tap.inbound(ipv4(src, net.ip), &net, src, false)
.await
.unwrap();
let frame = tap_rx.receiver.recv().await.unwrap();
assert_eq!(
parse_frame(frame.as_ref()).unwrap().ethertype,
ETHERTYPE_IPV4
);
let (tun_tx, mut tun_rx) = tun_channel();
let tun = EnhancedTunInbound::Tun(TunDataInbound::new(
tun_tx,
AllowSubnetExternalRoute::new(vec![]),
DeviceMode::Tun,
));
let frame = wrap_ipv4(ipv4(src, net.ip), src, &net).unwrap();
tun.inbound(frame, &net, src, true).await.unwrap();
let packet = tun_rx.receiver.recv().await.unwrap();
assert_eq!(packet[0] >> 4, 4);
assert_eq!(&packet[16..20], &net.ip.octets());
}
#[tokio::test]
async fn tap_keeps_arbitrary_ethernet_frame() {
let net = network();
let src = Ipv4Addr::new(10, 26, 0, 8);
let (tap_tx, mut tap_rx) = tun_channel();
let tap = EnhancedTunInbound::Tap(TunDataInbound::new(
tap_tx,
AllowSubnetExternalRoute::new(vec![]),
DeviceMode::Tap,
));
let mut raw = vec![0u8; 32];
raw[0..6].copy_from_slice(&[0xff; 6]);
raw[12..14].copy_from_slice(&0x88b5u16.to_be_bytes());
tap.inbound(raw.as_slice().into(), &net, src, true)
.await
.unwrap();
assert_eq!(tap_rx.receiver.recv().await.unwrap().as_ref(), raw);
}
}
vnt-core/src/tun/general.rs
+67 -20
View File
@@ -1,3 +1,4 @@
use crate::context::config::DeviceMode;
use crate::enhanced_tunnel::outbound::EnhancedOutbound;
use crate::protocol::ip_packet_protocol::HEAD_LENGTH;
use crate::protocol::transmission::TransmissionBytes;
@@ -10,9 +11,9 @@ use std::net::Ipv4Addr;
use std::sync::Arc;
use tokio::sync::mpsc::{Receiver, Sender};
use tun_rs::AsyncDevice;
#[cfg(not(target_os = "android"))]
use tun_rs::DeviceBuilder;
use tun_rs::async_framed::{Decoder, DeviceFramedRead, DeviceFramedWrite, Encoder};
#[cfg(not(target_os = "android"))]
use tun_rs::{DeviceBuilder, Layer};
#[derive(Clone)]
pub struct DeviceIOManager {
@@ -27,10 +28,12 @@ pub struct DeviceTask {
}
#[derive(Debug, Default)]
pub struct DeviceConfig {
pub device_mode: DeviceMode,
pub tun_name: Option<String>,
#[cfg(unix)]
pub tun_fd: Option<i32>,
pub mtu: Option<u16>,
pub mac_addr: Option<[u8; 6]>,
}
impl DeviceConfig {
@@ -47,6 +50,14 @@ impl DeviceConfig {
self.mtu = Some(mtu);
self
}
pub fn set_device_mode(mut self, device_mode: DeviceMode) -> Self {
self.device_mode = device_mode;
self
}
pub fn set_mac_addr(mut self, mac_addr: [u8; 6]) -> Self {
self.mac_addr = Some(mac_addr);
self
}
}
#[derive(Clone)]
pub struct TunInbound {
@@ -54,7 +65,7 @@ pub struct TunInbound {
}
pub struct TunReceiver {
receiver: Receiver<TransmissionBytes>,
pub(crate) receiver: Receiver<TransmissionBytes>,
}
pub fn tun_channel() -> (TunInbound, TunReceiver) {
let (sender, receiver) = tokio::sync::mpsc::channel(1024);
@@ -84,9 +95,10 @@ impl DeviceIOManager {
bail!("device task already started");
}
self.stop_task().await;
// 先执行可能失败的 tun 设备创建，成功后才消费 receiver/outbound，
// 先执行可能失败的 TUN/TAP 设备创建，成功后才消费 receiver/outbound，
// 保证失败时调用方状态完整、可以重试
let device = Arc::new(create_tun(device_config)?);
let device_mode = device_config.device_mode;
let device = Arc::new(create_device(device_config)?);
let receiver = receiver.take().unwrap();
let enhanced_outbound = enhanced_outbound.take().unwrap();
let task = create(
@@ -94,12 +106,13 @@ impl DeviceIOManager {
device,
receiver.receiver,
enhanced_outbound,
device_mode,
);
self.device.lock().await.0.replace(task);
Ok(())
}
#[cfg(not(target_os = "android"))]
pub async fn tun_if_index(&self) -> anyhow::Result<u32> {
pub async fn device_if_index(&self) -> anyhow::Result<u32> {
let guard = self.device.lock().await;
if let Some(v) = &guard.0 {
Ok(v.device.if_index()?)
@@ -111,7 +124,7 @@ impl DeviceIOManager {
pub async fn set_network(&self, ip: Ipv4Addr, prefix_len: u8) -> anyhow::Result<()> {
let mut guard = self.device.lock().await;
let Some(dev) = guard.0.as_ref() else {
bail!("未启动tun")
bail!("虚拟网卡尚未启动")
};
if let Some(v) = guard.1.as_ref()
&& v.0 == ip
@@ -127,7 +140,7 @@ impl DeviceIOManager {
}
}
fn create_tun(config: DeviceConfig) -> anyhow::Result<AsyncDevice> {
fn create_device(config: DeviceConfig) -> anyhow::Result<AsyncDevice> {
#[cfg(target_os = "android")]
{
let fd = config
@@ -146,21 +159,45 @@ fn create_tun(config: DeviceConfig) -> anyhow::Result<AsyncDevice> {
unsafe { return Ok(AsyncDevice::from_fd(fd)?) }
}
let mut builder = DeviceBuilder::new();
builder = builder.layer(match config.device_mode {
DeviceMode::Tap => Layer::L2,
DeviceMode::Tun => Layer::L3,
DeviceMode::No => bail!("cannot create a device in no mode"),
});
if let Some(tun_name) = config.tun_name {
builder = builder.name(tun_name);
}
if let Some(mtu) = config.mtu {
builder = builder.mtu(mtu);
}
#[cfg(any(
target_os = "windows",
target_os = "linux",
target_os = "freebsd",
target_os = "openbsd",
target_os = "macos",
target_os = "netbsd"
))]
if let Some(mac_addr) = config.mac_addr {
builder = builder.mac_addr(mac_addr);
}
#[cfg(windows)]
{
builder = builder.metric(1);
}
#[cfg(target_os = "linux")]
{
if config.device_mode == DeviceMode::Tun {
builder = builder.offload(true);
}
let dev = builder.build_async().context("创建tun失败")?;
let dev = builder.build_async().with_context(|| {
if config.device_mode == DeviceMode::Tap && cfg!(windows) {
"创建 TAP 失败；Windows TAP 模式需要预先安装 tap-windows (tap0901) 驱动"
} else if config.device_mode == DeviceMode::Tap {
"创建 TAP 失败"
} else {
"创建 TUN 失败"
}
})?;
#[cfg(target_os = "linux")]
{
_ = dev.set_tx_queue_len(1000);
@@ -173,26 +210,28 @@ fn create(
device: Arc<AsyncDevice>,
receiver: Receiver<TransmissionBytes>,
enhanced_outbound: EnhancedOutbound,
device_mode: DeviceMode,
) -> DeviceTask {
let device_framed_read = DeviceFramedRead::new(device.clone(), BytesCodec::new());
let device_framed_write = DeviceFramedWrite::new(device.clone(), BytesCodec::new());
let outbound_device = device.clone();
// 读写两个方向合并为一个任务：任一方向结束（出错或设备关闭）即
// 通过 select! 取消另一方向，避免单侧失败后另一侧继续运行的半开状态
let task = task_group.spawn(async move {
tokio::select! {
rs = in_tun_loop(receiver, device_framed_write) => {
rs = in_device_loop(receiver, device_framed_write) => {
if let Err(e) = rs {
log::error!("in_tun_loop error, stopping out_tun_loop: {e:?}");
log::error!("in_device_loop error, stopping out_device_loop: {e:?}");
} else {
log::warn!("in_tun_loop exited, stopping out_tun_loop");
log::warn!("in_device_loop exited, stopping out_device_loop");
}
}
rs = out_tun_loop(device_framed_read, enhanced_outbound) => {
rs = out_device_loop(device_framed_read, outbound_device, enhanced_outbound, device_mode) => {
if let Err(e) = rs {
log::error!("out_tun_loop error, stopping in_tun_loop: {e:?}");
log::error!("out_device_loop error, stopping in_device_loop: {e:?}");
} else {
log::warn!("out_tun_loop exited, stopping in_tun_loop");
log::warn!("out_device_loop exited, stopping in_device_loop");
}
}
}
@@ -201,7 +240,7 @@ fn create(
DeviceTask { device, task }
}
async fn in_tun_loop(
async fn in_device_loop(
mut receiver: Receiver<TransmissionBytes>,
mut device_framed_write: DeviceFramedWrite<BytesCodec, Arc<AsyncDevice>>,
) -> anyhow::Result<()> {
@@ -209,7 +248,7 @@ async fn in_tun_loop(
match device_framed_write.send(data).await {
Ok(_) => {}
Err(e) => {
log::error!("send to tun error: {:?}", e);
log::error!("send to virtual device error: {:?}", e);
return Err(anyhow::anyhow!(e));
}
}
@@ -217,13 +256,21 @@ async fn in_tun_loop(
Ok(())
}
async fn out_tun_loop(
async fn out_device_loop(
mut device_framed_read: DeviceFramedRead<BytesCodec, Arc<AsyncDevice>>,
device: Arc<AsyncDevice>,
enhanced_outbound: EnhancedOutbound,
device_mode: DeviceMode,
) -> anyhow::Result<()> {
while let Some(rs) = device_framed_read.next().await {
let bytes_mut = rs?;
enhanced_outbound.ipv4_outbound(bytes_mut).await;
if device_mode == DeviceMode::Tap {
if let Some(reply) = enhanced_outbound.ethernet_outbound(bytes_mut).await {
device.send(reply.as_ref()).await?;
}
} else {
enhanced_outbound.ipv4_outbound(bytes_mut).await;
}
}
Ok(())
}
vnt-core/src/tun/sender.rs
+44 -6
View File
@@ -1,25 +1,39 @@
use crate::context::NetworkAddr;
use crate::context::config::DeviceMode;
use crate::ethernet::wrap_ipv4;
use crate::nat::AllowSubnetExternalRoute;
use crate::protocol::transmission::TransmissionBytes;
use crate::tun::TunInbound;
use pnet_packet::ipv4::Ipv4Packet;
use std::net::Ipv4Addr;
#[derive(Clone)]
pub struct TunDataInbound {
allow_subnet: AllowSubnetExternalRoute,
tun_inbound: TunInbound,
device_mode: DeviceMode,
}
impl TunDataInbound {
pub fn new(tun_inbound: TunInbound, allow_subnet: AllowSubnetExternalRoute) -> Self {
pub fn new(
tun_inbound: TunInbound,
allow_subnet: AllowSubnetExternalRoute,
device_mode: DeviceMode,
) -> Self {
Self {
allow_subnet,
tun_inbound,
device_mode,
}
}
}
impl TunDataInbound {
pub async fn send(&self, data: TransmissionBytes, net: &NetworkAddr) -> anyhow::Result<()> {
pub async fn send_ip(
&self,
data: TransmissionBytes,
net: &NetworkAddr,
src_node: Ipv4Addr,
) -> anyhow::Result<()> {
if data.is_empty() || data[0] >> 4 != 4 {
return Ok(());
}
@@ -33,10 +47,23 @@ impl TunDataInbound {
|| dest.is_multicast()
|| self.allow_subnet.allow(&dest)
{
let data = if self.device_mode == DeviceMode::Tap {
let Some(frame) = wrap_ipv4(data, src_node, net) else {
return Ok(());
};
frame
} else {
data
};
self.tun_inbound.sender.send(data).await?;
}
Ok(())
}
pub async fn send_frame(&self, data: TransmissionBytes) -> anyhow::Result<()> {
self.tun_inbound.sender.send(data).await?;
Ok(())
}
}
#[cfg(test)]
@@ -44,7 +71,6 @@ mod tests {
use super::*;
use crate::nat::AllowSubnetExternalRoute;
use crate::tun::tun_channel;
use std::net::Ipv4Addr;
fn test_net() -> NetworkAddr {
NetworkAddr {
@@ -59,17 +85,29 @@ mod tests {
#[tokio::test]
async fn test_send_empty_or_short_packet_does_not_panic() {
let (tun_inbound, _receiver) = tun_channel();
let inbound = TunDataInbound::new(tun_inbound, AllowSubnetExternalRoute::new(vec![]));
let inbound = TunDataInbound::new(
tun_inbound,
AllowSubnetExternalRoute::new(vec![]),
DeviceMode::Tun,
);
// 零载荷包（头部被剥离后为空）
inbound
.send(TransmissionBytes::zeroed(0), &test_net())
.send_ip(
TransmissionBytes::zeroed(0),
&test_net(),
Ipv4Addr::new(10, 26, 0, 3),
)
.await
.unwrap();
// 过短的包（不足 IPv4 头）
inbound
.send(TransmissionBytes::zeroed(3), &test_net())
.send_ip(
TransmissionBytes::zeroed(3),
&test_net(),
Ipv4Addr::new(10, 26, 0, 3),
)
.await
.unwrap();
}
vnt-core/src/tunnel_core/outbound.rs
+85
View File
@@ -234,6 +234,60 @@ impl HybridOutbound {
self.traffic_stats.record_tx(dest, len);
Ok(())
}
pub async fn ethernet_ipv4_outbound(
&self,
net: NetworkAddr,
data: TransmissionBytes,
mut dest: Ipv4Addr,
) -> anyhow::Result<()> {
if dest == net.gateway {
let Some(ip) = crate::ethernet::strip_ipv4(data) else {
return Ok(());
};
return self.ipv4_gateway_outbound(net, ip).await;
}
if dest.is_multicast() || dest == net.broadcast || dest.is_broadcast() {
return self.ethernet_broadcast_outbound(net, data).await;
}
if !net.network().contains(&dest) {
if let Some(route) = self.external_route.route(&dest) {
dest = route;
} else {
return Ok(());
}
}
self.ethernet_unicast_outbound(net, dest, data).await
}
pub async fn ethernet_unicast_outbound(
&self,
net: NetworkAddr,
dest: Ipv4Addr,
mut data: TransmissionBytes,
) -> anyhow::Result<()> {
let len = data.len() as u64;
data.retreat_head(HEAD_LENGTH)?;
let mut packet = NetPacket::new(data)?;
packet.set_msg_type(MsgType::Turn);
packet.set_src_id(net.ip.into());
packet.set_dest_id(dest.into());
packet.set_ttl(5);
packet.set_ethernet_flag(true);
let packet = self
.packet_compression
.compress(packet, self.basic_outbound.encrypt_reserve())?;
let packet = if let Some(fec_encoder) = &self.fec_encoder {
fec_encoder.encode(packet)?
} else {
packet
};
self.basic_outbound
.send_encrypted_packet(dest, packet)
.await?;
self.traffic_stats.record_tx(dest, len);
Ok(())
}
pub async fn ipv4_gateway_outbound(
&self,
net: NetworkAddr,
@@ -279,6 +333,37 @@ impl HybridOutbound {
.send_raw_broadcast(exclude_ips, packet_bytes)
.await
}
pub async fn ethernet_broadcast_outbound(
&self,
net: NetworkAddr,
mut data: TransmissionBytes,
) -> anyhow::Result<()> {
data.retreat_head(HEAD_LENGTH)?;
let mut packet = NetPacket::new(data)?;
packet.set_msg_type(MsgType::Broadcast);
packet.set_src_id(net.ip.into());
packet.set_dest_id(Ipv4Addr::BROADCAST.into());
packet.set_ttl(5);
packet.set_ethernet_flag(true);
let mut packet = self
.packet_compression
.compress(packet, self.basic_outbound.encrypt_reserve())?;
self.basic_outbound.encrypt_in_place(&mut packet)?;
let packet_bytes = packet.into_bytes();
let list = self.server_info.client_online_ips();
let exclude_ips = self
.basic_outbound
.p2p_broadcast_transmission(&list, 16, &packet_bytes);
if let Some(exclude_ips) = &exclude_ips
&& exclude_ips.len() == list.len()
{
return Ok(());
}
self.basic_outbound
.send_raw_broadcast(exclude_ips, packet_bytes)
.await
}
#[allow(dead_code)]
pub fn has_route(&self, dest: &Ipv4Addr) -> bool {
self.basic_outbound.exists_route(dest)