syd 3.58.0

// Syd: rock-solid application kernel
// src/kernel/net/recvmsg.rs: recvmsg(2), recvmmsg(2), and recvmmsg_time64 handlers
//
// Copyright (c) 2025, 2026 Ali Polatel <alip@chesswob.org>
//
// SPDX-License-Identifier: GPL-3.0

use std::{io::IoSlice, os::fd::AsFd};

use libc::{c_int, c_uint, c_void, iovec, sockaddr, socklen_t, MSG_CTRUNC};
use libseccomp::ScmpNotifResp;
use nix::{
    errno::Errno,
    sys::{
        socket::{SockaddrLike, SockaddrStorage},
        uio::RemoteIoVec,
    },
};

use crate::{
    compat::{
        mmsghdr, mmsghdr32, msghdr, msghdr32, recvmmsg, recvmsg, try_from_bytes, AddressFamily,
        CmsgOwned, MmsgHdr, MsgFlags, RecvMsg, SockType, TimeSpec32, TimeSpec64, ToByteArray,
        UIO_MAXIOV,
    },
    confine::{scmp_arch_is_compat32, scmp_arch_is_compat_long32},
    fd::{fd_inode, has_recv_timeout},
    ip::SockInfo,
    kernel::net::{to_msgflags, SockOpts},
    proc::maps::ProcMaps,
    req::UNotifyEventRequest,
    unix::unix_addr_len2,
};

const SOCKADDR_SIZE: usize = size_of::<libc::sockaddr_storage>();

type IovBuf = (Vec<u8>, u64);

