crates/heimdall-broker/src/sys.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! Thin wrappers over the system calls the broker uses. Heimdall is a
//! defensive sandbox; nothing here decides anything.
use std::ffi::CString;
use std::io;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
pub fn errno() -> i32 {
io::Error::last_os_error()
.raw_os_error()
.unwrap_or(libc::EIO)
}
fn cvt(r: libc::c_long) -> io::Result<libc::c_long> {
if r < 0 {
Err(io::Error::last_os_error())
} else {
Ok(r)
}
}
fn cvt_int(r: libc::c_int) -> io::Result<libc::c_int> {
if r < 0 {
Err(io::Error::last_os_error())
} else {
Ok(r)
}
}
/// `struct open_how` (linux/openat2.h, VER0).
#[repr(C)]
#[derive(Clone, Copy, Debug, Default)]
pub struct OpenHow {
pub flags: u64,
pub mode: u64,
pub resolve: u64,
}
pub const OPEN_HOW_SIZE_VER0: usize = 24;
pub fn openat2(dirfd: RawFd, path: &std::path::Path, how: &OpenHow) -> io::Result<OwnedFd> {
use std::os::unix::ffi::OsStrExt;
let c = CString::new(path.as_os_str().as_bytes())
.map_err(|_| io::Error::from_raw_os_error(libc::EINVAL))?;
// SAFETY: valid C string and open_how of the stated size.
let fd = cvt(unsafe {
libc::syscall(
libc::SYS_openat2,
dirfd,
c.as_ptr(),
how as *const OpenHow,
std::mem::size_of::<OpenHow>(),
)
})?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd as RawFd) })
}
pub fn open_path_dir(path: &str) -> io::Result<OwnedFd> {
let c = CString::new(path).map_err(|_| io::Error::from_raw_os_error(libc::EINVAL))?;
// SAFETY: valid C string.
let fd = cvt_int(unsafe {
libc::open(
c.as_ptr(),
libc::O_PATH | libc::O_DIRECTORY | libc::O_CLOEXEC,
)
})?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}
// ---- processes ----
pub fn pidfd_open(pid: i32) -> io::Result<OwnedFd> {
// SAFETY: plain system call.
let fd = cvt(unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) })?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd as RawFd) })
}
pub fn pidfd_getfd(pidfd: &OwnedFd, target: i32) -> io::Result<OwnedFd> {
// SAFETY: plain system call.
let fd = cvt(unsafe { libc::syscall(libc::SYS_pidfd_getfd, pidfd.as_raw_fd(), target, 0) })?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd as RawFd) })
}
pub fn pidfd_kill(pidfd: &OwnedFd) -> io::Result<()> {
// SAFETY: plain system call.
cvt(unsafe {
libc::syscall(
libc::SYS_pidfd_send_signal,
pidfd.as_raw_fd(),
libc::SIGKILL,
std::ptr::null::<libc::siginfo_t>(),
0,
)
})?;
Ok(())
}
/// The thread-group id of thread `tid` (from /proc/<tid>/status).
pub fn tgid_of(tid: u32) -> io::Result<i32> {
let s = std::fs::read_to_string(format!("/proc/{tid}/status"))?;
s.lines()
.find_map(|l| l.strip_prefix("Tgid:"))
.and_then(|v| v.trim().parse().ok())
.ok_or_else(|| io::Error::from_raw_os_error(libc::ESRCH))
}
/// Reads `buf.len()` bytes at `addr` in process `pid` (one
/// `process_vm_readv`; short reads are errors).
pub fn read_mem(pid: u32, addr: u64, buf: &mut [u8]) -> io::Result<()> {
if buf.is_empty() {
return Ok(());
}
let local = libc::iovec {
iov_base: buf.as_mut_ptr() as *mut libc::c_void,
iov_len: buf.len(),
};
let remote = libc::iovec {
iov_base: addr as usize as *mut libc::c_void,
iov_len: buf.len(),
};
// SAFETY: local iovec points at `buf`; remote is only read by the kernel.
let n = unsafe { libc::process_vm_readv(pid as libc::pid_t, &local, 1, &remote, 1, 0) };
if n < 0 {
return Err(io::Error::last_os_error());
}
if n as usize != buf.len() {
return Err(io::Error::from_raw_os_error(libc::EFAULT));
}
Ok(())
}
/// Reads a NUL-terminated string of at most `max` bytes (NUL excluded) at
/// `addr`, page by page so a string ending just before an unmapped page reads.
pub fn read_cstr(pid: u32, addr: u64, max: usize) -> io::Result<Vec<u8>> {
const PAGE: u64 = 4096;
let mut out = Vec::new();
let mut at = addr;
while out.len() <= max {
let room = (PAGE - (at % PAGE)) as usize;
let take = room.min(max + 1 - out.len());
let mut chunk = vec![0u8; take];
read_mem(pid, at, &mut chunk)?;
if let Some(z) = chunk.iter().position(|&c| c == 0) {
out.extend_from_slice(&chunk[..z]);
return Ok(out);
}
out.extend_from_slice(&chunk);
at = at
.checked_add(take as u64)
.ok_or_else(|| io::Error::from_raw_os_error(libc::EFAULT))?;
}
Err(io::Error::from_raw_os_error(libc::ENAMETOOLONG))
}
// ---- files and sockets ----
/// `readlink` of a /proc link (a worker's cwd or fd), as bytes.
pub fn readlink(path: &str) -> io::Result<Vec<u8>> {
use std::os::unix::ffi::OsStrExt;
Ok(std::fs::read_link(path)?.as_os_str().as_bytes().to_vec())
}
/// The last field of `key` in /proc/<tid>/status (`NStgid:`, `NSpid:`): the
/// id as the innermost pid namespace sees it.
pub fn ns_id(tid: u32, key: &str) -> io::Result<u32> {
let s = std::fs::read_to_string(format!("/proc/{tid}/status"))?;
s.lines()
.find_map(|l| l.strip_prefix(key))
.and_then(|v| v.split_whitespace().last())
.and_then(|v| v.parse().ok())
.ok_or_else(|| io::Error::from_raw_os_error(libc::ESRCH))
}
/// The fs uid and fs gid of thread `tid` (the 4th field of `Uid:` / `Gid:`
/// in /proc/<tid>/status), as this process's user namespace sees them.
pub fn fs_ids(tid: u32) -> io::Result<(u32, u32)> {
let s = std::fs::read_to_string(format!("/proc/{tid}/status"))?;
let field = |k: &str| {
s.lines()
.find_map(|l| l.strip_prefix(k))
.and_then(|v| v.split_whitespace().nth(3))
.and_then(|v| v.parse::<u32>().ok())
};
match (field("Uid:"), field("Gid:")) {
(Some(u), Some(g)) => Ok((u, g)),
_ => Err(io::Error::from_raw_os_error(libc::ESRCH)),
}
}
/// Run `f` on this thread with the fs uid/gid `uid`/`gid` (`setfsuid(2)` and `setfsgid(2)`
/// are per thread), then put the thread's own back. A broker that CREATES a file for a job
/// creates it as the job: a job tmpfs mounted in the job's user namespace refuses an inode
/// whose owner it cannot map (`EOVERFLOW`), and host root is unmapped there when the runner
/// gives jobs a host id range. The kernel drops the fs capabilities with a non-root fs uid
/// and gives them back with fs uid 0, so the broker's own opens are unchanged afterwards.
pub fn as_fs_ids<T>(uid: u32, gid: u32, f: impl FnOnce() -> io::Result<T>) -> io::Result<T> {
/// Puts the thread's own fs ids back on every way out: a result, an error, a panic.
struct Restore {
uid: u32,
gid: u32,
}
impl Drop for Restore {
fn drop(&mut self) {
// SAFETY: plain per-thread credential calls restoring this thread's ids.
unsafe {
libc::syscall(libc::SYS_setfsuid, self.uid);
libc::syscall(libc::SYS_setfsgid, self.gid);
}
}
}
// SAFETY: plain per-thread credential calls; each returns the previous id.
let gid_was = unsafe { libc::syscall(libc::SYS_setfsgid, gid) } as u32;
let uid_was = unsafe { libc::syscall(libc::SYS_setfsuid, uid) } as u32;
let _restore = Restore {
uid: uid_was,
gid: gid_was,
};
f()
}
/// This thread's fs uid and gid (`/proc/thread-self/status`, the fourth field).
#[cfg(test)]
pub fn thread_fs_ids() -> io::Result<(u32, u32)> {
let s = std::fs::read_to_string("/proc/thread-self/status")?;
let field = |k: &str| {
s.lines()
.find_map(|l| l.strip_prefix(k))
.and_then(|v| v.split_whitespace().nth(3))
.and_then(|v| v.parse::<u32>().ok())
};
match (field("Uid:"), field("Gid:")) {
(Some(u), Some(g)) => Ok((u, g)),
_ => Err(io::Error::from_raw_os_error(libc::ESRCH)),
}
}
/// `fchown(fd, uid, gid)` unless the file already has those owners.
pub fn fchown_if_other(fd: RawFd, uid: u32, gid: u32) -> io::Result<()> {
// SAFETY: zeroed stat is valid; fstat fills it.
let mut st: libc::stat = unsafe { std::mem::zeroed() };
// SAFETY: as above.
cvt_int(unsafe { libc::fstat(fd, &mut st) })?;
if st.st_uid == uid && st.st_gid == gid {
return Ok(());
}
// SAFETY: plain system call on our own fd.
cvt_int(unsafe { libc::fchown(fd, uid, gid) })?;
Ok(())
}
pub fn fstat_mode(fd: RawFd) -> io::Result<u32> {
// SAFETY: zeroed stat is valid; fstat fills it.
let mut st: libc::stat = unsafe { std::mem::zeroed() };
// SAFETY: as above.
cvt_int(unsafe { libc::fstat(fd, &mut st) })?;
Ok(st.st_mode & libc::S_IFMT)
}
pub fn clear_nonblock(fd: RawFd) -> io::Result<()> {
// SAFETY: fcntl on our own fd.
let fl = cvt_int(unsafe { libc::fcntl(fd, libc::F_GETFL) })?;
// SAFETY: as above.
cvt_int(unsafe { libc::fcntl(fd, libc::F_SETFL, fl & !libc::O_NONBLOCK) })?;
Ok(())
}
pub fn socket(domain: i32, ty: i32, proto: i32) -> io::Result<OwnedFd> {
// SAFETY: plain system call.
let fd = cvt_int(unsafe { libc::socket(domain, ty, proto) })?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}
pub fn sockopt_int(fd: RawFd, level: i32, name: i32) -> io::Result<i32> {
let mut v: libc::c_int = 0;
let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
// SAFETY: getsockopt writes one int.
cvt_int(unsafe {
libc::getsockopt(
fd,
level,
name,
&mut v as *mut _ as *mut libc::c_void,
&mut len,
)
})?;
Ok(v)
}
/// `connect(fd, sockaddr)`; returns 0 or the errno.
pub fn connect_raw(fd: RawFd, sa: &[u8]) -> i32 {
// SAFETY: `sa` is a complete sockaddr of its length.
let r = unsafe {
libc::connect(
fd,
sa.as_ptr() as *const libc::sockaddr,
sa.len() as libc::socklen_t,
)
};
if r == 0 {
0
} else {
errno()
}
}
pub fn eventfd() -> io::Result<OwnedFd> {
// SAFETY: plain system call.
let fd = cvt_int(unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) })?;
// SAFETY: a fresh fd we own.
Ok(unsafe { OwnedFd::from_raw_fd(fd) })
}
pub fn eventfd_signal(fd: RawFd) {
let one: u64 = 1;
// SAFETY: writes 8 bytes from `one`.
unsafe {
libc::write(fd, &one as *const u64 as *const libc::c_void, 8);
}
}
/// `poll` on `fds` (POLLIN); returns their revents.
pub fn poll_in(fds: &[RawFd], timeout_ms: i32) -> io::Result<Vec<i16>> {
let mut p: Vec<libc::pollfd> = fds
.iter()
.map(|&fd| libc::pollfd {
fd,
events: libc::POLLIN,
revents: 0,
})
.collect();
loop {
// SAFETY: `p` is a valid pollfd array.
let r = unsafe { libc::poll(p.as_mut_ptr(), p.len() as libc::nfds_t, timeout_ms) };
if r >= 0 {
return Ok(p.iter().map(|x| x.revents).collect());
}
let e = io::Error::last_os_error();
if e.raw_os_error() != Some(libc::EINTR) {
return Err(e);
}
}
}
pub fn seqpacket_pair() -> io::Result<(OwnedFd, OwnedFd)> {
let mut fds = [0 as libc::c_int; 2];
// SAFETY: socketpair writes two fds.
cvt_int(unsafe {
libc::socketpair(
libc::AF_UNIX,
libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC,
0,
fds.as_mut_ptr(),
)
})?;
// SAFETY: fresh fds we own.
Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
}
/// Sends `fd` and a 4-byte `pid` over a unix socket. Async-signal-safe (stack
/// buffers, one `sendmsg`), for the `pre_exec` hook.
///
/// # Safety
/// `sock` and `fd` must be open.
pub unsafe fn send_fd_and_pid(sock: RawFd, fd: RawFd, pid: i32) -> io::Result<()> {
let mut payload = pid.to_ne_bytes();
let mut iov = libc::iovec {
iov_base: payload.as_mut_ptr() as *mut libc::c_void,
iov_len: payload.len(),
};
// Room for one fd: CMSG_SPACE(4) = 24 on 64-bit; 64 bytes, 8-aligned.
#[repr(C, align(8))]
struct Buf([u8; 64]);
let mut cbuf = Buf([0u8; 64]);
let space = libc::CMSG_SPACE(std::mem::size_of::<RawFd>() as u32) as usize;
let mut msg: libc::msghdr = std::mem::zeroed();
msg.msg_iov = &mut iov;
msg.msg_iovlen = 1;
msg.msg_control = cbuf.0.as_mut_ptr() as *mut libc::c_void;
msg.msg_controllen = space as _;
let cm = libc::CMSG_FIRSTHDR(&msg);
(*cm).cmsg_level = libc::SOL_SOCKET;
(*cm).cmsg_type = libc::SCM_RIGHTS;
(*cm).cmsg_len = libc::CMSG_LEN(std::mem::size_of::<RawFd>() as u32) as _;
std::ptr::copy_nonoverlapping(
&fd as *const RawFd as *const u8,
libc::CMSG_DATA(cm),
std::mem::size_of::<RawFd>(),
);
if libc::sendmsg(sock, &msg, libc::MSG_NOSIGNAL) < 0 {
return Err(io::Error::last_os_error());
}
Ok(())
}
/// Receives what [`send_fd_and_pid`] sent. `Ok(None)` = the peer closed.
pub fn recv_fd_and_pid(sock: RawFd) -> io::Result<Option<(OwnedFd, i32)>> {
let mut payload = [0u8; 4];
let mut iov = libc::iovec {
iov_base: payload.as_mut_ptr() as *mut libc::c_void,
iov_len: payload.len(),
};
#[repr(C, align(8))]
struct Buf([u8; 64]);
let mut cbuf = Buf([0u8; 64]);
// SAFETY: zeroed msghdr is valid; fields set below.
let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
msg.msg_iov = &mut iov;
msg.msg_iovlen = 1;
msg.msg_control = cbuf.0.as_mut_ptr() as *mut libc::c_void;
msg.msg_controllen = cbuf.0.len() as _;
// SAFETY: msg points at live buffers.
let n = unsafe { libc::recvmsg(sock, &mut msg, libc::MSG_CMSG_CLOEXEC) };
if n < 0 {
return Err(io::Error::last_os_error());
}
if n == 0 {
return Ok(None);
}
// SAFETY: walking the control buffer the kernel filled.
unsafe {
let cm = libc::CMSG_FIRSTHDR(&msg);
if cm.is_null()
|| (*cm).cmsg_level != libc::SOL_SOCKET
|| (*cm).cmsg_type != libc::SCM_RIGHTS
{
return Err(io::Error::from_raw_os_error(libc::EPROTO));
}
let mut fd: RawFd = -1;
std::ptr::copy_nonoverlapping(
libc::CMSG_DATA(cm),
&mut fd as *mut RawFd as *mut u8,
std::mem::size_of::<RawFd>(),
);
if fd < 0 || n != 4 {
return Err(io::Error::from_raw_os_error(libc::EPROTO));
}
Ok(Some((
OwnedFd::from_raw_fd(fd),
i32::from_ne_bytes(payload),
)))
}
}
#[cfg(test)]
mod fs_ids_tests {
use super::*;
/// The broker thread is back at its own fs ids after a create that failed or panicked
/// (as root the ids really change inside; unprivileged the kernel keeps them, and the
/// restore is checked all the same).
#[test]
fn the_fs_ids_come_back_on_an_error_and_on_a_panic() {
std::thread::spawn(|| {
let own = thread_fs_ids().unwrap();
let root = own.0 == 0;
let r: io::Result<()> = as_fs_ids(12_345, 23_456, || {
if root {
assert_eq!(thread_fs_ids().unwrap(), (12_345, 23_456));
}
Err(io::Error::from_raw_os_error(libc::EOVERFLOW))
});
assert_eq!(r.unwrap_err().raw_os_error(), Some(libc::EOVERFLOW));
assert_eq!(thread_fs_ids().unwrap(), own, "after an error");
let p = std::panic::catch_unwind(|| {
let _: io::Result<()> =
as_fs_ids(12_345, 23_456, || panic!("a create that panics"));
});
assert!(p.is_err());
assert_eq!(thread_fs_ids().unwrap(), own, "after a panic");
})
.join()
.unwrap();
}
}