More flakes work

Removed old git submodules
Started adding Windows cross compilation support

pcap-sys/build.rs

@@ -23,7 +23,8 @@ fn main() {
 
         let mut options = CopyOptions::new();
         options.copy_inside = true;
-        copy(std::env!("SPARSE_BUILD_LIBPCAP"), &libpcap_src, &options);
+        options.skip_exist = true;
+        copy(std::env!("SPARSE_BUILD_LIBPCAP"), &libpcap_src, &options).expect("could not copy libpcap source code to build");
 
         let dst = cmake::Config::new(&libpcap_src)
             .define("BUILD_SHARED_LIBS", "OFF")

pcap-sys/src/lib.rs

@@ -14,17 +14,15 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::{
-    collections::HashMap,
     ffi::{CStr, CString},
-    os::fd::{AsRawFd, RawFd},
-    pin::Pin,
-    ptr, slice,
-    task::{self, Poll},
+    ptr, slice
 };
 
 pub mod error;
 mod ffi;
 pub use packets;
+#[cfg(target_os = "linux")]
+pub mod stream;
 
 pub mod consts {
     pub use super::ffi::{
@@ -51,9 +49,6 @@ pub mod consts {
 }
 
 use ffi::PcapDevIf;
-use futures::{ready, StreamExt};
-use tokio::io::unix::AsyncFd;
-use tokio_stream::StreamMap;
 
 pub struct PcapDevIterator {
     dev_if: *const PcapDevIf,
@@ -409,327 +404,15 @@ impl<T: NotListening> Interface<T> {
         (info.fail_error, count)
     }
 
-    pub fn stream(mut self) -> error::Result<InterfaceStream<DevActivated>> {
+    #[cfg(target_os = "linux")]
+    pub fn stream(mut self) -> error::Result<stream::InterfaceStream<DevActivated>> {
         self.set_non_blocking(true)?;
 
-        Ok(InterfaceStream {
-            inner: AsyncFd::with_interest(
-                InternalInterfaceStream::<DevActivated>::new(unsafe { std::mem::transmute(self) })?,
+        Ok(stream::InterfaceStream {
+            inner: tokio::io::unix::AsyncFd::with_interest(
+                stream::InternalInterfaceStream::<DevActivated>::new(unsafe { std::mem::transmute(self) })?,
                 tokio::io::Interest::READABLE,
             )?,
         })
     }
 }
-
-struct InternalInterfaceStream<T: Activated> {
-    interface: Interface<T>,
-    fd: RawFd,
-}
-
-impl<T: Activated> InternalInterfaceStream<T> {
-    fn new(interface: Interface<T>) -> error::Result<InternalInterfaceStream<T>> {
-        let fd = unsafe { ffi::pcap_get_selectable_fd(interface.dev) };
-        if fd == -1 {
-            return Err(error::Error::InvalidPcapFd);
-        }
-        Ok(Self { interface, fd })
-    }
-}
-
-impl<T: Activated> AsRawFd for InternalInterfaceStream<T> {
-    fn as_raw_fd(&self) -> RawFd {
-        self.fd
-    }
-}
-
-pub struct InterfaceStream<T: Activated> {
-    inner: AsyncFd<InternalInterfaceStream<T>>,
-}
-
-impl<T: Activated> InterfaceStream<T> {
-    pub fn sendpacket(&mut self, packet: packets::EthernetPkt) -> error::Result<()> {
-        self.inner.get_mut().interface.sendpacket(packet)
-    }
-
-    pub fn set_filter(
-        &mut self,
-        filter: &str,
-        optimize: bool,
-        mask: Option<u32>,
-    ) -> error::Result<Box<ffi::BpfProgram>> {
-        self.inner
-            .get_mut()
-            .interface
-            .set_filter(filter, optimize, mask)
-    }
-}
-
-impl<T: Activated> Unpin for InterfaceStream<T> {}
-
-impl<T: Activated> futures::Stream for InterfaceStream<T> {
-    type Item = error::Result<packets::EthernetPacket>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
-        let stream = Pin::into_inner(self);
-
-        loop {
-            let mut guard = ready!(stream.inner.poll_read_ready_mut(cx))?;
-
-            match guard.try_io(|inner| match inner.get_mut().interface.next_packet() {
-                Ok(p) => Ok(Ok(p)),
-                Err(e) => Ok(Err(e)),
-            }) {
-                Ok(result) => {
-                    return Poll::Ready(Some(result?));
-                }
-                Err(_would_block) => continue,
-            }
-        }
-    }
-}
-
-pub fn new_aggregate_interface_filtered<F>(
-    crash: bool,
-    mut f: F,
-) -> error::Result<AggregateInterface<DevDisabled>>
-where
-    F: FnMut(&str) -> bool,
-{
-    let interfaces = if crash {
-        PcapDevIterator::new()?
-            .filter(|s| (f)(s))
-            .map(|if_name| {
-                let new_name = if_name.clone();
-                Interface::<DevDisabled>::new(&if_name)
-                    .map(|interface| (if_name, interface))
-                    .map_err(|e| e.add_ifname(&new_name))
-            })
-            .collect::<error::Result<HashMap<_, _>>>()?
-    } else {
-        PcapDevIterator::new()?
-            .filter(|s| (f)(s))
-            .filter_map(|if_name| {
-                let new_name = if_name.clone();
-                Interface::<DevDisabled>::new(&if_name)
-                    .map(|interface| (if_name, interface))
-                    .ok()
-                    .or_else(|| {
-                        println!("{} failed to create device", new_name);
-                        None
-                    })
-            })
-            .collect::<HashMap<_, _>>()
-    };
-
-    Ok(AggregateInterface { interfaces, crash })
-}
-
-pub fn new_aggregate_interface(crash: bool) -> error::Result<AggregateInterface<DevDisabled>> {
-    new_aggregate_interface_filtered(crash, |_| true)
-}
-
-pub struct AggregateInterface<T: State> {
-    interfaces: HashMap<String, Interface<T>>,
-    crash: bool,
-}
-
-impl<T: State> AggregateInterface<T> {
-    pub fn set_non_blocking(&mut self, nonblocking: bool) -> error::Result<()> {
-        for (n, i) in self.interfaces.iter_mut() {
-            i.set_non_blocking(nonblocking)
-                .map_err(|e| e.add_ifname(n))?;
-        }
-
-        Ok(())
-    }
-
-    pub fn lookupnets(&self) -> error::Result<HashMap<&str, (u32, u32)>> {
-        self.interfaces
-            .iter()
-            .map(|(name, interface)| {
-                interface
-                    .lookupnet()
-                    .map(|net| (&**name, net))
-                    .map_err(|e| e.add_ifname(&name))
-            })
-            .collect::<error::Result<_>>()
-    }
-
-    pub fn get_ifnames(&self) -> Vec<&str> {
-        self.interfaces.keys().map(|n| &**n).collect::<_>()
-    }
-}
-
-impl<T: Disabled> AggregateInterface<T> {
-    pub fn set_promisc(&mut self, promisc: bool) -> error::Result<()> {
-        for (n, i) in self.interfaces.iter_mut() {
-            i.set_promisc(promisc).map_err(|e| e.add_ifname(n))?;
-        }
-
-        Ok(())
-    }
-
-    pub fn set_buffer_size(&mut self, bufsize: i32) -> error::Result<()> {
-        for (n, i) in self.interfaces.iter_mut() {
-            i.set_buffer_size(bufsize).map_err(|e| e.add_ifname(n))?;
-        }
-
-        Ok(())
-    }
-
-    pub fn set_timeout(&mut self, timeout: i32) -> error::Result<()> {
-        for (n, i) in self.interfaces.iter_mut() {
-            i.set_timeout(timeout).map_err(|e| e.add_ifname(n))?;
-        }
-
-        Ok(())
-    }
-
-    pub fn activate(self) -> error::Result<AggregateInterface<DevActivated>> {
-        Ok(AggregateInterface {
-            interfaces: if self.crash {
-                self.interfaces
-                    .into_iter()
-                    .map(|(name, interface)| {
-                        let new_name = name.clone();
-                        interface
-                            .activate()
-                            .map(|interface| (name, interface))
-                            .map_err(|e| e.add_ifname(&new_name))
-                    })
-                    .collect::<error::Result<_>>()?
-            } else {
-                self.interfaces
-                    .into_iter()
-                    .filter_map(|(name, interface)| {
-                        let name_clone = name.clone();
-                        interface
-                            .activate()
-                            .map(|interface| (name, interface))
-                            .ok()
-                            .or_else(|| {
-                                println!("{} failed to activate", name_clone);
-                                None
-                            })
-                    })
-                    .collect::<_>()
-            },
-            crash: self.crash,
-        })
-    }
-}
-
-impl<T: Activated> AggregateInterface<T> {
-    pub fn datalinks(&self) -> HashMap<&str, i32> {
-        self.interfaces
-            .iter()
-            .map(|(name, interface)| (&**name, interface.datalink()))
-            .collect::<_>()
-    }
-
-    pub fn prune<F>(&mut self, mut f: F)
-    where
-        F: FnMut(&str, &mut Interface<T>) -> bool,
-    {
-        let to_prune = self
-            .interfaces
-            .iter_mut()
-            .filter_map(|(k, v)| if (f)(k, v) { Some(k.clone()) } else { None })
-            .collect::<Vec<_>>();
-
-        for name in to_prune {
-            self.interfaces.remove(&name);
-        }
-    }
-
-    pub fn set_filter(
-        &mut self,
-        filter: &str,
-        optimize: bool,
-        mask: Option<u32>,
-    ) -> error::Result<HashMap<&str, Box<ffi::BpfProgram>>> {
-        if self.crash {
-            self.interfaces
-                .iter_mut()
-                .map(|(name, interface)| {
-                    interface
-                        .set_filter(filter, optimize, mask)
-                        .map(|bpf| (&**name, bpf))
-                        .map_err(|e| e.add_ifname(&name))
-                })
-                .collect::<error::Result<_>>()
-        } else {
-            Ok(self
-                .interfaces
-                .iter_mut()
-                .filter_map(|(name, interface)| {
-                    let name_clone = name.clone();
-                    interface
-                        .set_filter(filter, optimize, mask)
-                        .map(|bpf| (&**name, bpf))
-                        .ok()
-                        .or_else(|| {
-                            println!("{} failed to set filter", name_clone);
-                            None
-                        })
-                })
-                .collect::<_>())
-        }
-    }
-
-    pub fn sendpacket(&self, ifname: &str, packet: packets::EthernetPkt) -> error::Result<()> {
-        if let Some(interface) = self.interfaces.get(ifname) {
-            interface
-                .sendpacket(packet)
-                .map_err(|e| e.add_ifname(ifname))?;
-        }
-
-        Ok(())
-    }
-}
-
-impl<T: NotListening> AggregateInterface<T> {
-    pub fn stream(self) -> error::Result<AggregateInterfaceStream<DevActivated>> {
-        Ok(AggregateInterfaceStream {
-            streams: self
-                .interfaces
-                .into_iter()
-                .map(|(ifname, interface)| {
-                    let new_name = ifname.clone();
-                    interface
-                        .stream()
-                        .map(|stream| (ifname, stream))
-                        .map_err(|e| e.add_ifname(&new_name))
-                })
-                .collect::<error::Result<_>>()?,
-        })
-    }
-}
-
-pub struct AggregateInterfaceStream<T: Activated> {
-    streams: StreamMap<String, InterfaceStream<T>>,
-}
-
-impl<T: Activated> AggregateInterfaceStream<T> {
-    pub fn get_ifnames(&self) -> Vec<&str> {
-        self.streams.keys().map(|n| &**n).collect::<_>()
-    }
-
-    pub fn sendpacket(&mut self, ifname: &str, packet: packets::EthernetPkt) -> error::Result<()> {
-        if let Some(interface) = self.streams.values_mut().find(|interface| {
-            interface.inner.get_ref().interface.dev_name.as_bytes() == ifname.as_bytes()
-        }) {
-            interface.sendpacket(packet)?;
-        }
-
-        Ok(())
-    }
-}
-
-impl<T: Activated> futures::Stream for AggregateInterfaceStream<T> {
-    type Item = (String, error::Result<packets::EthernetPacket>);
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
-        self.streams.poll_next_unpin(cx)
-    }
-}

pcap-sys/src/stream.rs

@@ -0,0 +1,340 @@
+// Copyright (C) 2023 Andrew Rioux
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::{
+    collections::HashMap,
+    os::fd::{AsRawFd, RawFd},
+    pin::Pin,
+    task::{self, Poll},
+};
+
+use futures::{ready, StreamExt};
+use tokio::io::unix::AsyncFd;
+use tokio_stream::StreamMap;
+
+use super::{error, ffi, packets, Activated, NotListening, Disabled, State, Interface, DevActivated, DevDisabled, PcapDevIterator};
+
+pub(crate) struct InternalInterfaceStream<T: Activated> {
+    interface: Interface<T>,
+    fd: RawFd,
+}
+
+impl<T: Activated> InternalInterfaceStream<T> {
+    pub(crate) fn new(interface: Interface<T>) -> error::Result<InternalInterfaceStream<T>> {
+        let fd = unsafe { ffi::pcap_get_selectable_fd(interface.dev) };
+        if fd == -1 {
+            return Err(error::Error::InvalidPcapFd);
+        }
+        Ok(Self { interface, fd })
+    }
+}
+
+impl<T: Activated> AsRawFd for InternalInterfaceStream<T> {
+    fn as_raw_fd(&self) -> RawFd {
+        self.fd
+    }
+}
+
+pub struct InterfaceStream<T: Activated> {
+    pub(crate) inner: AsyncFd<InternalInterfaceStream<T>>,
+}
+
+impl<T: Activated> InterfaceStream<T> {
+    pub fn sendpacket(&mut self, packet: packets::EthernetPkt) -> error::Result<()> {
+        self.inner.get_mut().interface.sendpacket(packet)
+    }
+
+    pub fn set_filter(
+        &mut self,
+        filter: &str,
+        optimize: bool,
+        mask: Option<u32>,
+    ) -> error::Result<Box<ffi::BpfProgram>> {
+        self.inner
+            .get_mut()
+            .interface
+            .set_filter(filter, optimize, mask)
+    }
+}
+
+impl<T: Activated> Unpin for InterfaceStream<T> {}
+
+impl<T: Activated> futures::Stream for InterfaceStream<T> {
+    type Item = error::Result<packets::EthernetPacket>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
+        let stream = Pin::into_inner(self);
+
+        loop {
+            let mut guard = ready!(stream.inner.poll_read_ready_mut(cx))?;
+
+            match guard.try_io(|inner| match inner.get_mut().interface.next_packet() {
+                Ok(p) => Ok(Ok(p)),
+                Err(e) => Ok(Err(e)),
+            }) {
+                Ok(result) => {
+                    return Poll::Ready(Some(result?));
+                }
+                Err(_would_block) => continue,
+            }
+        }
+    }
+}
+
+pub fn new_aggregate_interface_filtered<F>(
+    crash: bool,
+    mut f: F,
+) -> error::Result<AggregateInterface<DevDisabled>>
+where
+    F: FnMut(&str) -> bool,
+{
+    let interfaces = if crash {
+        PcapDevIterator::new()?
+            .filter(|s| (f)(s))
+            .map(|if_name| {
+                let new_name = if_name.clone();
+                Interface::<DevDisabled>::new(&if_name)
+                    .map(|interface| (if_name, interface))
+                    .map_err(|e| e.add_ifname(&new_name))
+            })
+            .collect::<error::Result<HashMap<_, _>>>()?
+    } else {
+        PcapDevIterator::new()?
+            .filter(|s| (f)(s))
+            .filter_map(|if_name| {
+                let new_name = if_name.clone();
+                Interface::<DevDisabled>::new(&if_name)
+                    .map(|interface| (if_name, interface))
+                    .ok()
+                    .or_else(|| {
+                        println!("{} failed to create device", new_name);
+                        None
+                    })
+            })
+            .collect::<HashMap<_, _>>()
+    };
+
+    Ok(AggregateInterface { interfaces, crash })
+}
+
+pub fn new_aggregate_interface(crash: bool) -> error::Result<AggregateInterface<DevDisabled>> {
+    new_aggregate_interface_filtered(crash, |_| true)
+}
+
+pub struct AggregateInterface<T: State> {
+    interfaces: HashMap<String, Interface<T>>,
+    crash: bool,
+}
+
+impl<T: State> AggregateInterface<T> {
+    pub fn set_non_blocking(&mut self, nonblocking: bool) -> error::Result<()> {
+        for (n, i) in self.interfaces.iter_mut() {
+            i.set_non_blocking(nonblocking)
+                .map_err(|e| e.add_ifname(n))?;
+        }
+
+        Ok(())
+    }
+
+    pub fn lookupnets(&self) -> error::Result<HashMap<&str, (u32, u32)>> {
+        self.interfaces
+            .iter()
+            .map(|(name, interface)| {
+                interface
+                    .lookupnet()
+                    .map(|net| (&**name, net))
+                    .map_err(|e| e.add_ifname(&name))
+            })
+            .collect::<error::Result<_>>()
+    }
+
+    pub fn get_ifnames(&self) -> Vec<&str> {
+        self.interfaces.keys().map(|n| &**n).collect::<_>()
+    }
+}
+
+impl<T: Disabled> AggregateInterface<T> {
+    pub fn set_promisc(&mut self, promisc: bool) -> error::Result<()> {
+        for (n, i) in self.interfaces.iter_mut() {
+            i.set_promisc(promisc).map_err(|e| e.add_ifname(n))?;
+        }
+
+        Ok(())
+    }
+
+    pub fn set_buffer_size(&mut self, bufsize: i32) -> error::Result<()> {
+        for (n, i) in self.interfaces.iter_mut() {
+            i.set_buffer_size(bufsize).map_err(|e| e.add_ifname(n))?;
+        }
+
+        Ok(())
+    }
+
+    pub fn set_timeout(&mut self, timeout: i32) -> error::Result<()> {
+        for (n, i) in self.interfaces.iter_mut() {
+            i.set_timeout(timeout).map_err(|e| e.add_ifname(n))?;
+        }
+
+        Ok(())
+    }
+
+    pub fn activate(self) -> error::Result<AggregateInterface<DevActivated>> {
+        Ok(AggregateInterface {
+            interfaces: if self.crash {
+                self.interfaces
+                    .into_iter()
+                    .map(|(name, interface)| {
+                        let new_name = name.clone();
+                        interface
+                            .activate()
+                            .map(|interface| (name, interface))
+                            .map_err(|e| e.add_ifname(&new_name))
+                    })
+                    .collect::<error::Result<_>>()?
+            } else {
+                self.interfaces
+                    .into_iter()
+                    .filter_map(|(name, interface)| {
+                        let name_clone = name.clone();
+                        interface
+                            .activate()
+                            .map(|interface| (name, interface))
+                            .ok()
+                            .or_else(|| {
+                                println!("{} failed to activate", name_clone);
+                                None
+                            })
+                    })
+                    .collect::<_>()
+            },
+            crash: self.crash,
+        })
+    }
+}
+
+impl<T: Activated> AggregateInterface<T> {
+    pub fn datalinks(&self) -> HashMap<&str, i32> {
+        self.interfaces
+            .iter()
+            .map(|(name, interface)| (&**name, interface.datalink()))
+            .collect::<_>()
+    }
+
+    pub fn prune<F>(&mut self, mut f: F)
+    where
+        F: FnMut(&str, &mut Interface<T>) -> bool,
+    {
+        let to_prune = self
+            .interfaces
+            .iter_mut()
+            .filter_map(|(k, v)| if (f)(k, v) { Some(k.clone()) } else { None })
+            .collect::<Vec<_>>();
+
+        for name in to_prune {
+            self.interfaces.remove(&name);
+        }
+    }
+
+    pub fn set_filter(
+        &mut self,
+        filter: &str,
+        optimize: bool,
+        mask: Option<u32>,
+    ) -> error::Result<HashMap<&str, Box<ffi::BpfProgram>>> {
+        if self.crash {
+            self.interfaces
+                .iter_mut()
+                .map(|(name, interface)| {
+                    interface
+                        .set_filter(filter, optimize, mask)
+                        .map(|bpf| (&**name, bpf))
+                        .map_err(|e| e.add_ifname(&name))
+                })
+                .collect::<error::Result<_>>()
+        } else {
+            Ok(self
+                .interfaces
+                .iter_mut()
+                .filter_map(|(name, interface)| {
+                    let name_clone = name.clone();
+                    interface
+                        .set_filter(filter, optimize, mask)
+                        .map(|bpf| (&**name, bpf))
+                        .ok()
+                        .or_else(|| {
+                            println!("{} failed to set filter", name_clone);
+                            None
+                        })
+                })
+                .collect::<_>())
+        }
+    }
+
+    pub fn sendpacket(&self, ifname: &str, packet: packets::EthernetPkt) -> error::Result<()> {
+        if let Some(interface) = self.interfaces.get(ifname) {
+            interface
+                .sendpacket(packet)
+                .map_err(|e| e.add_ifname(ifname))?;
+        }
+
+        Ok(())
+    }
+}
+
+impl<T: NotListening> AggregateInterface<T> {
+    pub fn stream(self) -> error::Result<AggregateInterfaceStream<DevActivated>> {
+        Ok(AggregateInterfaceStream {
+            streams: self
+                .interfaces
+                .into_iter()
+                .map(|(ifname, interface)| {
+                    let new_name = ifname.clone();
+                    interface
+                        .stream()
+                        .map(|stream| (ifname, stream))
+                        .map_err(|e| e.add_ifname(&new_name))
+                })
+                .collect::<error::Result<_>>()?,
+        })
+    }
+}
+
+pub struct AggregateInterfaceStream<T: Activated> {
+    streams: StreamMap<String, InterfaceStream<T>>,
+}
+
+impl<T: Activated> AggregateInterfaceStream<T> {
+    pub fn get_ifnames(&self) -> Vec<&str> {
+        self.streams.keys().map(|n| &**n).collect::<_>()
+    }
+
+    pub fn sendpacket(&mut self, ifname: &str, packet: packets::EthernetPkt) -> error::Result<()> {
+        if let Some(interface) = self.streams.values_mut().find(|interface| {
+            interface.inner.get_ref().interface.dev_name.as_bytes() == ifname.as_bytes()
+        }) {
+            interface.sendpacket(packet)?;
+        }
+
+        Ok(())
+    }
+}
+
+impl<T: Activated> futures::Stream for AggregateInterfaceStream<T> {
+    type Item = (String, error::Result<packets::EthernetPacket>);
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
+        self.streams.poll_next_unpin(cx)
+    }
+}