#[expect(clippy::cognitive_complexity)]
pub(crate) fn handle_recvmsg(
    request: &UNotifyEventRequest,
    mut sock: SockInfo,
    sock_opts: SockOpts,
    args: &[u64; 6],
) -> Result<ScmpNotifResp, Errno> {
    let SockOpts {
        flags,
        options,
        is_nonblock,
    } = sock_opts;

    // Truncate flags to 32-bit keeping unknown flags.
    let call_flags = to_msgflags(args[2]);

    // Reject MSG_OOB as necessary.
    if !options.allow_unsafe_oob() && call_flags.contains(MsgFlags::MSG_OOB) {
        // Signal no support to let the sandbox process handle the error
        // gracefully. This is consistent with the Linux kernel.
        return Err(Errno::EOPNOTSUPP);
    }

    // Determine bitness of sandbox process.
    let req = request.scmpreq;
    let is32 = scmp_arch_is_compat32(req.data.arch);

    // Linux rejects MSG_CMSG_COMPAT on 64-bit.
    if !is32 && call_flags.contains(MsgFlags::MSG_CMSG_COMPAT) {
        return Err(Errno::EINVAL);
    }
    let call_flags = call_flags.difference(MsgFlags::MSG_CMSG_COMPAT);

    // Read user msghdr.
    let hdr_sz = if is32 {
        size_of::<msghdr32>()
    } else {
        size_of::<msghdr>()
    };

    let hdr = request.read_vec_all(args[1], hdr_sz)?;
    let mut hdr: msghdr = if is32 {
        let m32: msghdr32 = try_from_bytes(&hdr)?;
        msghdr::from(m32)
    } else {
        try_from_bytes(&hdr)?
    };

    // Mirror sandbox process iovecs to local, bounded buffers.
    let mut msg_bufs: Vec<(Vec<u8>, u64)> = Vec::new();
    let mut msg_iovs: Vec<iovec> = Vec::new();
    let mut nam_buf: Vec<u8> = Vec::new();
    let mut ctl_buf: Vec<u8> = Vec::new();

    // Handle msg_name.
    let (user_nam_base, user_nam_size) = request.setup_msghdr_name(&mut hdr, &mut nam_buf)?;

    // Linux discards bytes with MSG_TRUNC.
    let discard = call_flags.contains(MsgFlags::MSG_TRUNC)
        && matches!(sock.get_dom()?, AddressFamily::Inet | AddressFamily::Inet6)
        && sock.get_stype()? == SockType::Stream
        && sock.get_trunc_discard()?;

    // Handle msg_iov.
    let user_iov_base =
        request.read_msghdr_iov(&mut hdr, (!discard).then_some(&mut msg_bufs), &mut msg_iovs)?;

    // Handle msg_control.
    let (user_ctl_base, user_ctl_size) = request.setup_msghdr_ctl(&mut hdr, &mut ctl_buf)?;

    // Linux consumes only bytes it can copy, rest stays queued.
    if !msg_bufs.is_empty() && sock.get_stype()? == SockType::Stream {
        let mut bufs = msg_bufs.iter();
        let mut short = false;
        let mut total: usize = 0;
        for iov in &mut msg_iovs {
            if iov.iov_len == 0 {
                continue;
            }
            let ptr = match bufs.next() {
                Some((_, ptr)) => *ptr,
                None => break,
            };
            if short {
                iov.iov_len = 0;
                continue;
            }
            let bad = request.get_fault(ptr, iov.iov_len)?;
            short = bad > 0;
            iov.iov_len = iov.iov_len.saturating_sub(bad);
            total = total.checked_add(iov.iov_len).ok_or(Errno::EOVERFLOW)?;
        }
        if total == 0 {
            return Err(Errno::EFAULT);
        }
    }

    // Handle scatter buffers to be used post-syscall.
    //
    // Buffer consists of message payload, header, and address.
    let buf_len = msg_bufs.len().checked_add(2).ok_or(Errno::EOVERFLOW)?;
    let mut iovs_l: Vec<IoSlice<'_>> = Vec::new();
    let mut iovs_r: Vec<RemoteIoVec> = Vec::new();
    iovs_l.try_reserve_exact(buf_len).or(Err(Errno::ENOMEM))?;
    iovs_r.try_reserve_exact(buf_len).or(Err(Errno::ENOMEM))?;

    let mmsghdr_size = if is32 {
        size_of::<msghdr32>()
    } else {
        size_of::<msghdr>()
    };
    let mut hdr_buf: Vec<u8> = Vec::new();
    hdr_buf
        .try_reserve_exact(mmsghdr_size)
        .or(Err(Errno::ENOMEM))?;

    // Handle address buffer.
    let mut addr_buf = [0u8; SOCKADDR_SIZE];

    // Track blocking call for invalidation semantics.
    let is_blocking = !is_nonblock && !call_flags.contains(MsgFlags::MSG_DONTWAIT);
    let ignore_restart = if is_blocking {
        has_recv_timeout(sock.fd())?
    } else {
        false
    };
    if is_blocking {
        request.cache.add_sys_block(req, ignore_restart)?;
    }

    // Perform recvmsg(2).
    let result = recvmsg(sock.fd(), hdr.as_mut(), call_flags);

    // Remove invalidation record.
    if is_blocking {
        request.cache.del_sys_block(req.id)?;
    }

    // Check result after critical block.
    let recv = result?;
    let r_bytes = recv.bytes;

    // Parse control buffer allowing truncations.
    let cmsgs = recv.cmsgs(true /*trunc_ok*/)?;

    // Set each buffer's length to number of bytes received.
    let mut rem = r_bytes;
    let mut bufs = msg_bufs.iter_mut();

    for iov in &msg_iovs {
        if iov.iov_len == 0 {
            continue;
        }

        let (buf, _ptr) = match bufs.next() {
            Some(buf) => buf,
            None => break,
        };
        let n = rem.min(iov.iov_len);

        // SAFETY: recvmsg(2) wrote "n" bytes into this buffer.
        unsafe { buf.set_len(n) };

        rem = rem.saturating_sub(n);
    }

    // Scatter payload into sandbox process iov buffers.
    scatter_iov(r_bytes, &msg_bufs, &mut iovs_l, &mut iovs_r)?;

    // Handle peer address logic.
    //
    // Linux touches msg_name/msg_namelen only when msg_name is not NULL.
    // Linux rejects negative values for msg_namelen.
    let (namelen_out, addr_len) = if user_nam_base != 0 {
        let namelen = socklen_t::try_from(hdr.msg_namelen).or(Err(Errno::EINVAL))?;
        let (namelen_out, addr_len) = if sock.get_dom()? == AddressFamily::Unix {
            fixup_unix_addr(request, sock.fd(), hdr.msg_name, namelen, &mut addr_buf)?
        } else {
            copy_addr(hdr.msg_name, namelen, &mut addr_buf)?
        };
        hdr.msg_namelen = c_int::try_from(namelen_out).or(Err(Errno::EINVAL))?;
        (namelen_out, addr_len)
    } else {
        (0, 0)
    };

    // Handle control messages.
    //
    // Pass unsupported control messages unchanged.
    //
    // Linux doesn't fail on undeliverable control messages.
    // Linux sets MSG_CTRUNC for undelivered file descriptors.
    if !cmsgs.is_empty() {
        let close_on_exec =
            flags.force_cloexec() || call_flags.contains(MsgFlags::MSG_CMSG_CLOEXEC);
        let rand_fd = flags.force_rand_fd();

        // Ensure memory is writable before installing fds.
        //
        // This is best effort, we can still leak fds if page protections
        // change after this call but before the next write memory call.
        let has_fd_cmsg = has_fd_cmsg(&cmsgs);
        let ctl_fault = if has_fd_cmsg {
            let maps = ProcMaps::new(req.pid())?;
            if !request.is_valid() {
                return Err(Errno::ESRCH);
            }
            maps.get_fault(user_ctl_base, hdr.msg_controllen.min(user_ctl_size))? > 0
        } else {
            false
        };

        if ctl_fault {
            drop(cmsgs); // close owned fds.
            hdr.msg_flags |= MSG_CTRUNC as c_uint;
            hdr.msg_controllen = 0;
        } else {
            let (cmsgs, cmsgs_truncated) =
                request.fixup_cmsgs(sock.fd(), cmsgs, user_ctl_size, close_on_exec, rand_fd)?;
            let (out_buf, cmsg_len, truncated) = request.setup_cmsgs(&cmsgs, user_ctl_size)?;

            if truncated || cmsgs_truncated {
                hdr.msg_flags |= MSG_CTRUNC as c_uint;
            }
            hdr.msg_controllen = if cmsg_len > 0
                && request
                    .write_mem_all(&out_buf[..cmsg_len], user_ctl_base)
                    .is_err()
            {
                if has_fd_cmsg {
                    hdr.msg_flags |= MSG_CTRUNC as c_uint;
                }
                0
            } else {
                cmsg_len
            };
        }
    } else {
        hdr.msg_controllen = 0;
    }

    // Copy message header into pre-allocated buffer.
    //
    // Replace local pointers with sandbox process pointers.
    hdr.msg_iov = user_iov_base as *mut iovec;
    hdr.msg_name = user_nam_base as *mut c_void;
    hdr.msg_control = user_ctl_base as *mut c_void;

    // Copy message header handling 32-bit as necessary.
    if is32 {
        let m32: msghdr32 = hdr.try_into()?;
        let buf: [u8; size_of::<msghdr32>()] = m32.to_byte_array();
        hdr_buf.extend_from_slice(&buf);
    } else {
        let buf: [u8; size_of::<msghdr>()] = hdr.to_padded_bytes();
        hdr_buf.extend_from_slice(&buf);
    }

    // Gather message header and address into batch.
    //
    // Gather message header.
    iovs_l.push(IoSlice::new(&hdr_buf));
    iovs_r.push(RemoteIoVec {
        base: usize::try_from(args[1]).or(Err(Errno::EOVERFLOW))?,
        len: hdr_buf.len(),
    });

    // Gather peer address.
    #[expect(clippy::cast_possible_truncation)]
    let out_len = (namelen_out.min(user_nam_size as socklen_t)) as usize;
    let out_len = out_len.min(addr_len);
    if out_len > 0 {
        iovs_l.push(IoSlice::new(&addr_buf[..out_len]));
        iovs_r.push(RemoteIoVec {
            base: usize::try_from(user_nam_base).or(Err(Errno::EOVERFLOW))?,
            len: out_len,
        });
    }

    // Write in single batch.
    //
    // Linux rejects copy failures with EFAULT.
    if !iovs_l.is_empty() {
        let siz: usize = iovs_r.iter().map(|v| v.len).sum();
        let len = request.write_mem_many_all(&iovs_l, &iovs_r)?;
        if len != siz {
            return Err(Errno::EFAULT);
        }
    }

    // Return number of payload bytes received.
    #[expect(clippy::cast_possible_wrap)]
    Ok(request.return_syscall(r_bytes as i64))
}

