use std::net::Ipv4Addr; use cidr::Ipv4Inet; use pnet::packet::{ ip::IpNextHeaderProtocols, ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet}, udp::{self, MutableUdpPacket, UdpPacket}, }; #[cfg(any(windows, test))] use { crate::{ common::global_ctx::GlobalCtxEvent, common::stats_manager::{CounterHandle, LabelSet, LabelType, MetricName}, peers::peer_manager::PeerManager, tunnel::packet_def::ZCPacket, }, anyhow::Context, network_interface::{Addr, NetworkInterface, NetworkInterfaceConfig}, socket2::{Domain, Protocol, SockAddr, Socket, Type}, std::{ io, net::{IpAddr, SocketAddrV4, UdpSocket as StdUdpSocket}, sync::Arc, }, tokio_util::task::AbortOnDropHandle, }; #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] use windivert::{ WinDivert, error::WinDivertError, layer, packet::WinDivertPacket, prelude::{WinDivertFlags, WinDivertShutdownMode}, }; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] struct PhysicalInterface { addr: Ipv4Addr, directed_broadcast: Ipv4Addr, } impl PhysicalInterface { fn from_ip_and_prefix(addr: Ipv4Addr, prefix: u8) -> Option { if should_ignore_interface_addr(addr) || prefix > 30 { return None; } Some(Self { addr, directed_broadcast: directed_broadcast(addr, prefix)?, }) } } #[derive(Debug, Clone)] struct BroadcastRelayConfig { virtual_ipv4: Ipv4Inet, physical_interfaces: Vec, } impl BroadcastRelayConfig { fn new(virtual_ipv4: Ipv4Inet, physical_interfaces: Vec) -> Self { Self { virtual_ipv4, physical_interfaces, } } fn is_physical_source(&self, addr: Ipv4Addr) -> bool { self.physical_interfaces .iter() .any(|iface| iface.addr == addr) } fn normalize_destination(&self, dst: Ipv4Addr) -> Option { if dst.is_broadcast() || dst.is_multicast() { return Some(dst); } self.physical_interfaces .iter() .any(|iface| iface.directed_broadcast == dst) .then_some(self.virtual_ipv4.last_address()) } } #[derive(Debug, Clone, PartialEq, Eq)] struct NormalizedPacket { packet: Vec, destination: Ipv4Addr, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct UdpPacketSummary { src: Ipv4Addr, dst: Ipv4Addr, src_port: u16, dst_port: u16, ip_len: usize, udp_len: usize, payload_len: usize, } impl UdpPacketSummary { fn parse(packet: &[u8]) -> Option { let ipv4_packet = Ipv4Packet::new(packet)?; if ipv4_packet.get_version() != 4 || ipv4_packet.get_next_level_protocol() != IpNextHeaderProtocols::Udp { return None; } let header_len = usize::from(ipv4_packet.get_header_length()) * 4; let total_len = usize::from(ipv4_packet.get_total_length()); if header_len < Ipv4Packet::minimum_packet_size() || total_len < header_len + UdpPacket::minimum_packet_size() || total_len > packet.len() { return None; } let udp_packet = UdpPacket::new(&packet[header_len..total_len])?; let udp_len = usize::from(udp_packet.get_length()); if udp_len < UdpPacket::minimum_packet_size() || header_len + udp_len != total_len { return None; } Some(Self { src: ipv4_packet.get_source(), dst: ipv4_packet.get_destination(), src_port: udp_packet.get_source(), dst_port: udp_packet.get_destination(), ip_len: total_len, udp_len, payload_len: udp_len - UdpPacket::minimum_packet_size(), }) } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct ParsedUdpBroadcastPacket { header_len: usize, udp_len: usize, normalized_destination: Ipv4Addr, summary: UdpPacketSummary, } #[cfg(any(windows, test))] #[derive(Clone)] struct BroadcastRelayStats { packets_captured: CounterHandle, packets_ignored: CounterHandle, packets_forwarded: CounterHandle, packets_forward_failed: CounterHandle, } #[cfg(any(windows, test))] impl BroadcastRelayStats { fn new(peer_manager: &PeerManager) -> Self { let global_ctx = peer_manager.get_global_ctx(); let label_set = LabelSet::new().with_label_type(LabelType::NetworkName(global_ctx.get_network_name())); let stats_manager = global_ctx.stats_manager(); Self { packets_captured: stats_manager.get_counter( MetricName::UdpBroadcastRelayPacketsCaptured, label_set.clone(), ), packets_ignored: stats_manager.get_counter( MetricName::UdpBroadcastRelayPacketsIgnored, label_set.clone(), ), packets_forwarded: stats_manager.get_counter( MetricName::UdpBroadcastRelayPacketsForwarded, label_set.clone(), ), packets_forward_failed: stats_manager .get_counter(MetricName::UdpBroadcastRelayPacketsForwardFailed, label_set), } } fn record_captured(&self) { self.packets_captured.inc(); } fn record_ignored(&self) { self.packets_ignored.inc(); } fn record_forwarded(&self) { self.packets_forwarded.inc(); } fn record_forward_failed(&self) { self.packets_forward_failed.inc(); } } fn should_ignore_interface_addr(addr: Ipv4Addr) -> bool { addr.is_unspecified() || addr.is_loopback() || addr.is_multicast() || addr.is_broadcast() } fn prefix_len_from_netmask(mask: Ipv4Addr) -> Option { let raw = u32::from(mask); let prefix = raw.count_ones() as u8; let expected = if prefix == 0 { 0 } else { u32::MAX << (32 - prefix) }; (raw == expected).then_some(prefix) } fn directed_broadcast(addr: Ipv4Addr, prefix: u8) -> Option { if prefix > 32 { return None; } let mask = if prefix == 0 { 0 } else { u32::MAX << (32 - prefix) }; Some(Ipv4Addr::from(u32::from(addr) | !mask)) } fn parse_udp_broadcast( packet: &[u8], config: &BroadcastRelayConfig, ) -> Result { let ipv4_packet = Ipv4Packet::new(packet).ok_or("malformed_ipv4")?; if ipv4_packet.get_version() != 4 || ipv4_packet.get_next_level_protocol() != IpNextHeaderProtocols::Udp { return Err("not_udp_ipv4"); } if ipv4_packet.get_fragment_offset() != 0 || ipv4_packet.get_flags() & Ipv4Flags::MoreFragments != 0 { return Err("fragmented"); } let header_len = usize::from(ipv4_packet.get_header_length()) * 4; let total_len = usize::from(ipv4_packet.get_total_length()); if header_len < Ipv4Packet::minimum_packet_size() || total_len < header_len + UdpPacket::minimum_packet_size() || total_len > packet.len() { return Err("bad_ipv4_length"); } let src = ipv4_packet.get_source(); let dst = ipv4_packet.get_destination(); if should_ignore_interface_addr(src) { return Err("ignored_source"); } if src == config.virtual_ipv4.address() { return Err("virtual_source_duplicate"); } if !config.is_physical_source(src) { return Err("non_physical_source"); } let normalized_destination = config .normalize_destination(dst) .ok_or("unsupported_destination")?; if normalized_destination.is_loopback() { return Err("loopback_destination"); } let udp_packet = UdpPacket::new(&packet[header_len..total_len]).ok_or("malformed_udp")?; let udp_len = usize::from(udp_packet.get_length()); if udp_len < UdpPacket::minimum_packet_size() || header_len + udp_len != total_len { return Err("bad_udp_length"); } Ok(ParsedUdpBroadcastPacket { header_len, udp_len, normalized_destination, summary: UdpPacketSummary { src, dst, src_port: udp_packet.get_source(), dst_port: udp_packet.get_destination(), ip_len: total_len, udp_len, payload_len: udp_len - UdpPacket::minimum_packet_size(), }, }) } fn log_ignored_udp_packet(packet: &[u8], reason: &'static str) { if let Some(summary) = UdpPacketSummary::parse(packet) { tracing::debug!( src = %summary.src, dst = %summary.dst, src_port = summary.src_port, dst_port = summary.dst_port, ip_len = summary.ip_len, udp_len = summary.udp_len, payload_len = summary.payload_len, reason, "ignored Windows UDP broadcast packet" ); } else { tracing::debug!( packet_len = packet.len(), reason, "ignored malformed Windows UDP raw packet" ); } } fn normalize_udp_broadcast_packet( packet: &[u8], config: &BroadcastRelayConfig, ) -> Option { let parsed = match parse_udp_broadcast(packet, config) { Ok(parsed) => parsed, Err(reason) => { if tracing::enabled!(tracing::Level::DEBUG) { log_ignored_udp_packet(packet, reason); } return None; } }; let header_len = parsed.header_len; let udp_len = parsed.udp_len; let destination = parsed.normalized_destination; let summary = parsed.summary; let packet_len = header_len + udp_len; let virtual_ipv4 = config.virtual_ipv4.address(); let mut normalized = packet[..packet_len].to_vec(); { let mut ipv4_packet = MutableIpv4Packet::new(&mut normalized)?; ipv4_packet.set_source(virtual_ipv4); ipv4_packet.set_destination(destination); ipv4_packet.set_total_length(packet_len as u16); ipv4_packet.set_checksum(0); } { let mut udp_packet = MutableUdpPacket::new(&mut normalized[header_len..packet_len])?; udp_packet.set_checksum(0); let checksum = udp::ipv4_checksum(&udp_packet.to_immutable(), &virtual_ipv4, &destination); udp_packet.set_checksum(checksum); } { let mut ipv4_packet = MutableIpv4Packet::new(&mut normalized)?; let checksum = ipv4::checksum(&ipv4_packet.to_immutable()); ipv4_packet.set_checksum(checksum); } tracing::debug!( src = %summary.src, dst = %summary.dst, src_port = summary.src_port, dst_port = summary.dst_port, ip_len = summary.ip_len, udp_len = summary.udp_len, payload_len = summary.payload_len, normalized_src = %virtual_ipv4, normalized_dst = %destination, "normalized Windows UDP broadcast packet" ); Some(NormalizedPacket { packet: normalized, destination, }) } #[cfg(any(windows, test))] fn log_captured_udp_packet(packet: &[u8]) { if let Some(summary) = UdpPacketSummary::parse(packet) { tracing::debug!( src = %summary.src, dst = %summary.dst, src_port = summary.src_port, dst_port = summary.dst_port, ip_len = summary.ip_len, udp_len = summary.udp_len, payload_len = summary.payload_len, "captured Windows UDP broadcast candidate" ); } else { tracing::debug!( packet_len = packet.len(), "captured malformed Windows UDP broadcast candidate" ); } } #[cfg(any(windows, test))] fn collect_physical_interfaces(virtual_ipv4: Ipv4Inet) -> anyhow::Result> { let mut ret = Vec::new(); for iface in NetworkInterface::show().context("failed to list Windows network interfaces")? { if iface.internal { continue; } for addr in iface.addr { let Addr::V4(v4) = addr else { continue; }; if v4.ip == virtual_ipv4.address() { continue; } let Some(netmask) = v4.netmask else { continue; }; let Some(prefix) = prefix_len_from_netmask(netmask) else { tracing::debug!( iface = %iface.name, ip = %v4.ip, mask = %netmask, "ignoring interface with non-contiguous IPv4 netmask" ); continue; }; let Some(physical) = PhysicalInterface::from_ip_and_prefix(v4.ip, prefix) else { continue; }; if !ret.contains(&physical) { ret.push(physical); } } } Ok(ret) } #[cfg(any(windows, test))] fn join_addr_equals(field: &str, addrs: &[Ipv4Addr]) -> String { addrs .iter() .map(|addr| format!("{field} == {addr}")) .collect::>() .join(" or ") } #[cfg(any(windows, test))] fn build_windivert_udp_filter(physical_interfaces: &[PhysicalInterface]) -> String { let mut src_addrs = Vec::new(); let mut directed_broadcasts = Vec::new(); for iface in physical_interfaces { if !src_addrs.contains(&iface.addr) { src_addrs.push(iface.addr); } if !directed_broadcasts.contains(&iface.directed_broadcast) { directed_broadcasts.push(iface.directed_broadcast); } } if src_addrs.is_empty() { return "false".to_owned(); } let src_filter = join_addr_equals("ip.SrcAddr", &src_addrs); let mut dst_filters = vec!["ip.DstAddr == 255.255.255.255".to_owned()]; if !directed_broadcasts.is_empty() { dst_filters.push(join_addr_equals("ip.DstAddr", &directed_broadcasts)); } dst_filters.push("(ip.DstAddr >= 224.0.0.0 and ip.DstAddr <= 239.255.255.255)".to_owned()); format!( "outbound and ip and udp and ({}) and ({})", src_filter, dst_filters.join(" or ") ) } #[cfg(any(windows, test))] fn open_raw_udp_socket() -> io::Result { let socket = Socket::new(Domain::IPV4, Type::RAW, Some(Protocol::UDP))?; // Match ubihazard/broadcast: use one raw UDP listener on loopback, then // inspect the IPv4 header to identify the real physical source interface. socket.bind(&SockAddr::from(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)))?; socket.set_nonblocking(true)?; Ok(socket) } #[cfg(windows)] fn socket2_into_udp_socket(socket: Socket) -> StdUdpSocket { use std::os::windows::io::{FromRawSocket, IntoRawSocket}; // The raw socket handle came from socket2 and is transferred exactly once. unsafe { StdUdpSocket::from_raw_socket(socket.into_raw_socket()) } } #[cfg(all(not(windows), unix))] fn socket2_into_udp_socket(socket: Socket) -> StdUdpSocket { use std::os::fd::{FromRawFd, IntoRawFd}; // The raw socket fd came from socket2 and is transferred exactly once. unsafe { StdUdpSocket::from_raw_fd(socket.into_raw_fd()) } } #[cfg(any(windows, test))] struct RawUdpCaptureSocket { socket: tokio::net::UdpSocket, buf: Vec, } #[cfg(any(windows, test))] impl RawUdpCaptureSocket { const MAX_PACKET_LEN: usize = 65_535; fn open() -> anyhow::Result { let socket = open_raw_udp_socket().with_context(|| { "failed to open Windows raw UDP broadcast listener; administrator privileges are required" })?; let socket = socket2_into_udp_socket(socket); let socket = tokio::net::UdpSocket::from_std(socket) .context("failed to register Windows raw UDP broadcast listener with Tokio")?; Ok(Self { socket, buf: vec![0; Self::MAX_PACKET_LEN], }) } async fn recv(&mut self) -> io::Result<&[u8]> { let len = self.socket.recv(&mut self.buf).await?; Ok(&self.buf[..len]) } } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] struct WinDivertCaptureReader { inner: std::cell::UnsafeCell>, } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] unsafe impl Send for WinDivertCaptureReader {} #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] unsafe impl Sync for WinDivertCaptureReader {} #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] impl WinDivertCaptureReader { fn new(inner: WinDivert) -> Self { Self { inner: std::cell::UnsafeCell::new(inner), } } fn recv<'a>( &self, buffer: Option<&'a mut [u8]>, ) -> Result, WinDivertError> { let inner = unsafe { &*self.inner.get() }; inner.recv(buffer) } fn shutdown(&self) -> anyhow::Result<()> { let inner = unsafe { &mut *self.inner.get() }; inner .shutdown(WinDivertShutdownMode::Recv) .with_context(|| "WinDivert UDP broadcast capture shutdown failed")?; Ok(()) } fn close(&self) -> anyhow::Result<()> { let inner = unsafe { &mut *self.inner.get() }; inner .close(windivert::CloseAction::Nothing) .with_context(|| "WinDivert UDP broadcast capture close failed")?; Ok(()) } } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] impl Drop for WinDivertCaptureReader { fn drop(&mut self) { if let Err(err) = self.close() { tracing::error!(?err, "WinDivert UDP broadcast capture close failed"); } } } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] struct WinDivertCaptureSocket { rx: tokio::sync::mpsc::Receiver>, reader: Arc, buf: Vec, } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] impl WinDivertCaptureSocket { const CHANNEL_CAPACITY: usize = 1024; const MAX_PACKET_LEN: usize = 65_535; fn open(config: &BroadcastRelayConfig) -> anyhow::Result { let filter = build_windivert_udp_filter(&config.physical_interfaces); tracing::debug!( filter = %filter, "opening WinDivert UDP broadcast capture backend" ); let flags = WinDivertFlags::default().set_sniff(); let reader = WinDivert::network(&filter, 0, flags) .map_err(io::Error::other) .with_context(|| "failed to open WinDivert UDP broadcast capture")?; let reader = Arc::new(WinDivertCaptureReader::new(reader)); let reader_clone = reader.clone(); let (tx, rx) = tokio::sync::mpsc::channel(Self::CHANNEL_CAPACITY); std::thread::Builder::new() .name("easytier-udp-broadcast-windivert".to_owned()) .spawn(move || { let mut buffer = vec![0; Self::MAX_PACKET_LEN]; loop { match reader_clone.recv(Some(&mut buffer)) { Ok(packet) => { if tx.blocking_send(packet.data.to_vec()).is_err() { break; } } Err(err) => { tracing::warn!(?err, "WinDivert UDP broadcast capture receive failed"); break; } } } }) .with_context(|| "failed to spawn WinDivert UDP broadcast capture thread")?; Ok(Self { rx, reader, buf: Vec::new(), }) } async fn recv(&mut self) -> io::Result<&[u8]> { self.buf = self.rx.recv().await.ok_or_else(|| { io::Error::new( io::ErrorKind::BrokenPipe, "WinDivert UDP broadcast capture stopped", ) })?; Ok(&self.buf) } } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] impl Drop for WinDivertCaptureSocket { fn drop(&mut self) { if let Err(err) = self.reader.shutdown() { tracing::debug!(?err, "WinDivert UDP broadcast capture shutdown failed"); } } } #[cfg(any(windows, test))] enum CaptureSocket { Raw(RawUdpCaptureSocket), #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] WinDivert(WinDivertCaptureSocket), } #[cfg(any(windows, test))] impl CaptureSocket { async fn recv(&mut self) -> io::Result<&[u8]> { match self { Self::Raw(socket) => socket.recv().await, #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] Self::WinDivert(socket) => socket.recv().await, } } fn backend_name(&self) -> &'static str { match self { Self::Raw(_) => "raw_socket", #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] Self::WinDivert(_) => "windivert", } } fn fallback_to_raw(&mut self) -> anyhow::Result { #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] { if matches!(self, Self::WinDivert(_)) { *self = Self::Raw(RawUdpCaptureSocket::open()?); return Ok(true); } } Ok(false) } } #[cfg(all(windows, any(target_arch = "x86_64", target_arch = "x86")))] fn open_capture_socket(config: &BroadcastRelayConfig) -> anyhow::Result { match WinDivertCaptureSocket::open(config) { Ok(socket) => Ok(CaptureSocket::WinDivert(socket)), Err(err) => { tracing::warn!( ?err, "WinDivert UDP broadcast capture unavailable; falling back to raw socket" ); RawUdpCaptureSocket::open().map(CaptureSocket::Raw) } } } #[cfg(all( any(windows, test), not(all(windows, any(target_arch = "x86_64", target_arch = "x86"))) ))] fn open_capture_socket(_config: &BroadcastRelayConfig) -> anyhow::Result { RawUdpCaptureSocket::open().map(CaptureSocket::Raw) } #[cfg(any(windows, test))] fn issue_start_result_event( peer_manager: &PeerManager, capture_backend: Option<&str>, error: Option, ) { peer_manager .get_global_ctx() .issue_event(GlobalCtxEvent::UdpBroadcastRelayStartResult { capture_backend: capture_backend.map(str::to_owned), error, }); } #[cfg(any(windows, test))] async fn forward_normalized_packet( peer_manager: &PeerManager, normalized: NormalizedPacket, stats: &BroadcastRelayStats, ) { let packet = ZCPacket::new_with_payload(&normalized.packet); let ret = peer_manager .send_msg_by_ip(packet, IpAddr::V4(normalized.destination), true) .await; let summary = UdpPacketSummary::parse(&normalized.packet); match ret { Ok(_) => { stats.record_forwarded(); if let Some(summary) = summary { tracing::debug!( src = %summary.src, dst = %summary.dst, src_port = summary.src_port, dst_port = summary.dst_port, ip_len = summary.ip_len, udp_len = summary.udp_len, payload_len = summary.payload_len, peer_dst = %normalized.destination, broadcast = true, "forwarded Windows UDP broadcast packet" ); } else { tracing::debug!( packet_len = normalized.packet.len(), peer_dst = %normalized.destination, broadcast = true, "forwarded Windows UDP broadcast packet" ); } } Err(err) => { stats.record_forward_failed(); if let Some(summary) = summary { tracing::debug!( src = %summary.src, dst = %summary.dst, src_port = summary.src_port, dst_port = summary.dst_port, ip_len = summary.ip_len, udp_len = summary.udp_len, payload_len = summary.payload_len, peer_dst = %normalized.destination, broadcast = true, ?err, "failed to forward Windows UDP broadcast packet" ); } else { tracing::debug!( packet_len = normalized.packet.len(), peer_dst = %normalized.destination, broadcast = true, ?err, "failed to forward Windows UDP broadcast packet" ); } } } } #[cfg(any(windows, test))] async fn capture_loop( peer_manager: Arc, config: BroadcastRelayConfig, mut socket: CaptureSocket, stats: BroadcastRelayStats, ) { let mut capture_backend = socket.backend_name(); loop { let normalized = match socket.recv().await { Ok(packet) => { stats.record_captured(); if tracing::enabled!(tracing::Level::DEBUG) { log_captured_udp_packet(packet); } let normalized = normalize_udp_broadcast_packet(packet, &config); if normalized.is_none() { stats.record_ignored(); } normalized } Err(err) => { tracing::warn!( ?err, capture_backend, "Windows UDP broadcast capture receive failed" ); match socket.fallback_to_raw() { Ok(true) => { let old_backend = capture_backend; capture_backend = socket.backend_name(); tracing::warn!( old_backend, new_backend = capture_backend, "Windows UDP broadcast capture backend fell back" ); } Ok(false) => {} Err(fallback_err) => { tracing::error!( ?fallback_err, "Windows UDP broadcast raw socket fallback failed; stopping relay" ); break; } } continue; } }; if let Some(normalized) = normalized { forward_normalized_packet(&peer_manager, normalized, &stats).await; } } } #[cfg(any(windows, test))] pub(crate) fn start( peer_manager: Arc, virtual_ipv4: Ipv4Inet, ) -> anyhow::Result> { let physical_interfaces = match collect_physical_interfaces(virtual_ipv4) { Ok(interfaces) => interfaces, Err(err) => { issue_start_result_event(&peer_manager, None, Some(format!("{err:#}"))); return Err(err); } }; if physical_interfaces.is_empty() { let msg = "no physical IPv4 interface is available for UDP broadcast relay"; issue_start_result_event(&peer_manager, None, Some(msg.to_owned())); anyhow::bail!(msg); } let config = BroadcastRelayConfig::new(virtual_ipv4, physical_interfaces); let socket = match open_capture_socket(&config) { Ok(socket) => socket, Err(err) => { issue_start_result_event(&peer_manager, None, Some(format!("{err:#}"))); return Err(err); } }; let capture_backend = socket.backend_name(); issue_start_result_event(&peer_manager, Some(capture_backend), None); tracing::debug!( virtual_ipv4 = %config.virtual_ipv4, physical_interfaces = ?config.physical_interfaces, capture_backend, "starting Windows UDP broadcast relay" ); let stats = BroadcastRelayStats::new(&peer_manager); let task = tokio::spawn(capture_loop(peer_manager, config, socket, stats)); Ok(AbortOnDropHandle::new(task)) } #[cfg(test)] mod tests { use super::*; use pnet::packet::{MutablePacket, Packet}; fn config() -> BroadcastRelayConfig { BroadcastRelayConfig::new( "10.144.144.1/24".parse().unwrap(), vec![PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(192, 168, 1, 7), 24).unwrap()], ) } fn build_udp_packet(src: Ipv4Addr, dst: Ipv4Addr, payload: &[u8]) -> Vec { let mut packet = vec![0; 20 + 8 + payload.len()]; { let mut ipv4_packet = MutableIpv4Packet::new(&mut packet).unwrap(); ipv4_packet.set_version(4); ipv4_packet.set_header_length(5); ipv4_packet.set_total_length((20 + 8 + payload.len()) as u16); ipv4_packet.set_ttl(64); ipv4_packet.set_next_level_protocol(IpNextHeaderProtocols::Udp); ipv4_packet.set_source(src); ipv4_packet.set_destination(dst); } { let mut udp_packet = MutableUdpPacket::new(&mut packet[20..]).unwrap(); udp_packet.set_source(12345); udp_packet.set_destination(37020); udp_packet.set_length((8 + payload.len()) as u16); udp_packet.payload_mut().copy_from_slice(payload); let checksum = udp::ipv4_checksum(&udp_packet.to_immutable(), &src, &dst); udp_packet.set_checksum(checksum); } { let mut ipv4_packet = MutableIpv4Packet::new(&mut packet).unwrap(); let checksum = ipv4::checksum(&ipv4_packet.to_immutable()); ipv4_packet.set_checksum(checksum); } packet } fn assert_valid_checksums(packet: &[u8]) { let ipv4_packet = Ipv4Packet::new(packet).unwrap(); assert_eq!(ipv4::checksum(&ipv4_packet), ipv4_packet.get_checksum()); let udp_packet = UdpPacket::new(ipv4_packet.payload()).unwrap(); assert_eq!( udp::ipv4_checksum( &udp_packet, &ipv4_packet.get_source(), &ipv4_packet.get_destination() ), udp_packet.get_checksum() ); } #[test] fn windows_udp_broadcast_rewrites_limited_broadcast() { let packet = build_udp_packet(Ipv4Addr::new(192, 168, 1, 7), Ipv4Addr::BROADCAST, b"hello"); let normalized = normalize_udp_broadcast_packet(&packet, &config()).unwrap(); let ipv4_packet = Ipv4Packet::new(&normalized.packet).unwrap(); assert_eq!(normalized.destination, Ipv4Addr::BROADCAST); assert_eq!(ipv4_packet.get_source(), Ipv4Addr::new(10, 144, 144, 1)); assert_eq!(ipv4_packet.get_destination(), Ipv4Addr::BROADCAST); assert_eq!(&ipv4_packet.payload()[8..], b"hello"); assert_valid_checksums(&normalized.packet); } #[test] fn windows_udp_broadcast_rewrites_directed_broadcast() { let packet = build_udp_packet( Ipv4Addr::new(192, 168, 1, 7), Ipv4Addr::new(192, 168, 1, 255), b"directed", ); let normalized = normalize_udp_broadcast_packet(&packet, &config()).unwrap(); let ipv4_packet = Ipv4Packet::new(&normalized.packet).unwrap(); assert_eq!(normalized.destination, Ipv4Addr::new(10, 144, 144, 255)); assert_eq!(ipv4_packet.get_source(), Ipv4Addr::new(10, 144, 144, 1)); assert_eq!( ipv4_packet.get_destination(), Ipv4Addr::new(10, 144, 144, 255) ); assert_eq!(&ipv4_packet.payload()[8..], b"directed"); assert_valid_checksums(&normalized.packet); } #[test] fn windows_udp_broadcast_preserves_multicast_destination() { let multicast = Ipv4Addr::new(239, 255, 255, 250); let packet = build_udp_packet(Ipv4Addr::new(192, 168, 1, 7), multicast, b"multicast"); let normalized = normalize_udp_broadcast_packet(&packet, &config()).unwrap(); let ipv4_packet = Ipv4Packet::new(&normalized.packet).unwrap(); assert_eq!(normalized.destination, multicast); assert_eq!(ipv4_packet.get_source(), Ipv4Addr::new(10, 144, 144, 1)); assert_eq!(ipv4_packet.get_destination(), multicast); assert_eq!(&ipv4_packet.payload()[8..], b"multicast"); assert_valid_checksums(&normalized.packet); } #[test] fn windows_udp_broadcast_rejects_malformed_packets() { assert!(normalize_udp_broadcast_packet(&[], &config()).is_none()); let mut packet = build_udp_packet(Ipv4Addr::new(192, 168, 1, 7), Ipv4Addr::BROADCAST, b"bad"); packet[2..4].copy_from_slice(&10u16.to_be_bytes()); assert!(normalize_udp_broadcast_packet(&packet, &config()).is_none()); } #[test] fn windows_udp_broadcast_rejects_fragments() { let mut packet = build_udp_packet( Ipv4Addr::new(192, 168, 1, 7), Ipv4Addr::BROADCAST, b"fragment", ); { let mut ipv4_packet = MutableIpv4Packet::new(&mut packet).unwrap(); ipv4_packet.set_flags(Ipv4Flags::MoreFragments); } assert!(normalize_udp_broadcast_packet(&packet, &config()).is_none()); } #[test] fn windows_udp_broadcast_rejects_non_broadcast_destinations() { let packet = build_udp_packet( Ipv4Addr::new(192, 168, 1, 7), Ipv4Addr::new(192, 168, 1, 10), b"unicast", ); assert!(normalize_udp_broadcast_packet(&packet, &config()).is_none()); } #[test] fn windows_udp_broadcast_rejects_virtual_source_duplicates() { let packet = build_udp_packet(Ipv4Addr::new(10, 144, 144, 1), Ipv4Addr::BROADCAST, b"loop"); assert!(normalize_udp_broadcast_packet(&packet, &config()).is_none()); } #[test] fn windows_udp_broadcast_detects_directed_broadcast_from_prefix() { let physical = PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(172, 16, 5, 10), 20).unwrap(); assert_eq!(physical.directed_broadcast, Ipv4Addr::new(172, 16, 15, 255)); assert_eq!( prefix_len_from_netmask(Ipv4Addr::new(255, 255, 240, 0)), Some(20) ); assert_eq!(prefix_len_from_netmask(Ipv4Addr::new(255, 0, 255, 0)), None); } #[test] fn windows_udp_broadcast_keeps_link_local_interfaces() { let physical = PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(169, 254, 13, 10), 16).unwrap(); assert_eq!( physical.directed_broadcast, Ipv4Addr::new(169, 254, 255, 255) ); } #[test] fn windows_udp_broadcast_windivert_filter_is_constrained() { let interfaces = vec![ PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(192, 168, 1, 7), 24).unwrap(), PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(169, 254, 13, 10), 16).unwrap(), PhysicalInterface::from_ip_and_prefix(Ipv4Addr::new(169, 254, 156, 121), 16).unwrap(), ]; let filter = build_windivert_udp_filter(&interfaces); assert!(filter.starts_with("outbound and ip and udp and ")); assert!(filter.contains("ip.SrcAddr == 192.168.1.7")); assert!(filter.contains("ip.SrcAddr == 169.254.13.10")); assert!(filter.contains("ip.DstAddr == 255.255.255.255")); assert!(filter.contains("ip.DstAddr == 192.168.1.255")); assert!(filter.contains("ip.DstAddr == 169.254.255.255")); assert!(filter.contains("ip.DstAddr >= 224.0.0.0")); assert!(filter.contains("ip.DstAddr <= 239.255.255.255")); assert_eq!(filter.matches("ip.DstAddr == 169.254.255.255").count(), 1); } }