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

支持tap网卡,优化tun网卡配置

Browse Source
This commit is contained in:
lubeilin
2023-05-07 18:32:11 +08:00
parent 068580e036
commit 35ed7f7e45
68 changed files with 2810 additions and 2393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch/src/handle/heartbeat_handler.rs
+48 -21
View File
@@ -1,45 +1,60 @@
use std::{io, thread};
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
use std::{io, thread};
use chrono::Local;
use crossbeam::atomic::AtomicCell;
use parking_lot::Mutex;
use rand::prelude::SliceRandom;
use p2p_channel::channel::Route;
use p2p_channel::channel::sender::Sender;
use p2p_channel::channel::Route;
use p2p_channel::idle::Idle;
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, Version};
use crate::protocol::control_packet::PingPacket;
use crate::protocol::{control_packet, NetPacket, Protocol, Version, MAX_TTL};
pub fn start_idle(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) {
thread::Builder::new().name("idle".into()).spawn(move || {
if let Err(e) = start_idle_(idle, sender) {
log::info!("空闲检测线程停止:{:?}",e);
}
}).unwrap();
thread::Builder::new()
.name("idle".into())
.spawn(move || {
if let Err(e) = start_idle_(idle, sender) {
log::info!("空闲检测线程停止:{:?}", e);
}
})
.unwrap();
}
fn start_idle_(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) -> io::Result<()> {
loop {
let (idle_status, peer_ips, route) = idle.next_idle()?;
log::warn!("peer_ip:{:?},route:{:?},idle_status:{:?}",peer_ips,route,idle_status);
log::warn!(
"peer_ip:{:?},route:{:?},idle_status:{:?}",
peer_ips,
route,
idle_status
);
for peer_ip in peer_ips {
sender.remove_route(&peer_ip);
}
}
}
pub fn start_heartbeat(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::Builder::new().name("heartbeat".into()).spawn(move || {
if let Err(e) = start_heartbeat_(sender, device_list, current_device) {
log::info!("空闲检测线程停止:{:?}",e);
}
}).unwrap();
pub fn start_heartbeat(
sender: Sender<Ipv4Addr>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
) {
thread::Builder::new()
.name("heartbeat".into())
.spawn(move || {
if let Err(e) = start_heartbeat_(sender, device_list, current_device) {
log::info!("空闲检测线程停止:{:?}", e);
}
})
.unwrap();
}
fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> {
@@ -49,7 +64,11 @@ fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> {
Ok(())
}
fn start_heartbeat_(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<()> {
fn start_heartbeat_(
sender: Sender<Ipv4Addr>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
) -> io::Result<()> {
let mut net_packet = NetPacket::new([0u8; 16])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Control);
@@ -71,7 +90,10 @@ fn start_heartbeat_(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<P
set_now_time(&mut net_packet)?;
net_packet.first_set_ttl(MAX_TTL);
net_packet.set_destination(peer.virtual_ip);
if sender.send_to_id(net_packet.buffer(), &peer.virtual_ip).is_err() {
if sender
.send_to_id(net_packet.buffer(), &peer.virtual_ip)
.is_err()
{
//没有路由则发送到网关
let _ = sender.send_to_addr(net_packet.buffer(), current_device.connect_server);
//再随机发送到其他地址，看有没有客户端符合转发条件
@@ -97,15 +119,20 @@ fn start_heartbeat_(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<P
}
set_now_time(&mut net_packet)?;
net_packet.set_destination(current_device.virtual_gateway());
if let Err(e) = sender.send_to_addr(net_packet.buffer(), current_device.connect_server) {
log::warn!("connect_server:{:?},e:{:?}",current_device.connect_server,e);
if let Err(e) = sender.send_to_addr(net_packet.buffer(), current_device.connect_server)
{
log::warn!(
"connect_server:{:?},e:{:?}",
current_device.connect_server,
e
);
}
} else {
for (peer_ip, route) in sender.route_table().iter() {
set_now_time(&mut net_packet)?;
net_packet.set_destination(*peer_ip);
if let Err(e) = sender.send_to_route(net_packet.buffer(), &route.route_key()) {
log::warn!("peer_ip:{:?},route:{:?},e:{:?}",peer_ip,route,e);
log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e);
}
thread::sleep(Duration::from_millis(1));
}
@@ -114,4 +141,4 @@ fn start_heartbeat_(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<P
count += 1;
thread::sleep(Duration::from_millis(5000));
}
}
}
switch/src/handle/mod.rs
+6 -6
View File
@@ -1,10 +1,11 @@
use std::net::{Ipv4Addr, SocketAddr};
pub mod heartbeat_handler;
pub mod punch_handler;
pub mod registration_handler;
pub mod tun_handler;
pub mod tap_handler;
pub mod punch_handler;
pub mod recv_handler;
pub mod registration_handler;
/// 是否在一个网段
fn check_dest(dest: Ipv4Addr, virtual_netmask: Ipv4Addr, virtual_network: Ipv4Addr) -> bool {
@@ -70,6 +71,7 @@ pub struct CurrentDeviceInfo {
pub broadcast_address: Ipv4Addr,
//链接的服务器地址
pub connect_server: SocketAddr,
pub mac:[u8;6]
}
impl CurrentDeviceInfo {
@@ -78,6 +80,7 @@ impl CurrentDeviceInfo {
virtual_gateway: Ipv4Addr,
virtual_netmask: Ipv4Addr,
connect_server: SocketAddr,
mac:[u8;6],
) -> Self {
let broadcast_address = (!u32::from_be_bytes(virtual_netmask.octets()))
| u32::from_be_bytes(virtual_gateway.octets());
@@ -92,6 +95,7 @@ impl CurrentDeviceInfo {
virtual_network,
broadcast_address,
connect_server,
mac
}
}
#[inline]
@@ -103,7 +107,3 @@ impl CurrentDeviceInfo {
self.virtual_gateway
}
}
switch/src/handle/punch_handler.rs
+67 -40
View File
@@ -1,35 +1,45 @@
use std::{io, thread};
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;
use std::time::Duration;
use crossbeam::atomic::AtomicCell;
use parking_lot::Mutex;
use protobuf::Message;
use rand::prelude::SliceRandom;
use p2p_channel::channel::sender::Sender;
use p2p_channel::punch::{NatInfo, NatType, Punch};
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
use crate::nat::NatTest;
use crate::proto::message::{PunchInfo, PunchNatType};
use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, turn_packet, Version};
use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version, MAX_TTL};
use crossbeam::atomic::AtomicCell;
use p2p_channel::channel::sender::Sender;
use p2p_channel::punch::{NatInfo, NatType, Punch};
use parking_lot::Mutex;
use protobuf::Message;
use rand::prelude::SliceRandom;
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;
use std::time::Duration;
use std::{io, thread};
pub fn start_cone(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::Builder::new().name("punch-cone".into()).spawn(move || {
if let Err(e) = start_(true, punch, current_device) {
log::warn!("锥形网络打洞处理线程停止 {:?}",e);
}
}).unwrap();
thread::Builder::new()
.name("punch-cone".into())
.spawn(move || {
if let Err(e) = start_(true, punch, current_device) {
log::warn!("锥形网络打洞处理线程停止 {:?}", e);
}
})
.unwrap();
}
pub fn start_symmetric(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::Builder::new().name("punch-symmetric".into()).spawn(move || {
if let Err(e) = start_(false, punch, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}",e);
}
}).unwrap();
thread::Builder::new()
.name("punch-symmetric".into())
.spawn(move || {
if let Err(e) = start_(false, punch, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}", e);
}
})
.unwrap();
}
fn start_(is_cone: bool, mut punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<()> {
fn start_(
is_cone: bool,
mut punch: Punch<Ipv4Addr>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
) -> io::Result<()> {
let mut packet = NetPacket::new([0u8; 12])?;
packet.set_version(Version::V1);
packet.first_set_ttl(1);
@@ -49,22 +59,35 @@ fn start_(is_cone: bool, mut punch: Punch<Ipv4Addr>, current_device: Arc<AtomicC
}
packet.set_source(current_device.load().virtual_ip());
packet.set_destination(peer_ip);
log::info!("发起打洞，目标:{:?},{:?}",peer_ip,nat_info);
log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info);
if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info) {
log::warn!("peer_ip:{:?},e:{:?}",peer_ip,e);
log::warn!("peer_ip:{:?},e:{:?}", peer_ip, e);
}
}
}
pub fn start_punch(nat_test: NatTest, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, sender: Sender<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::Builder::new().name("punch-send-request".into()).spawn(move || {
if let Err(e) = start_punch_(nat_test, device_list, sender, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}",e);
}
}).unwrap();
pub fn start_punch(
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
sender: Sender<Ipv4Addr>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
) {
thread::Builder::new()
.name("punch-send-request".into())
.spawn(move || {
if let Err(e) = start_punch_(nat_test, device_list, sender, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}", e);
}
})
.unwrap();
}
fn start_punch_(nat_test: NatTest, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, sender: Sender<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> crate::Result<()> {
fn start_punch_(
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
sender: Sender<Ipv4Addr>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
) -> crate::Result<()> {
loop {
if sender.is_close() {
return Ok(());
@@ -104,19 +127,23 @@ fn start_punch_(nat_test: NatTest, device_list: Arc<Mutex<(u16, Vec<PeerDeviceIn
}
}
pub fn punch_packet(virtual_ip: Ipv4Addr, nat_info: &NatInfo, dest: Ipv4Addr) -> crate::Result<Vec<u8>> {
pub fn punch_packet(
virtual_ip: Ipv4Addr,
nat_info: &NatInfo,
dest: Ipv4Addr,
) -> crate::Result<Vec<u8>> {
let mut punch_reply = PunchInfo::new();
punch_reply.reply = false;
punch_reply.public_ip_list = nat_info.public_ips.iter().map(|i| {
match i {
IpAddr::V4(ip) => {
u32::from_be_bytes(ip.octets())
}
punch_reply.public_ip_list = nat_info
.public_ips
.iter()
.map(|i| match i {
IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
IpAddr::V6(_) => {
panic!()
}
}
}).collect();
})
.collect();
punch_reply.public_port = nat_info.public_port as u32;
punch_reply.public_port_range = nat_info.public_port_range as u32;
punch_reply.local_ip = match nat_info.local_ip {
@@ -137,4 +164,4 @@ pub fn punch_packet(virtual_ip: Ipv4Addr, nat_info: &NatInfo, dest: Ipv4Addr) ->
net_packet.set_destination(dest);
net_packet.set_payload(&bytes);
Ok(net_packet.into_buffer())
}
}
switch/src/handle/recv_handler.rs
+47 -19
View File
@@ -10,6 +10,7 @@ use protobuf::Message;
use p2p_channel::channel::{Channel, Route, RouteKey};
use p2p_channel::punch::NatInfo;
use packet::ethernet;
use packet::icmp::{icmp, Kind};
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
@@ -23,6 +24,7 @@ use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationRes
use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, turn_packet, Version};
use crate::protocol::control_packet::ControlPacket;
use crate::protocol::error_packet::InErrorPacket;
use crate::tap_device::TapWriter;
use crate::tun_device::TunWriter;
pub fn start(mut handler: RecvHandler) {
@@ -57,7 +59,8 @@ pub struct RecvHandler {
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
register: Arc<Register>,
nat_test: NatTest,
tun_writer: TunWriter,
tun_writer: Option<TunWriter>,
tap_writer: Option<TapWriter>,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
}
@@ -68,7 +71,8 @@ impl RecvHandler {
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
register: Arc<Register>,
nat_test: NatTest,
tun_writer: TunWriter,
tun_writer: Option<TunWriter>,
tap_writer: Option<TapWriter>,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
) -> Self {
@@ -79,6 +83,7 @@ impl RecvHandler {
register,
nat_test,
tun_writer,
tap_writer,
connect_status,
peer_nat_info_map,
}
@@ -91,6 +96,7 @@ impl RecvHandler {
register: self.register.clone(),
nat_test: self.nat_test.clone(),
tun_writer: self.tun_writer.clone(),
tap_writer: self.tap_writer.clone(),
connect_status: self.connect_status.clone(),
peer_nat_info_map: self.peer_nat_info_map.clone(),
})
@@ -103,6 +109,7 @@ impl RecvHandler {
if net_packet.ttl() == 0 {
return Ok(());
}
net_packet.set_ttl(net_packet.ttl() - 1);
let source = net_packet.source();
let current_device = self.current_device.load();
if source == current_device.virtual_ip() {
@@ -122,7 +129,6 @@ impl RecvHandler {
let ttl = net_packet.ttl();
if ttl > 1 {
// 转发
net_packet.set_ttl(ttl - 1);
if let Some(route) = self.channel.route(&destination) {
if route.metric <= net_packet.ttl() {
self.channel.send_to_route(net_packet.buffer(), &route.route_key())?;
@@ -138,22 +144,38 @@ impl RecvHandler {
match net_packet.protocol() {
Protocol::Ipv4Turn => {
let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?;
if ipv4.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
if icmp_packet.kind() == Kind::EchoRequest {
//开启ping
icmp_packet.set_kind(Kind::EchoReply);
icmp_packet.update_checksum();
ipv4.set_source_ip(destination);
ipv4.set_destination_ip(source);
ipv4.update_checksum();
net_packet.set_source(destination);
net_packet.set_destination(source);
self.channel.send_to_route(net_packet.buffer(), route_key)?;
return Ok(());
if ipv4.destination_ip() != destination {
//todo 外部数据转发
} else {
if ipv4.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
if icmp_packet.kind() == Kind::EchoRequest {
//开启ping
icmp_packet.set_kind(Kind::EchoReply);
icmp_packet.update_checksum();
ipv4.set_source_ip(destination);
ipv4.set_destination_ip(source);
ipv4.update_checksum();
net_packet.set_source(destination);
net_packet.set_destination(source);
self.channel.send_to_route(net_packet.buffer(), route_key)?;
return Ok(());
}
}
if let Some(tun_writer) = &self.tun_writer {
tun_writer.write(net_packet.payload())?;
} else {
if let Some(tap_writer) = &self.tap_writer {
let mut ethernet_packet = ethernet::packet::EthernetPacket::unchecked(vec![0; 14 + ipv4.buffer.len()]);
let source = source.octets();
ethernet_packet.set_source(&[source[0], source[1], source[2], source[3], 123, 234]);
ethernet_packet.set_destination(&current_device.mac);
ethernet_packet.set_protocol(ethernet::protocol::Protocol::Ipv4);
ethernet_packet.payload_mut().copy_from_slice(ipv4.buffer);
tap_writer.write(&ethernet_packet.buffer)?;
}
}
}
self.tun_writer.write(net_packet.payload())?;
}
Protocol::Service => {
self.service(current_device, source, net_packet, route_key)?;
@@ -195,9 +217,15 @@ impl RecvHandler {
let virtual_ip = Ipv4Addr::from(response.virtual_ip);
let virtual_gateway = Ipv4Addr::from(response.virtual_gateway);
let virtual_netmask = Ipv4Addr::from(response.virtual_netmask);
self.tun_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
if let Some(tun_writer) = &self.tun_writer {
tun_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
} else {
if let Some(tap_writer) = &self.tap_writer {
tap_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
}
}
let new_current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway,
virtual_netmask, current_device.connect_server);
virtual_netmask, current_device.connect_server, current_device.mac);
if let Err(e) = self.current_device.compare_exchange(current_device, new_current_device) {
log::warn!("替换失败:{:?}",e);
}
switch/src/handle/registration_handler.rs
+24 -26
View File
@@ -4,9 +4,9 @@ use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;
use chrono::Local;
use protobuf::Message;
use p2p_channel::channel::Channel;
use p2p_channel::channel::sender::Sender;
use p2p_channel::channel::Channel;
use protobuf::Message;
use crate::error::*;
use crate::proto::message::{RegistrationRequest, RegistrationResponse};
@@ -35,20 +35,14 @@ pub fn registration(
Protocol::Service => {
match service_packet::Protocol::from(net_packet.transport_protocol()) {
service_packet::Protocol::RegistrationResponse => {
let response =
RegistrationResponse::parse_from_bytes(net_packet.payload())?;
let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
Ok(response)
}
_ => {
Err(Error::Warn(format!("数据错误：{:?}", net_packet)))
}
_ => Err(Error::Warn(format!("数据错误：{:?}", net_packet))),
}
}
Protocol::Error => {
match InErrorPacket::new(
net_packet.transport_protocol(),
net_packet.payload(),
) {
match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) {
Ok(e) => match e {
InErrorPacket::TokenError => Err(Error::Stop("token错误".to_string())),
InErrorPacket::Disconnect => Err(Error::Warn("断开连接".to_string())),
@@ -61,9 +55,7 @@ pub fn registration(
Err(e) => Err(Error::Warn(format!("{:?}", e))),
}
}
_ => {
Err(Error::Warn(format!("数据错误：{:?}", net_packet)))
}
_ => Err(Error::Warn(format!("数据错误：{:?}", net_packet))),
};
}
@@ -99,11 +91,13 @@ pub struct Register {
}
impl Register {
pub fn new(sender: Sender<Ipv4Addr>,
server_address: SocketAddr,
token: String,
device_id: String,
name: String, ) -> Self {
pub fn new(
sender: Sender<Ipv4Addr>,
server_address: SocketAddr,
token: String,
device_id: String,
name: String,
) -> Self {
Self {
sender,
server_address,
@@ -117,18 +111,22 @@ impl Register {
let last = self.time.load(Ordering::Relaxed);
let new = Local::now().timestamp_millis();
if new - last < 1000
|| self.time
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
|| self
.time
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
{
//短时间不重复注册
return Ok(());
}
log::info!("重新连接");
let request_packet =
registration_request_packet(self.token.clone(),
self.device_id.clone(),
self.name.clone(), false).unwrap();
let request_packet = registration_request_packet(
self.token.clone(),
self.device_id.clone(),
self.name.clone(),
false,
)
.unwrap();
let buf = request_packet.buffer();
self.sender.send_to_addr(buf, self.server_address)?;
Ok(())
switch/src/handle/tap_handler.rs
+112
View File
@@ -0,0 +1,112 @@
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::{io, thread};
use crossbeam::atomic::AtomicCell;
use p2p_channel::channel::sender::Sender;
use packet::arp::arp::ArpPacket;
use packet::ethernet;
use packet::ethernet::packet::EthernetPacket;
use packet::icmp::icmp::IcmpPacket;
use packet::icmp::Kind;
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use crate::handle::{check_dest, CurrentDeviceInfo};
use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
use crate::tap_device::{TapReader, TapWriter};
pub fn start(sender: Sender<Ipv4Addr>,
tap_reader: TapReader,
tap_writer: TapWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) {
thread::Builder::new().name("tap-handler".into()).spawn(move || {
if let Err(e) = start_(sender, tap_reader, tap_writer, current_device) {
log::warn!("{:?}",e);
}
}).unwrap();
}
fn start_(sender: Sender<Ipv4Addr>,
tap_reader: TapReader,
tap_writer: TapWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) -> io::Result<()> {
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500]).unwrap();
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Ipv4Turn);
net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
net_packet.set_ttl(MAX_TTL);
let mut buf = [0; 2048];
loop {
let len = tap_reader.read(&mut buf)?;
if len == 0 {
continue;
}
let mut ethernet_packet = EthernetPacket::unchecked(&mut buf[..len]);
if let Err(e) = handle(&mut net_packet, &current_device, &tap_writer, &mut ethernet_packet, &sender) {
log::error!("tap handle{:?}",e);
}
}
}
fn handle(net_packet: &mut NetPacket<Vec<u8>>, current_device: &AtomicCell<CurrentDeviceInfo>, tap_writer: &TapWriter, ethernet_packet: &mut EthernetPacket<&mut [u8]>, sender: &Sender<Ipv4Addr>) -> io::Result<()> {
let current_device = current_device.load();
match ethernet_packet.protocol() {
ethernet::protocol::Protocol::Arp => {
let mut out_ethernet_packet = ethernet::packet::EthernetPacket::unchecked(ethernet_packet.buffer.to_vec());
let arp_packet = ArpPacket::unchecked(ethernet_packet.payload());
let mut out_arp_packet = ArpPacket::unchecked(out_ethernet_packet.payload_mut());
let sender_h = arp_packet.sender_hardware_addr();
let sender_p = arp_packet.sender_protocol_addr();
let target_p = arp_packet.target_protocol_addr();
if target_p == &[0, 0, 0, 0] || sender_p == &[0, 0, 0, 0] || target_p == sender_p {
return Ok(());
}
//回复一个虚假的MAC地址
out_arp_packet.set_sender_hardware_addr(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]);
out_arp_packet.set_sender_protocol_addr(target_p);
out_arp_packet.set_target_hardware_addr(sender_h);
out_arp_packet.set_target_protocol_addr(sender_p);
out_arp_packet.set_op_code(2);
out_ethernet_packet.set_source(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]);
out_ethernet_packet.set_destination(sender_h);
tap_writer.write(&out_ethernet_packet.buffer)?;
}
ethernet::protocol::Protocol::Ipv4 => {
// println!("in ethernet_packet {:?}", ethernet_packet);
let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut());
let src_ip = ipv4_packet.source_ip();
let dest_ip = ipv4_packet.destination_ip();
if src_ip != current_device.virtual_ip() || (!check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) && !dest_ip.is_broadcast()) {
return Ok(());
}
if src_ip == dest_ip {
if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp = IcmpPacket::unchecked(ipv4_packet.payload_mut());
if icmp.kind() == Kind::EchoRequest {
icmp.set_kind(Kind::EchoReply);
icmp.update_checksum();
let src = ipv4_packet.source_ip();
ipv4_packet.set_source_ip(ipv4_packet.destination_ip());
ipv4_packet.set_destination_ip(src);
ipv4_packet.update_checksum();
tap_writer.write(ethernet_packet.buffer)?;
return Ok(());
}
}
}
net_packet.set_source(src_ip);
net_packet.set_destination(dest_ip);
let data_len = ipv4_packet.buffer.len();
net_packet.set_payload(ipv4_packet.buffer);
//优先发到直连到地址
if sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip).is_err() {
sender.send_to_addr(&net_packet.buffer()[..(12 + data_len)], current_device.connect_server)?;
}
}
p => {
log::warn!("不支持的二层协议：{:?}",p)
}
}
Ok(())
}
switch/src/handle/tun_handler.rs
+6 -6
View File
@@ -1,7 +1,7 @@
use std::{io, thread};
/// 接收tun数据，并且转发到udp上
use std::net::Ipv4Addr;
use std::sync::Arc;
use crossbeam::atomic::AtomicCell;
use p2p_channel::channel::sender::Sender;
@@ -15,7 +15,6 @@ use crate::handle::{check_dest, CurrentDeviceInfo};
use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
use crate::tun_device::{TunReader, TunWriter};
fn icmp(tun_writer: &TunWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
@@ -32,6 +31,7 @@ fn icmp(tun_writer: &TunWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Resul
Ok(())
}
/// 接收tun数据，并且转发到udp上
#[inline]
fn handle(sender: &Sender<Ipv4Addr>, data: &mut [u8], tun_writer: &TunWriter, current_device: CurrentDeviceInfo, net_packet: &mut NetPacket<Vec<u8>>) -> Result<()> {
let data_len = data.len();
@@ -68,7 +68,7 @@ fn handle(sender: &Sender<Ipv4Addr>, data: &mut [u8], tun_writer: &TunWriter, cu
pub fn start(sender: Sender<Ipv4Addr>,
tun_reader: TunReader,
tun_writer: TunWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) {
current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::Builder::new().name("tun-handler".into()).spawn(move || {
if let Err(e) = start_(sender, tun_reader, tun_writer, current_device) {
log::warn!("{:?}",e);
@@ -80,7 +80,7 @@ pub fn start(sender: Sender<Ipv4Addr>,
fn start_(sender: Sender<Ipv4Addr>,
tun_reader: TunReader,
tun_writer: TunWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) -> io::Result<()> {
current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<()> {
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Ipv4Turn);
@@ -109,8 +109,8 @@ fn start_(sender: Sender<Ipv4Addr>,
net_packet.set_ttl(MAX_TTL);
let mut buf = [0; 4096];
loop {
let data = tun_reader.read(&mut buf)?;
match handle(&sender, data, &tun_writer, current_device.load(), &mut net_packet) {
let len = tun_reader.read(&mut buf)?;
match handle(&sender, &mut buf[..len], &tun_writer, current_device.load(), &mut net_packet) {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)