pub(crate) fn handle_recvmmsg(
    request: &UNotifyEventRequest,
    sock: SockInfo,
    sock_opts: SockOpts,
    args: &[u64; 6],
) -> Result<ScmpNotifResp, Errno> {
    // Determine if timeout is 32-bit or 64-bit.
    let timeout_is32 = scmp_arch_is_compat_long32(request.scmpreq.data.arch);

    do_recvmmsg(request, sock, sock_opts, args, timeout_is32)
}

pub(crate) fn handle_recvmmsg64(
    request: &UNotifyEventRequest,
    sock: SockInfo,
    sock_opts: SockOpts,
    args: &[u64; 6],
) -> Result<ScmpNotifResp, Errno> {
    // timeout is always 64-bit for recvmmsg_time64(2).
    do_recvmmsg(request, sock, sock_opts, args, false)
}

// Helper to handle both recvmmsg(2) and recvmmsg_time64(2) syscalls.
fn do_recvmmsg(
    request: &UNotifyEventRequest,
    mut sock: SockInfo,
    sock_opts: SockOpts,
    args: &[u64; 6],
    timeout_is32: bool,
) -> Result<ScmpNotifResp, Errno> {
    // Truncate flags to 32-bit keeping unknown flags.
    let call_flags = to_msgflags(args[3]);

    // Determine bitness of sandbox process.
    let req = request.scmpreq;
    let is32 = scmp_arch_is_compat32(req.data.arch);

    // Linux rejects MSG_CMSG_COMPAT on 64-bit before reading timeout.
    if !is32 && call_flags.contains(MsgFlags::MSG_CMSG_COMPAT) {
        return Err(Errno::EINVAL);
    }
    let call_flags = call_flags.difference(MsgFlags::MSG_CMSG_COMPAT);

    // Read timespec structure for timeout (32-bit or 64-bit).
    let mut timeout = if args[4] != 0 {
        if timeout_is32 {
            Some(request.remote_timespec32(args[4])?)
        } else {
            Some(request.remote_timespec64(args[4])?)
        }
    } else {
        None
    };

    // Reject invalid timeout.
    if let Some(timeout) = timeout {
        if !timeout.is_valid() {
            return Err(Errno::EINVAL);
        }
    }

    // Linux rejects with EOPNOTSUPP when MSG_OOB isn't supported.
    if !sock_opts.options.allow_unsafe_oob() && call_flags.contains(MsgFlags::MSG_OOB) {
        return Err(Errno::EOPNOTSUPP);
    }

    // Linux truncates message count to unsigned int.
    #[expect(clippy::cast_possible_truncation)]
    let msg_count = args[2] as c_uint as usize;
    let msgs_offset = args[1];

    let hdr_sz = if is32 {
        size_of::<mmsghdr32>()
    } else {
        size_of::<mmsghdr>()
    };

    let mut call_flags = call_flags;
    let mut total: usize = 0;

    loop {
        #[expect(clippy::arithmetic_side_effects)]
        let batch = (msg_count - total).min(UIO_MAXIOV);
        let off = (total as u64)
            .checked_mul(hdr_sz as u64)
            .and_then(|off| msgs_offset.checked_add(off))
            .ok_or(Errno::EOVERFLOW)?;

        let (nmsg, oob) = match do_recvmmsg_one(
            request,
            &mut sock,
            sock_opts,
            call_flags,
            off,
            batch,
            &mut timeout,
        ) {
            Ok(result) => result,
            Err(errno) if total == 0 => return Err(errno),
            Err(_) => break,
        };
        total = total.checked_add(nmsg).ok_or(Errno::EOVERFLOW)?;

        // Linux stops at vlen, on short delivery, and on OOB data.
        if oob || nmsg < batch || total >= msg_count {
            break;
        }

        // Linux stops when the timeout expires.
        if let Some(timeout) = timeout {
            if timeout.tv_sec == 0 && timeout.tv_nsec == 0 {
                break;
            }
        }

        // Replace MSG_{WAITFORONE->DONTWAIT} on next batches.
        if call_flags.contains(MsgFlags::MSG_WAITFORONE) {
            call_flags.remove(MsgFlags::MSG_WAITFORONE);
            call_flags.insert(MsgFlags::MSG_DONTWAIT);
        }
    }

    // Copy remaining timeout back with any received message.
    if total > 0 && args[4] != 0 {
        if let Some(timeout) = timeout {
            if timeout_is32 {
                let t32: TimeSpec32 = timeout.try_into()?;
                let buf: [u8; size_of::<TimeSpec32>()] = t32.to_byte_array();
                request.write_mem_all(&buf, args[4])?;
            } else {
                let buf: [u8; size_of::<TimeSpec64>()] = timeout.to_byte_array();
                request.write_mem_all(&buf, args[4])?;
            }
        }
    }

    // Return number of messages received.
    #[expect(clippy::cast_possible_wrap)]
    Ok(request.return_syscall(total as i64))
}

