feat: added AggregateInterface

AggregateInterface can be used to gather multiple libpcap interfaces
together in order to listen to all simultaneously and also selectively
send on different interfaces

pcap-sys/src/error.rs

@@ -22,6 +22,7 @@ use std::{
 #[derive(Debug)]
 pub enum Error {
     PcapError(CString),
+    PcapErrorIf(String, CString),
     StringParse,
     UnknownPacketType(u16),
     PacketLengthInvalid,
@@ -42,6 +43,13 @@ impl Display for Error {
 
                 write!(f, "unknown pcap error")
             },
+            Error::PcapErrorIf(ifname, err) => {
+                if let Ok(err_str) = std::str::from_utf8(err.as_bytes()) {
+                    return write!(f, "pcap error on interface {ifname}: {err_str}");
+                }
+
+                write!(f, "unknown pcap error with interface {ifname}")
+            },
             Error::StringParse => write!(f, "unable to parse a string from pcap"),
             Error::UnknownPacketType(ptype) => write!(f, "unknown packet type ({ptype})"),
             Error::PacketLengthInvalid => write!(f, "received a packet with a length that mismatched the header"),
@@ -52,6 +60,15 @@ impl Display for Error {
     }
 }
 
+impl Error {
+    pub fn add_ifname(self, ifname: &str) -> Self {
+        match self {
+            Error::PcapError(err) => Error::PcapErrorIf(ifname.to_string(), err),
+            other => other
+        }
+    }
+}
+
 impl error::Error for Error {
     fn source(&self) -> Option<&(dyn error::Error + 'static)> {
         match self {

pcap-sys/src/lib.rs

@@ -18,7 +18,7 @@ use std::{
     os::fd::{AsRawFd, RawFd},
     pin::Pin,
     ptr, slice,
-    task::{self, Poll},
+    task::{self, Poll}, collections::HashMap,
 };
 
 pub mod error;
@@ -50,8 +50,9 @@ pub mod consts {
 }
 
 use ffi::PcapDevIf;
-use futures::ready;
+use futures::{ready, Stream, StreamExt};
 use tokio::io::unix::AsyncFd;
+use tokio_stream::StreamMap;
 
 pub struct PcapDevIterator {
     dev_if: *const PcapDevIf,
@@ -288,8 +289,6 @@ impl<T: Activated> Interface<T> {
     }
 
     pub fn sendpacket(&self, packet: packets::EthernetPkt) -> error::Result<()> {
-        dbg!(packet.data.len());
-
         if unsafe {
             ffi::pcap_sendpacket(
                 self.dev,
@@ -470,3 +469,215 @@ impl<T: Activated> futures::Stream for InterfaceStream<T> {
         }
     }
 }
+
+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()
+            })
+            .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<_>>()
+    }
+}
+
+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: 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<_>>()?
+        })
+    }
+}
+
+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>>> {
+        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<_>>()
+    }
+
+    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 sendpacket(&mut self, ifname: &str, packet: packets::EthernetPkt) -> error::Result<()> {
+        if let Some(interface) = self.streams
+            .values_mut()
+            .find(|interface| {
+                let dev_name = interface.inner
+                        .get_ref()
+                        .interface
+                        .dev_name
+                        .clone();
+                
+                CString::new(ifname).map(|ifname|
+                    ifname == dev_name)
+                    .unwrap_or(false)
+            }) {
+            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)
+    }
+}