refactor(broker): use rustix for descriptors and processes

Replace the hand-built sendmsg and recvmsg control messages, the sealed
memfd handling, and the polling, signalling, and child status calls with
typed rustix calls. Broker functions now take borrowed descriptors instead
of raw integers, so fd lifetimes are checked by the compiler.

The packet and descriptor caps, the truncation checks, and the signal
allow-list stay. Passed descriptors now also close when a packet with no
data is rejected.

One check narrows: rustix skips ancillary types it does not know instead
of reporting them, so only descriptors and credentials are visible here,
and credentials are still rejected. The skipped case is SCM_SECURITY,
which needs SO_PASSSEC on the socket, and nothing here sets it.

The post-clone child stub stays on raw syscalls by design, and the
broker entry keeps one borrow of its inherited channels, which live until
the process exits.

src/daemon/overlay/lifecycle.rs

@@ -1,4 +1,4 @@
-use std::os::fd::{AsRawFd, BorrowedFd};
+use std::os::fd::{AsFd, BorrowedFd};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
@@ -108,7 +108,7 @@ impl OverlayLifecycle {
                 &candidate.program,
                 &launch.arguments,
                 launch.environment,
-                daemon_watchdog.as_raw_fd(),
+                daemon_watchdog.as_fd(),
             )?;
             self.child.start(child)?;
             self.child

src/process_broker/bootstrap.rs

@@ -1,5 +1,5 @@
 use std::io;
-use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
+use std::os::fd::{AsRawFd, OwnedFd};
 use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
 
@@ -17,6 +17,13 @@ use std::thread::Builder;
 use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result};
+#[cfg(not(test))]
+use rustix::io as rustix_io;
+use rustix::io::Errno;
+use rustix::net::{self, AddressFamily, SocketFlags, SocketType};
+#[cfg(not(test))]
+use rustix::process::Signal;
+use rustix::process::{self as rustix_process, Pid, WaitOptions};
 
 use crate::daemon::protocol_v2::ProtocolToken;
 #[cfg(not(test))]