// Do one recvmmsg over "msg_count" entries at "msg_offset".
//
// Return number of messages delivered and whether OOB data stopped loop.
#[expect(clippy::cognitive_complexity)]
fn do_recvmmsg_one(
    request: &UNotifyEventRequest,
    sock: &mut SockInfo,
    sock_opts: SockOpts,
    call_flags: MsgFlags,
    msgs_offset: u64,
    msg_count: usize,
    timeout: &mut Option<TimeSpec64>,
) -> Result<(usize, bool), Errno> {
    let SockOpts {
        flags,
        options: _,
        is_nonblock,
    } = sock_opts;

    // Determine bitness of sandbox process.
    let req = request.scmpreq;
    let is32 = scmp_arch_is_compat32(req.data.arch);

    // Preallocate memory for mmsghdr array.
    let hdr_sz = if is32 {
        size_of::<mmsghdr32>()
    } else {
        size_of::<mmsghdr>()
    };
    let total_sz = hdr_sz.checked_mul(msg_count).ok_or(Errno::EOVERFLOW)?;

    // Linux copies each mmsghdr eagerly.
    // Linux returns 0 for empty message vector.
    let hdr = request.read_vec(msgs_offset, total_sz)?;
    if msg_count > 0 && hdr.len() < hdr_sz {
        return Err(Errno::EFAULT);
    }

    // Convert to native format to pass to recvmmsg(2).
    let mut msgs = Vec::new();
    #[expect(clippy::type_complexity)]
    let mut msg_bufs: Vec<Option<Vec<(Vec<u8>, u64)>>> = Vec::new();
    let mut nam_bufs: Vec<Option<Vec<u8>>> = Vec::new();
    let mut ctl_bufs: Vec<Option<Vec<u8>>> = Vec::new();
    let mut msg_iovs: Vec<Vec<iovec>> = Vec::new();
    let mut user_iov_bases: Vec<Option<u64>> = Vec::new();
    let mut user_nam_bases: Vec<Option<(u64, usize)>> = Vec::new();
    let mut user_ctl_bases: Vec<Option<(u64, usize)>> = Vec::new();
    msgs.try_reserve_exact(msg_count).or(Err(Errno::ENOMEM))?;
    msg_bufs
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    nam_bufs
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    ctl_bufs
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    msg_iovs
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    user_iov_bases
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    user_nam_bases
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;
    user_ctl_bases
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;

    for chunk in hdr.chunks_exact(hdr_sz) {
        let inner: libc::mmsghdr = if is32 {
            let m32: mmsghdr32 = try_from_bytes(chunk)?;
            mmsghdr::from(m32).into()
        } else {
            let m64: mmsghdr = try_from_bytes(chunk)?;
            m64.into()
        };
        msgs.push(MmsgHdr::from_raw(inner));
    }

    // Linux discards bytes with MSG_TRUNC.
    let discard = call_flags.contains(MsgFlags::MSG_TRUNC)
        && matches!(sock.get_dom()?, AddressFamily::Inet | AddressFamily::Inet6)
        && sock.get_stype()? == SockType::Stream
        && sock.get_trunc_discard()?;

    // Linux parses messages eagerly.
    let mut msg_errno: Option<Errno> = None;
    for idx in 0..msgs.len() {
        let mut result = request.setup_mmsghdr_name(
            msgs[idx].as_inner_mut(),
            &mut nam_bufs,
            &mut user_nam_bases,
        );
        if result.is_ok() {
            result = request.read_mmsghdr_iovs(
                &mut msgs[idx..=idx],
                (!discard).then_some(&mut msg_bufs),
                &mut msg_iovs,
                &mut user_iov_bases,
            );
        }
        if result.is_ok() {
            result = request.setup_mmsghdr_ctl(
                msgs[idx].as_inner_mut(),
                &mut ctl_bufs,
                &mut user_ctl_bases,
            );
        }
        match result {
            Ok(()) => {}
            Err(errno) if idx == 0 => return Err(errno),
            Err(errno) => {
                msg_errno = Some(errno);
                msgs.truncate(idx);
                break; // partial success.
            }
        }
    }

    // Linux rejects short array read after message parse.
    if msg_errno.is_none() && hdr.len() < total_sz && !msgs.is_empty() {
        msg_errno = Some(Errno::EFAULT);
    }

    // Linux receives messages eagerly.
    let is_stream = sock.get_stype()? == SockType::Stream;
    if !call_flags.contains(MsgFlags::MSG_PEEK) && !msgs.is_empty() && (is_stream || msgs.len() > 1)
    {
        let maps = ProcMaps::new(req.pid())?;
        if !request.is_valid() {
            return Err(Errno::ESRCH);
        }

        let iovs = msg_iovs.get_mut(..msgs.len()).ok_or(Errno::EINVAL)?;
        let nmsg = msg_fault_count(
            &maps,
            (msgs_offset, hdr_sz),
            iovs,
            &msg_bufs,
            &user_nam_bases,
            is_stream,
        )?;
        if nmsg == 0 {
            return Err(Errno::EFAULT);
        }
        if nmsg < msgs.len() {
            msgs.truncate(nmsg);
            msg_errno = Some(Errno::EFAULT);
        }
    }

    // Handle scatter buffers to be used post-syscall.
    //
    // Buffer consists of the following items:
    // 1. Payload iov total
    // 2. Header + address: +2 per message
    // 3. Timeout.
    let buf_len: usize = msg_bufs
        .iter()
        .filter_map(Option::as_ref)
        .map(Vec::len)
        .try_fold(0usize, |acc, n| acc.checked_add(n))
        .ok_or(Errno::EOVERFLOW)?;
    let buf_len = msg_count
        .checked_mul(2)
        .and_then(|n| n.checked_add(buf_len))
        .and_then(|n| n.checked_add(1)) // timeout
        .ok_or(Errno::EOVERFLOW)?;

    let mut iovs_l: Vec<IoSlice<'_>> = Vec::new();
    let mut iovs_r: Vec<RemoteIoVec> = Vec::new();
    iovs_l.try_reserve_exact(buf_len).or(Err(Errno::ENOMEM))?;
    iovs_r.try_reserve_exact(buf_len).or(Err(Errno::ENOMEM))?;

    let mmsghdr_size = if is32 {
        size_of::<mmsghdr32>()
    } else {
        size_of::<mmsghdr>()
    };
    let hdr_len = msg_count
        .checked_mul(mmsghdr_size)
        .ok_or(Errno::EOVERFLOW)?;

    let mut hdr_buf: Vec<u8> = Vec::new();
    hdr_buf.try_reserve_exact(hdr_len).or(Err(Errno::ENOMEM))?;

    // Allocate address buffer which is one sockaddr_storage per message.
    let addr_buf_len = msg_count
        .checked_mul(SOCKADDR_SIZE)
        .ok_or(Errno::EOVERFLOW)?;
    let mut addr_buf: Vec<u8> = Vec::new();
    addr_buf
        .try_reserve_exact(addr_buf_len)
        .or(Err(Errno::ENOMEM))?;
    addr_buf.resize(addr_buf_len, 0);

    // Allocate per-message address metadata.
    #[expect(clippy::type_complexity)]
    let mut addr_meta: Vec<Option<(u64, usize, usize)>> = Vec::new();
    addr_meta
        .try_reserve_exact(msg_count)
        .or(Err(Errno::ENOMEM))?;

    // Inject a bad message to reproduce deferred errno(3).
    let mut nam_buf = [0u8; SOCKADDR_SIZE];
    if let Some(errno @ (Errno::EFAULT | Errno::EMSGSIZE | Errno::EINVAL)) = msg_errno {
        // SAFETY: An all-zero mmsghdr is a valid value.
        let mut inner: libc::mmsghdr = unsafe { std::mem::zeroed() };
        match errno {
            Errno::EFAULT => inner.msg_hdr.msg_iovlen = 1, // NULL iov faults on copy
            Errno::EMSGSIZE => inner.msg_hdr.msg_iovlen = 1025, // > UIO_MAXIOV (1024)
            _ => {
                // EINVAL: Negative msg_namelen with a valid msg_name.
                inner.msg_hdr.msg_name = nam_buf.as_mut_ptr().cast();
                inner.msg_hdr.msg_namelen = socklen_t::MAX;
            }
        }
        msgs.push(MmsgHdr::from_raw(inner));
    }

    // Track blocking call for invalidation semantics.
    let is_blocking = !is_nonblock && !call_flags.contains(MsgFlags::MSG_DONTWAIT);
    let ignore_restart = if is_blocking {
        timeout.is_some() || has_recv_timeout(sock.fd())?
    } else {
        false
    };
    if is_blocking {
        request.cache.add_sys_block(req, ignore_restart)?;
    }

    // Perform recvmmsg(2).
    let result = recvmmsg(sock.fd(), &mut msgs, call_flags, timeout.as_mut());

    // Remove invalidation record.
    if is_blocking {
        request.cache.del_sys_block(req.id)?;
    }

    // Check result after critical block.
    let msg_count = result?;

    // Parse control buffers allowing truncations.
    let mut msg_cmsgs: Vec<Vec<CmsgOwned>> = Vec::new();
    if msgs
        .iter()
        .take(msg_count)
        .any(|msg| !msg.msg_hdr().cmsg_bytes().is_empty())
    {
        msg_cmsgs
            .try_reserve_exact(msg_count)
            .or(Err(Errno::ENOMEM))?;

        let mut cmsg_errno: Option<Errno> = None;
        for msg in msgs.iter().take(msg_count) {
            let recv = RecvMsg {
                msghdr: msg.msg_hdr(),
                bytes: msg.msg_len() as usize,
                flags: msg.msg_hdr().msg_flags(),
            };
            match recv.cmsgs(true /*trunc_ok*/) {
                Ok(cmsgs) => msg_cmsgs.push(cmsgs),
                Err(errno) => {
                    cmsg_errno.get_or_insert(errno);
                }
            }
        }
        if let Some(errno) = cmsg_errno {
            return Err(errno);
        }
    }
    let msg_cmsgs = msg_cmsgs
        .into_iter()
        .chain(std::iter::repeat_with(Vec::new));

    // Iterate over raw mmsghdr results with each message's owned cmsgs.
    //
    // Linux returns number of processed messages when a later iteration fails.
    let mut ctl_maps: Option<ProcMaps> = None;
    for ((idx, mmsg_hdr), cmsgs) in msgs.iter_mut().enumerate().take(msg_count).zip(msg_cmsgs) {
        let mmsg_hdr = mmsg_hdr.as_inner_mut();

        // Restore msg_iov pointer.
        if let Some(iov_ptr) = user_iov_bases.get(idx).copied().flatten() {
            mmsg_hdr.msg_hdr.msg_iov = iov_ptr as *mut iovec;
        }

        // Prepare peer address.
        let addr_out = if let Some((nam_ptr, nam_len)) = user_nam_bases.get(idx).copied().flatten()
        {
            let addr_off = idx.checked_mul(SOCKADDR_SIZE).ok_or(Errno::EOVERFLOW)?;
            #[expect(clippy::arithmetic_side_effects)]
            let addr_buf = &mut addr_buf[addr_off..addr_off + SOCKADDR_SIZE];
            let (namelen_out, addr_len) = if sock.get_dom()? == AddressFamily::Unix {
                fixup_unix_addr(
                    request,
                    sock.fd(),
                    mmsg_hdr.msg_hdr.msg_name,
                    mmsg_hdr.msg_hdr.msg_namelen,
                    addr_buf,
                )?
            } else {
                copy_addr(
                    mmsg_hdr.msg_hdr.msg_name,
                    mmsg_hdr.msg_hdr.msg_namelen,
                    addr_buf,
                )?
            };

            // Fixup message header pointers.
            mmsg_hdr.msg_hdr.msg_namelen = namelen_out;
            mmsg_hdr.msg_hdr.msg_name = nam_ptr as *mut c_void;

            if addr_len > 0 {
                Some((nam_ptr, nam_len, addr_len))
            } else {
                None
            }
        } else {
            None
        };

        // Prepare control messages.
        //
        // Linux doesn't fail on undeliverable control messages.
        // Linux sets MSG_CTRUNC for undelivered file descriptors.
        #[expect(clippy::disallowed_methods)]
        #[expect(clippy::useless_conversion)]
        if let Some((ctl_ptr, ctl_len)) = user_ctl_bases.get(idx).copied().flatten() {
            if !cmsgs.is_empty() {
                let close_on_exec =
                    flags.force_cloexec() || call_flags.contains(MsgFlags::MSG_CMSG_CLOEXEC);
                let rand_fd = flags.force_rand_fd();

                // Ensure memory is writable before installing fds.
                //
                // This is best effort, we can still leak fds if page protections
                // change after this call but before the next write memory call.
                let has_fd_cmsg = has_fd_cmsg(&cmsgs);
                let ctl_fault = if has_fd_cmsg {
                    if ctl_maps.is_none() {
                        ctl_maps = Some(ProcMaps::new(req.pid())?);
                        if !request.is_valid() {
                            return Err(Errno::ESRCH);
                        }
                    }

                    if let Some(ref maps) = ctl_maps {
                        #[expect(clippy::unnecessary_cast)]
                        let r_ctl_len = mmsg_hdr.msg_hdr.msg_controllen as usize;

                        maps.get_fault(ctl_ptr, r_ctl_len.min(ctl_len))? > 0
                    } else {
                        false
                    }
                } else {
                    false
                };

                let cmsg_len = if ctl_fault {
                    drop(cmsgs); // close owned fds.
                    mmsg_hdr.msg_hdr.msg_flags |= MsgFlags::MSG_CTRUNC.bits();
                    0
                } else {
                    let (cmsgs, cmsgs_truncated) =
                        request.fixup_cmsgs(sock.fd(), cmsgs, ctl_len, close_on_exec, rand_fd)?;
                    let (out_buf, cmsg_len, truncated) = request.setup_cmsgs(&cmsgs, ctl_len)?;
                    if truncated || cmsgs_truncated {
                        mmsg_hdr.msg_hdr.msg_flags |= MsgFlags::MSG_CTRUNC.bits();
                    }

                    if cmsg_len > 0
                        && request
                            .write_mem_all(&out_buf[..cmsg_len], ctl_ptr)
                            .is_err()
                    {
                        if has_fd_cmsg {
                            mmsg_hdr.msg_hdr.msg_flags |= MsgFlags::MSG_CTRUNC.bits();
                        }
                        0
                    } else {
                        cmsg_len
                    }
                };

                // unwrap is for musl compat.
                mmsg_hdr.msg_hdr.msg_control = ctl_ptr as *mut c_void;
                mmsg_hdr.msg_hdr.msg_controllen = cmsg_len.try_into().unwrap();
            } else {
                mmsg_hdr.msg_hdr.msg_control = ctl_ptr as *mut c_void;
                mmsg_hdr.msg_hdr.msg_controllen = 0;
            }
        } else {
            mmsg_hdr.msg_hdr.msg_controllen = 0;
        }

        // Copy header into pre-allocated buffer.
        if is32 {
            let m32: mmsghdr32 = (*mmsg_hdr).try_into()?;
            let buf: [u8; size_of::<mmsghdr32>()] = m32.to_byte_array();
            hdr_buf.extend_from_slice(&buf);
        } else {
            let m64: mmsghdr = (*mmsg_hdr).into();
            let buf: [u8; size_of::<mmsghdr>()] = m64.to_padded_bytes();
            hdr_buf.extend_from_slice(&buf);
        }

        addr_meta.push(addr_out);
    }

    // Set each buffer's length to number of bytes received.
    for idx in 0..msg_count {
        let msg_len = msgs[idx].msg_len() as usize;
        if let Some(bufs) = msg_bufs.get_mut(idx).and_then(Option::as_mut) {
            let mut rem = msg_len;
            let mut bufs = bufs.iter_mut();

            for iov in &msg_iovs[idx] {
                if iov.iov_len == 0 {
                    continue;
                }

                let (buf, _ptr) = match bufs.next() {
                    Some(buf) => buf,
                    None => break,
                };
                let n = rem.min(iov.iov_len);

                // SAFETY: recvmmsg(2) wrote "n" bytes into this buffer.
                unsafe { buf.set_len(n) };

                rem = rem.saturating_sub(n);
            }
        }
    }

    // Build per-message iovs.
    //
    // For each message push payload, header, and address.
    for idx in 0..msg_count {
        let msg = msgs[idx].as_inner_mut();

        // Copy iov payload scatter entries.
        if let Some(bufs) = msg_bufs.get(idx).and_then(Option::as_ref) {
            scatter_iov(msg.msg_len as usize, bufs, &mut iovs_l, &mut iovs_r)?;
        }

        // Copy header.
        let hdr_off = idx.checked_mul(mmsghdr_size).ok_or(Errno::EOVERFLOW)?;
        let hdr_len = (idx as u64)
            .checked_mul(mmsghdr_size as u64)
            .ok_or(Errno::EOVERFLOW)?;
        let off = msgs_offset.checked_add(hdr_len).ok_or(Errno::EOVERFLOW)?;

        #[expect(clippy::arithmetic_side_effects)]
        iovs_l.push(IoSlice::new(&hdr_buf[hdr_off..hdr_off + mmsghdr_size]));
        iovs_r.push(RemoteIoVec {
            base: usize::try_from(off).or(Err(Errno::EOVERFLOW))?,
            len: mmsghdr_size,
        });

        // Copy peer address.
        if let Some((nam_ptr, nam_len, addr_len)) = addr_meta[idx] {
            let namelen_out = msg.msg_hdr.msg_namelen;

            #[expect(clippy::cast_possible_truncation)]
            let out_len = (namelen_out.min(nam_len as socklen_t)) as usize;
            let out_len = out_len.min(addr_len);

            if out_len > 0 {
                let addr_off = idx.checked_mul(SOCKADDR_SIZE).ok_or(Errno::EOVERFLOW)?;
                #[expect(clippy::arithmetic_side_effects)]
                iovs_l.push(IoSlice::new(&addr_buf[addr_off..addr_off + out_len]));
                iovs_r.push(RemoteIoVec {
                    base: usize::try_from(nam_ptr).or(Err(Errno::EOVERFLOW))?,
                    len: out_len,
                });
            }
        }
    }

    // Write in a single batch.
    //
    // Linux returns count of messages on partial writes.
    let mut msg_count = msg_count;
    if !iovs_l.is_empty() {
        let len = request.write_mem_many_all(&iovs_l, &iovs_r)?;
        let siz: usize = iovs_r.iter().map(|v| v.len).sum();

        // Detect partial writes an return count of messages.
        if len != siz {
            let mut off = 0usize;
            let mut nbytes = 0usize;
            let mut nwrite = 0usize;

            #[expect(clippy::arithmetic_side_effects)]
            for idx in 0..msg_count {
                let n = msg_iov_count(
                    &msgs[idx],
                    msg_bufs.get(idx).and_then(Option::as_ref),
                    &addr_meta[idx],
                );

                nbytes += iovs_r[off..off + n].iter().map(|v| v.len).sum::<usize>();
                if nbytes > len {
                    break;
                }

                nwrite += 1;
                off += n;
            }

            if nwrite == 0 {
                return Err(Errno::EFAULT);
            }

            msg_count = nwrite;
        }
    }

    // Linux returns right away on OOB data.
    let oob = msg_count
        .checked_sub(1)
        .and_then(|idx| msgs.get(idx))
        .map(|msg| (msg.as_inner().msg_hdr.msg_flags & libc::MSG_OOB) != 0)
        .unwrap_or(false);

    Ok((msg_count, oob))
}

