fix(socket): restore buffer capacities before reusing recvmmsg headers

changelog/2813.fixed.md

@@ -0,0 +1,4 @@
+Fixed `recvmmsg` losing control messages and truncating sender addresses when a
+`MultiHeaders` is reused: the capacities in `msg_namelen`/`msg_controllen`,
+which the kernel overwrites with the lengths it used, are now restored before
+every `recvmmsg` and `sendmmsg` call.

src/sys/socket/mod.rs

@@ -1797,10 +1797,10 @@ pub fn sendmmsg<'a, XS, AS, C, I, S>(
         C: AsRef<[ControlMessage<'a>]> + 'a,
         S: SockaddrLike + 'a,
 {
+    data.restore_capacities();
 
     let mut count = 0;
 
-
     for (i, ((slice, addr), mmsghdr)) in slices.into_iter().zip(addrs.as_ref()).zip(data.items.iter_mut() ).enumerate() {
         let p = &mut mmsghdr.msg_hdr;
         p.msg_iov = slice.as_ref().as_ptr().cast_mut().cast();
@@ -1860,7 +1860,10 @@ pub struct MultiHeaders<S> {
     // while we are not using it directly - this is used to store control messages
     // and we retain pointers to them inside items array
     _cmsg_buffers: Option<Box<[u8]>>,
+    // the capacities of the control and address buffers of every header,
+    // see `restore_capacities`
     msg_controllen: usize,
+    msg_namelen: libc::socklen_t,
 }
 
 #[cfg(any(linux_android, target_os = "freebsd", target_os = "netbsd"))]
@@ -1904,6 +1907,29 @@ impl<S> MultiHeaders<S> {
             addresses,
             _cmsg_buffers: cmsg_buffers,
             msg_controllen,
+            msg_namelen: S::size(),
+        }
+    }
+
+    /// `msg_namelen`, `msg_controllen` and `msg_flags` are in-out parameters of
+    /// `recvmsg(2)`: going in, the first two are the capacities of the address
+    /// and control buffers, coming out they are the number of bytes the kernel
+    /// actually wrote there. Every call has to start from the capacities again,
+    /// or a slot that once received a datagram with a short address or without
+    /// any control message would be stuck with the shrunken value for good and
+    /// silently truncate what a later datagram carries.
+    ///
+    /// `sendmsg(2)` leaves the header alone, but a `MultiHeaders` may go through
+    /// [`recvmmsg`] and [`sendmmsg`] in turn, and `sendmmsg` encodes its control
+    /// messages against `msg_controllen` too.
+    // The cast is not unnecessary on all platforms.
+    #[allow(clippy::unnecessary_cast)]
+    fn restore_capacities(&mut self) {
+        for mmsghdr in self.items.iter_mut() {
+            let p = &mut mmsghdr.msg_hdr;
+            p.msg_namelen = self.msg_namelen;
+            p.msg_controllen = self.msg_controllen as _;
+            p.msg_flags = 0;
         }
     }
 }
@@ -1943,6 +1969,8 @@ where
     XS: IntoIterator<Item = &'a mut I>,
     I: AsMut<[IoSliceMut<'a>]> + 'a,
 {
+    data.restore_capacities();
+
     let mut count = 0;
     for (i, (slice, mmsghdr)) in slices.into_iter().zip(data.items.iter_mut()).enumerate() {
         let p = &mut mmsghdr.msg_hdr;

test/sys/test_socket.rs

@@ -2504,6 +2504,251 @@ fn test_recvmmsg_timestampns() {
     assert!(rduration <= time1.duration_since(UNIX_EPOCH).unwrap());
 }
 
+// A `MultiHeaders` that has already received a datagram without any control
+// message must still be able to report one on the next receive: the kernel
+// shrinks `msg_controllen` to the length it used, and `recvmmsg` has to restore
+// the capacity before reusing the header.
+#[cfg_attr(qemu, ignore)]
+#[cfg(target_os = "linux")]
+#[test]
+fn test_recvmmsg_cmsgs_after_reuse() {
+    use nix::sys::socket::*;
+    use nix::sys::time::*;
+    use std::io::{IoSlice, IoSliceMut};
+
+    let message = "Ohayō!".as_bytes();
+    let in_socket = socket(
+        AddressFamily::Inet,
+        SockType::Datagram,
+        SockFlag::empty(),
+        None,
+    )
+    .unwrap();
+    let localhost = SockaddrIn::from_str("127.0.0.1:0").unwrap();
+    bind(in_socket.as_raw_fd(), &localhost).unwrap();
+    let address: SockaddrIn = getsockname(in_socket.as_raw_fd()).unwrap();
+
+    let flags = MsgFlags::empty();
+    let send = || {
+        let iov = [IoSlice::new(message)];
+        let sent =
+            sendmsg(in_socket.as_raw_fd(), &iov, &[], flags, Some(&address))
+                .unwrap();
+        assert_eq!(message.len(), sent);
+    };
+
+    let mut buffer = vec![0u8; message.len()];
+    let mut data =
+        MultiHeaders::<()>::preallocate(1, Some(nix::cmsg_space!(TimeSpec)));
+
+    // The buffers are borrowed for as long as the results live, so each receive
+    // has to build its own `iov` in its own scope in order to reuse `data`.
+
+    // The first datagram carries no control message, so the kernel reports a
+    // used control length of zero.
+    send();
+    {
+        let mut iov = [[IoSliceMut::new(&mut buffer)]];
+        let received: Vec<RecvMsg<()>> = recvmmsg(
+            in_socket.as_raw_fd(),
+            &mut data,
+            iov.iter_mut(),
+            flags,
+            None,
+        )
+        .unwrap()
+        .collect();
+        assert_eq!(received.len(), 1);
+        assert!(received[0].cmsgs().unwrap().next().is_none());
+    }
+
+    // The second one does, and reusing the same headers must not lose it.
+    setsockopt(&in_socket, sockopt::ReceiveTimestampns, &true).unwrap();
+    send();
+    {
+        let mut iov = [[IoSliceMut::new(&mut buffer)]];
+        let received: Vec<RecvMsg<()>> = recvmmsg(
+            in_socket.as_raw_fd(),
+            &mut data,
+            iov.iter_mut(),
+            flags,
+            None,
+        )
+        .unwrap()
+        .collect();
+        assert_eq!(received.len(), 1);
+        assert!(
+            !received[0].flags.contains(MsgFlags::MSG_CTRUNC),
+            "the control message did not fit: the buffer capacity was not restored"
+        );
+        match received[0].cmsgs().unwrap().next() {
+            Some(ControlMessageOwned::ScmTimestampns(_)) => (),
+            Some(other) => panic!("Unexpected control message {other:?}"),
+            None => panic!("No control message"),
+        }
+    }
+}
+
+// A `MultiHeaders` slot that has received a datagram from a peer with a short
+// address must still report the whole address of the next peer: the kernel
+// copies at most `msg_namelen` bytes of the sender address before reporting its
+// real length, and `recvmmsg` has to restore the capacity before reusing the
+// header.
+#[cfg_attr(qemu, ignore)]
+#[cfg(target_os = "linux")]
+#[test]
+fn test_recvmmsg_address_after_reuse() {
+    use nix::sys::socket::*;
+    use std::io::{IoSlice, IoSliceMut};
+
+    let tempdir = tempfile::tempdir().unwrap();
+    let receiver_addr =
+        UnixAddr::new(&tempdir.path().join("receiver")).unwrap();
+    let receiver = socket(
+        AddressFamily::Unix,
+        SockType::Datagram,
+        SockFlag::empty(),
+        None,
+    )
+    .unwrap();
+    bind(receiver.as_raw_fd(), &receiver_addr).unwrap();
+
+    let sender = |name: &str| {
+        let addr = UnixAddr::new(&tempdir.path().join(name)).unwrap();
+        let sock = socket(
+            AddressFamily::Unix,
+            SockType::Datagram,
+            SockFlag::empty(),
+            None,
+        )
+        .unwrap();
+        bind(sock.as_raw_fd(), &addr).unwrap();
+        (sock, addr)
+    };
+    let (short_sender, short_addr) = sender("s");
+    let (long_sender, long_addr) = sender("a-much-longer-socket-name");
+    assert!(short_addr.len() < long_addr.len());
+
+    let message = "Ohayō!".as_bytes();
+    let flags = MsgFlags::empty();
+    let send = |sock: &std::os::fd::OwnedFd| {
+        let iov = [IoSlice::new(message)];
+        let sent =
+            sendmsg(sock.as_raw_fd(), &iov, &[], flags, Some(&receiver_addr))
+                .unwrap();
+        assert_eq!(message.len(), sent);
+    };
+
+    let mut buffer = vec![0u8; message.len()];
+    let mut data = MultiHeaders::<UnixAddr>::preallocate(1, None);
+    // The results borrow the headers, so the address is copied out and the
+    // results are dropped before the headers are reused.
+    let mut receive = |data: &mut MultiHeaders<UnixAddr>| -> UnixAddr {
+        let mut iov = [[IoSliceMut::new(&mut buffer)]];
+        let received: Vec<RecvMsg<UnixAddr>> =
+            recvmmsg(receiver.as_raw_fd(), data, iov.iter_mut(), flags, None)
+                .unwrap()
+                .collect();
+        assert_eq!(received.len(), 1);
+        received[0].address.unwrap()
+    };
+
+    // After the first datagram the header holds the short length.
+    send(&short_sender);
+    assert_eq!(receive(&mut data).path(), short_addr.path());
+
+    // The second sender has a longer address, and it must arrive whole.
+    send(&long_sender);
+    assert_eq!(
+        receive(&mut data).path(),
+        long_addr.path(),
+        "the sender address was truncated: the buffer capacity was not restored"
+    );
+}
+
+// `sendmmsg` encodes its control messages against `msg_controllen`, so a
+// `MultiHeaders` that has gone through `recvmmsg` must get the capacity back
+// before it is reused for sending.
+#[cfg_attr(qemu, ignore)]
+#[cfg(target_os = "linux")]
+#[test]
+fn test_sendmmsg_cmsgs_after_recvmmsg() {
+    use nix::sys::socket::*;
+    use std::io::{IoSlice, IoSliceMut};
+    use std::os::fd::FromRawFd;
+
+    let (sender, receiver) = socketpair(
+        AddressFamily::Unix,
+        SockType::Datagram,
+        None,
+        SockFlag::empty(),
+    )
+    .unwrap();
+    let (pipe_read, pipe_write) = nix::unistd::pipe().unwrap();
+
+    let message = "Ohayō!".as_bytes();
+    let flags = MsgFlags::empty();
+    let mut buffer = vec![0u8; message.len()];
+    let mut data =
+        MultiHeaders::<()>::preallocate(1, Some(nix::cmsg_space!([RawFd; 1])));
+
+    // A plain datagram leaves the header with a used control length of zero.
+    let iov = [IoSlice::new(message)];
+    sendmsg::<()>(sender.as_raw_fd(), &iov, &[], flags, None).unwrap();
+    {
+        let mut iov = [[IoSliceMut::new(&mut buffer)]];
+        let received: Vec<RecvMsg<()>> = recvmmsg(
+            receiver.as_raw_fd(),
+            &mut data,
+            iov.iter_mut(),
+            flags,
+            None,
+        )
+        .unwrap()
+        .collect();
+        assert_eq!(received.len(), 1);
+        assert!(received[0].cmsgs().unwrap().next().is_none());
+    }
+
+    // Sending a control message through the same headers must still find
+    // room for it.
+    let fds = [pipe_write.as_raw_fd()];
+    let cmsgs = [ControlMessage::ScmRights(&fds)];
+    let addrs: [Option<()>; 1] = [None];
+    let iovs = [iov];
+    let sent: Vec<RecvMsg<()>> =
+        sendmmsg(sender.as_raw_fd(), &mut data, &iovs, addrs, cmsgs, flags)
+            .unwrap()
+            .collect();
+    assert_eq!(sent.len(), 1);
+    assert_eq!(sent[0].bytes, message.len());
+
+    let mut cmsgspace = nix::cmsg_space!([RawFd; 1]);
+    let mut iov = [IoSliceMut::new(&mut buffer)];
+    let received = recvmsg::<()>(
+        receiver.as_raw_fd(),
+        &mut iov,
+        Some(&mut cmsgspace),
+        flags,
+    )
+    .unwrap();
+    assert_eq!(received.bytes, message.len());
+    match received.cmsgs().unwrap().next() {
+        Some(ControlMessageOwned::ScmRights(received_fds)) => {
+            assert_eq!(received_fds.len(), 1);
+            let received_write =
+                unsafe { std::os::fd::OwnedFd::from_raw_fd(received_fds[0]) };
+            nix::unistd::write(&received_write, b"x").unwrap();
+        }
+        Some(other) => panic!("Unexpected control message {other:?}"),
+        None => panic!("No control message"),
+    }
+    drop(pipe_write);
+    let mut byte = [0u8; 1];
+    assert_eq!(nix::unistd::read(&pipe_read, &mut byte).unwrap(), 1);
+    assert_eq!(&byte, b"x");
+}
+
 // Disable the test on emulated platforms because it fails in Cirrus-CI.  Lack
 // of QEMU support is suspected.
 #[cfg_attr(qemu, ignore)]