diff --git a/Cargo.lock b/Cargo.lock index 7b70f93..8ddabbf 100644 --- a/Cargo.lock +++ b/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" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffef37f26cf649b9f473ccc00fa52d419511225a7b3a2093bf16dc8673ab9cf1" +version = "0.3.0" dependencies = [ "libc", "maybe-async", diff --git a/Cargo.toml b/Cargo.toml index d17e2a7..f5373c9 100644 --- a/Cargo.toml +++ b/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-socket2 = "0.2" +netlink-bindings = { version = "0.3", features = ["rt-link", "tc"] } +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" } diff --git a/src/netlink.rs b/src/netlink.rs index f0b9e48..0623167 100644 --- a/src/netlink.rs +++ b/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 { let mut socket = NetlinkSocket::new(); - let mut request = rt_link::Request::new() - .op_getlink_do_request(&Default::default()); + let mut request = rt_link::Request::new().op_getlink_do(&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() - .op_getlink_do_request(&Default::default()); - request.encode() + let mut request = rt_link::Request::new().op_getlink_do(&Default::default()); + request + .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(rt_link::OpGetlinkDoReply::Stats64(stats)) = attr { - return Ok((stats.rx_bytes(), stats.tx_bytes())); - } + if let Ok(stats) = attrs.get_stats64() { + return Ok((stats.rx_bytes, stats.tx_bytes)); } } @@ -77,30 +74,26 @@ impl Netlink { pub fn qdisc_from_ifindex(ifindex: i32, ifname: &str) -> Result { let mut socket = NetlinkSocket::new(); - let header = tc::PushTcmsg::new(); - let request = tc::Request::new().op_getqdisc_dump_request(&header); + let header = tc::Tcmsg::new(); + 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 { - let mut is_cake = false; - for attr in attrs { - if let Ok(tc::OpGetqdiscDumpReply::Kind(kind)) = attr { - if kind.to_str().map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))? == "cake" { - is_cake = true; - break; - } + if header.ifindex == ifindex { + if let Ok(kind) = attrs.get_kind() { + if kind + .to_str() + .map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))? + == "cake" + { + 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(); - header.set_ifindex(qdisc.ifindex); - header.set_parent(qdisc.parent); - - let mut request = tc::Request::new() - .set_change() - .op_newqdisc_do_request(&header); - request.encode() + let mut header = tc::Tcmsg::new(); + header.ifindex = qdisc.ifindex; + header.parent = qdisc.parent; + let mut request = tc::Request::new().set_change().op_newqdisc_do(&header); + request + .encode() .push_kind(c"cake") .nested_options_cake() .push_base_rate64(bandwidth) diff --git a/vendor/netlink-socket2/Cargo.toml b/vendor/netlink-socket2/Cargo.toml new file mode 100644 index 0000000..0e53aa0 --- /dev/null +++ b/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 " ] +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 = [ ] } diff --git a/vendor/netlink-socket2/README.md b/vendor/netlink-socket2/README.md new file mode 100644 index 0000000..dbc88f1 --- /dev/null +++ b/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). diff --git a/vendor/netlink-socket2/src/chained.rs b/vendor/netlink-socket2/src/chained.rs new file mode 100644 index 0000000..0897672 --- /dev/null +++ b/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> + 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> { + 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), +} + +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> {} +} diff --git a/vendor/netlink-socket2/src/error.rs b/vendor/netlink-socket2/src/error.rs new file mode 100644 index 0000000..386acc1 --- /dev/null +++ b/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>, + 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 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 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> { + 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 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(()) + } +} diff --git a/vendor/netlink-socket2/src/lib.rs b/vendor/netlink-socket2/src/lib.rs new file mode 100644 index 0000000..dd44e42 --- /dev/null +++ b/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; + +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, + 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, + 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 { + 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> + 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 { + 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> + 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 { + 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, +} + +impl NetlinkReply<'_, Request> { + #[cfg_attr(not(feature = "async"), maybe_async::maybe_async)] + pub async fn recv_one(&mut self) -> Result, 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, 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 {} +} diff --git a/vendor/netlink-socket2/src/multicast.rs b/vendor/netlink-socket2/src/multicast.rs new file mode 100644 index 0000000..e507ab2 --- /dev/null +++ b/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, +} + +impl MulticastSocketRaw { + pub fn new(protonum: u16) -> io::Result { + 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, + ) -> Result { + 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); + } + } +}