fn msg_iov_count(
    mmsg: &MmsgHdr,
    bufs: Option<&Vec<(Vec<u8>, u64)>>,
    addr_meta: &Option<(u64, usize, usize)>,
) -> usize {
    let mut n = 0usize;

    // Count payload:
    // One iov per scatter buffer entry up to msg_len bytes.
    if let Some(bufs) = bufs {
        let mut nrem = mmsg.msg_len() as usize;
        if nrem > 0 {
            #[expect(clippy::arithmetic_side_effects)]
            for (buf, _) in bufs {
                if nrem == 0 {
                    break;
                }
                n += 1;
                nrem = nrem.saturating_sub(buf.len());
            }
        }
    }

    // Count message header.
    #[expect(clippy::arithmetic_side_effects)]
    {
        n += 1;
    }

    // Count address if present.
    if let Some((_, nam_len, addr_len)) = addr_meta {
        let namelen_out = mmsg.as_inner().msg_hdr.msg_namelen;
        #[expect(clippy::cast_possible_truncation)]
        let out_len = (namelen_out.min(*nam_len as socklen_t)) as usize;
        let out_len = out_len.min(*addr_len);
        #[expect(clippy::arithmetic_side_effects)]
        if out_len > 0 {
            n += 1;
        }
    }

    n
}

