优化tun\tap处理

2023-06-23 15:31:44 +08:00
parent 11d1ff4b42
commit 17f3fcf9b0
3 changed files with 372 additions and 0 deletions
@@ -0,0 +1,166 @@
+use std::net::{Ipv4Addr, SocketAddrV4};
+use packet::ip::ipv4::packet::IpV4Packet;
+use packet::ip::ipv4::protocol::Protocol;
+use packet::tcp::tcp::TcpPacket;
+use packet::udp::udp::UdpPacket;
+use crate::channel::sender::ChannelSender;
+use crate::external_route::ExternalRoute;
+use crate::handle::{check_dest, CurrentDeviceInfo};
+use crate::ip_proxy::IpProxyMap;
+use crate::protocol::{ip_turn_packet, NetPacket, Version};
+use crate::error::*;
+use crate::igmp_server::IgmpServer;
+use crate::protocol;
+use crate::protocol::ip_turn_packet::BroadcastPacketEnd;
+
+pub mod tun_handler;
+pub mod tap_handler;
+
+async fn broadcast(sender: &ChannelSender, net_packet: &mut NetPacket<&mut [u8]>, data_len: usize, current_device: &CurrentDeviceInfo) -> Result<()> {
+    let mut peer_ips = Vec::with_capacity(8);
+    let vec = sender.direct_route_table_one();
+    for (peer_ip, route) in vec {
+        if sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() {
+            peer_ips.push(peer_ip);
+        }
+    }
+    if peer_ips.is_empty() {
+        sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?;
+    } else {
+        let end_len = 1 + peer_ips.len() * 4;
+        //剩余的发送到服务端，需要告知哪些已发送过
+        //放在末尾可以减少复制次数
+        let mut broadcast = BroadcastPacketEnd::unchecked(&mut net_packet.buffer_mut()[data_len..data_len + end_len]);
+        broadcast.set_address(&peer_ips)?;
+        net_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
+        sender.send_main(&net_packet.buffer()[..(data_len + end_len)], current_device.connect_server).await?;
+    }
+    Ok(())
+}
+
+async fn multicast(igmp_server: &IgmpServer, multicast_addr: Ipv4Addr, sender: &ChannelSender, net_packet: &mut NetPacket<&mut [u8]>, data_len: usize, current_device: &CurrentDeviceInfo) -> Result<()> {
+    let mut peer_ips = Vec::with_capacity(8);
+    let vec = sender.direct_route_table_one();
+    if let Some(members) = igmp_server.load(&multicast_addr) {
+        let members_guard = members.read();
+        for (peer_ip, route) in vec {
+            if members_guard.is_send(&peer_ip) {
+                if sender.send_by_key(&net_packet.buffer()[..data_len], &route.route_key()).await.is_ok() {
+                    peer_ips.push(peer_ip);
+                    if peer_ips.len() == u8::MAX as usize {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+    if peer_ips.is_empty() {
+        sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?;
+    } else {
+        let end_len = 1 + peer_ips.len() * 4;
+        //剩余的发送到服务端，需要告知哪些已发送过
+        //放在末尾可以减少复制次数
+        let mut broadcast = BroadcastPacketEnd::unchecked(&mut net_packet.buffer_mut()[data_len..data_len + end_len]);
+        broadcast.set_address(&peer_ips)?;
+        net_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into());
+        sender.send_main(&net_packet.buffer()[..(data_len + end_len)], current_device.connect_server).await?;
+    }
+    Ok(())
+}
+
+/// 实现一个原地发送，必须保证是如下结构
+/// |12字节开头|ip报文|至少1024字节结尾|
+///
+#[inline]
+pub async fn base_handle(sender: &ChannelSender, buf: &mut [u8],
+                   data_len: usize,//数据总长度=ip长度+12
+                   igmp_server: &IgmpServer,
+                   current_device: CurrentDeviceInfo,
+                   ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> Result<()> {
+    let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?;
+    let protocol = ipv4_packet.protocol();
+    let ip_head_len = ipv4_packet.header_len() as usize * 4;
+    let src_ip = ipv4_packet.source_ip();
+    let mut dest_ip = ipv4_packet.destination_ip();
+    let mut net_packet = NetPacket::new(buf)?;
+    net_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4.into());
+    net_packet.set_version(Version::V1);
+    net_packet.set_protocol(protocol::Protocol::IpTurn);
+    net_packet.first_set_ttl(3);
+    net_packet.set_source(src_ip);
+    net_packet.set_destination(dest_ip);
+    if dest_ip.is_broadcast() || current_device.broadcast_address == dest_ip {
+        // 广播 发送到直连目标
+        if Protocol::Udp == protocol {
+            broadcast(sender, &mut net_packet, data_len, &current_device).await?;
+        }
+        return Ok(());
+    } else if dest_ip.is_multicast() {
+        match protocol {
+            Protocol::Igmp => {
+                //发送到服务端
+                sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?;
+            }
+            Protocol::Udp => {
+                multicast(igmp_server, dest_ip, sender, &mut net_packet, data_len, &current_device).await?;
+            }
+            _ => {}
+        }
+        return Ok(());
+    } else {
+        if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) {
+            if let Some(r_dest_ip) = ip_route.route(&dest_ip) {
+                //路由的目标不能是自己
+                if r_dest_ip == src_ip {
+                    return Ok(());
+                }
+                //需要修改目的地址
+                dest_ip = r_dest_ip;
+                net_packet.set_destination(r_dest_ip);
+            } else {
+                return Ok(());
+            }
+        } else {
+            match protocol {
+                Protocol::Tcp => {
+                    let dest_addr = {
+                        let tcp_packet = TcpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?;
+                        SocketAddrV4::new(dest_ip, tcp_packet.destination_port())
+                    };
+                    if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) {
+                        let source_addr = entry.value().1;
+                        let source_ip = *source_addr.ip();
+                        let mut tcp_packet = TcpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?;
+                        tcp_packet.set_source_port(source_addr.port());
+                        tcp_packet.update_checksum();
+                        let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?;
+                        ipv4_packet.set_source_ip(source_ip);
+                        ipv4_packet.update_checksum();
+                    }
+                }
+                Protocol::Udp => {
+                    let dest_addr = {
+                        let udp_packet = UdpPacket::new(src_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?;
+                        SocketAddrV4::new(dest_ip, udp_packet.destination_port())
+                    };
+                    if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) {
+                        let source_addr = entry.value().1;
+                        let source_ip = *source_addr.ip();
+                        let mut udp_packet = UdpPacket::new(source_ip, dest_ip, &mut net_packet.buffer_mut()[12 + ip_head_len..data_len])?;
+                        udp_packet.set_source_port(source_addr.port());
+                        udp_packet.update_checksum();
+                        let mut ipv4_packet = IpV4Packet::new(&mut net_packet.buffer_mut()[12..data_len])?;
+                        ipv4_packet.set_source_ip(source_ip);
+                        ipv4_packet.update_checksum();
+                    }
+                }
+                _ => {}
+            }
+        }
+    }
+    //优先发到直连到地址
+    if sender.send_by_id(&net_packet.buffer()[..data_len], &dest_ip).await.is_err() {
+        sender.send_main(&net_packet.buffer()[..data_len], current_device.connect_server).await?;
+    }
+    return Ok(());
+}
@@ -0,0 +1,116 @@
+use std::sync::Arc;
+use std::{io, thread};
+use crossbeam::atomic::AtomicCell;
+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::channel::sender::ChannelSender;
+use crate::external_route::ExternalRoute;
+use crate::handle::CurrentDeviceInfo;
+use crate::igmp_server::IgmpServer;
+use crate::ip_proxy::IpProxyMap;
+use crate::tun_tap_device::{DeviceReader, DeviceWriter};
+
+pub fn start(sender: ChannelSender,
+             device_reader: DeviceReader,
+             device_writer: DeviceWriter,
+             igmp_server: IgmpServer,
+             current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+             ip_route: ExternalRoute,
+             ip_proxy_map: IpProxyMap) {
+    thread::Builder::new().name("tap-handler".into()).spawn(move || {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all().build().unwrap()
+            .block_on(async move {
+                if let Err(e) = start_(sender, device_reader,
+                                       device_writer, igmp_server,
+                                       current_device, ip_route, ip_proxy_map).await {
+                    log::warn!("tap:{:?}",e);
+                }
+            });
+
+    }).unwrap();
+}
+
+async fn start_(sender: ChannelSender,
+          device_reader: DeviceReader,
+          device_writer: DeviceWriter,
+          igmp_server: IgmpServer,
+          current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+          ip_route: ExternalRoute,
+          ip_proxy_map: IpProxyMap) -> io::Result<()> {
+    let mut buf = [0; 4096];
+    loop {
+        let len = device_reader.read(&mut buf)?;
+        if let Err(e) = handle(&mut buf, len, &igmp_server, &current_device, &device_writer, &sender, &ip_route, &ip_proxy_map).await {
+            log::error!("tap handle{:?}",e);
+        }
+    }
+}
+
+async fn handle(buf: &mut [u8], len: usize, igmp_server: &IgmpServer, current_device: &AtomicCell<CurrentDeviceInfo>,
+          device_writer: &DeviceWriter, sender: &ChannelSender, ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> crate::Result<()> {
+    let mut ethernet_packet = EthernetPacket::new(&mut buf[..len])?;
+    let current_device = current_device.load();
+    match ethernet_packet.protocol() {
+        ethernet::protocol::Protocol::Arp => {
+            let mut out_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);
+            device_writer.write_ethernet_tap(&out_ethernet_packet.buffer)?;
+        }
+        ethernet::protocol::Protocol::Ipv4 => {
+            let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut());
+            let src_ip = ipv4_packet.source_ip();
+            if src_ip != current_device.virtual_ip() {
+                return Ok(());
+            }
+            let dest_ip = ipv4_packet.destination_ip();
+            let protocol = ipv4_packet.protocol();
+            if src_ip == dest_ip {
+                if protocol == ipv4::protocol::Protocol::Icmp {
+                    let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
+                    if icmp.kind() == Kind::EchoRequest {
+                        icmp.set_kind(Kind::EchoReply);
+                        icmp.update_checksum();
+                        ipv4_packet.set_source_ip(dest_ip);
+                        ipv4_packet.set_destination_ip(src_ip);
+                        ipv4_packet.update_checksum();
+                        let source = ethernet_packet.source().to_vec();
+                        let dest = ethernet_packet.destination().to_vec();
+                        ethernet_packet.set_source(&dest);
+                        ethernet_packet.set_destination(&source);
+                        device_writer.write_ethernet_tap(&ethernet_packet.buffer)?;
+                    }
+                }
+                return Ok(());
+            }
+            // 以太网帧头部14字节，预留12字节
+            return crate::handle::tun_tap::base_handle(sender, &mut buf[2..], len - 2, igmp_server, current_device,
+                                                       ip_route, proxy_map).await;
+        }
+        _ => {
+            // log::warn!("不支持的二层协议：{:?}",p)
+        }
+    }
+    Ok(())
+}
+
@@ -0,0 +1,90 @@
+use std::{io, thread};
+use std::sync::Arc;
+
+use crossbeam::atomic::AtomicCell;
+
+use packet::icmp::Kind;
+use packet::icmp::icmp::IcmpPacket;
+use packet::ip::ipv4;
+use packet::ip::ipv4::packet::IpV4Packet;
+use crate::channel::sender::ChannelSender;
+
+use crate::error::*;
+use crate::external_route::ExternalRoute;
+use crate::handle::CurrentDeviceInfo;
+use crate::igmp_server::IgmpServer;
+use crate::ip_proxy::IpProxyMap;
+use crate::tun_tap_device::{DeviceReader, DeviceWriter};
+
+fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
+    if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
+        let mut icmp = IcmpPacket::new(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();
+            device_writer.write_ipv4_tun(ipv4_packet.buffer)?;
+        }
+    }
+    Ok(())
+}
+
+/// 接收tun数据，并且转发到udp上
+#[inline]
+async fn handle(sender: &ChannelSender, data: &mut [u8], len: usize, device_writer: &DeviceWriter, igmp_server: &IgmpServer, current_device: CurrentDeviceInfo,
+                ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> Result<()> {
+    let ipv4_packet = if let Ok(ipv4_packet) = IpV4Packet::new(&mut data[12..len]) {
+        ipv4_packet
+    } else {
+        return Ok(());
+    };
+    let src_ip = ipv4_packet.source_ip();
+    let dest_ip = ipv4_packet.destination_ip();
+    if src_ip != current_device.virtual_ip() {
+        return Ok(());
+    }
+    if src_ip == dest_ip {
+        return icmp(&device_writer, ipv4_packet);
+    }
+    return crate::handle::tun_tap::base_handle(sender, data, len, igmp_server, current_device, ip_route, proxy_map).await;
+}
+
+pub fn start(sender: ChannelSender,
+             device_reader: DeviceReader,
+             device_writer: DeviceWriter,
+             igmp_server: IgmpServer,
+             current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+             ip_route: ExternalRoute,
+             ip_proxy_map: IpProxyMap) {
+    thread::Builder::new().name("tun-handler".into()).spawn(move || {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all().build().unwrap()
+            .block_on(async move {
+                if let Err(e) = start_(sender, device_reader, device_writer, igmp_server, current_device, ip_route, ip_proxy_map).await {
+                    log::warn!("tun:{:?}",e);
+                }
+            })
+    }).unwrap();
+}
+
+async fn start_(sender: ChannelSender,
+                device_reader: DeviceReader,
+                device_writer: DeviceWriter,
+                igmp_server: IgmpServer,
+                current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
+                ip_route: ExternalRoute,
+                ip_proxy_map: IpProxyMap) -> io::Result<()> {
+    let mut buf = [0; 4096];
+    loop {
+        let len = device_reader.read(&mut buf[12..])? + 12;
+        match handle(&sender, &mut buf, len, &device_writer, &igmp_server, current_device.load(), &ip_route, &ip_proxy_map).await {
+            Ok(_) => {}
+            Err(e) => {
+                log::warn!("{:?}", e)
+            }
+        }
+    }
+}