diff --git a/vnt/src/igmp_server/mod.rs b/vnt/src/igmp_server/mod.rs deleted file mode 100644 index 5877b05..0000000 --- a/vnt/src/igmp_server/mod.rs +++ /dev/null @@ -1,233 +0,0 @@ -use std::collections::{HashMap, HashSet}; -use std::net::Ipv4Addr; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use parking_lot::RwLock; - -use packet::igmp::igmp_v2::IgmpV2Packet; -use packet::igmp::igmp_v3::{IgmpV3QueryPacket, IgmpV3RecordType, IgmpV3ReportPacket}; -use packet::igmp::IgmpType; -use packet::ip::ipv4::protocol::Protocol; - -use crate::tun_tap_device::DeviceWriter; - -//1. 定时发送query，启动时20秒一次，连发3次，之后8分钟一次 -//2. 接收网关的igmp report 维护组播源信息 -#[derive(Clone, Debug)] -pub struct Multicast { - //成员虚拟ip - members: HashMap, - //是否是过滤模式 - //成员过滤或包含的源ip - map: HashMap)>, -} - -impl Multicast { - pub fn new() -> Self { - Self { - members: Default::default(), - map: Default::default(), - } - } - pub fn is_send(&self, ip: &Ipv4Addr) -> bool { - if self.members.contains_key(ip) { - if let Some((is_include, set)) = self.map.get(ip) { - if *is_include { - set.contains(ip) - } else { - !set.contains(ip) - } - } else { - true - } - } else { - false - } - } -} - -#[derive(Clone)] -pub struct IgmpServer { - multicast: Arc>>>>, -} - -impl IgmpServer { - pub fn new(device_writer: DeviceWriter) -> Self { - let multicast: Arc>>>> = - Arc::new(RwLock::new(HashMap::with_capacity(16))); - std::thread::spawn(move || { - //预留以太网帧头和ip头 - let mut buf = [0; 14 + 24 + 12]; - let dest = Ipv4Addr::new(224, 0, 0, 1); - let src = Ipv4Addr::new(10, 26, 0, 1); - { - let buf = &mut buf[14..]; - let len = buf.len(); - // ipv4 头部20字节 - buf[0] = 0b0100_0110; - //写入总长度 - buf[2..4].copy_from_slice(&(len as u16).to_be_bytes()); - //ttl - buf[8] = 1; - buf[20] = 0x94; - buf[21] = 0x04; - let mut ipv4 = packet::ip::ipv4::packet::IpV4Packet::unchecked(buf); - ipv4.set_flags(2); - ipv4.set_protocol(Protocol::Igmp); - ipv4.set_source_ip(src); - ipv4.set_destination_ip(dest); - ipv4.update_checksum(); - } - { - let mut igmp_query = IgmpV3QueryPacket::unchecked(&mut buf[14 + 24..]); - igmp_query.set_igmp_type(); - igmp_query.set_max_resp_code(50); - igmp_query.set_group_address(Ipv4Addr::UNSPECIFIED); - igmp_query.set_qrv(2); - igmp_query.set_qqic(10); - igmp_query.update_checksum(); - } - loop { - let _ = device_writer.write_ipv4(&mut buf); - std::thread::sleep(Duration::from_secs(20)) - } - }); - Self { multicast } - } - pub fn load(&self, multicast_addr: &Ipv4Addr) -> Option>> { - self.multicast.read().get(multicast_addr).cloned() - } - pub fn handle(&self, buf: &[u8], source: Ipv4Addr) -> crate::Result<()> { - let multicast = self.multicast.read(); - for (_, v) in multicast.iter() { - let mut list = Vec::new(); - let mut write_guard = v.write(); - for (ip, time) in &write_guard.members { - if time.elapsed() > Duration::from_secs(30) { - list.push(*ip); - } - } - for ip in list { - write_guard.members.remove(&ip); - write_guard.map.remove(&ip); - } - } - match IgmpType::from(buf[0]) { - IgmpType::Query => {} - IgmpType::ReportV1 | IgmpType::ReportV2 => { - //加入组播，v1和v2差不多 - let report = IgmpV2Packet::new(buf)?; - let multicast_addr = report.group_address(); - if !multicast_addr.is_multicast() { - return Ok(()); - } - let multi = self.add_multicast(multicast_addr); - let mut guard = multi.write(); - guard.members.insert(source, Instant::now()); - } - IgmpType::LeaveV2 => { - //退出组播 - let leave = IgmpV2Packet::new(buf)?; - let multicast_addr = leave.group_address(); - if !multicast_addr.is_multicast() { - return Ok(()); - } - if let Some(entry) = self.load(&multicast_addr) { - let mut guard = entry.write(); - guard.map.remove(&source); - guard.members.remove(&source); - } - } - IgmpType::ReportV3 => { - let report = IgmpV3ReportPacket::new(buf)?; - if let Some(group_records) = report.group_records() { - for group_record in group_records { - let multicast_addr = group_record.multicast_address(); - if !multicast_addr.is_multicast() { - return Ok(()); - } - let multi = self.add_multicast(multicast_addr); - let mut guard = multi.write(); - - match group_record.record_type() { - IgmpV3RecordType::ModeIsInclude - | IgmpV3RecordType::ChangeToIncludeMode => { - match group_record.source_addresses() { - None => { - //不接收所有 - guard.members.remove(&source); - guard.map.remove(&source); - } - Some(src) => { - guard.members.insert(source, Instant::now()); - guard.map.insert(source, (true, HashSet::from_iter(src))); - } - } - } - - IgmpV3RecordType::ModeIsExclude - | IgmpV3RecordType::ChangeToExcludeMode => { - match group_record.source_addresses() { - None => { - //接收所有 - guard.members.insert(source, Instant::now()); - guard.map.remove(&source); - } - Some(src) => { - guard.members.insert(source, Instant::now()); - guard.map.insert(source, (false, HashSet::from_iter(src))); - } - } - } - IgmpV3RecordType::AllowNewSources => { - //在已有源的基础上，接收目标源，如果是排除模式，则删除；是包含模式则添加 - match group_record.source_addresses() { - None => {} - Some(src) => match guard.map.get_mut(&source) { - None => {} - Some((is_include, set)) => { - for ip in src { - if *is_include { - set.insert(ip); - } else { - set.remove(&ip); - } - } - } - }, - } - } - IgmpV3RecordType::BlockOldSources => { - //在已有源的基础上，不接收目标源 - match group_record.source_addresses() { - None => {} - Some(src) => match guard.map.get_mut(&source) { - None => {} - Some((is_include, set)) => { - for ip in src { - if *is_include { - set.remove(&ip); - } else { - set.insert(ip); - } - } - } - }, - } - } - IgmpV3RecordType::Unknown(_) => {} - } - } - } - } - IgmpType::Unknown(_) => {} - } - Ok(()) - } - fn add_multicast(&self, multicast_addr: Ipv4Addr) -> Arc> { - let value = Arc::new(RwLock::new(Multicast::new())); - self.multicast.write().insert(multicast_addr, value.clone()); - value - } -} diff --git a/vnt/src/nat/stun_test.rs b/vnt/src/nat/stun_test.rs deleted file mode 100644 index e2e5866..0000000 --- a/vnt/src/nat/stun_test.rs +++ /dev/null @@ -1,137 +0,0 @@ -use std::collections::HashSet; -use std::io; -use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; -use std::time::Duration; - -use crate::channel::punch::NatType; -use stun_format::Attr; -use tokio::net::UdpSocket; - -pub async fn stun_test_nat(stun_servers: Vec) -> io::Result<(NatType, Vec, u16)> { - let mut h = Vec::new(); - for x in stun_servers { - let handle = tokio::spawn(test_nat(x)); - h.push(handle); - } - let mut nat_type = NatType::Cone; - let mut port_range = 0; - let mut hash_set = HashSet::new(); - for x in h { - if let Ok(rs) = x.await { - if let Ok((nat_type_t, ip_list_t, port_range_t)) = rs { - if nat_type_t == NatType::Symmetric { - nat_type = NatType::Symmetric; - } - for x in ip_list_t { - hash_set.insert(x); - } - if port_range < port_range_t { - port_range = port_range_t; - } - } - } - } - Ok((nat_type, hash_set.into_iter().collect(), port_range)) -} - -async fn test_nat(stun_server: String) -> io::Result<(NatType, Vec, u16)> { - let udp = UdpSocket::bind("0.0.0.0:0").await?; - udp.connect(stun_server).await?; - let mut nat_type = NatType::Cone; - let mut port_range = 0; - let mut hash_set = HashSet::new(); - match test_nat_(&udp, true, true).await { - Ok((mapped_addr1, changed_addr1)) => { - match mapped_addr1.ip() { - IpAddr::V4(ip) => { - hash_set.insert(ip); - } - IpAddr::V6(_) => {} - } - if udp.connect(changed_addr1).await.is_ok() { - if let Ok((mapped_addr2, _)) = test_nat_(&udp, false, false).await { - match mapped_addr2.ip() { - IpAddr::V4(ip) => { - hash_set.insert(ip); - } - IpAddr::V6(_) => {} - } - port_range = mapped_addr2.port().abs_diff(mapped_addr1.port()); - if mapped_addr1 != mapped_addr2 { - nat_type = NatType::Symmetric; - } - } - } - } - Err(_) => {} - } - Ok((nat_type, hash_set.into_iter().collect(), port_range)) -} - -async fn test_nat_( - udp: &UdpSocket, - change_ip: bool, - change_port: bool, -) -> io::Result<(SocketAddr, SocketAddr)> { - for _ in 0..2 { - let mut buf = [0u8; 28]; - let mut msg = stun_format::MsgBuilder::from(buf.as_mut_slice()); - msg.typ(stun_format::MsgType::BindingRequest).unwrap(); - msg.tid(1).unwrap(); - msg.add_attr(Attr::ChangeRequest { - change_ip, - change_port, - }) - .unwrap(); - udp.send(msg.as_bytes()).await?; - let mut buf = [0; 10240]; - let (len, addr) = - match tokio::time::timeout(Duration::from_millis(300), udp.recv_from(&mut buf)).await { - Ok(rs) => rs?, - Err(_) => { - continue; - } - }; - let msg = stun_format::Msg::from(&buf[..len]); - let mut mapped_addr = None; - let mut changed_addr = None; - for x in msg.attrs_iter() { - match x { - Attr::MappedAddress(addr) => { - if mapped_addr.is_none() { - let _ = mapped_addr.insert(stun_addr(addr)); - } - } - Attr::ChangedAddress(addr) => { - if changed_addr.is_none() { - let _ = changed_addr.insert(stun_addr(addr)); - } - } - Attr::XorMappedAddress(addr) => { - if mapped_addr.is_none() { - let _ = mapped_addr.insert(stun_addr(addr)); - } - } - _ => {} - } - if changed_addr.is_some() && mapped_addr.is_some() { - return Ok((mapped_addr.unwrap(), changed_addr.unwrap())); - } - } - if mapped_addr.is_some() { - return Ok((mapped_addr.unwrap(), changed_addr.unwrap_or(addr))); - } - } - Err(io::Error::new(io::ErrorKind::Other, "stun response err")) -} - -fn stun_addr(addr: stun_format::SocketAddr) -> SocketAddr { - match addr { - stun_format::SocketAddr::V4(ip, port) => { - SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::from(ip), port)) - } - stun_format::SocketAddr::V6(ip, port) => { - SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::from(ip), port, 0, 0)) - } - } -} diff --git a/vnt/src/util/wait.rs b/vnt/src/util/wait.rs deleted file mode 100644 index 655e938..0000000 --- a/vnt/src/util/wait.rs +++ /dev/null @@ -1,44 +0,0 @@ -use std::sync::atomic::{AtomicIsize, Ordering}; -use std::sync::Arc; -use tokio::sync::watch::{channel, Receiver, Sender}; - -#[derive(Clone)] -pub struct WaitGroup { - count: Arc, - receiver: Receiver, - sender: Arc>, -} - -impl WaitGroup { - pub fn new() -> Self { - let (sender, receiver) = channel(1); - Self { - count: Arc::new(Default::default()), - receiver, - sender: Arc::new(sender), - } - } - pub fn add(&self) { - let _ = self.count.fetch_add(1, Ordering::Relaxed); - } - pub fn done(&self) { - let i = self.count.fetch_sub(1, Ordering::Relaxed); - if i == 1 { - let _ = self.sender.send(0); - } - } - pub async fn wait(&mut self) { - loop { - if 0 == *self.receiver.borrow() { - return; - } - if self.receiver.changed().await.is_ok() { - if 0 == *self.receiver.borrow() { - return; - } - } else { - return; - } - } - } -}