// Gather iov payload scatter entries into pre-allocated buffers.
fn scatter_iov<'a>(
    size: usize,
    bufs: &'a [(Vec<u8>, u64)],
    local_iovs: &mut Vec<IoSlice<'a>>,
    remote_iovs: &mut Vec<RemoteIoVec>,
) -> Result<(), Errno> {
    if size == 0 || bufs.is_empty() {
        return Ok(());
    }

    let mut nrem = size;
    for (buf, ptr) in bufs {
        if nrem == 0 {
            break;
        }
        let take = nrem.min(buf.len());

        local_iovs.push(IoSlice::new(&buf[..take]));
        remote_iovs.push(RemoteIoVec {
            base: usize::try_from(*ptr).or(Err(Errno::EFAULT))?,
            len: take,
        });

        nrem = nrem.checked_sub(take).ok_or(Errno::EOVERFLOW)?;
    }

    Ok(())
}

// Count leading messages with writable copy-out destinations.
fn msg_fault_count(
    maps: &ProcMaps,
    hdr: (u64, usize),
    msg_iovs: &mut [Vec<iovec>],
    msg_bufs: &[Option<Vec<IovBuf>>],
    nam_bases: &[Option<(u64, usize)>],
    is_stream: bool,
) -> Result<usize, Errno> {
    let (hdr_off, hdr_sz) = hdr;

    for (idx, iovs) in msg_iovs.iter_mut().enumerate() {
        let off = (idx as u64)
            .checked_mul(hdr_sz as u64)
            .and_then(|off| hdr_off.checked_add(off))
            .ok_or(Errno::EOVERFLOW)?;
        if maps.get_fault(off, hdr_sz)? > 0 {
            return Ok(idx);
        }
        let bufs = msg_bufs.get(idx).and_then(|bufs| bufs.as_deref());
        if msg_iov_fault(maps, iovs, bufs, is_stream)? {
            return Ok(idx);
        }
        if let Some((ptr, len)) = nam_bases.get(idx).copied().flatten() {
            if maps.get_fault(ptr, len)? > 0 {
                return Ok(idx);
            }
        }
    }

    Ok(msg_iovs.len())
}

