ethtool: Add pause support of ethtool through netlink

Utilizing genetlink, this patch introduce new crate for ethtool pause
support through kernel netlink interface.

Unlike other crates in this project, it is required to use `.await` for
`execute()` function due to requirement of caching family ID of generic
netlink. For example:

* rtnetlink:

```rust
    let mut links = handle.link().get().execute();
    while let Some(nl_msg) = links.try_next().await? {
    }
```

* ethtool:

```rust
    let mut pause_handle = handle.pause().get(iface_name).execute().await;
    while let Some(msg) = pause_handle.try_next().await? {
    }
```

Example code been places as `ethtool/examples/dump_pause.rs`.

Signed-off-by: Gris Ge <cnfourt@gmail.com>

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "netlink-packet-audit/fuzz",
     "netlink-packet-sock-diag",
     "netlink-proto",
+    "ethtool",
     "genetlink",
     "rtnetlink",
     "audit",
@@ -26,6 +27,7 @@ default-members = [
     "netlink-packet-audit",
     "netlink-packet-sock-diag",
     "netlink-proto",
+    "ethtool",
     "genetlink",
     "rtnetlink",
     "audit",
@@ -43,3 +45,4 @@ netlink-proto = { path = "netlink-proto" }
 genetlink = { path = "genetlink" }
 rtnetlink = { path = "rtnetlink" }
 audit = { path = "audit" }
+ethtool = { path = "ethtool" }

ethtool/Cargo.toml

@@ -0,0 +1,43 @@
+[package]
+name = "ethtool"
+version = "0.1.0"
+authors = ["Gris Ge <fge@redhat.com>"]
+license = "MIT"
+edition = "2018"
+description = "Linux Ethtool Communication Library"
+keywords = ["network"]
+categories = ["network-programming", "os"]
+readme = "../README.md"
+
+[lib]
+name = "ethtool"
+path = "src/lib.rs"
+crate-type = ["lib"]
+
+[features]
+default = ["tokio_socket"]
+tokio_socket = ["netlink-proto/tokio_socket", "tokio"]
+smol_socket = ["netlink-proto/smol_socket", "async-std"]
+
+[dependencies]
+anyhow = "1.0.44"
+async-std = { version = "1.9.0", optional = true}
+byteorder = "1.4.3"
+futures = "0.3.17"
+genetlink = { default-features = false, version = "0.1.0"}
+log = "0.4.14"
+netlink-packet-core = "0.2.4"
+netlink-packet-generic = "0.1.0"
+netlink-packet-utils = "0.4.1"
+netlink-proto = { default-features = false, version = "0.7.0" }
+netlink-sys = "0.7.0"
+thiserror = "1.0.29"
+tokio = { version = "1.0.1", features = ["rt"], optional = true}
+
+[dev-dependencies]
+tokio = { version = "1.11.0", features = ["macros", "rt", "rt-multi-thread"] }
+env_logger = "0.9.0"
+
+[[example]]
+name = "dump_pause"
+required-features = ["tokio_socket"]

ethtool/README.md

@@ -0,0 +1 @@
+../README.md

ethtool/examples/dump_pause.rs

@@ -0,0 +1,29 @@
+use env_logger;
+use ethtool;
+use futures::stream::TryStreamExt;
+use tokio;
+
+// Once we find a way to load netsimdev kernel module in CI, we can convert this
+// to a test
+fn main() {
+    env_logger::init();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_io()
+        .build()
+        .unwrap();
+    rt.block_on(get_pause(None));
+}
+
+async fn get_pause(iface_name: Option<&str>) {
+    let (connection, mut handle, _) = ethtool::new_connection().unwrap();
+    tokio::spawn(connection);
+
+    let mut pause_handle = handle.pause().get(iface_name).execute().await;
+
+    let mut msgs = Vec::new();
+    while let Some(msg) = pause_handle.try_next().await.unwrap() {
+        msgs.push(msg);
+    }
+    assert!(msgs.len() > 0);
+    println!("{:?}", msgs);
+}

ethtool/src/connection.rs

@@ -0,0 +1,19 @@
+use std::io;
+
+use futures::channel::mpsc::UnboundedReceiver;
+use genetlink::message::RawGenlMessage;
+use netlink_packet_core::NetlinkMessage;
+use netlink_proto::Connection;
+use netlink_sys::SocketAddr;
+
+use crate::EthtoolHandle;
+
+#[allow(clippy::type_complexity)]
+pub fn new_connection() -> io::Result<(
+    Connection<RawGenlMessage>,
+    EthtoolHandle,
+    UnboundedReceiver<(NetlinkMessage<RawGenlMessage>, SocketAddr)>,
+)> {
+    let (conn, handle, messages) = genetlink::new_connection()?;
+    Ok((conn, EthtoolHandle::new(handle), messages))
+}

ethtool/src/error.rs

@@ -0,0 +1,21 @@
+use thiserror::Error;
+
+use netlink_packet_core::{ErrorMessage, NetlinkMessage};
+use netlink_packet_generic::GenlMessage;
+
+use crate::EthtoolMessage;
+
+#[derive(Clone, Eq, PartialEq, Debug, Error)]
+pub enum EthtoolError {
+    #[error("Received an unexpected message {0:?}")]
+    UnexpectedMessage(NetlinkMessage<GenlMessage<EthtoolMessage>>),
+
+    #[error("Received a netlink error message {0}")]
+    NetlinkError(ErrorMessage),
+
+    #[error("A netlink request failed")]
+    RequestFailed(String),
+
+    #[error("A bug in this crate")]
+    Bug(String),
+}

ethtool/src/handle.rs

@@ -0,0 +1,35 @@
+use futures::Stream;
+use genetlink::GenetlinkHandle;
+use netlink_packet_core::NetlinkMessage;
+use netlink_packet_generic::GenlMessage;
+use netlink_packet_utils::DecodeError;
+
+use crate::{EthtoolError, EthtoolMessage, EthtoolPauseHandle};
+
+#[derive(Clone, Debug)]
+pub struct EthtoolHandle {
+    pub handle: GenetlinkHandle,
+}
+
+impl EthtoolHandle {
+    pub(crate) fn new(handle: GenetlinkHandle) -> Self {
+        EthtoolHandle { handle }
+    }
+
+    pub fn pause(&mut self) -> EthtoolPauseHandle {
+        EthtoolPauseHandle::new(self.clone())
+    }
+
+    pub async fn request(
+        &mut self,
+        message: NetlinkMessage<GenlMessage<EthtoolMessage>>,
+    ) -> Result<
+        impl Stream<Item = Result<NetlinkMessage<GenlMessage<EthtoolMessage>>, DecodeError>>,
+        EthtoolError,
+    > {
+        self.handle
+            .request(message)
+            .await
+            .map_err(|e| EthtoolError::RequestFailed(format!("BUG: Request failed with {}", e)))
+    }
+}

ethtool/src/header.rs

@@ -0,0 +1,85 @@
+use std::ffi::CString;
+
+use anyhow::Context;
+use byteorder::{ByteOrder, NativeEndian};
+use netlink_packet_utils::{
+    nla::{self, DefaultNla, NlaBuffer},
+    parsers::{parse_string, parse_u32},
+    DecodeError,
+    Parseable,
+};
+
+const ALTIFNAMSIZ: usize = 128;
+const ETHTOOL_A_HEADER_DEV_INDEX: u16 = 1;
+const ETHTOOL_A_HEADER_DEV_NAME: u16 = 2;
+const ETHTOOL_A_HEADER_FLAGS: u16 = 3;
+
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum EthtoolHeader {
+    DevIndex(u32),
+    DevName(String),
+    Flags(u32),
+    Other(DefaultNla),
+}
+
+impl nla::Nla for EthtoolHeader {
+    fn value_len(&self) -> usize {
+        match self {
+            Self::DevIndex(_) | Self::Flags(_) => 4,
+            Self::DevName(s) => {
+                if s.len() + 1 > ALTIFNAMSIZ {
+                    ALTIFNAMSIZ
+                } else {
+                    s.len() + 1
+                }
+            }
+            Self::Other(attr) => attr.value_len(),
+        }
+    }
+
+    fn kind(&self) -> u16 {
+        match self {
+            Self::DevIndex(_) => ETHTOOL_A_HEADER_DEV_INDEX,
+            Self::DevName(_) => ETHTOOL_A_HEADER_DEV_NAME,
+            Self::Flags(_) => ETHTOOL_A_HEADER_FLAGS,
+            Self::Other(attr) => attr.kind(),
+        }
+    }
+
+    fn emit_value(&self, buffer: &mut [u8]) {
+        match self {
+            Self::DevIndex(value) | Self::Flags(value) => NativeEndian::write_u32(buffer, *value),
+            Self::DevName(s) => str_to_zero_ended_u8_array(s, buffer, ALTIFNAMSIZ),
+            Self::Other(ref attr) => attr.emit_value(buffer),
+        }
+    }
+}
+
+impl<'a, T: AsRef<[u8]> + ?Sized> Parseable<NlaBuffer<&'a T>> for EthtoolHeader {
+    fn parse(buf: &NlaBuffer<&'a T>) -> Result<Self, DecodeError> {
+        let payload = buf.value();
+        Ok(match buf.kind() {
+            ETHTOOL_A_HEADER_DEV_INDEX => Self::DevIndex(
+                parse_u32(payload).context("invalid ETHTOOL_A_HEADER_DEV_INDEX value")?,
+            ),
+            ETHTOOL_A_HEADER_FLAGS => {
+                Self::Flags(parse_u32(payload).context("invalid ETHTOOL_A_HEADER_FLAGS value")?)
+            }
+            ETHTOOL_A_HEADER_DEV_NAME => Self::DevName(
+                parse_string(payload).context("invalid ETHTOOL_A_HEADER_DEV_NAME value")?,
+            ),
+            _ => Self::Other(DefaultNla::parse(buf).context("invalid NLA (unknown kind)")?),
+        })
+    }
+}
+
+fn str_to_zero_ended_u8_array(src_str: &str, buffer: &mut [u8], max_size: usize) {
+    if let Ok(src_cstring) = CString::new(src_str.as_bytes()) {
+        let src_null_ended_str = src_cstring.into_bytes_with_nul();
+        if src_null_ended_str.len() > max_size {
+            buffer[..max_size].clone_from_slice(&src_null_ended_str[..max_size])
+        } else {
+            buffer[..src_null_ended_str.len()].clone_from_slice(&src_null_ended_str)
+        }
+    }
+}

ethtool/src/lib.rs

@@ -0,0 +1,19 @@
+mod connection;
+mod error;
+mod handle;
+mod header;
+mod macros;
+mod message;
+mod pause;
+
+pub use connection::new_connection;
+pub use error::EthtoolError;
+pub use handle::EthtoolHandle;
+pub use header::EthtoolHeader;
+pub use message::{EthtoolAttr, EthtoolMessage};
+pub use pause::{
+    EthtoolPauseAttr,
+    EthtoolPauseGetRequest,
+    EthtoolPauseHandle,
+    EthtoolPauseStatAttr,
+};

ethtool/src/macros.rs

@@ -0,0 +1,23 @@
+#[macro_export]
+macro_rules! try_ethtool {
+    ($msg: expr) => {{
+        use netlink_packet_core::{NetlinkMessage, NetlinkPayload};
+        use $crate::EthtoolError;
+
+        match $msg {
+            Ok(msg) => {
+                let (header, payload) = msg.into_parts();
+                match payload {
+                    NetlinkPayload::InnerMessage(msg) => msg,
+                    NetlinkPayload::Error(err) => return Err(EthtoolError::NetlinkError(err)),
+                    _ => {
+                        return Err(EthtoolError::UnexpectedMessage(NetlinkMessage::new(
+                            header, payload,
+                        )))
+                    }
+                }
+            }
+            Err(e) => return Err(EthtoolError::Bug(format!("BUG: decode error {:?}", e))),
+        }
+    }};
+}