@@ -152,9 +159,8 @@ pub(super) fn start() -> Result<ProcessBroker> {
         }),
     };
     if let Err(error) = broker.request(BrokerOperation::Ping) {
-        // SAFETY: pid is the raw-clone broker child created above.
-        unsafe {
-            libc::kill(pid, libc::SIGKILL);
+        if let Some(broker_pid) = Pid::from_raw(pid) {
+            let _ = rustix_process::kill_process(broker_pid, Signal::KILL);
         }
         wait_for_broker_process(pid);
         return Err(error).context("process broker exec/authentication handshake failed");
@@ -164,37 +170,20 @@ pub(super) fn start() -> Result<ProcessBroker> {
 
 #[cfg(not(test))]
 fn duplicate_for_exec(descriptor: &OwnedFd) -> Result<OwnedFd> {
-    // SAFETY: F_DUPFD_CLOEXEC duplicates the live descriptor at or above five.
-    let duplicate = unsafe { libc::fcntl(descriptor.as_raw_fd(), libc::F_DUPFD_CLOEXEC, 5) };
-    if duplicate < 0 {
-        return Err(io::Error::last_os_error()).context("failed to stage broker descriptor");
-    }
-    // SAFETY: fcntl returned a fresh owned descriptor.
-    Ok(unsafe { OwnedFd::from_raw_fd(duplicate) })
+    rustix_io::fcntl_dupfd_cloexec(descriptor, 5)
+        .map_err(io::Error::from)
+        .context("failed to stage broker descriptor")
 }
 
 fn socket_pair(label: &str) -> Result<(OwnedFd, OwnedFd)> {
-    let mut sockets = [0; 2];
-    // SAFETY: sockets has room for the returned descriptor pair.
-    if unsafe {
-        libc::socketpair(
-            libc::AF_UNIX,
-            libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC,
-            0,
-            sockets.as_mut_ptr(),
-        )
-    } != 0
-    {
-        return Err(io::Error::last_os_error())
-            .with_context(|| format!("failed to create {label} socketpair"));
-    }
-    // SAFETY: socketpair returned two new descriptors.
-    Ok(unsafe {
-        (
-            OwnedFd::from_raw_fd(sockets[0]),
-            OwnedFd::from_raw_fd(sockets[1]),
-        )
-    })
+    net::socketpair(
+        AddressFamily::UNIX,
+        SocketType::SEQPACKET,
+        SocketFlags::CLOEXEC,
+        None,
+    )
+    .map_err(io::Error::from)
+    .with_context(|| format!("failed to create {label} socketpair"))
 }
 
 /// Whether the broker process exits within `grace`, reaping it if it does.
@@ -203,20 +192,17 @@ pub(super) fn broker_exits_within(child_pid: libc::pid_t, grace: Duration) -> bo
         return true;
     }
 
+    let Some(pid) = Pid::from_raw(child_pid) else {
+        return true;
+    };
     let deadline = Instant::now() + grace;
-    let mut status = 0;
     loop {
-        // SAFETY: child_pid names the raw-clone broker child owned by its guard.
-        let result = unsafe { libc::waitpid(child_pid, &raw mut status, libc::WNOHANG) };
-        if result == child_pid {
-            return true;
-        }
-        if result < 0 {
-            if io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
-                continue;
-            }
+        match rustix_process::waitpid(Some(pid), WaitOptions::NOHANG) {
+            Ok(Some(_)) => return true,
+            Ok(None) => {}
+            Err(Errno::INTR) => continue,
             // Already reaped: nothing is left to wait for.
-            return true;
+            Err(_) => return true,
         }
         if Instant::now() >= deadline {
             return false;
@@ -229,16 +215,11 @@ pub(super) fn wait_for_broker_process(child_pid: libc::pid_t) {
     if child_pid <= 0 {
         return;
     }
-    let mut status = 0;
-    loop {
-        // SAFETY: child_pid names the raw-clone broker child owned by its guard.
-        let result = unsafe { libc::waitpid(child_pid, &raw mut status, 0) };
-        if result == child_pid {
-            break;
-        }
-        if result < 0 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
-            continue;
-        }
-        break;
-    }
+    let Some(pid) = Pid::from_raw(child_pid) else {
+        return;
+    };
+    while matches!(
+        rustix_process::waitpid(Some(pid), WaitOptions::empty()),
+        Err(Errno::INTR)
+    ) {}
 }

src/process_broker/client.rs

@@ -2,7 +2,7 @@ use std::collections::VecDeque;
 use std::ffi::{OsStr, OsString};
 use std::io;
 use std::io::ErrorKind;
-use std::os::fd::{AsRawFd, OwnedFd, RawFd};
+use std::os::fd::{AsFd, BorrowedFd, OwnedFd};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard, OnceLock, TryLockError, Weak};
 #[cfg(test)]
@@ -10,6 +10,9 @@ use std::thread::JoinHandle;
 use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
+use rustix::io::Errno;
+use rustix::net::{self, SendFlags};
+use rustix::process::{self as rustix_process, Pid, Signal};
 
 use crate::daemon::protocol_v2::ProtocolId;
 
@@ -87,8 +90,8 @@ enum ExchangeWait {
 /// The transport policy and watchdog a spawn request carries, kept together so
 /// the private spawn path stays inside the argument-count lint.
 #[derive(Debug, Clone, Copy)]
-struct SpawnOptions {
-    watchdog: Option<RawFd>,
+struct SpawnOptions<'descriptor> {
+    watchdog: Option<BorrowedFd<'descriptor>>,
     wait: ExchangeWait,
 }
 
@@ -119,7 +122,7 @@ pub(crate) fn current() -> Result<ProcessBroker> {
 impl Drop for ProcessBrokerGuard {
     fn drop(&mut self) {
         let was_healthy = self.broker.inner.healthy.swap(false, Ordering::AcqRel);
-        let asked_to_stop = signal_shutdown(self.broker.inner.shutdown.as_raw_fd()).is_ok();
+        let asked_to_stop = signal_shutdown(self.broker.inner.shutdown.as_fd()).is_ok();
         // A broker whose exchange failed may be stuck and never read the
         // shutdown request. Waiting on it without a bound would hang the very
         // exit that lets the service manager start a fresh daemon.
@@ -129,11 +132,11 @@ impl Drop for ProcessBrokerGuard {
                     self.broker.inner.child_pid,
                     UNHEALTHY_BROKER_EXIT_GRACE,
                 ));
-        if must_kill && self.broker.inner.child_pid > 0 {
-            // SAFETY: child_pid is the broker process owned by this guard.
-            unsafe {
-                libc::kill(self.broker.inner.child_pid, libc::SIGKILL);
-            }
+        if must_kill
+            && self.broker.inner.child_pid > 0
+            && let Some(pid) = Pid::from_raw(self.broker.inner.child_pid)
+        {
+            let _ = rustix_process::kill_process(pid, Signal::KILL);
         }
         #[cfg(test)]
         if let Some(thread) = self
@@ -163,29 +166,20 @@ impl Drop for ProcessBrokerGuard {
     }
 }
 
-fn signal_shutdown(descriptor: RawFd) -> io::Result<()> {
+fn signal_shutdown(descriptor: BorrowedFd<'_>) -> io::Result<()> {
     let byte = [GRACEFUL_SHUTDOWN_BYTE];
     loop {
-        // SAFETY: descriptor is the broker shutdown socket and byte is readable.
-        let written = unsafe {
-            libc::send(
-                descriptor,
-                byte.as_ptr().cast(),
-                byte.len(),
-                libc::MSG_NOSIGNAL,
-            )
-        };
-        if written == 1 {
-            return Ok(());
+        match net::send(descriptor, &byte, SendFlags::NOSIGNAL) {
+            Ok(1) => return Ok(()),
+            Ok(_) => {
+                return Err(io::Error::new(
+                    ErrorKind::WriteZero,
+                    "short broker shutdown write",
+                ));
+            }
+            Err(Errno::INTR) => {}
+            Err(errno) => return Err(errno.into()),
         }
-        if written < 0 && io::Error::last_os_error().kind() == ErrorKind::Interrupted {
-            continue;
-        }
-        return Err(if written < 0 {
-            io::Error::last_os_error()
-        } else {
-            io::Error::new(ErrorKind::WriteZero, "short broker shutdown write")
-        });
     }
 }
 
@@ -253,7 +247,7 @@ impl ProcessBroker {
     fn request_with_descriptors(
         &self,
         operation: BrokerOperation,
-        descriptors: &[RawFd],
+        descriptors: &[BorrowedFd<'_>],
         wait: ExchangeWait,
     ) -> Result<(BrokerOutcome, Vec<OwnedFd>)> {
         let request_id = ProtocolId::generate()?.to_string();
@@ -271,9 +265,9 @@ impl ProcessBroker {
             bail!("process broker transport is no longer usable");
         }
         let exchange = (|| -> Result<(BrokerResponse, Vec<OwnedFd>)> {
-            set_socket_timeout(self.inner.socket.as_raw_fd(), exchange_timeout)?;
-            send_packet(self.inner.socket.as_raw_fd(), &packet, descriptors)?;
-            let (packet, descriptors) = recv_packet(self.inner.socket.as_raw_fd())?;
+            set_socket_timeout(self.inner.socket.as_fd(), exchange_timeout)?;
+            send_packet(self.inner.socket.as_fd(), &packet, descriptors)?;
+            let (packet, descriptors) = recv_packet(self.inner.socket.as_fd())?;
             let response: BrokerResponse = serde_json::from_slice(&packet)?;
             if response.request_id != request_id {
                 bail!("broker response identity mismatch");
@@ -368,7 +362,7 @@ impl ProcessBroker {
         let (input, input_descriptor) = encode_blob(input, MAX_INPUT_BYTES)?;
         let request_descriptors = input_descriptor
             .as_ref()
-            .map(|descriptor| vec![descriptor.as_raw_fd()])
+            .map(|descriptor| vec![descriptor.as_fd()])
             .unwrap_or_default();
         let (outcome, descriptors) = self.request_with_descriptors(
             BrokerOperation::Run {
@@ -428,7 +422,7 @@ impl ProcessBroker {
         let (input, input_descriptor) = encode_blob(input, super::manifest::input_cap(kind))?;
         let request_descriptors = input_descriptor
             .as_ref()
-            .map(|descriptor| vec![descriptor.as_raw_fd()])
+            .map(|descriptor| vec![descriptor.as_fd()])
             .unwrap_or_default();
         let (outcome, descriptors) = self.request_with_descriptors(
             BrokerOperation::Publish {
@@ -530,7 +524,7 @@ impl ProcessBroker {
         program: &OsStr,
         arguments: I,
         environment: Vec<(OsString, Option<OsString>)>,
-        watchdog: RawFd,
+        watchdog: BorrowedFd<'_>,
     ) -> Result<BrokerChild>
     where
         I: IntoIterator<Item = S>,
@@ -556,7 +550,7 @@ impl ProcessBroker {
         program: &OsStr,
         arguments: I,
         environment: Vec<(OsString, Option<OsString>)>,
-        options: SpawnOptions,
+        options: SpawnOptions<'_>,
     ) -> Result<BrokerChild>
     where
         I: IntoIterator<Item = S>,

src/process_broker/execution.rs

@@ -1,7 +1,6 @@
 use std::collections::BTreeMap;
 use std::io::{self, Read, Write};
-use std::mem::MaybeUninit;
-use std::os::fd::{AsRawFd, RawFd};
+use std::os::fd::{AsFd, BorrowedFd};
 use std::os::unix::process::CommandExt;
 use std::process::{Child, ChildStdin, Command, ExitStatus, Stdio};
 use std::sync::Arc;
@@ -10,6 +9,10 @@ use std::time::Duration;
 use std::{mem, thread};
 
 use anyhow::{Context, Result, anyhow};
+use rustix::event::{self, PollFd, PollFlags, Timespec};
+use rustix::fs::{self as rustix_fs, OFlags};
+use rustix::io::Errno;
+use rustix::process::{self as rustix_process, Pid, Signal, WaitId, WaitIdOptions};
 
 use crate::daemon::protocol_v2::{BootClock, BootDeadline};
 
@@ -25,7 +28,10 @@ use super::wire::{HelperKind, MAX_STDERR_BYTES, OutputMode, STDOUT_CAP_EXCEEDED}
 const PIPE_DRAIN_GRACE: Duration = Duration::from_secs(2);
 
 /// How often a pipe thread waiting on an idle pipe checks whether to give up.
-const PIPE_POLL_INTERVAL_MS: libc::c_int = 20;
+const PIPE_POLL_INTERVAL: Timespec = Timespec {
+    tv_sec: 0,
+    tv_nsec: 20_000_000,
+};
 
 const PIPE_HELD_OPEN: &str = "broker helper pipe is still open after the helper exited";
 
@@ -63,12 +69,9 @@ pub(super) fn terminate_owned_children(children: &mut BTreeMap<String, Child>) {
 }
 
 pub(super) fn kill_child_process_group(child: &mut Child) {
-    let pid = child.id();
-    if let Ok(group) = i32::try_from(pid) {
-        // SAFETY: broker children use a fresh process group equal to their PID.
-        if unsafe { libc::kill(-group, libc::SIGKILL) } == 0 {
-            return;
-        }
+    // Broker children use a fresh process group equal to their PID.
+    if rustix_process::kill_process_group(Pid::from_child(child), Signal::KILL).is_ok() {
+        return;
     }
     let _ = child.kill();
 }
@@ -145,48 +148,28 @@ impl Drop for OwnedProcess {
 }
 
 fn child_status_unreaped(child: &Child) -> io::Result<Option<i32>> {
-    let mut info = MaybeUninit::<libc::siginfo_t>::zeroed();
     // WNOWAIT pins the zombie leader and therefore its process-group identity.
-    // SAFETY: info is writable and child is owned by this broker.
-    if unsafe {
-        libc::waitid(
-            libc::P_PID,
-            child.id(),
-            info.as_mut_ptr(),
-            libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
-        )
-    } != 0
-    {
-        return Err(io::Error::last_os_error());
-    }
-    // SAFETY: waitid initialized info after success.
-    let info = unsafe { info.assume_init() };
-    // SAFETY: info was zeroed before the call and waitid only writes a SIGCHLD
-    // record into it, so the pid field it reads is initialized; it stays zero
-    // when WNOHANG found no waitable child.
-    if unsafe { info.si_pid() } == 0 {
+    let options = WaitIdOptions::EXITED | WaitIdOptions::NOHANG | WaitIdOptions::NOWAIT;
+    let Some(status) = rustix_process::waitid(WaitId::Pid(Pid::from_child(child)), options)? else {
         return Ok(None);
-    }
-    // SAFETY: CLD terminal records initialize si_status.
-    let detail = unsafe { info.si_status() };
-    let status = match info.si_code {
-        libc::CLD_EXITED => detail,
-        libc::CLD_KILLED | libc::CLD_DUMPED => 128_i32.saturating_add(detail),
-        _ => {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "waitid returned a nonterminal child status",
-            ));
-        }
     };
-    Ok(Some(status))
+    if let Some(code) = status.exit_status() {
+        return Ok(Some(code));
+    }
+    if let Some(signal) = status.terminating_signal() {
+        return Ok(Some(128_i32.saturating_add(signal)));
+    }
+    Err(io::Error::new(
+        io::ErrorKind::InvalidData,
+        "waitid returned a nonterminal child status",
+    ))
 }
 
 pub(super) fn publish_bounded(
     mut command: Command,
     input: Vec<u8>,
     timeout: Duration,
-    shutdown_fd: RawFd,
+    shutdown_fd: BorrowedFd<'static>,
 ) -> Result<PublishOutput> {
     if shutdown_requested(shutdown_fd)? {
         return Err(anyhow!("broker publication cancelled during shutdown"));
@@ -254,17 +237,10 @@ fn write_input_until(
     mut stdin: ChildStdin,
     input: &[u8],
     deadline: BootDeadline,
-    shutdown_fd: RawFd,
+    shutdown_fd: BorrowedFd<'static>,
 ) -> io::Result<()> {
-    let descriptor = stdin.as_raw_fd();
-    // SAFETY: fcntl reads and updates descriptor-local flags.
-    let flags = unsafe { libc::fcntl(descriptor, libc::F_GETFL) };
-    // SAFETY: `stdin` keeps the descriptor open, and F_SETFL only adds
-    // O_NONBLOCK to the flags just read from it.
-    if flags < 0 || unsafe { libc::fcntl(descriptor, libc::F_SETFL, flags | libc::O_NONBLOCK) } != 0
-    {
-        return Err(io::Error::last_os_error());
-    }
+    let flags = rustix_fs::fcntl_getfl(&stdin)?;
+    rustix_fs::fcntl_setfl(&stdin, flags | OFlags::NONBLOCK)?;
     let mut offset = 0;
     while offset < input.len() {
         if shutdown_requested(shutdown_fd)? {
@@ -298,7 +274,7 @@ pub(super) fn run_bounded(
     timeout: Duration,
     output_cap: usize,
     output_mode: OutputMode,
-    shutdown_fd: RawFd,
+    shutdown_fd: BorrowedFd<'static>,
 ) -> Result<BoundedOutput> {
     if shutdown_requested(shutdown_fd)? {
         return Err(anyhow!("broker helper cancelled during shutdown"));
@@ -421,7 +397,7 @@ struct AbandonablePipe<P> {
     abandon: Arc<AtomicBool>,
 }
 
-impl<P: AsRawFd> AbandonablePipe<P> {
+impl<P: AsFd> AbandonablePipe<P> {
     fn new(pipe: P, abandon: &Arc<AtomicBool>) -> Self {
         Self {
             pipe,
@@ -429,42 +405,31 @@ impl<P: AsRawFd> AbandonablePipe<P> {
         }
     }
 
-    fn wait_until_ready(&self, events: libc::c_short) -> io::Result<()> {
-        let mut pollfd = libc::pollfd {
-            fd: self.pipe.as_raw_fd(),
-            events,
-            revents: 0,
-        };
+    fn wait_until_ready(&self, events: PollFlags) -> io::Result<()> {
+        let mut pollfds = [PollFd::new(&self.pipe, events)];
         loop {
             if self.abandon.load(Ordering::Acquire) {
                 return Err(io::Error::new(io::ErrorKind::TimedOut, PIPE_HELD_OPEN));
             }
-            // SAFETY: pollfd points to one initialized entry, and the pipe
-            // stays open for as long as `self` owns it.
-            let result = unsafe { libc::poll(&raw mut pollfd, 1, PIPE_POLL_INTERVAL_MS) };
-            if result > 0 {
-                return Ok(());
-            }
-            if result < 0 {
-                let error = io::Error::last_os_error();
-                if error.kind() != io::ErrorKind::Interrupted {
-                    return Err(error);
-                }
+            match event::poll(&mut pollfds, Some(&PIPE_POLL_INTERVAL)) {
+                Ok(0) | Err(Errno::INTR) => {}
+                Ok(_) => return Ok(()),
+                Err(errno) => return Err(errno.into()),
             }
         }
     }
 }
 
-impl<P: Read + AsRawFd> Read for AbandonablePipe<P> {
+impl<P: Read + AsFd> Read for AbandonablePipe<P> {
     fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
-        self.wait_until_ready(libc::POLLIN)?;
+        self.wait_until_ready(PollFlags::IN)?;
         self.pipe.read(buffer)
     }
 }
 
-impl<P: Write + AsRawFd> Write for AbandonablePipe<P> {
+impl<P: Write + AsFd> Write for AbandonablePipe<P> {
     fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
-        self.wait_until_ready(libc::POLLOUT)?;
+        self.wait_until_ready(PollFlags::OUT)?;
         // A writable pipe has room for at least PIPE_BUF bytes, so a write of
         // no more than that cannot block on a reader that stopped reading.
         self.pipe.write(&buffer[..buffer.len().min(libc::PIPE_BUF)])

src/process_broker/server.rs

@@ -1,12 +1,17 @@
 use std::collections::{BTreeMap, VecDeque};
-use std::os::fd::{AsRawFd, OwnedFd, RawFd};
+use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd, RawFd};
 use std::os::unix::process::CommandExt;
 use std::process::{Child, ExitCode, Stdio};
 use std::thread::Builder;
 use std::time::Duration;
-use std::{env, io, mem};
+use std::{env, io};
 
 use anyhow::{Context, Result, anyhow, bail};
+use rustix::event::{self, PollFd, PollFlags};
+use rustix::io::{self as rustix_io, Errno, FdFlags};
+use rustix::net::SocketType;
+use rustix::net::sockopt;
+use rustix::process::{self as rustix_process, Pid, Signal};
 
 use crate::daemon::protocol_v2::ProtocolId;
 use crate::env_vars;
@@ -33,18 +38,24 @@ pub(crate) fn run_internal_broker_if_requested() -> Option<ExitCode> {
         .parse::<RawFd>()
         .ok()?;
     let token = env::var(BROKER_TOKEN_ENV).ok()?;
-    if fd != BROKER_FD
-        || shutdown_fd != BROKER_SHUTDOWN_FD
-        || !canonical_lower_hex(&token, 64)
-        || validate_broker_socket(fd).is_err()
-        || validate_broker_socket(shutdown_fd).is_err()
-    {
+    if fd != BROKER_FD || shutdown_fd != BROKER_SHUTDOWN_FD || !canonical_lower_hex(&token, 64) {
+        return Some(ExitCode::from(126));
+    }
+    // SAFETY: the fixed numbers are never -1, the socket checks below reject a
+    // number that is not an open broker channel before anything else uses it,
+    // and nothing in this process closes those channels before it exits.
+    let (fd, shutdown_fd) = unsafe {
+        (
+            BorrowedFd::borrow_raw(fd),
+            BorrowedFd::borrow_raw(shutdown_fd),
+        )
+    };
+    if validate_broker_socket(fd).is_err() || validate_broker_socket(shutdown_fd).is_err() {
         return Some(ExitCode::from(126));
     }
     // Restore CLOEXEC before any runtime helper can inherit broker internals.
     for descriptor in [fd, shutdown_fd] {
-        // SAFETY: both descriptors were validated as inherited broker channels.
-        if unsafe { libc::fcntl(descriptor, libc::F_SETFD, libc::FD_CLOEXEC) } != 0 {
+        if rustix_io::fcntl_setfd(descriptor, FdFlags::CLOEXEC).is_err() {
             return Some(ExitCode::from(126));
         }
     }
@@ -60,23 +71,8 @@ pub(crate) fn run_internal_broker_if_requested() -> Option<ExitCode> {
     })
 }
 
-fn validate_broker_socket(fd: RawFd) -> io::Result<()> {
-    let mut socket_type = 0_i32;
-    let mut length = mem::size_of::<i32>() as libc::socklen_t;
-    // SAFETY: the output slots are valid for getsockopt.
-    if unsafe {
-        libc::getsockopt(
-            fd,
-            libc::SOL_SOCKET,
-            libc::SO_TYPE,
-            (&raw mut socket_type).cast(),
-            &raw mut length,
-        )
-    } != 0
-    {
-        return Err(io::Error::last_os_error());
-    }
-    if socket_type != libc::SOCK_SEQPACKET {
+fn validate_broker_socket(fd: BorrowedFd<'_>) -> io::Result<()> {
+    if sockopt::socket_type(fd)? != SocketType::SEQPACKET {
         return Err(io::Error::new(
             io::ErrorKind::InvalidInput,
             "broker descriptor is not SOCK_SEQPACKET",
@@ -85,7 +81,11 @@ fn validate_broker_socket(fd: RawFd) -> io::Result<()> {
     Ok(())
 }
 
-fn broker_loop(socket: RawFd, shutdown_fd: RawFd, token: &str) -> Result<()> {
+fn broker_loop(
+    socket: BorrowedFd<'static>,
+    shutdown_fd: BorrowedFd<'static>,
+    token: &str,
+) -> Result<()> {
     let mut ownership = BrokerOwnership::default();
     loop {
         if wait_for_request(socket, shutdown_fd)? == BrokerWake::Shutdown {
@@ -139,7 +139,7 @@ fn broker_loop(socket: RawFd, shutdown_fd: RawFd, token: &str) -> Result<()> {
         let response_descriptors = wire_response
             .descriptors
             .iter()
-            .map(AsRawFd::as_raw_fd)
+            .map(AsFd::as_fd)
             .collect::<Vec<_>>();
         send_packet(socket, &bytes, &response_descriptors)?;
         if take_graceful_shutdown_signal(shutdown_fd)? {
@@ -155,39 +155,27 @@ enum BrokerWake {
     Shutdown,
 }
 
-fn wait_for_request(socket: RawFd, shutdown_fd: RawFd) -> io::Result<BrokerWake> {
+fn wait_for_request(socket: BorrowedFd<'_>, shutdown_fd: BorrowedFd<'_>) -> io::Result<BrokerWake> {
     let mut descriptors = [
-        libc::pollfd {
-            fd: socket,
-            events: libc::POLLIN,
-            revents: 0,
-        },
-        libc::pollfd {
-            fd: shutdown_fd,
-            events: libc::POLLIN,
-            revents: 0,
-        },
+        PollFd::from_borrowed_fd(socket, PollFlags::IN),
+        PollFd::from_borrowed_fd(shutdown_fd, PollFlags::IN),
     ];
+    let unusable = PollFlags::HUP | PollFlags::ERR | PollFlags::NVAL;
     loop {
-        // SAFETY: descriptors points to two initialized pollfd entries.
-        let result = unsafe { libc::poll(descriptors.as_mut_ptr(), descriptors.len() as _, -1) };
-        if result < 0 {
-            let error = io::Error::last_os_error();
-            if error.kind() == io::ErrorKind::Interrupted {
-                continue;
-            }
-            return Err(error);
+        match event::poll(&mut descriptors, None) {
+            Ok(_) => {}
+            Err(Errno::INTR) => continue,
+            Err(errno) => return Err(errno.into()),
         }
-        if descriptors[1].revents & libc::POLLIN != 0 && take_graceful_shutdown_signal(shutdown_fd)?
-        {
+        let socket_events = descriptors[0].revents();
+        let shutdown_events = descriptors[1].revents();
+        if shutdown_events.contains(PollFlags::IN) && take_graceful_shutdown_signal(shutdown_fd)? {
             return Ok(BrokerWake::Shutdown);
         }
-        if descriptors[0].revents & libc::POLLIN != 0 {
+        if socket_events.contains(PollFlags::IN) {
             return Ok(BrokerWake::Request);
         }
-        if descriptors[0].revents & (libc::POLLHUP | libc::POLLERR | libc::POLLNVAL) != 0
-            || descriptors[1].revents & (libc::POLLHUP | libc::POLLERR | libc::POLLNVAL) != 0
-        {
+        if socket_events.intersects(unusable) || shutdown_events.intersects(unusable) {
             return Err(io::Error::new(
                 io::ErrorKind::BrokenPipe,
                 "broker control channel became unusable",
@@ -198,6 +186,14 @@ fn wait_for_request(socket: RawFd, shutdown_fd: RawFd) -> io::Result<BrokerWake>
 
 #[cfg(test)]
 pub(super) fn run_loop_for_test(socket: RawFd, shutdown_fd: RawFd, token: &str) -> Result<()> {
+    // SAFETY: the test broker thread owns both descriptors and keeps them open
+    // until this loop returns.
+    let (socket, shutdown_fd) = unsafe {
+        (
+            BorrowedFd::borrow_raw(socket),
+            BorrowedFd::borrow_raw(shutdown_fd),
+        )
+    };
     broker_loop(socket, shutdown_fd, token)
 }
 
@@ -242,7 +238,7 @@ fn handle_operation(
     operation: BrokerOperation,
     descriptors: &mut VecDeque<OwnedFd>,
     ownership: &mut BrokerOwnership,
-    shutdown_fd: RawFd,
+    shutdown_fd: BorrowedFd<'static>,
 ) -> Result<BrokerWireResponse> {
     if shutdown_requested(shutdown_fd)? {
         bail!("broker operation cancelled during shutdown");
@@ -350,20 +346,20 @@ fn handle_operation(
         ),
         BrokerOperation::Signal { handle, signal } => {
             reject_descriptors(descriptors)?;
-            if !matches!(
-                signal,
-                libc::SIGUSR1 | libc::SIGUSR2 | libc::SIGTERM | libc::SIGKILL
-            ) {
-                bail!("signal is not allowed by broker manifest");
-            }
+            let signal = match signal {
+                libc::SIGUSR1 => Signal::USR1,
+                libc::SIGUSR2 => Signal::USR2,
+                libc::SIGTERM => Signal::TERM,
+                libc::SIGKILL => Signal::KILL,
+                _ => bail!("signal is not allowed by broker manifest"),
+            };
             let child = ownership
                 .children
                 .get(&handle)
                 .ok_or_else(|| anyhow!("unknown broker child handle"))?;
-            // SAFETY: the broker retains the exact unreaped child handle.
-            if unsafe { libc::kill(child.id() as i32, signal) } != 0 {
-                return Err(io::Error::last_os_error()).context("broker child signal failed");
-            }
+            rustix_process::kill_process(Pid::from_child(child), signal)
+                .map_err(io::Error::from)
+                .context("broker child signal failed")?;
             Ok(wire_outcome(BrokerOutcome::Acknowledged))
         }
         BrokerOperation::TryWait { handle } => {
@@ -409,7 +405,7 @@ fn spawn_helper(
     request: SpawnRequest,
     descriptors: &mut VecDeque<OwnedFd>,
     children: &mut BTreeMap<String, Child>,
-    shutdown_fd: RawFd,
+    shutdown_fd: BorrowedFd<'static>,
 ) -> Result<BrokerWireResponse> {
     let SpawnRequest {
         kind,
@@ -504,22 +500,17 @@ fn reject_descriptors(descriptors: &VecDeque<OwnedFd>) -> Result<()> {
 }
 
 fn set_cloexec(descriptor: &OwnedFd, enabled: bool) -> Result<()> {
-    // SAFETY: fcntl reads and updates descriptor-local flags.
-    let current = unsafe { libc::fcntl(descriptor.as_raw_fd(), libc::F_GETFD) };
-    if current < 0 {
-        return Err(io::Error::last_os_error()).context("failed to read descriptor flags");
-    }
+    let current = rustix_io::fcntl_getfd(descriptor)
+        .map_err(io::Error::from)
+        .context("failed to read descriptor flags")?;
     let updated = if enabled {
-        current | libc::FD_CLOEXEC
+        current | FdFlags::CLOEXEC
     } else {
-        current & !libc::FD_CLOEXEC
+        current - FdFlags::CLOEXEC
     };
-    // SAFETY: `descriptor` is borrowed open for the call, and `updated` only
-    // changes FD_CLOEXEC in the flags just read from it.
-    if unsafe { libc::fcntl(descriptor.as_raw_fd(), libc::F_SETFD, updated) } != 0 {
-        return Err(io::Error::last_os_error()).context("failed to update descriptor flags");
-    }
-    Ok(())
+    rustix_io::fcntl_setfd(descriptor, updated)
+        .map_err(io::Error::from)
+        .context("failed to update descriptor flags")
 }
 
 fn wire_outcome(outcome: BrokerOutcome) -> BrokerWireResponse {
@@ -551,30 +542,18 @@ fn canonical_lower_hex(value: &str, length: usize) -> bool {
 mod tests {
     use std::process::Command;
 
+    use rustix::net::{self, AddressFamily, SendFlags, SocketFlags};
+
     use super::*;
-    use std::os::fd::FromRawFd;
 
     fn test_socket_pair() -> (OwnedFd, OwnedFd) {
-        let mut sockets = [0; 2];
-        assert_eq!(
-            // SAFETY: sockets has room for both returned descriptors.
-            unsafe {
-                libc::socketpair(
-                    libc::AF_UNIX,
-                    libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC,
-                    0,
-                    sockets.as_mut_ptr(),
-                )
-            },
-            0
-        );
-        // SAFETY: socketpair returned two fresh owned descriptors.
-        unsafe {
-            (
-                OwnedFd::from_raw_fd(sockets[0]),
-                OwnedFd::from_raw_fd(sockets[1]),
-            )
-        }
+        net::socketpair(
+            AddressFamily::UNIX,
+            SocketType::SEQPACKET,
+            SocketFlags::CLOEXEC,
+            None,
+        )
+        .unwrap()
     }
 
     #[test]
@@ -583,8 +562,7 @@ mod tests {
         let (shutdown_writer, shutdown_reader) = test_socket_pair();
         drop(shutdown_writer);
 
-        let Err(error) = wait_for_request(control_reader.as_raw_fd(), shutdown_reader.as_raw_fd())
-        else {
+        let Err(error) = wait_for_request(control_reader.as_fd(), shutdown_reader.as_fd()) else {
             panic!("hangup must unwind through destructive broker cleanup");
         };
 
@@ -597,21 +575,13 @@ mod tests {
         let (shutdown_writer, shutdown_reader) = test_socket_pair();
         let signal = [super::super::transport::GRACEFUL_SHUTDOWN_BYTE];
         assert_eq!(
-            // SAFETY: both the descriptor and one-byte buffer are valid.
-            unsafe {
-                libc::send(
-                    shutdown_writer.as_raw_fd(),
-                    signal.as_ptr().cast(),
-                    signal.len(),
-                    libc::MSG_NOSIGNAL,
-                )
-            },
+            net::send(&shutdown_writer, &signal, SendFlags::NOSIGNAL).unwrap(),
             1
         );
         drop(shutdown_writer);
 
         assert!(matches!(
-            wait_for_request(control_reader.as_raw_fd(), shutdown_reader.as_raw_fd()),
+            wait_for_request(control_reader.as_fd(), shutdown_reader.as_fd()),
             Ok(BrokerWake::Shutdown)
         ));
     }
@@ -622,20 +592,11 @@ mod tests {
         let (shutdown_writer, shutdown_reader) = test_socket_pair();
         let invalid = [2_u8];
         assert_eq!(
-            // SAFETY: both the descriptor and one-byte buffer are valid.
-            unsafe {
-                libc::send(
-                    shutdown_writer.as_raw_fd(),
-                    invalid.as_ptr().cast(),
-                    invalid.len(),
-                    libc::MSG_NOSIGNAL,
-                )
-            },
+            net::send(&shutdown_writer, &invalid, SendFlags::NOSIGNAL).unwrap(),
             1
         );
 
-        let Err(error) = wait_for_request(control_reader.as_raw_fd(), shutdown_reader.as_raw_fd())
-        else {
+        let Err(error) = wait_for_request(control_reader.as_fd(), shutdown_reader.as_fd()) else {
             panic!("invalid packet must not authorize ownership release");
         };
 
@@ -648,20 +609,11 @@ mod tests {
         let (shutdown_writer, shutdown_reader) = test_socket_pair();
         let invalid = [super::super::transport::GRACEFUL_SHUTDOWN_BYTE; 2];
         assert_eq!(
-            // SAFETY: both the descriptor and two-byte buffer are valid.
-            unsafe {
-                libc::send(
-                    shutdown_writer.as_raw_fd(),
-                    invalid.as_ptr().cast(),
-                    invalid.len(),
-                    libc::MSG_NOSIGNAL,
-                )
-            },
+            net::send(&shutdown_writer, &invalid, SendFlags::NOSIGNAL).unwrap(),
             2
         );
 
-        let Err(error) = wait_for_request(control_reader.as_raw_fd(), shutdown_reader.as_raw_fd())
-        else {
+        let Err(error) = wait_for_request(control_reader.as_fd(), shutdown_reader.as_fd()) else {
             panic!("oversized packet must not authorize ownership release");
         };
 

src/process_broker/tests.rs

@@ -1,5 +1,5 @@
 use std::ffi::OsStr;
-use std::os::fd::AsRawFd;
+use std::os::fd::{AsFd, AsRawFd};
 use std::os::unix::process::CommandExt;
 use std::path::Path;
 use std::process::Command;
@@ -778,7 +778,7 @@ fn owned_child_inherits_daemon_pidfd_without_leaking_broker_copy() {
             OsStr::new("sleep"),
             [OsStr::new("30")],
             Vec::new(),
-            watchdog.as_raw_fd(),
+            watchdog.as_fd(),
         )
         .unwrap();
     let inherited_pidfd = fs::read_dir(format!("/proc/{}/fd", child.id()))

src/process_broker/transport.rs

@@ -1,10 +1,18 @@
 use std::collections::VecDeque;
+use std::io::{self, IoSlice, IoSliceMut};
 use std::mem::MaybeUninit;
-use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
+use std::os::fd::{BorrowedFd, OwnedFd};
 use std::time::Duration;
-use std::{io, mem, ptr};
 
 use anyhow::{Context, Result, anyhow, bail};
+use rustix::event::{self, PollFd, PollFlags, Timespec};
+use rustix::fs::{self as rustix_fs, FileType, MemfdFlags};
+use rustix::io::{self as rustix_io, Errno};
+use rustix::net::sockopt::{self, Timeout};
+use rustix::net::{
+    self, RecvAncillaryBuffer, RecvAncillaryMessage, RecvFlags, ReturnFlags, SendAncillaryBuffer,
+    SendAncillaryMessage, SendFlags,
+};
 
 use super::wire::{
     BlobWire, INLINE_BLOB_BYTES, MAX_OUTPUT_BYTES, MAX_PACKET_BYTES, MAX_PACKET_DESCRIPTORS,
@@ -13,80 +21,47 @@ use super::wire::{
 
 pub(super) const GRACEFUL_SHUTDOWN_BYTE: u8 = 1;
 
-pub(super) fn set_socket_timeout(fd: RawFd, timeout: Duration) -> io::Result<()> {
+pub(super) fn set_socket_timeout(fd: BorrowedFd<'_>, timeout: Duration) -> io::Result<()> {
     let timeout = timeout.max(Duration::from_millis(1));
-    let value = libc::timeval {
-        tv_sec: libc::time_t::try_from(timeout.as_secs())
-            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "socket timeout overflow"))?,
-        tv_usec: libc::suseconds_t::from(timeout.subsec_micros()),
-    };
-    for option in [libc::SO_SNDTIMEO, libc::SO_RCVTIMEO] {
-        // SAFETY: value is initialized and setsockopt copies it.
-        if unsafe {
-            libc::setsockopt(
-                fd,
-                libc::SOL_SOCKET,
-                option,
-                (&raw const value).cast(),
-                mem::size_of::<libc::timeval>() as libc::socklen_t,
-            )
-        } != 0
-        {
-            return Err(io::Error::last_os_error());
-        }
+    for option in [Timeout::Send, Timeout::Recv] {
+        sockopt::set_socket_timeout(fd, option, Some(timeout))?;
     }
     Ok(())
 }
 
-pub(super) fn shutdown_requested(descriptor: RawFd) -> io::Result<bool> {
-    let mut pollfd = libc::pollfd {
-        fd: descriptor,
-        events: libc::POLLIN,
-        revents: 0,
-    };
-    // SAFETY: pollfd points to one initialized entry.
-    let result = unsafe { libc::poll(&raw mut pollfd, 1, 0) };
-    if result < 0 {
-        return Err(io::Error::last_os_error());
-    }
-    if pollfd.revents & libc::POLLNVAL != 0 {
+pub(super) fn shutdown_requested(descriptor: BorrowedFd<'_>) -> io::Result<bool> {
+    let mut pollfds = [PollFd::from_borrowed_fd(descriptor, PollFlags::IN)];
+    event::poll(&mut pollfds, Some(&Timespec::default()))?;
+    let revents = pollfds[0].revents();
+    if revents.contains(PollFlags::NVAL) {
         return Err(io::Error::new(
             io::ErrorKind::BrokenPipe,
             "broker shutdown channel became invalid",
         ));
     }
-    Ok(pollfd.revents & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0)
+    Ok(revents.intersects(PollFlags::IN | PollFlags::HUP | PollFlags::ERR))
 }
 
-pub(super) fn take_graceful_shutdown_signal(descriptor: RawFd) -> io::Result<bool> {
-    let mut pollfd = libc::pollfd {
-        fd: descriptor,
-        events: libc::POLLIN,
-        revents: 0,
-    };
+pub(super) fn take_graceful_shutdown_signal(descriptor: BorrowedFd<'_>) -> io::Result<bool> {
+    let mut pollfds = [PollFd::from_borrowed_fd(descriptor, PollFlags::IN)];
     loop {
-        // SAFETY: pollfd points to one initialized entry.
-        let result = unsafe { libc::poll(&raw mut pollfd, 1, 0) };
-        if result < 0 {
-            let error = io::Error::last_os_error();
-            if error.kind() == io::ErrorKind::Interrupted {
-                continue;
-            }
-            return Err(error);
+        match event::poll(&mut pollfds, Some(&Timespec::default())) {
+            Ok(0) => return Ok(false),
+            Ok(_) => {}
+            Err(Errno::INTR) => continue,
+            Err(errno) => return Err(errno.into()),
         }
-        if result == 0 {
-            return Ok(false);
-        }
-        if pollfd.revents & libc::POLLNVAL != 0 {
+        let revents = pollfds[0].revents();
+        if revents.contains(PollFlags::NVAL) {
             return Err(io::Error::new(
                 io::ErrorKind::BrokenPipe,
                 "broker shutdown channel became invalid",
             ));
         }
-        if pollfd.revents & libc::POLLIN != 0 {
+        if revents.contains(PollFlags::IN) {
             return receive_graceful_shutdown_signal(descriptor);
         }
-        if pollfd.revents & (libc::POLLHUP | libc::POLLERR) != 0 {
+        if revents.intersects(PollFlags::HUP | PollFlags::ERR) {
             return Err(io::Error::new(
                 io::ErrorKind::BrokenPipe,
                 "broker shutdown channel closed without a graceful signal",
@@ -96,27 +71,17 @@ pub(super) fn take_graceful_shutdown_signal(descriptor: RawFd) -> io::Result<boo
     }
 }
 
-fn receive_graceful_shutdown_signal(descriptor: RawFd) -> io::Result<bool> {
-    let mut byte = 0_u8;
+fn receive_graceful_shutdown_signal(descriptor: BorrowedFd<'_>) -> io::Result<bool> {
+    let mut byte = [0_u8];
     loop {
-        // MSG_TRUNC makes an oversized SOCK_SEQPACKET message report its full length.
-        // SAFETY: byte is a valid one-byte destination for this live broker socket.
-        let received = unsafe {
-            libc::recv(
-                descriptor,
-                (&raw mut byte).cast(),
-                1,
-                libc::MSG_DONTWAIT | libc::MSG_TRUNC,
-            )
+        // TRUNC makes an oversized SOCK_SEQPACKET message report its full length.
+        let flags = RecvFlags::DONTWAIT | RecvFlags::TRUNC;
+        let received = match net::recv(descriptor, &mut byte, flags) {
+            Ok((_, received)) => received,
+            Err(Errno::INTR) => continue,
+            Err(errno) => return Err(errno.into()),
         };
-        if received < 0 {
-            let error = io::Error::last_os_error();
-            if error.kind() == io::ErrorKind::Interrupted {
-                continue;
-            }
-            return Err(error);
-        }
-        if received == 1 && byte == GRACEFUL_SHUTDOWN_BYTE {
+        if received == 1 && byte[0] == GRACEFUL_SHUTDOWN_BYTE {
             return Ok(true);
         }
         return Err(io::Error::new(
@@ -166,28 +131,11 @@ pub(super) fn decode_blob(
             let mut bytes = vec![0_u8; length];
             let mut offset = 0;
             while offset < bytes.len() {
-                // SAFETY: the destination range is writable and the validated
-                // memfd has exactly the declared immutable length.
-                let read = unsafe {
-                    libc::pread(
-                        descriptor.as_raw_fd(),
-                        bytes[offset..].as_mut_ptr().cast(),
-                        bytes.len() - offset,
-                        offset as libc::off_t,
-                    )
-                };
-                if read > 0 {
-                    offset += read as usize;
-                } else if read < 0
-                    && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted
-                {
-                    continue;
-                } else {
-                    return Err(if read == 0 {
-                        anyhow!("broker memfd ended before its declared length")
-                    } else {
-                        io::Error::last_os_error().into()
-                    });
+                match rustix_io::pread(&descriptor, &mut bytes[offset..], offset as u64) {
+                    Ok(0) => bail!("broker memfd ended before its declared length"),
+                    Ok(read) => offset += read,
+                    Err(Errno::INTR) => {}
+                    Err(errno) => return Err(io::Error::from(errno).into()),
                 }
             }
             Ok(bytes)
@@ -196,206 +144,120 @@ pub(super) fn decode_blob(
 }
 
 fn sealed_memfd(bytes: &[u8]) -> Result<OwnedFd> {
-    let name = c"wayscriber-broker-payload";
-    // SAFETY: name is a valid C string and flags request a private, sealable fd.
-    let raw =
-        unsafe { libc::memfd_create(name.as_ptr(), libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING) };
-    if raw < 0 {
-        return Err(io::Error::last_os_error()).context("broker memfd creation failed");
-    }
-    // SAFETY: memfd_create returned a new owned descriptor.
-    let descriptor = unsafe { OwnedFd::from_raw_fd(raw) };
+    let descriptor = rustix_fs::memfd_create(
+        c"wayscriber-broker-payload",
+        MemfdFlags::CLOEXEC | MemfdFlags::ALLOW_SEALING,
+    )
+    .map_err(io::Error::from)
+    .context("broker memfd creation failed")?;
     let mut offset = 0;
     while offset < bytes.len() {
-        // SAFETY: the source range is readable for the requested length.
-        let written = unsafe {
-            libc::write(
-                descriptor.as_raw_fd(),
-                bytes[offset..].as_ptr().cast(),
-                bytes.len() - offset,
-            )
-        };
-        if written > 0 {
-            offset += written as usize;
-        } else if written < 0 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
-            continue;
-        } else {
-            return Err(if written == 0 {
-                anyhow!("short write while creating broker memfd")
-            } else {
-                io::Error::last_os_error().into()
-            });
+        match rustix_io::write(&descriptor, &bytes[offset..]) {
+            Ok(0) => bail!("short write while creating broker memfd"),
+            Ok(written) => offset += written,
+            Err(Errno::INTR) => {}
+            Err(errno) => return Err(io::Error::from(errno).into()),
         }
     }
-    // SAFETY: fcntl changes only seal metadata on the owned memfd.
-    if unsafe {
-        libc::fcntl(
-            descriptor.as_raw_fd(),
-            libc::F_ADD_SEALS,
-            REQUIRED_MEMFD_SEALS,
-        )
-    } < 0
-    {
-        return Err(io::Error::last_os_error()).context("failed to seal broker memfd");
-    }
+    rustix_fs::fcntl_add_seals(&descriptor, REQUIRED_MEMFD_SEALS)
+        .map_err(io::Error::from)
+        .context("failed to seal broker memfd")?;
     Ok(descriptor)
 }
 
 fn validate_sealed_memfd(descriptor: &OwnedFd, length: usize) -> Result<()> {
-    let mut stat = MaybeUninit::<libc::stat>::uninit();
-    // SAFETY: stat points to writable storage for fstat.
-    if unsafe { libc::fstat(descriptor.as_raw_fd(), stat.as_mut_ptr()) } != 0 {
-        return Err(io::Error::last_os_error()).context("failed to inspect broker memfd");
-    }
-    // SAFETY: fstat initialized stat after success.
-    let stat = unsafe { stat.assume_init() };
-    if (stat.st_mode & libc::S_IFMT) != libc::S_IFREG || stat.st_size != length as libc::off_t {
+    let stat = rustix_fs::fstat(descriptor)
+        .map_err(io::Error::from)
+        .context("failed to inspect broker memfd")?;
+    if FileType::from_raw_mode(stat.st_mode) != FileType::RegularFile
+        || u64::try_from(stat.st_size).ok() != Some(length as u64)
+    {
         bail!("broker memfd shape does not match its declaration");
     }
-    // SAFETY: F_GET_SEALS reads descriptor metadata.
-    let seals = unsafe { libc::fcntl(descriptor.as_raw_fd(), libc::F_GET_SEALS) };
-    if seals < 0 || seals & REQUIRED_MEMFD_SEALS != REQUIRED_MEMFD_SEALS {
+    let sealed = rustix_fs::fcntl_get_seals(descriptor)
+        .is_ok_and(|seals| seals.contains(REQUIRED_MEMFD_SEALS));
+    if !sealed {
         bail!("broker memfd is not immutably sealed");
     }
     Ok(())
 }
 
-pub(super) fn send_packet(fd: RawFd, packet: &[u8], descriptors: &[RawFd]) -> io::Result<()> {
+pub(super) fn send_packet(
+    fd: BorrowedFd<'_>,
+    packet: &[u8],
+    descriptors: &[BorrowedFd<'_>],
+) -> io::Result<()> {
     if descriptors.len() > MAX_PACKET_DESCRIPTORS {
         return Err(io::Error::new(
             io::ErrorKind::InvalidInput,
             "too many broker packet descriptors",
         ));
     }
+    let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(MAX_PACKET_DESCRIPTORS))];
+    let mut control = SendAncillaryBuffer::new(&mut space);
+    if !descriptors.is_empty() && !control.push(SendAncillaryMessage::ScmRights(descriptors)) {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "broker packet descriptors do not fit their control buffer",
+        ));
+    }
+    let payload = [IoSlice::new(packet)];
     loop {
-        let mut iovec = libc::iovec {
-            iov_base: packet.as_ptr().cast_mut().cast(),
-            iov_len: packet.len(),
-        };
-        let control_len = if descriptors.is_empty() {
-            0
-        } else {
-            // SAFETY: CMSG_SPACE only performs arithmetic on its argument.
-            unsafe { libc::CMSG_SPACE(mem::size_of_val(descriptors) as libc::c_uint) as usize }
-        };
-        let mut control = vec![0_usize; control_len.div_ceil(mem::size_of::<usize>())];
-        // SAFETY: zero is a valid initial state for msghdr.
-        let mut message: libc::msghdr = unsafe { mem::zeroed() };
-        message.msg_iov = &raw mut iovec;
-        message.msg_iovlen = 1;
-        if !descriptors.is_empty() {
-            message.msg_control = control.as_mut_ptr().cast();
-            message.msg_controllen = control_len;
-            // SAFETY: the control buffer has CMSG_SPACE bytes.
-            unsafe {
-                let header = libc::CMSG_FIRSTHDR(&raw const message);
-                (*header).cmsg_level = libc::SOL_SOCKET;
-                (*header).cmsg_type = libc::SCM_RIGHTS;
-                (*header).cmsg_len =
-                    libc::CMSG_LEN(mem::size_of_val(descriptors) as libc::c_uint) as usize;
-                ptr::copy_nonoverlapping(
-                    descriptors.as_ptr(),
-                    libc::CMSG_DATA(header).cast::<RawFd>(),
-                    descriptors.len(),
-                );
+        match net::sendmsg(fd, &payload, &mut control, SendFlags::NOSIGNAL) {
+            Ok(sent) if sent == packet.len() => return Ok(()),
+            Ok(_) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::WriteZero,
+                    "short seqpacket send",
+                ));
             }
+            Err(Errno::INTR) => {}
+            Err(errno) => return Err(errno.into()),
         }
-        // SAFETY: message references live packet/control buffers.
-        let sent = unsafe { libc::sendmsg(fd, &raw const message, libc::MSG_NOSIGNAL) };
-        if sent == packet.len() as isize {
-            return Ok(());
-        }
-        if sent < 0 && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted {
-            continue;
-        }
-        return Err(if sent < 0 {
-            io::Error::last_os_error()
-        } else {
-            io::Error::new(io::ErrorKind::WriteZero, "short seqpacket send")
-        });
     }
 }
 
-pub(super) fn recv_packet(fd: RawFd) -> io::Result<(Vec<u8>, Vec<OwnedFd>)> {
+pub(super) fn recv_packet(fd: BorrowedFd<'_>) -> io::Result<(Vec<u8>, Vec<OwnedFd>)> {
     let mut buffer = vec![0_u8; MAX_PACKET_BYTES + 1];
-    loop {
-        let mut iovec = libc::iovec {
-            iov_base: buffer.as_mut_ptr().cast(),
-            iov_len: buffer.len(),
-        };
-        // SAFETY: CMSG_SPACE only performs arithmetic on its argument.
-        let control_len = unsafe {
-            libc::CMSG_SPACE((MAX_PACKET_DESCRIPTORS * mem::size_of::<RawFd>()) as libc::c_uint)
-                as usize
-        };
-        let mut control = vec![0_usize; control_len.div_ceil(mem::size_of::<usize>())];
-        // SAFETY: zero is a valid initial state for msghdr.
-        let mut message: libc::msghdr = unsafe { mem::zeroed() };
-        message.msg_iov = &raw mut iovec;
-        message.msg_iovlen = 1;
-        message.msg_control = control.as_mut_ptr().cast();
-        message.msg_controllen = control_len;
-        // SAFETY: message references writable packet/control buffers.
-        let read = unsafe { libc::recvmsg(fd, &raw mut message, libc::MSG_CMSG_CLOEXEC) };
-        if read > 0 {
-            let read = read as usize;
-            let mut descriptors = Vec::new();
-            // SAFETY: recvmsg initialized ancillary headers in the buffer.
-            unsafe {
-                let mut header = libc::CMSG_FIRSTHDR(&raw const message);
-                while !header.is_null() {
-                    if (*header).cmsg_level != libc::SOL_SOCKET
-                        || (*header).cmsg_type != libc::SCM_RIGHTS
-                    {
-                        return Err(io::Error::new(
-                            io::ErrorKind::InvalidData,
-                            "unexpected broker packet ancillary data",
-                        ));
-                    }
-                    let base_len = libc::CMSG_LEN(0) as usize;
-                    if (*header).cmsg_len < base_len {
-                        return Err(io::Error::new(
-                            io::ErrorKind::InvalidData,
-                            "malformed broker descriptor payload",
-                        ));
-                    }
-                    let data_len = (*header).cmsg_len - base_len;
-                    if !data_len.is_multiple_of(mem::size_of::<RawFd>()) {
-                        return Err(io::Error::new(
-                            io::ErrorKind::InvalidData,
-                            "misaligned broker descriptor payload",
-                        ));
-                    }
-                    let count = data_len / mem::size_of::<RawFd>();
-                    let data = libc::CMSG_DATA(header).cast::<RawFd>();
-                    for index in 0..count {
-                        descriptors.push(OwnedFd::from_raw_fd(*data.add(index)));
-                    }
-                    header = libc::CMSG_NXTHDR(&raw const message, header);
-                }
-            }
-            if read > MAX_PACKET_BYTES
-                || message.msg_flags & (libc::MSG_TRUNC | libc::MSG_CTRUNC) != 0
-                || descriptors.len() > MAX_PACKET_DESCRIPTORS
-            {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    "broker packet exceeds its data or descriptor cap",
-                ));
-            }
-            buffer.truncate(read);
-            return Ok((buffer, descriptors));
+    let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(MAX_PACKET_DESCRIPTORS))];
+    let mut control = RecvAncillaryBuffer::new(&mut space);
+    let received = loop {
+        let mut payload = [IoSliceMut::new(&mut buffer)];
+        match net::recvmsg(fd, &mut payload, &mut control, RecvFlags::CMSG_CLOEXEC) {
+            Ok(received) => break received,
+            Err(Errno::INTR) => {}
+            Err(errno) => return Err(errno.into()),
         }
-        if read == 0 {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                "broker socket closed",
-            ));
-        }
-        let error = io::Error::last_os_error();
-        if error.kind() == io::ErrorKind::Interrupted {
-            continue;
-        }
-        return Err(error);
+    };
+    if received.bytes == 0 {
+        return Err(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "broker socket closed",
+        ));
     }
+    // Descriptors left in `control` close when it drops, so every rejection
+    // below releases what the packet carried.
+    let mut descriptors = Vec::new();
+    for message in control.drain() {
+        let RecvAncillaryMessage::ScmRights(passed) = message else {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "unexpected broker packet ancillary data",
+            ));
+        };
+        descriptors.extend(passed);
+    }
+    if received.bytes > MAX_PACKET_BYTES
+        || received
+            .flags
+            .intersects(ReturnFlags::TRUNC | ReturnFlags::CTRUNC)
+        || descriptors.len() > MAX_PACKET_DESCRIPTORS
+    {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "broker packet exceeds its data or descriptor cap",
+        ));
+    }
+    buffer.truncate(received.bytes);
+    Ok((buffer, descriptors))
 }

src/process_broker/wire.rs

@@ -3,6 +3,7 @@ use std::os::fd::OwnedFd;
 use std::os::unix::ffi::OsStrExt;
 
 use anyhow::{Result, bail};
+use rustix::fs::SealFlags;
 use serde::{Deserialize, Serialize};
 
 pub(super) const BROKER_FD: i32 = 3;
@@ -19,8 +20,10 @@ pub(super) const MAX_OUTPUT_BYTES: usize = 256 * 1024 * 1024;
 pub(super) const MAX_STDERR_BYTES: usize = 64 * 1024;
 pub(super) const MAX_OWNED_CHILDREN: usize = 64;
 pub(super) const MAX_PACKET_DESCRIPTORS: usize = 2;
-pub(super) const REQUIRED_MEMFD_SEALS: i32 =
-    libc::F_SEAL_WRITE | libc::F_SEAL_GROW | libc::F_SEAL_SHRINK | libc::F_SEAL_SEAL;
+pub(super) const REQUIRED_MEMFD_SEALS: SealFlags = SealFlags::WRITE
+    .union(SealFlags::GROW)
+    .union(SealFlags::SHRINK)
+    .union(SealFlags::SEAL);
 
 /// Failure text used when a `Complete`-mode helper writes past its stdout cap.
 ///