// Check one message's payload iovecs for writability.
//
// Return true if message cannot be delivered.
fn msg_iov_fault(
    maps: &ProcMaps,
    iovs: &mut [iovec],
    bufs: Option<&[IovBuf]>,
    is_stream: bool,
) -> Result<bool, Errno> {
    // Linux discards receives with MSG_TRUNC.
    let bufs = match bufs {
        Some(bufs) => bufs,
        None => return Ok(false),
    };

    let mut bufs = bufs.iter();
    let mut short = false;
    let mut total: usize = 0;
    let mut nlen: usize = 0;

    for iov in iovs.iter_mut() {
        if iov.iov_len == 0 {
            continue;
        }
        let ptr = match bufs.next() {
            Some((_, ptr)) => *ptr,
            None => break,
        };
        total = total.checked_add(iov.iov_len).ok_or(Errno::EOVERFLOW)?;
        if short {
            iov.iov_len = 0;
            continue;
        }
        let bad = maps.get_fault(ptr, iov.iov_len)?;
        if bad > 0 {
            if !is_stream {
                return Ok(true);
            }
            short = true;
            iov.iov_len = iov.iov_len.saturating_sub(bad);
        }
        nlen = nlen.checked_add(iov.iov_len).ok_or(Errno::EOVERFLOW)?;
    }

    Ok(total > 0 && nlen == 0)
}

