Cargo.lock

@@ -266,15 +266,13 @@ checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
 
 [[package]]
 name = "netlink-bindings"
-version = "0.2.4"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e639b3a966932d8ab55cfc1e606e5421aef3d30a2610284535b854f11f6e4b44"
+checksum = "b3707287f6171a6c70944b59089bfe0316613926df2e0d9ef28d108248709851"
 
 [[package]]
 name = "netlink-socket2"
-version = "0.2.4"
+version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ffef37f26cf649b9f473ccc00fa52d419511225a7b3a2093bf16dc8673ab9cf1"
 dependencies = [
  "libc",
  "maybe-async",

Cargo.toml

@@ -21,8 +21,8 @@ strip = "symbols"
 [dependencies]
 anyhow = "1.0"
 icmp-socket2 = "0.4.2"
-netlink-bindings = { version = "0.2", features = ["rt-link", "tc"] }
+netlink-bindings = { version = "0.3", features = ["rt-link", "tc"] }
-netlink-socket2 = "0.2"
+netlink-socket2 = "0.3"
 fastrand = "2"
 rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 rust-uci = { version = "0.1", optional = true }
@@ -35,3 +35,10 @@ version = "0.4"
 
 [features]
 uci = ["dep:rust-uci"]
+
+[patch.crates-io]
+# netlink-socket2 0.3.0 uses struct-literal syntax for libc::msghdr/cmsghdr which
+# fails on musl (private __pad1/__pad2 fields, u32 vs usize for cmsg_len).
+# Remove this patch once a fixed version is published upstream.
+# Upstream PR: https://github.com/one-d-wide/netlink-bindings/pull/12
+netlink-socket2 = { path = "vendor/netlink-socket2" }

src/netlink.rs

@@ -28,7 +28,6 @@ pub enum NetlinkError {
 
     #[error("Error happened while parsing UTF-8 string")]
     Utf8Error(#[from] Utf8Error),
-
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -43,32 +42,30 @@ impl Netlink {
     pub fn find_interface(ifname: &str) -> Result<i32, NetlinkError> {
         let mut socket = NetlinkSocket::new();
 
-        let mut request = rt_link::Request::new()
+        let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
-            .op_getlink_do_request(&Default::default());
         request.encode().push_ifname_bytes(ifname.as_bytes());
 
         let mut iter = socket.request(&request)?;
-        let (header, _) = iter.recv_one()
+        let (header, _) = iter
+            .recv_one()
             .map_err(|_| NetlinkError::InterfaceNotFound(ifname.to_string()))?;
-        Ok(header.ifi_index())
+        Ok(header.ifi_index)
     }
 
     pub fn get_interface_stats(ifname: &str) -> Result<(u64, u64), NetlinkError> {
         let mut socket = NetlinkSocket::new();
 
-        let mut request = rt_link::Request::new()
+        let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
-            .op_getlink_do_request(&Default::default());
+        request
-        request.encode()
+            .encode()
             .push_ifname_bytes(ifname.as_bytes())
             .push_ext_mask(1 /* RTEXT_FILTER_VF */);
 
         let mut iter = socket.request(&request)?;
         while let Some(reply) = iter.recv() {
             let (_, attrs) = reply?;
-            for attr in attrs {
+            if let Ok(stats) = attrs.get_stats64() {
-                if let Ok(rt_link::OpGetlinkDoReply::Stats64(stats)) = attr {
+                return Ok((stats.rx_bytes, stats.tx_bytes));
-                    return Ok((stats.rx_bytes(), stats.tx_bytes()));
-                }
             }
         }
 
@@ -77,30 +74,26 @@ impl Netlink {
 
     pub fn qdisc_from_ifindex(ifindex: i32, ifname: &str) -> Result<Qdisc, NetlinkError> {
         let mut socket = NetlinkSocket::new();
-        let header = tc::PushTcmsg::new();
+        let header = tc::Tcmsg::new();
-        let request = tc::Request::new().op_getqdisc_dump_request(&header);
+        let request = tc::Request::new().op_getqdisc_dump(&header);
 
         let mut iter = socket.request(&request)?;
         while let Some(reply) = iter.recv() {
             let (header, attrs) = reply?;
 
-            if header.ifindex() == ifindex {
+            if header.ifindex == ifindex {
-                let mut is_cake = false;
+                if let Ok(kind) = attrs.get_kind() {
-                for attr in attrs {
+                    if kind
-                    if let Ok(tc::OpGetqdiscDumpReply::Kind(kind)) = attr {
+                        .to_str()
-                        if kind.to_str().map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))? == "cake" {
+                        .map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))?
-                            is_cake = true;
+                        == "cake"
-                            break;
+                    {
-                        }
+                        return Ok(Qdisc {
+                            ifindex,
+                            parent: header.parent,
+                        });
                     }
                 }
-
-                if is_cake {
-                    return Ok(Qdisc {
-                        ifindex,
-                        parent: header.parent(),
-                    });
-                }
             }
         }
 
@@ -112,7 +105,11 @@ impl Netlink {
         Netlink::qdisc_from_ifindex(ifindex, ifname)
     }
 
-    pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64, dry_run: bool) -> Result<(), NetlinkError> {
+    pub fn set_qdisc_rate(
+        qdisc: Qdisc,
+        bandwidth_kbit: u64,
+        dry_run: bool,
+    ) -> Result<(), NetlinkError> {
         if dry_run {
             info!(
                 "dry-run: would set qdisc ifindex={} to {} kbit/s",
@@ -124,14 +121,12 @@ impl Netlink {
         let mut socket = NetlinkSocket::new();
         let bandwidth = bandwidth_kbit * 1000 / 8;
 
-        let mut header = tc::PushTcmsg::new();
+        let mut header = tc::Tcmsg::new();
-        header.set_ifindex(qdisc.ifindex);
+        header.ifindex = qdisc.ifindex;
-        header.set_parent(qdisc.parent);
+        header.parent = qdisc.parent;
-
+        let mut request = tc::Request::new().set_change().op_newqdisc_do(&header);
-        let mut request = tc::Request::new()
+        request
-            .set_change()
+            .encode()
-            .op_newqdisc_do_request(&header);
-        request.encode()
             .push_kind(c"cake")
             .nested_options_cake()
             .push_base_rate64(bandwidth)

vendor/netlink-socket2/Cargo.toml

@@ -0,0 +1,75 @@
+[package]
+name = "netlink-socket2"
+description = "Type-safe Rust bindings for Netlink generated from YAML specifications"
+version = "0.3.0"
+authors = [ "Remy D. Farley <one-d-wide@protonmail.com>" ]
+categories = [ "network-programming", "os", "api-bindings" ]
+repository = "https://github.com/one-d-wide/netlink-bindings"
+license = "MIT OR Apache-2.0"
+edition = "2021"
+
+[[example]]
+name = "wireguard-setup"
+required-features = [ "netlink-bindings/wireguard", "netlink-bindings/rt-link", "netlink-bindings/rt-addr" ]
+
+[[example]]
+name = "conntrack"
+required-features = [ "netlink-bindings/conntrack" ]
+
+[[example]]
+name = "extack"
+required-features = [ "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "nftables"
+required-features = [ "netlink-bindings/nftables" ]
+
+[[example]]
+name = "nftables-api"
+required-features = [ "netlink-bindings/nftables" ]
+
+[[example]]
+name = "nl80211"
+required-features = [ "netlink-bindings/nl80211", "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "nl80211-raw"
+required-features = [ "netlink-bindings/nl80211", "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "tc-prio"
+required-features = [ "netlink-bindings/tc", "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "tcp-rtt"
+required-features = [ "netlink-bindings/inet-diag" ]
+
+[[example]]
+name = "multicast-simple"
+required-features = [ "netlink-bindings/netdev", "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "multicast-generic"
+required-features = [ "netlink-bindings/netdev", "netlink-bindings/rt-link" ]
+
+[[example]]
+name = "multicast-rtnetlink"
+required-features = [ "netlink-bindings/rt-link" ]
+
+[features]
+async = []
+tokio = [ "async", "dep:tokio"]
+smol = [ "async", "dep:smol", "dep:smol-macros", "dep:macro_rules_attribute" ]
+
+[dependencies]
+libc = "0.2"
+# netlink-bindings = { path = "../netlink-bindings", features = [ "nlctrl" ] }
+netlink-bindings = { version = "0.3", features = [ "nlctrl" ] }
+
+maybe-async = { version = "0", features = [ "is_sync" ] }
+
+tokio = { optional = true, version = "1", features = [ "net", "io-util", "rt", "macros" ] }
+
+smol = { optional = true, version = "2", features = [ ] }
+smol-macros = { optional = true, version = "0" }
+macro_rules_attribute = { optional = true, version = "0", features = [ ] }

vendor/netlink-socket2/README.md

@@ -0,0 +1,376 @@
+## Netlink-bindings
+
+Type-safe Rust bindings for encoding/decoding Netlink messages generated from
+YAML [specifications].
+
+[specifications]: https://docs.kernel.org/userspace-api/netlink/specs.html
+
+## Overview
+
+Netlink is a collection of APIs exposed by the kernel, unified by a similar
+style of encoding data. The general list of Netlink families can be found in
+[the kernel
+documentation](https://docs.kernel.org/networking/netlink_spec/index.html), or
+at least those families, that were possible to condense to a machine readable
+description. This list very likely includes all the families that you would
+want to interact with.
+
+The goal of this project is to provide an easy-to-use type-safe interface,
+while also being reasonably fast and supporting all properties of all
+_sensible_ Netlink families.
+
+## Features
+
+- Streamlined type-safe interface.
+- All the properties described in yaml specifications are available in Rust.
+- Decoded Netlink messages can be Debug-printed, with enum values and flags
+annotated.
+- There is [a tool](#working-off-of-existing-tools) to simplify working off of
+existing tools by decoding Netlink messages of other programs.
+- Unified interface for both generic and classic flavors of netlink. Family
+resolution is automatic.
+- Many quirks of netlink-legacy and netlink-raw protocols are supported,
+including: binary structures, nested types, multi-attribute arrays, indexed
+arrays, sub messages.
+- Zero unsafe code in encoder/decoder logic.
+
+## Installation
+
+```toml
+[dependencies]
+netlink-bindings = { version = "0.3", features = [ "wireguard" ] }
+netlink-socket2 = { version = "0.3", features = [ ] }
+```
+
+## Making requests
+
+A typical Netlink family, say wireguard, support multiple operations, for
+example, "get-device" and "set-device". Each operation may also be of "do" or
+"dump" kind.
+
+For example, to get info about a device you would use a "dump" kind of
+"get-device" request. That's usually what it means, although different
+subsystems may imply different things. A typical request looks like this:
+
+```rust
+use netlink_bindings::wireguard;
+use netlink_socket2::NetlinkSocket;
+
+let mut sock = NetlinkSocket::new();
+
+let ifname = "wg0";
+
+// All available requests are conveniently accessible using `family::Request`
+let mut request = wireguard::Request::new()
+    .op_get_device_dump();
+
+// Add contents to the request
+request.encode()
+    .push_ifname_bytes(ifname.as_bytes());
+
+let mut iter = sock.request(&request).unwrap();
+while let Some(res) = iter.recv() {
+    // Each request may return an error (literal error code), in some cases
+    // with some additional info from the kernel, e.g. lacking a permission,
+    // if you missing CAP_NET_ADMIN capability for querying wireguard info.
+    let attrs = res.unwrap();
+
+    // A simple approach to get a specific property from an attribute set is
+    // following. Note that it's not guaranteed that the property was supplied,
+    // nor that it can be parsed correctly. If either occurs, the error will
+    // include error context, i.e. name of the attribute and it's parent set.
+    let listen_port = attrs.get_listen_port().unwrap();
+    println!("Interface {ifname:?} is listening on {listen_port}");
+
+    // Print out all the attributes using the debug formatter.
+    println!("{:#?}", attrs);
+}
+```
+
+Your LSP should be able to nicely suggest appropriate methods both for encoding
+and decoding as you type.
+
+## More complicated requests
+
+Let's say you have a network interface and want to assign an ip address to it.
+This is the domain of "rt-addr" family. It was one of the first ones created,
+inheriting some now-discouraged quirks, like a fixed-header - a struct that
+always present and may also carry some relevant data in some cases or may be
+just ignored (zeroed-out) for other requests.
+
+The relevant operation is "newaddr" with "do" kind. You may also notice
+".set_change()". This specifies an additional request flags. Similar to
+fixed-header, theses flags may invoke some additional behavior in certain
+operations, or do nothing in others.
+
+```rust,should_panic
+use std::net::IpAddr;
+use netlink_bindings::rt_addr;
+use netlink_socket2::NetlinkSocket;
+
+let mut sock = NetlinkSocket::new();
+
+let addr: IpAddr = "10.0.0.1".parse().unwrap();
+
+// Create fixed-header for the request
+let header = rt_addr::Ifaddrmsg {
+    ifa_index: 1234, // Acquired via "get-addr" request
+    ifa_family: libc::AF_INET as u8, // aka ipv4
+    ifa_prefixlen: 32, // stands for "/32" in "10.0.0.1/32"
+    ..Default::default()
+};
+
+let mut request = rt_addr::Request::new()
+    .set_change() // Don't fail if address already assigned
+    .op_newaddr_do(&header);
+
+request.encode()
+    .push_local(addr);
+
+sock.request(&request).unwrap()
+    .recv_ack().unwrap();
+```
+
+See full code in the [example](./netlink-socket/examples/wireguard-setup.rs).
+
+## Async sockets
+
+Async functionality is available using the same interface, you just need to
+enable it, and to add `.await` keyword in all places where async IO is expected.
+
+```toml
+[dependencies]
+netlink-socket2 = { ... , features = [ "tokio" ] } # or "smol"
+```
+
+An earlier example, but using async, would look like this:
+
+```rust,compile_fail
+use netlink_bindings::wireguard;
+use netlink_socket2::NetlinkSocket;
+
+let mut sock = NetlinkSocket::new();
+
+let mut request = wireguard::Request::new()
+    .op_get_device_dump();
+
+request.encode()
+    .push_ifname_bytes(b"wg0");
+
+let mut iter = sock.request(&request).await.unwrap();
+while let Some(res) = iter.recv().await {
+    println!("{:#?}", res);
+}
+```
+
+## Other examples
+
+- [wireguard-setup](./netlink-socket/examples/wireguard-setup.rs) - Create and
+configure wireguard interface.
+- [conntrack](./netlink-socket/examples/conntrack.rs) - Dump tracked network
+connections, similar to `conntrack -L`.
+- [extack](./netlink-socket/examples/extack.rs) - Showcase handing of extended
+ack attributes in error reporting.
+- [nftables](./netlink-socket/examples/nftables.rs) - Create nftables rules.
+- [nftables-api](./netlink-socket/examples/nftables-api.rs) - A high-level
+wrapper for creating nftables rules.
+- [nl80211](./netlink-socket/examples/nl80211.rs) - Basic interactions with nl80211.
+- [nl80211-raw](./netlink-socket/examples/nl80211-raw.rs) - Same as nl80211,
+but manually encoding/decoding netlink messages.
+- [tc-prio](./netlink-socket/examples/tc-prio.rs) - Add, show, and delete
+traffic control queueing discipline.
+- [tcp-rtt](./netlink-socket/examples/tcp-rtt.rs) - Dump socket information,
+including RTT of a TCP socket.
+- [multicast-simple](./netlink-socket/examples/multicast-simple.rs) - Listen
+for multicast notifications emitted when a network device is created, changed,
+or deleted.
+- [multicast-generic](./netlink-socket/examples/multicast-generic.rs) - Listen
+for all multicast notifications on a generic netlink subsystem.
+- [multicast-raw](./netlink-socket/examples/multicast-rtnetlink.rs) - Listen
+for multicast notifications on legacy rtnetlink subsystem.
+
+## Attribute encoding
+
+Under the hood, calling `.encode()` is just a convenience to pass the lead to
+the correct encoding struct. The ecoder's job is to actually write the
+attributes directly into provided buffer. All types relevant for encoding are
+prefixed with `Push`. For example, directly encoding a "do" request of
+"set-device" operation looks like the following:
+
+```rust
+use netlink_bindings::wireguard::{OpSetDeviceDo, WgdeviceFlags};
+
+let mut vec = Vec::new();
+
+// Do set-device (request)
+OpSetDeviceDo::encode_request(&mut vec)
+    .push_ifname(c"wg0") // &CStr
+    // .push_ifname_bytes("wg0".as_bytes()) // &[u8]
+    .push_flags(WgdeviceFlags::ReplacePeers as u32) // Remove existing peers
+    .array_peers()
+        .entry_nested()
+        .push_public_key(&[/* ... */]) // &[u8]
+        .push_endpoint("127.0.0.1:12345".parse().unwrap()) // SocketAddr
+        .array_allowedips()
+            .entry_nested()
+            .push_family(libc::AF_INET as u16) // aka ipv4
+            .push_ipaddr("0.0.0.0".parse().unwrap()) // IpAddr
+            .push_cidr_mask(0) // stands for "/0" in "0.0.0.0/0"
+            .end_nested()
+            // More allowed ips...
+        .end_array() // Explicitly closing isn't necessary
+        .end_nested()
+        // More peers...
+    .end_array();
+```
+
+Additionally, check out [all available
+methods](./netlink-bindings/src/wireguard/wireguard.md), along with the in-line
+documentation.
+
+## Attribute decoding
+
+Similarly, under the hood, receiving a reply yield an attribute decoder. The
+decoder itself is just a slice, therefore it can be cheaply cloned, copying
+it's frame. The low-level interface is based on iterators, with nice-to-use
+wrapper on top.
+
+```rust,should_panic
+use netlink_bindings::traits::NetlinkRequest;
+use netlink_bindings::wireguard::OpGetDeviceDump;
+
+let buf = vec![/* ... */];
+
+// Dump get-device (reply)
+let attrs = OpGetDeviceDump::decode_reply(&buf);
+
+println!("Ifname: {:?}", attrs.get_ifname().unwrap()); // &CStr
+for peer in attrs.get_peers().unwrap() {
+    println!("Endpoint: {}", peer.get_endpoint().unwrap()); // SockAddr
+
+    for addr in peer.get_allowedips().unwrap() {
+        let ip = addr.get_ipaddr().unwrap(); // IpAddr
+        let mask = addr.get_cidr_mask().unwrap(); // u8
+        println!("Allowed ip: {ip}/{mask}");
+    }
+}
+```
+
+See full code in the [example](./netlink-bindings/examples/wireguard.rs). And
+as previously, check out [all available
+methods](./netlink-bindings/src/wireguard/wireguard.md), along with the in-line
+documentation.
+
+## Low-level decoding
+
+A low-level decoding interface is exposed as an iterator, that yields enum
+variants, containing either a target type, e.g. SockAddr, or another iterator,
+in case of a nested attribute set. But as you can see, using it directly
+quickly turns very ugly.
+
+```rust
+use netlink_bindings::traits::NetlinkRequest;
+use netlink_bindings::wireguard::{OpGetDeviceDump, Wgdevice, Wgpeer};
+
+let buf = vec![/* ... */];
+
+for attr in OpGetDeviceDump::decode_reply(&buf) {
+    match attr.unwrap() {
+        Wgdevice::Ifname(n) => println!("Ifname: {n:?}"),
+        Wgdevice::Peers(iter) => {
+            for entry in iter {
+                for attr in entry.unwrap() {
+                    match attr.unwrap() {
+                        Wgpeer::Endpoint(e) => println!("Endpoint: {e:?}"),
+                        _ => {}
+                    }
+                }
+            }
+        }
+        _ => {}
+    }
+}
+```
+
+## Alternatives
+
+- [Rust-netlink](https://github.com/rust-netlink).
+- [Neli](https://github.com/jbaublitz/neli).
+
+Both don't use codegen to generate bindings, hence many Netlink families are
+not supported.
+
+Another difference is that they represent netlink messages as lists of rust
+enums, while this project works with the binary representation directly, with a
+separate interfaces for encoding and decoding: a builder pattern-like interface
+for encoding, and an iterator interface for decoding (internally).
+
+## Support status
+
+All [upstream specifications](https://www.kernel.org/doc/html/latest/netlink/specs/) are supported as of Linux 6.19.
+
+- ✅ supported, has tests:
+[conntrack](./netlink-socket/examples/conntrack.rs),
+[inet-diag](./netlink-socket/examples/tcp-rtt.rs),
+[nftables](./netlink-socket/examples/nftables.rs),
+[nl80211](./netlink-socket/examples/nl80211.rs),
+[nlctrl](./netlink-bindings/src/nlctrl/nlctrl.md),
+[rt-addr](./netlink-socket/examples/wireguard-setup.rs),
+[rt-link](./netlink-socket/examples/wireguard-setup.rs),
+[tc](./netlink-socket/examples/tc-prio.rs),
+[wireguard](./netlink-socket/examples/wireguard-setup.rs).
+- ✔️ compiles, testing needed: binder, dev-energymodel, devlink, dpll, ethtool,
+fou, handshake, lockd, mptcp_pm, netdev, net-shaper, nfsd, ovpn, ovs_datapath,
+ovs_flow, ovs_vport, psp, rt-neigh, rt-route, rt-rule tcp_metrics, team,
+unix-diag.
+
+## Working off of existing tools
+
+If there's an existing tool using Netlink, you can use `reverse-lookup` binary
+from this project to decipher it's Netlink communications, and work off of
+that. Let's say you want to inspect what `wg` command does.
+
+```sh
+$ strace -o ./output_file --decode-fd=socket -e %network --{write,read}=$(seq -s, 0 100) -- wg
+$ cargo run --bin reverse-lookup --features=wireguard,nlctrl,rt-link -- ./output_file
+Decoding request in family ROUTE flags=[REQUEST,ACK,DUMP,REPLACE,EXCL] Raw { protonum: 0, request_type: 0 }
+...
+Decoding reply in genl family wireguard flags=[MULTI] Generic("wireguard")
+Wgdevice {
+    ListenPort: 0,
+    Fwmark: 0,
+    Ifindex: 10,
+    Ifname: "wg0",
+}
+...
+```
+
+This way you can study the structure of the messages the kernel expects/replies
+with. In order to translate it into code, you simply need to convert attributes
+from CamelCase to snake_case, and add occasional .get_\*() or .push_\*()
+prefix.
+
+You can quickly build and install the tool using:
+
+```sh
+$ cargo install --git https://github.com/one-d-wide/netlink-bindings reverse-lookup --features all-subsystems
+$ ~/.cargo/bin/reverse-lookup --help
+```
+
+## Contribute
+
+See [./CONTRIBUTING.md](./CONTRIBUTING.md) for information on how to work with
+this repo, its structure, codegen stuff, etc.
+
+If your want to contribute, you can, for example:
+
+- Just use netlink-bindings. If you encounter a shortcoming of the current API, report it.
+- Write straightforward higher-level abstractions on top of netlink-bindings.
+- Add testing: collect netlink messages and check that they are parsed
+correctly. See wireguard [tests](./netlink-bindings/src/wireguard/tests.rs) as
+an example. Additional [examples](./netlink-socket/examples) are also very
+welcome.
+- Implement yet unsupported netlink functionality.
+- Improve compilation time, reduce the size of generated bindings.
+- Experiment with a better Rust interface (for encoding/decoding and the sockets).
+- Sponsor the project (contact the author for options).

vendor/netlink-socket2/src/chained.rs

@@ -0,0 +1,173 @@
+use std::{
+    fmt,
+    io::{self, IoSlice},
+    marker::{Send, Sync},
+    sync::Arc,
+};
+
+use netlink_bindings::traits::NetlinkChained;
+
+use crate::{NetlinkReplyInner, NetlinkSocket, ReplyError, Socket, RECV_BUF_SIZE};
+
+impl NetlinkSocket {
+    /// Execute a chained request (experimental)
+    ///
+    /// Some subsystems have special requirements for related requests,
+    /// expecting certain types of messages to be sent within a single write
+    /// operation. For example transactions in nftables subsystem.
+    ///
+    /// Chained requests currently don't support replies carrying data.
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn request_chained<'a, Chained>(
+        &'a mut self,
+        request: &'a Chained,
+    ) -> io::Result<NetlinkReplyChained<'a>>
+    where
+        Chained: NetlinkChained + Send + Sync,
+    {
+        let sock = Self::get_socket_cached(&mut self.sock, request.protonum())?;
+
+        Self::write_buf(sock, &[IoSlice::new(request.payload())]).await?;
+
+        let mut done = Bits::with_len(request.chain_len());
+        for i in 0..request.chain_len() {
+            let Some(has_ack) = request.supports_ack(i) else {
+                break;
+            };
+            if !has_ack {
+                done.set(i);
+            }
+        }
+
+        Ok(NetlinkReplyChained {
+            sock,
+            buf: &mut self.buf,
+            request,
+            inner: NetlinkReplyInner {
+                buf_offset: 0,
+                buf_read: 0,
+            },
+            done,
+        })
+    }
+}
+
+pub struct NetlinkReplyChained<'sock> {
+    inner: NetlinkReplyInner,
+    request: &'sock (dyn NetlinkChained + Send + Sync),
+    sock: &'sock mut Socket,
+    buf: &'sock mut Arc<[u8; RECV_BUF_SIZE]>,
+    done: Bits,
+}
+
+impl NetlinkReplyChained<'_> {
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv_all(&mut self) -> Result<(), ReplyError> {
+        while let Some(res) = self.recv().await {
+            res?;
+        }
+        Ok(())
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv(&mut self) -> Option<Result<(), ReplyError>> {
+        if self.done.is_all() {
+            return None;
+        }
+
+        let buf = Arc::make_mut(self.buf);
+
+        loop {
+            match self.inner.recv(self.sock, buf).await {
+                Err(io_err) => {
+                    self.done.set_all();
+                    return Some(Err(io_err.into()));
+                }
+                Ok((seq, _type, res)) => {
+                    let Some(index) = self.request.get_index(seq) else {
+                        continue;
+                    };
+                    match res {
+                        Ok(_) => return Some(Ok(())),
+                        Err(mut err) => {
+                            if err.code.raw_os_error().unwrap() == 0 {
+                                self.done.set(index);
+                                return Some(Ok(()));
+                            } else {
+                                self.done.set_all();
+                                err.chained_name = Some(self.request.name(index));
+                                if err.has_context() {
+                                    err.lookup = self.request.lookup(index);
+                                    err.reply_buf = Some(self.buf.clone());
+                                }
+                                return Some(Err(err));
+                            };
+                        }
+                    }
+                }
+            };
+        }
+    }
+}
+
+#[derive(Clone)]
+enum Bits {
+    Inline(u64),
+    Vec(Vec<u64>),
+}
+
+impl fmt::Debug for Bits {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let n = self.count_zeros();
+        write!(f, "{n} replies pending")
+    }
+}
+
+impl Bits {
+    fn with_len(len: usize) -> Self {
+        if len < 64 {
+            Self::Inline(u64::MAX << (len % 64))
+        } else {
+            let mut vec = vec![0; len.div_ceil(64)];
+            *vec.last_mut().unwrap() |= u64::MAX << (len % 64);
+            Self::Vec(vec)
+        }
+    }
+
+    fn set(&mut self, index: usize) {
+        match self {
+            Self::Inline(w) => *w |= 1u64 << index,
+            Self::Vec(bits) => bits[index / 64] |= 1u64 << (index % 64),
+        }
+    }
+
+    fn is_all(&self) -> bool {
+        match self {
+            Self::Inline(w) => *w == u64::MAX,
+            Self::Vec(bits) => bits.iter().all(|w| *w == u64::MAX),
+        }
+    }
+
+    fn set_all(&mut self) {
+        match self {
+            Self::Inline(w) => *w = u64::MAX,
+            Self::Vec(bits) => bits.iter_mut().for_each(|w| *w = u64::MAX),
+        }
+    }
+
+    fn count_zeros(&self) -> usize {
+        match self {
+            Self::Inline(w) => w.count_zeros() as usize,
+            Self::Vec(bits) => bits.iter().map(|s| s.count_zeros() as usize).sum(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[allow(unused)]
+    trait SpawnCompatible: Send {}
+    impl<'a> SpawnCompatible for NetlinkReplyChained<'a> {}
+}

vendor/netlink-socket2/src/error.rs

@@ -0,0 +1,157 @@
+use std::{fmt, io, sync::Arc};
+
+use netlink_bindings::{
+    builtin::{IterableNlmsgerrAttrs, NlmsgerrAttrs, Nlmsghdr},
+    traits::LookupFn,
+    utils::ErrorContext,
+};
+
+use crate::RECV_BUF_SIZE;
+
+// For an error type to be convenient to use it has to be unconstrained by
+// lifetime bounds and generics, so it can be freely passed around in a call
+// chain, therefore the data buffers are ref counter.
+pub struct ReplyError {
+    pub(crate) code: io::Error,
+    pub(crate) reply_buf: Option<Arc<[u8; RECV_BUF_SIZE]>>,
+    pub(crate) ext_ack_bounds: (u32, u32),
+    pub(crate) request_bounds: (u32, u32),
+    pub(crate) lookup: LookupFn,
+    pub(crate) chained_name: Option<&'static str>,
+}
+
+impl From<ErrorContext> for ReplyError {
+    fn from(value: ErrorContext) -> Self {
+        Self {
+            code: io::Error::other(value),
+            reply_buf: None,
+            request_bounds: (0, 0),
+            ext_ack_bounds: (0, 0),
+            lookup: |_, _, _| Default::default(),
+            chained_name: None,
+        }
+    }
+}
+
+impl From<io::Error> for ReplyError {
+    fn from(value: io::Error) -> Self {
+        Self {
+            code: value,
+            reply_buf: None,
+            request_bounds: (0, 0),
+            ext_ack_bounds: (0, 0),
+            lookup: |_, _, _| Default::default(),
+            chained_name: None,
+        }
+    }
+}
+
+impl ReplyError {
+    pub fn as_io_error(&self) -> &io::Error {
+        &self.code
+    }
+
+    pub fn ext_ack(&self) -> Option<IterableNlmsgerrAttrs<'_>> {
+        let Some(reply_buf) = &self.reply_buf else {
+            return None;
+        };
+        let (l, r) = self.ext_ack_bounds;
+        Some(NlmsgerrAttrs::new(&reply_buf[l as usize..r as usize]))
+    }
+
+    pub fn request(&self) -> Option<&[u8]> {
+        let Some(reply_buf) = &self.reply_buf else {
+            return None;
+        };
+        let (l, r) = self.request_bounds;
+        Some(&reply_buf[l as usize..r as usize])
+    }
+
+    pub(crate) fn has_context(&self) -> bool {
+        let mut res = false;
+        let (l, r) = self.ext_ack_bounds;
+        res |= l != r;
+
+        let (l, r) = self.request_bounds;
+        res |= l != r;
+
+        res
+    }
+}
+
+impl From<ReplyError> for io::Error {
+    fn from(value: ReplyError) -> Self {
+        value.code
+    }
+}
+
+impl std::error::Error for ReplyError {}
+
+impl fmt::Debug for ReplyError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(self, f)
+    }
+}
+
+impl fmt::Display for ReplyError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.code)?;
+
+        let Some(ext_ack) = self.ext_ack() else {
+            return Ok(());
+        };
+
+        if let Ok(msg) = ext_ack.get_msg() {
+            f.write_str(": ")?;
+            match msg.to_str() {
+                Ok(m) => write!(f, "{m}")?,
+                Err(_) => write!(f, "{msg:?}")?,
+            }
+        }
+
+        if let Some(chained) = self.chained_name {
+            write!(f, " in {chained:?}")?;
+        }
+
+        if let Ok(missing_offset) = ext_ack.get_missing_nest() {
+            let missing_attr = ext_ack.get_missing_type().ok();
+
+            let (trace, attr) = (self.lookup)(
+                self.request().unwrap(),
+                missing_offset as usize - Nlmsghdr::len(),
+                missing_attr,
+            );
+
+            if let Some(attr) = attr {
+                write!(f, ": missing {attr:?}")?;
+            }
+            for (attrs, _) in trace.iter() {
+                write!(f, " in {attrs:?}")?;
+            }
+        }
+
+        if let Ok(invalid_offset) = ext_ack.get_offset() {
+            let (trace, _) = (self.lookup)(
+                self.request().unwrap(),
+                invalid_offset as usize - Nlmsghdr::len(),
+                None,
+            );
+
+            if let Some((attr, _)) = trace.first() {
+                write!(f, ": attribute {attr:?}")?;
+            }
+            for (attrs, _) in trace.iter().skip(1) {
+                write!(f, " in {attrs:?}")?;
+            }
+            if let Ok(policy) = ext_ack.get_policy() {
+                write!(f, ": {policy:?}")?;
+            }
+        }
+
+        if ext_ack.get_buf().is_empty() {
+            write!(f, " (no extended ack)")?;
+        }
+
+        Ok(())
+    }
+}

vendor/netlink-socket2/src/lib.rs

@@ -0,0 +1,445 @@
+#![allow(clippy::doc_lazy_continuation)]
+#![doc = include_str!("../README.md")]
+
+use std::{
+    collections::{hash_map::Entry, HashMap},
+    io::{self, ErrorKind, IoSlice},
+    marker::PhantomData,
+    os::fd::{AsRawFd, FromRawFd, OwnedFd},
+    sync::Arc,
+};
+
+#[cfg(not(feature = "async"))]
+use std::{
+    io::{Read, Write},
+    net::TcpStream as Socket,
+};
+
+#[cfg(feature = "tokio")]
+use tokio::net::TcpStream as Socket;
+
+#[cfg(feature = "smol")]
+use smol::io::{AsyncReadExt, AsyncWriteExt};
+#[cfg(feature = "smol")]
+type Socket = smol::Async<std::net::TcpStream>;
+
+use netlink_bindings::{
+    builtin::Nlmsghdr,
+    nlctrl,
+    traits::{NetlinkRequest, Protocol},
+    utils,
+};
+
+mod chained;
+mod error;
+mod multicast;
+
+pub use chained::NetlinkReplyChained;
+pub use error::ReplyError;
+pub use multicast::{MulticastRecv, MulticastSocketRaw};
+
+/// Netlink documentation recommends max(8192, page_size)
+pub const RECV_BUF_SIZE: usize = 8192;
+
+pub struct NetlinkSocket {
+    buf: Arc<[u8; RECV_BUF_SIZE]>,
+    cache: HashMap<&'static [u8], u16>,
+    sock: HashMap<u16, Socket>,
+    seq: u32,
+}
+
+impl NetlinkSocket {
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        Self {
+            buf: Arc::new([0u8; RECV_BUF_SIZE]),
+            cache: HashMap::default(),
+            sock: HashMap::new(),
+            seq: 1,
+        }
+    }
+
+    fn get_socket_cached(
+        cache: &mut HashMap<u16, Socket>,
+        protonum: u16,
+    ) -> io::Result<&mut Socket> {
+        match cache.entry(protonum) {
+            Entry::Occupied(sock) => Ok(sock.into_mut()),
+            Entry::Vacant(ent) => {
+                let sock = Self::get_socket_new(protonum)?;
+                Ok(ent.insert(sock))
+            }
+        }
+    }
+
+    fn get_socket_new(family: u16) -> io::Result<Socket> {
+        let fd = unsafe {
+            libc::socket(
+                libc::AF_NETLINK,
+                libc::SOCK_RAW | libc::SOCK_CLOEXEC,
+                family as i32,
+            )
+        };
+        if fd < 0 {
+            return Err(io::Error::from_raw_os_error(-fd));
+        }
+        let fd = unsafe { OwnedFd::from_raw_fd(fd) };
+
+        // Enable extended attributes in libc::NLMSG_ERROR and libc::NLMSG_DONE
+        let res = unsafe {
+            libc::setsockopt(
+                fd.as_raw_fd(),
+                libc::SOL_NETLINK,
+                libc::NETLINK_EXT_ACK,
+                &1u32 as *const u32 as *const libc::c_void,
+                4,
+            )
+        };
+        if res < 0 {
+            return Err(io::Error::from_raw_os_error(-res));
+        }
+
+        let sock: std::net::TcpStream = fd.into();
+
+        #[cfg(feature = "async")]
+        {
+            sock.set_nonblocking(true)?;
+            Socket::try_from(sock)
+        }
+
+        #[cfg(not(feature = "async"))]
+        Ok(sock)
+    }
+
+    /// Reserve a sequential chunk of `seq` values, so chained messages don't
+    /// get confused. A random `seq` number might be used just as well.
+    pub fn reserve_seq(&mut self, len: u32) -> u32 {
+        let seq = self.seq;
+        self.seq = self.seq.wrapping_add(len);
+        seq
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn request<'sock, Request>(
+        &'sock mut self,
+        request: &Request,
+    ) -> io::Result<NetlinkReply<'sock, Request>>
+    where
+        Request: NetlinkRequest,
+    {
+        let (protonum, request_type) = match request.protocol() {
+            Protocol::Raw {
+                protonum,
+                request_type,
+            } => (protonum, request_type),
+            Protocol::Generic(name) => (libc::GENL_ID_CTRL as u16, self.resolve(name).await?),
+        };
+
+        self.request_raw(request, protonum, request_type).await
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn resolve(&mut self, family_name: &'static [u8]) -> io::Result<u16> {
+        if let Some(id) = self.cache.get(family_name) {
+            return Ok(*id);
+        }
+
+        let mut request = nlctrl::Request::new().op_getfamily_do();
+        request.encode().push_family_name_bytes(family_name);
+
+        let Protocol::Raw {
+            protonum,
+            request_type,
+        } = request.protocol()
+        else {
+            unreachable!()
+        };
+        assert_eq!(protonum, libc::NETLINK_GENERIC as u16);
+        assert_eq!(request_type, libc::GENL_ID_CTRL as u16);
+
+        let mut iter = self.request_raw(&request, protonum, request_type).await?;
+        if let Some(reply) = iter.recv().await {
+            let Ok(id) = reply?.get_family_id() else {
+                return Err(ErrorKind::Unsupported.into());
+            };
+            self.cache.insert(family_name, id);
+            return Ok(id);
+        }
+
+        Err(ErrorKind::UnexpectedEof.into())
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn request_raw<'sock, Request>(
+        &'sock mut self,
+        request: &Request,
+        protonum: u16,
+        request_type: u16,
+    ) -> io::Result<NetlinkReply<'sock, Request>>
+    where
+        Request: NetlinkRequest,
+    {
+        let seq = self.reserve_seq(1);
+        let sock = Self::get_socket_cached(&mut self.sock, protonum)?;
+
+        let header = Nlmsghdr {
+            len: Nlmsghdr::len() as u32 + request.payload().len() as u32,
+            r#type: request_type,
+            flags: request.flags() | libc::NLM_F_REQUEST as u16 | libc::NLM_F_ACK as u16,
+            seq,
+            pid: 0,
+        };
+
+        Self::write_buf(
+            sock,
+            &[
+                IoSlice::new(header.as_slice()),
+                IoSlice::new(request.payload()),
+            ],
+        )
+        .await?;
+
+        Ok(NetlinkReply {
+            sock,
+            buf: &mut self.buf,
+            inner: NetlinkReplyInner {
+                buf_offset: 0,
+                buf_read: 0,
+            },
+            seq: header.seq,
+            done: false,
+            phantom: PhantomData,
+        })
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn write_buf(sock: &mut Socket, payload: &[IoSlice<'_>]) -> io::Result<()> {
+        loop {
+            #[cfg(not(feature = "tokio"))]
+            let res = sock.write_vectored(payload).await;
+
+            #[cfg(feature = "tokio")]
+            let res = loop {
+                // Some subsystems don't correctly implement io notifications, which tokio runtime
+                // expects to receive before doing any actual io, hence we instead always attempt an io
+                // operation first.
+                let res = sock.try_write_vectored(payload);
+                if matches!(&res, Err(err) if err.kind() == ErrorKind::WouldBlock) {
+                    sock.writable().await?;
+                    continue;
+                }
+                break res;
+            };
+
+            match res {
+                Ok(sent) if sent != payload.iter().map(|s| s.len()).sum() => {
+                    return Err(io::Error::other("Couldn't send the whole message"));
+                }
+                Ok(_) => return Ok(()),
+                Err(err) if err.kind() == ErrorKind::Interrupted => continue,
+                Err(err) => return Err(err),
+            }
+        }
+    }
+}
+
+struct NetlinkReplyInner {
+    buf_offset: usize,
+    buf_read: usize,
+}
+
+impl NetlinkReplyInner {
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn read_buf(sock: &mut Socket, buf: &mut [u8]) -> io::Result<usize> {
+        loop {
+            #[cfg(not(feature = "tokio"))]
+            let res = sock.read(&mut buf[..]).await;
+
+            #[cfg(feature = "tokio")]
+            let res = {
+                // Some subsystems don't correctly implement io notifications, which tokio
+                // runtime expects to receive before doing any actual io, hence we instead
+                // always attempt an io operation first.
+                let res = sock.try_read(&mut buf[..]);
+                if matches!(&res, Err(err) if err.kind() == ErrorKind::WouldBlock) {
+                    sock.readable().await?;
+                    continue;
+                }
+                res
+            };
+
+            match res {
+                Ok(read) => return Ok(read),
+                Err(err) if err.kind() == ErrorKind::Interrupted => continue,
+                Err(err) => return Err(err),
+            }
+        }
+    }
+
+    #[allow(clippy::type_complexity)]
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv(
+        &mut self,
+        sock: &mut Socket,
+        buf: &mut [u8; RECV_BUF_SIZE],
+    ) -> io::Result<(u32, u16, Result<(usize, usize), ReplyError>)> {
+        if self.buf_offset == self.buf_read {
+            self.buf_read = Self::read_buf(sock, &mut buf[..]).await?;
+            self.buf_offset = 0;
+        }
+        self.parse_next(buf).await
+    }
+
+    #[allow(clippy::type_complexity)]
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn parse_next(
+        &mut self,
+        buf: &[u8; RECV_BUF_SIZE],
+    ) -> io::Result<(u32, u16, Result<(usize, usize), ReplyError>)> {
+        let packet = &buf[self.buf_offset..self.buf_read];
+
+        let too_short_err = || io::Error::other("Received packet is too short");
+
+        let Some(header) = packet.get(..Nlmsghdr::len()) else {
+            return Err(too_short_err());
+        };
+        let header = Nlmsghdr::from_slice(header);
+
+        let payload_start = self.buf_offset + Nlmsghdr::len();
+        self.buf_offset += header.len as usize;
+
+        match header.r#type as i32 {
+            libc::NLMSG_DONE | libc::NLMSG_ERROR => {
+                let Some(code) = packet.get(16..20) else {
+                    return Err(too_short_err());
+                };
+                let code = utils::parse_i32(code).unwrap();
+
+                let (echo_start, echo_end) =
+                    if code == 0 || header.r#type == libc::NLMSG_DONE as u16 {
+                        (20, 20)
+                    } else {
+                        let Some(echo_header) = packet.get(20..(20 + Nlmsghdr::len())) else {
+                            return Err(too_short_err());
+                        };
+                        let echo_header = Nlmsghdr::from_slice(echo_header);
+
+                        if echo_header.flags & libc::NLM_F_CAPPED as u16 == 0 {
+                            let start = echo_header.len;
+                            if packet.len() < start as usize + 20 {
+                                return Err(too_short_err());
+                            }
+
+                            (20 + 16, 20 + start as usize)
+                        } else {
+                            let ext_ack_start = 20 + Nlmsghdr::len();
+                            (ext_ack_start, ext_ack_start)
+                        }
+                    };
+
+                Ok((
+                    header.seq,
+                    header.r#type,
+                    Err(ReplyError {
+                        code: io::Error::from_raw_os_error(-code),
+                        request_bounds: (echo_start as u32, echo_end as u32),
+                        ext_ack_bounds: (echo_end as u32, self.buf_offset as u32),
+                        reply_buf: None,
+                        chained_name: None,
+                        lookup: |_, _, _| Default::default(),
+                    }),
+                ))
+            }
+            libc::NLMSG_NOOP => Ok((
+                header.seq,
+                header.r#type,
+                Err(io::Error::other("Received NLMSG_NOOP").into()),
+            )),
+            libc::NLMSG_OVERRUN => Ok((
+                header.seq,
+                header.r#type,
+                Err(io::Error::other("Received NLMSG_OVERRUN").into()),
+            )),
+            _ => Ok((
+                header.seq,
+                header.r#type,
+                Ok((payload_start, self.buf_offset)),
+            )),
+        }
+    }
+}
+
+pub struct NetlinkReply<'sock, Request: NetlinkRequest> {
+    inner: NetlinkReplyInner,
+    sock: &'sock mut Socket,
+    buf: &'sock mut Arc<[u8; RECV_BUF_SIZE]>,
+    seq: u32,
+    done: bool,
+    phantom: PhantomData<Request>,
+}
+
+impl<Request: NetlinkRequest> NetlinkReply<'_, Request> {
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv_one(&mut self) -> Result<Request::ReplyType<'_>, ReplyError> {
+        if let Some(res) = self.recv().await {
+            return res;
+        }
+        Err(io::Error::other("Reply didn't contain data").into())
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv_ack(&mut self) -> Result<(), ReplyError> {
+        if let Some(res) = self.recv().await {
+            res?;
+            return Err(io::Error::other("Reply isn't just an ack").into());
+        }
+        Ok(())
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv(&mut self) -> Option<Result<Request::ReplyType<'_>, ReplyError>> {
+        if self.done {
+            return None;
+        }
+
+        let buf = Arc::make_mut(self.buf);
+
+        loop {
+            match self.inner.recv(self.sock, buf).await {
+                Err(io_err) => {
+                    self.done = true;
+                    return Some(Err(io_err.into()));
+                }
+                Ok((seq, _type, res)) => {
+                    if seq != self.seq {
+                        continue;
+                    }
+                    return match res {
+                        Ok((l, r)) => Some(Ok(Request::decode_reply(&self.buf[l..r]))),
+                        Err(mut err) => {
+                            self.done = true;
+                            if err.code.raw_os_error().unwrap() == 0 {
+                                None
+                            } else {
+                                if err.has_context() {
+                                    err.lookup = Request::lookup;
+                                    err.reply_buf = Some(self.buf.clone());
+                                }
+                                Some(Err(err))
+                            }
+                        }
+                    };
+                }
+            };
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[allow(unused)]
+    trait SpawnCompatible: Send {}
+    impl<'a> SpawnCompatible for NetlinkSocket {}
+}

vendor/netlink-socket2/src/multicast.rs

@@ -0,0 +1,203 @@
+use std::{io, os::fd::AsRawFd, sync::Arc};
+
+use netlink_bindings::utils;
+
+use crate::{NetlinkReplyInner, NetlinkSocket, ReplyError, Socket, RECV_BUF_SIZE};
+
+#[derive(Debug, Clone)]
+pub struct MulticastRecv {
+    pub multicast_group: u32,
+    pub message_type: u16,
+}
+
+pub struct MulticastSocketRaw {
+    buf: Arc<[u8; RECV_BUF_SIZE]>,
+    sock: Socket,
+    reply: NetlinkReplyInner,
+    last_group: Option<u32>,
+}
+
+impl MulticastSocketRaw {
+    pub fn new(protonum: u16) -> io::Result<Self> {
+        let sock = NetlinkSocket::get_socket_new(protonum)?;
+
+        // Enable multicast group number via recvmsg ancillary messages
+        let res = unsafe {
+            libc::setsockopt(
+                sock.as_raw_fd(),
+                libc::SOL_NETLINK,
+                libc::NETLINK_PKTINFO,
+                &1u32 as *const u32 as *const libc::c_void,
+                4,
+            )
+        };
+        if res < 0 {
+            return Err(io::Error::from_raw_os_error(-res));
+        }
+
+        let mut buf: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
+        buf.nl_family = libc::AF_NETLINK as u16;
+        buf.nl_groups = 0;
+
+        let res = unsafe {
+            libc::bind(
+                sock.as_raw_fd(),
+                &buf as *const _ as *const libc::sockaddr,
+                std::mem::size_of_val(&buf) as libc::socklen_t,
+            )
+        };
+        if res < 0 {
+            return Err(io::Error::last_os_error());
+        }
+
+        Ok(Self {
+            buf: Arc::new([0u8; RECV_BUF_SIZE]),
+            sock,
+            reply: NetlinkReplyInner {
+                buf_offset: 0,
+                buf_read: 0,
+            },
+            last_group: None,
+        })
+    }
+
+    pub fn listen(&mut self, group_id: u32) -> io::Result<()> {
+        let res = unsafe {
+            libc::setsockopt(
+                self.sock.as_raw_fd(),
+                libc::SOL_NETLINK,
+                libc::NETLINK_ADD_MEMBERSHIP,
+                &group_id as *const u32 as *const libc::c_void,
+                4,
+            )
+        };
+        if res < 0 {
+            return Err(io::Error::last_os_error());
+        }
+
+        Ok(())
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    pub async fn recv(&mut self) -> Result<(MulticastRecv, &[u8]), ReplyError> {
+        let buf = Arc::make_mut(&mut self.buf);
+
+        loop {
+            if self.reply.buf_offset == self.reply.buf_read {
+                let read = Self::read_buf(&mut self.sock, buf, &mut self.last_group).await?;
+                self.reply.buf_read = read;
+                self.reply.buf_offset = 0;
+            }
+
+            match self.reply.parse_next(buf).await {
+                Err(io_err) => {
+                    return Err(io_err.into());
+                }
+                Ok((seq, message_type, res)) => {
+                    if seq != 0 {
+                        continue;
+                    }
+
+                    let Some(multicast_group) = self.last_group else {
+                        continue;
+                    };
+
+                    match res {
+                        Ok((l, r)) => {
+                            return Ok((
+                                MulticastRecv {
+                                    multicast_group,
+                                    message_type,
+                                },
+                                &self.buf[l..r],
+                            ));
+                        }
+                        Err(mut err) => {
+                            if err.code.raw_os_error().unwrap() == 0 {
+                                continue;
+                            }
+
+                            if err.has_context() {
+                                // err.lookup = Request::lookup;
+                                err.reply_buf = Some(self.buf.clone());
+                            }
+
+                            return Err(err);
+                        }
+                    };
+                }
+            };
+        }
+    }
+
+    #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)]
+    async fn read_buf(
+        sock: &Socket,
+        buf: &mut [u8],
+        last_group: &mut Option<u32>,
+    ) -> Result<usize, ReplyError> {
+        loop {
+            #[cfg(feature = "async")]
+            sock.readable().await?;
+
+            let mut addr: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
+            let mut iov = libc::iovec {
+                iov_base: buf.as_mut_ptr() as *mut libc::c_void,
+                iov_len: buf.len(),
+            };
+
+            let mut control_buf = [0u8; 128];
+
+            // Use zeroed init + field assignment to avoid private-field struct
+            // literal restrictions on musl targets (__pad1, __pad2 in msghdr).
+            let mut msghdr: libc::msghdr = unsafe { std::mem::zeroed() };
+            msghdr.msg_name = &mut addr as *mut libc::sockaddr_nl as *mut libc::c_void;
+            msghdr.msg_namelen = std::mem::size_of_val(&addr) as u32;
+            msghdr.msg_iov = &mut iov as *mut libc::iovec;
+            msghdr.msg_iovlen = 1;
+            msghdr.msg_control = control_buf.as_mut_ptr() as *mut libc::c_void;
+            // msg_controllen is usize on glibc but u32 on musl — let the compiler infer.
+            msghdr.msg_controllen = control_buf.len() as _;
+            msghdr.msg_flags = 0;
+
+            let read = unsafe { libc::recvmsg(sock.as_raw_fd(), &mut msghdr, 0) };
+            if read < 0 {
+                let err = io::Error::last_os_error();
+                #[cfg(feature = "async")]
+                if err.kind() == std::io::ErrorKind::WouldBlock {
+                    continue;
+                }
+                return Err(err.into());
+            }
+
+            *last_group = None;
+            unsafe {
+                let msghdr_ptr = &msghdr as *const libc::msghdr;
+                let mut cmsg_ptr: *const libc::cmsghdr = libc::CMSG_FIRSTHDR(msghdr_ptr);
+                while !cmsg_ptr.is_null() {
+                    let libc::cmsghdr {
+                        cmsg_len,
+                        cmsg_level,
+                        cmsg_type,
+                        .. // ignore __pad1 on musl
+                    } = *cmsg_ptr;
+
+                    match (cmsg_level, cmsg_type) {
+                        (libc::SOL_NETLINK, libc::NETLINK_PKTINFO) => {
+                            let data = std::slice::from_raw_parts(
+                                libc::CMSG_DATA(cmsg_ptr),
+                                cmsg_len as usize - libc::CMSG_LEN(0) as usize,
+                            );
+                            *last_group = Some(utils::parse_u32(&data[..4]).unwrap());
+                        }
+                        _ => {}
+                    }
+
+                    cmsg_ptr = libc::CMSG_NXTHDR(msghdr_ptr, cmsg_ptr);
+                }
+            }
+
+            return Ok(read as usize);
+        }
+    }
+}