// Resolve Unix peer address from a kernel-returned message header,
// and copy into given buffer.
//
// Returns new "msg_namelen" value and address bytes to copy.
fn fixup_unix_addr<Fd: AsFd>(
    request: &UNotifyEventRequest,
    fd: Fd,
    msg_name: *mut c_void,
    msg_namelen: socklen_t,
    addr_buf: &mut [u8],
) -> Result<(socklen_t, usize), Errno> {
    let r_addr = if !msg_name.is_null() && msg_namelen > 0 {
        // SAFETY: msg_name and msg_namelen are from a kernel-returned msghdr.
        unsafe { SockaddrStorage::from_raw(msg_name as *const sockaddr, Some(msg_namelen)) }
    } else {
        None
    };

    if let Some(mut addr) = r_addr {
        let hdr_namelen = if let Ok(ino) = fd_inode(fd) {
            if let Ok(peer_addr) = request.resolve_unix_peer(&addr, ino) {
                addr = peer_addr;
                addr.as_unix_addr().map_or(addr.len(), unix_addr_len2)
            } else {
                msg_namelen
            }
        } else {
            msg_namelen
        };

        // SAFETY:
        // 1. SockaddrStorage is initialized.
        // 2. as_ptr() and len() return valid bounds.
        let src =
            unsafe { std::slice::from_raw_parts(addr.as_ptr().cast::<u8>(), addr.len() as usize) };

        let namelen_out = addr.len().min(hdr_namelen);
        let len = src.len().min(addr_buf.len());

        addr_buf[..len].copy_from_slice(&src[..len]);

        Ok((namelen_out, len))
    } else {
        Ok((0, 0))
    }
}

// Copy address bytes from a kernel-returned message header into given buffer.
//
// Returns new "msg_namelen" value and address bytes to copy.
fn copy_addr(
    msg_name: *mut c_void,
    msg_namelen: socklen_t,
    addr_buf: &mut [u8],
) -> Result<(socklen_t, usize), Errno> {
    if msg_name.is_null() || msg_namelen == 0 {
        return Ok((0, 0));
    }

    let len = msg_namelen as usize;
    if len > addr_buf.len() {
        return Err(Errno::EINVAL);
    }

    // SAFETY: msg_name and msg_namelen are from a kernel-returned msghdr.
    let src = unsafe { std::slice::from_raw_parts(msg_name as *const u8, len) };
    addr_buf[..len].copy_from_slice(src);

    Ok((msg_namelen, len))
}

// Check control messages for ones that install a file descriptor.
fn has_fd_cmsg(cmsgs: &[CmsgOwned]) -> bool {
    cmsgs
        .iter()
        .any(|cmsg| matches!(cmsg, CmsgOwned::ScmRights(_) | CmsgOwned::ScmPidFd(_)))
}