crates/heimdall-broker/src/broker.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! The supervisor loop (spec §4). Heimdall is a defensive sandbox.
//!
//! For every notification the broker
//! 1. copies the arguments out of the worker's memory ONCE,
//! 2. checks `SECCOMP_IOCTL_NOTIF_ID_VALID` (the task is still the one that
//! made the call, so the copy came from the right process),
//! 3. decides with [`BrokerPolicy`] on its own copy,
//! 4. opens the file (`openat2` beneath the prefix it holds open, no magic
//! links, no mount crossing) or creates the socket ITSELF, or connects the
//! worker's socket itself (via `pidfd_getfd`) to the address it checked,
//! and
//! 5. injects the fd with `SECCOMP_IOCTL_NOTIF_ADDFD` (`FLAG_SEND`) or answers
//! with the result.
//!
//! The worker's own call never proceeds on a pointer it can change: the only
//! `CONTINUE` answers are exec (whose decision never reads the worker's
//! memory) and the [`Mode::CheckOnly`] fallback, which warns.
//!
//! The prefixes are opened once, as the WORKER sees them (through
//! `/proc/<worker>/root`), so the fds the worker gets belong to its own file
//! tree, inside its container.
use std::net::SocketAddr;
use std::os::fd::{AsFd, AsRawFd, OwnedFd, RawFd};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use heimdall_core::{BrokerEvent, Decision, JobId};
use heimdall_seccomp::notif::{AddFd, Listener, Notif, Response};
use crate::filter;
use crate::policy::{
join_relative, parse_sockaddr, proc_rewrite, render_bytes, render_open_flags, sockaddr_bytes,
Access, BrokerPolicy, ExecPolicy, OnRefuse, OpenPlan, Refusal, Resolve,
};
use crate::support::{broker_feature, Mode, Support};
use crate::sys;
use crate::BrokerError;
const PATH_MAX: usize = 4096;
const SOCKADDR_MAX: usize = 128;
/// `RESOLVE_*` (linux/openat2.h).
pub const RESOLVE_NO_XDEV: u64 = 0x01;
pub const RESOLVE_NO_MAGICLINKS: u64 = 0x02;
pub const RESOLVE_NO_SYMLINKS: u64 = 0x04;
pub const RESOLVE_BENEATH: u64 = 0x08;
pub const RESOLVE_IN_ROOT: u64 = 0x10;
pub const RESOLVE_CACHED: u64 = 0x20;
/// What the broker did during one job.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BrokerReport {
pub job_id: JobId,
pub mode: Mode,
/// Every decision, in time order (into `JobResult.events`).
pub events: Vec<BrokerEvent>,
/// Events beyond `max_events`, counted but not kept.
pub dropped_events: u64,
/// Things the operator must know (e.g. the check-only fallback).
pub warnings: Vec<String>,
}
impl BrokerReport {
/// `Features.broker` for this run: true only when the broker really ran
/// in [`Mode::Full`].
pub fn broker_feature(&self) -> bool {
broker_feature(self.mode)
}
}
/// Test knobs. Not for production use.
#[derive(Clone, Debug, Default)]
pub struct DebugKnobs {
/// Sleep this long after reading a call's arguments, before answering
/// (lets a test race the worker against a brokered call).
pub hold_before_answer: Option<Duration>,
}
struct Log {
start: Instant,
max: usize,
events: Vec<BrokerEvent>,
dropped: u64,
}
impl Log {
fn push(&mut self, call: &str, args: Vec<String>, decision: Decision) {
if self.events.len() >= self.max {
self.dropped += 1;
return;
}
self.events.push(BrokerEvent {
at_ms: self.start.elapsed().as_millis() as u64,
call: call.to_string(),
args,
decision,
});
}
}
/// The broker for one job.
pub struct Broker {
listener: Arc<Listener>,
policy: BrokerPolicy,
mode: Mode,
support: Support,
/// The worker's root, as an `O_PATH` dir.
root: OwnedFd,
/// One per prefix: its directory (`O_PATH`), `None` when the worker's
/// tree has no such directory or the prefix is a device.
roots: Vec<Option<OwnedFd>>,
initial_pid: u32,
exec_used: bool,
log: Arc<Mutex<Log>>,
warnings: Vec<String>,
knobs: DebugKnobs,
pending: Vec<JoinHandle<()>>,
}
impl Broker {
/// A broker for the worker whose first process is `worker_pid` (as this
/// process sees it), reading `listener` (the filter
/// [`filter::program`] installed in that worker).
pub fn new(
listener: Listener,
worker_pid: u32,
policy: BrokerPolicy,
mode: Mode,
knobs: DebugKnobs,
) -> Result<Self, BrokerError> {
policy.validate().map_err(BrokerError::Policy)?;
let support = Support::probe();
if !support.allows(mode) {
return Err(BrokerError::Unsupported(format!(
"the broker cannot run in {mode:?} on this kernel ({support:?})"
)));
}
let root = sys::open_path_dir(&format!("/proc/{worker_pid}/root")).map_err(|e| {
BrokerError::Io {
what: "open the worker's root".into(),
source: e,
}
})?;
let mut roots = Vec::with_capacity(policy.prefixes.len());
for p in &policy.prefixes {
if p.resolve.exact() {
roots.push(None);
continue;
}
let rel = p.path.strip_prefix("/").unwrap_or(&p.path);
let rel = if rel.as_os_str().is_empty() {
Path::new(".")
} else {
rel
};
match sys::openat2(
root.as_raw_fd(),
rel,
&sys::OpenHow {
flags: (libc::O_PATH | libc::O_DIRECTORY | libc::O_CLOEXEC) as u64,
mode: 0,
resolve: RESOLVE_IN_ROOT | RESOLVE_NO_MAGICLINKS,
},
) {
Ok(fd) => roots.push(Some(fd)),
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => roots.push(None),
Err(e) => {
return Err(BrokerError::Io {
what: format!("open prefix {}", p.path.display()),
source: e,
})
}
}
}
let mut warnings = Vec::new();
if mode == Mode::CheckOnly {
warnings.push(format!(
"job {}: the kernel has no SECCOMP_IOCTL_NOTIF_ADDFD; the broker checks calls and \
lets them proceed (allow/deny only, not TOCTOU-safe)",
policy.job_id
));
}
let max = policy.max_events;
Ok(Self {
listener: Arc::new(listener),
policy,
mode,
support,
root,
roots,
initial_pid: worker_pid,
exec_used: false,
log: Arc::new(Mutex::new(Log {
start: Instant::now(),
max,
events: Vec::new(),
dropped: 0,
})),
warnings,
knobs,
pending: Vec::new(),
})
}
/// Serves notifications until every task using the filter is gone or
/// `stop` becomes readable.
pub fn serve(mut self, stop: RawFd) -> BrokerReport {
let lfd = self.listener.as_raw_fd();
loop {
let rev = match sys::poll_in(&[lfd, stop], -1) {
Ok(r) => r,
Err(e) => {
self.warnings.push(format!("broker poll failed: {e}"));
break;
}
};
if rev[1] != 0 {
break;
}
if rev[0] & libc::POLLIN == 0 {
if rev[0] & (libc::POLLHUP | libc::POLLERR | libc::POLLNVAL) != 0 {
break;
}
continue;
}
let n = match self.listener.recv() {
Ok(n) => n,
// The caller died between poll and recv.
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => continue,
Err(e) => {
self.warnings.push(format!("broker NOTIF_RECV failed: {e}"));
break;
}
};
self.dispatch(&n);
}
self.finish()
}
fn finish(self) -> BrokerReport {
// Connects still running get a short grace; ones that hang are left.
let deadline = Instant::now() + Duration::from_secs(2);
for h in self.pending {
while !h.is_finished() && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(5));
}
if h.is_finished() {
let _ = h.join();
}
}
let mut log = self.log.lock().unwrap_or_else(|e| e.into_inner());
let mut events = std::mem::take(&mut log.events);
events.sort_by_key(|e| e.at_ms);
BrokerReport {
job_id: self.policy.job_id,
mode: self.mode,
events,
dropped_events: log.dropped,
warnings: self.warnings,
}
}
fn valid(&self, id: u64) -> bool {
self.listener.id_valid(id).unwrap_or(false)
}
fn hold(&self) {
if let Some(d) = self.knobs.hold_before_answer {
std::thread::sleep(d);
}
}
fn push(&self, call: &str, args: Vec<String>, decision: Decision) {
self.log
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(call, args, decision);
}
fn answer(&self, id: u64, resp: Response, args: &mut Vec<String>) {
if let Err(e) = self.listener.respond(id, resp) {
args.push(gone_note(&e));
}
}
/// Answers with `-err` and logs.
fn answer_err_logged(
&self,
id: u64,
call: &str,
mut args: Vec<String>,
decision: Decision,
err: i32,
) {
self.answer(id, Response::Errno(err), &mut args);
self.push(call, args, decision);
}
/// Refuses per policy: EACCES, or kill the worker's process.
fn refuse(&self, n: &Notif, call: &str, mut args: Vec<String>, why: Refusal) {
args.push(format!("refused: {}", why.0));
let decision = match self.policy.on_refuse {
OnRefuse::Eacces => Decision::Refused,
OnRefuse::Kill => {
if self.kill_caller(n) {
Decision::Killed
} else {
args.push("kill failed".into());
Decision::Refused
}
}
};
self.answer_err_logged(n.id, call, args, decision, libc::EACCES);
}
/// The caller's whole process, as a pidfd, checked against pid reuse.
fn caller_pidfd(&self, n: &Notif) -> Option<OwnedFd> {
let tgid = sys::tgid_of(n.pid).ok()?;
let pfd = sys::pidfd_open(tgid).ok()?;
// The task is still the caller: its tgid (and so the pidfd) is right.
self.valid(n.id).then_some(pfd)
}
fn kill_caller(&self, n: &Notif) -> bool {
match self.caller_pidfd(n) {
Some(p) => sys::pidfd_kill(&p).is_ok(),
None => false,
}
}
fn dispatch(&mut self, n: &Notif) {
let call = filter::name_of(n.data.nr);
let a = n.data.args;
match call {
"openat" => self.on_open(n, call, a[0] as i32, a[1], a[2], a[3], 0),
"openat2" => self.on_openat2(n, call),
"open" => self.on_open(n, call, libc::AT_FDCWD, a[0], a[1], a[2], 0),
"creat" => self.on_open(
n,
call,
libc::AT_FDCWD,
a[0],
(libc::O_CREAT | libc::O_WRONLY | libc::O_TRUNC) as u64,
a[1],
0,
),
"socket" => self.on_socket(n, call),
"connect" => self.on_connect(n, call),
"execve" | "execveat" => self.on_exec(n, call),
_ => {
// bind, listen, accept, io_uring_setup, MSG_FASTOPEN sends, and
// anything unexpected: refused.
let why = match call {
"bind" => Refusal("bind is never allowed"),
"listen" | "accept" | "accept4" => Refusal("no listening sockets"),
"io_uring_setup" => Refusal("io_uring bypasses the broker"),
"sendto" | "sendmsg" | "sendmmsg" => {
Refusal("MSG_FASTOPEN connects around the broker")
}
_ => Refusal("unexpected call"),
};
let args = vec![format!("pid={}", n.pid), format!("nr={}", n.data.nr)];
self.hold();
self.refuse(n, call, args, why);
}
}
}
fn on_openat2(&mut self, n: &Notif, call: &str) {
let a = n.data.args;
let size = a[3] as usize;
let mut args = vec![format!("pid={}", n.pid)];
if !(sys::OPEN_HOW_SIZE_VER0..=4096).contains(&size) {
self.refuse(n, call, args, Refusal("bad open_how size"));
return;
}
let mut buf = vec![0u8; size];
if sys::read_mem(n.pid, a[2], &mut buf).is_err() || !self.valid(n.id) {
args.push("open_how unreadable".into());
self.answer_err_logged(n.id, call, args, Decision::Refused, libc::EFAULT);
return;
}
if buf[sys::OPEN_HOW_SIZE_VER0..].iter().any(|&b| b != 0) {
self.refuse(n, call, args, Refusal("unknown open_how fields"));
return;
}
let w = |i: usize| u64::from_ne_bytes(buf[i * 8..i * 8 + 8].try_into().unwrap());
let (flags, mode, resolve) = (w(0), w(1), w(2));
let known = RESOLVE_NO_XDEV
| RESOLVE_NO_MAGICLINKS
| RESOLVE_NO_SYMLINKS
| RESOLVE_BENEATH
| RESOLVE_IN_ROOT
| RESOLVE_CACHED;
if resolve & !known != 0 {
self.refuse(n, call, args, Refusal("unknown RESOLVE_ flags"));
return;
}
if resolve & (RESOLVE_BENEATH | RESOLVE_IN_ROOT) != 0 {
// The worker's own scoping is relative to its dirfd; the broker
// scopes every open to a prefix and does not emulate these.
self.refuse(
n,
call,
args,
Refusal("RESOLVE_BENEATH/RESOLVE_IN_ROOT from the worker"),
);
return;
}
self.on_open(n, call, a[0] as i32, a[1], flags, mode, resolve);
}
/// The absolute path (as the worker sees it) of a relative open: its
/// cwd's or dirfd's path, from `/proc/<tid>/…` (read after the copy;
/// the worker can only make it name another path, which is then checked
/// as such).
fn base_dir(&self, n: &Notif, dirfd: i32) -> Result<Vec<u8>, i32> {
let link = if dirfd == libc::AT_FDCWD {
format!("/proc/{}/cwd", n.pid)
} else if dirfd >= 0 {
format!("/proc/{}/fd/{dirfd}", n.pid)
} else {
return Err(libc::EBADF);
};
sys::readlink(&link).map_err(|e| match e.raw_os_error() {
Some(libc::ENOENT) => libc::EBADF,
Some(e) => e,
None => libc::EBADF,
})
}
#[allow(clippy::too_many_arguments)]
fn on_open(
&mut self,
n: &Notif,
call: &str,
dirfd: i32,
path_ptr: u64,
flags: u64,
mode: u64,
resolve: u64,
) {
let mut args = vec![format!("pid={}", n.pid)];
let path = match sys::read_cstr(n.pid, path_ptr, PATH_MAX) {
Ok(p) => p,
Err(e) => {
args.push(format!("path unreadable: {e}"));
let err = e.raw_os_error().unwrap_or(libc::EFAULT);
let err = if err == libc::ESRCH || err == libc::EIO {
libc::EFAULT
} else {
err
};
self.answer_err_logged(n.id, call, args, Decision::Refused, err);
return;
}
};
// The copy came from the caller only if it is still waiting.
if !self.valid(n.id) {
args.push(format!("path={}", render_bytes(&path)));
args.push("worker gone before the check".into());
self.push(call, args, Decision::Refused);
return;
}
args.push(format!("path={}", render_bytes(&path)));
args.push(format!("flags={}", render_open_flags(flags)));
self.hold();
if path.is_empty() {
args.push("result=ENOENT".into());
self.answer_err_logged(n.id, call, args, Decision::Refused, libc::ENOENT);
return;
}
let abs = if path[0] == b'/' {
path.clone()
} else {
let base = match self.base_dir(n, dirfd) {
Ok(b) => b,
Err(e) => {
args.push(format!("result={}", errno_name(e)));
self.answer_err_logged(n.id, call, args, Decision::Refused, e);
return;
}
};
match join_relative(&base, &path) {
Ok(p) => {
args.push(format!("resolved={}", render_bytes(&p)));
p
}
Err(why) => return self.refuse(n, call, args, why),
}
};
let plan = match self.policy.check_open(&abs, flags, mode) {
Ok(p) => p,
Err(why) => return self.refuse(n, call, args, why),
};
if self.mode == Mode::CheckOnly {
args.push("check-only: the worker's own call proceeds".into());
self.answer(n.id, Response::Continue, &mut args);
self.push(call, args, Decision::Allowed);
return;
}
match self.open_for(n, &plan, resolve) {
Ok(fd) => self.inject(n, call, args, fd, plan.cloexec()),
Err(OpenFail::Policy(why)) => self.refuse(n, call, args, why),
Err(OpenFail::Errno(e)) => {
args.push(format!("result={}", errno_name(e)));
self.answer_err_logged(n.id, call, args, Decision::Allowed, e);
}
}
}
/// Opens `plan` beneath its prefix, as the broker.
fn open_for(
&self,
n: &Notif,
plan: &OpenPlan,
worker_resolve: u64,
) -> Result<OwnedFd, OpenFail> {
let prefix = &self.policy.prefixes[plan.prefix];
let errno = |e: std::io::Error| OpenFail::Errno(e.raw_os_error().unwrap_or(libc::EIO));
let (dir, rel, mut resolve) = match prefix.resolve {
Resolve::Device | Resolve::File => {
let rel = prefix
.path
.strip_prefix("/")
.unwrap_or(&prefix.path)
.to_path_buf();
let how = if prefix.resolve == Resolve::Device {
RESOLVE_IN_ROOT | RESOLVE_NO_MAGICLINKS | RESOLVE_NO_SYMLINKS
} else {
RESOLVE_IN_ROOT | RESOLVE_NO_MAGICLINKS
};
(self.root.as_raw_fd(), rel, how)
}
r => {
let Some(dir) = &self.roots[plan.prefix] else {
return Err(OpenFail::Errno(libc::ENOENT));
};
let base = RESOLVE_NO_MAGICLINKS | RESOLVE_NO_XDEV;
let (rel, how) = match r {
Resolve::NoSymlinks => (
plan.rel.clone(),
base | RESOLVE_BENEATH | RESOLVE_NO_SYMLINKS,
),
Resolve::Beneath => (plan.rel.clone(), base | RESOLVE_BENEATH),
Resolve::InRoot => (plan.rel.clone(), base | RESOLVE_IN_ROOT),
Resolve::Proc => {
let tgid = sys::ns_id(n.pid, "NStgid:").map_err(errno)?;
let tid = sys::ns_id(n.pid, "NSpid:").map_err(errno)?;
(proc_rewrite(&plan.rel, tgid, tid), base | RESOLVE_BENEATH)
}
Resolve::Device | Resolve::File => unreachable!(),
};
(dir.as_raw_fd(), rel, how)
}
};
// The worker may only ask for more strictness.
resolve |= worker_resolve & (RESOLVE_NO_XDEV | RESOLVE_NO_MAGICLINKS | RESOLVE_NO_SYMLINKS);
// Always non-blocking (a FIFO would block the broker) and
// close-on-exec for the broker's own copy; the worker's O_NONBLOCK
// is restored below and its O_CLOEXEC goes on the injected fd.
let flags = if plan.flags & libc::O_PATH != 0 {
// ADDFD cannot inject an O_PATH file (the kernel's fget skips
// them), so the broker opens it for reading instead: only under
// a prefix that allows reading anyway (check_open refuses O_PATH
// under a write-only one).
plan.flags & (libc::O_DIRECTORY | libc::O_NOFOLLOW)
| libc::O_RDONLY
| libc::O_NONBLOCK
| libc::O_CLOEXEC
} else {
(plan.flags | libc::O_NONBLOCK | libc::O_CLOEXEC) & !libc::O_NOCTTY
};
let how = sys::OpenHow {
flags: flags as u32 as u64,
mode: plan.mode as u64,
resolve,
};
let nofollow = plan.flags & libc::O_NOFOLLOW != 0;
let map = |e: std::io::Error| match e.raw_os_error().unwrap_or(libc::EIO) {
libc::ELOOP if nofollow => OpenFail::Errno(libc::ELOOP),
libc::ELOOP => OpenFail::Policy(Refusal("symbolic link")),
libc::EXDEV => OpenFail::Policy(Refusal("escapes its prefix or crosses a mount")),
e => OpenFail::Errno(e),
};
let fd = if flags & libc::O_CREAT != 0 && prefix.resolve != Resolve::Device {
self.open_or_create(n, dir, &rel, &how).map_err(map)?
} else {
sys::openat2(dir, &rel, &how).map_err(map)?
};
let kind = sys::fstat_mode(fd.as_raw_fd()).map_err(errno)?;
let ok = match prefix.resolve {
Resolve::Device => kind == libc::S_IFCHR,
Resolve::File => kind == libc::S_IFREG,
_ => {
kind == libc::S_IFREG
|| kind == libc::S_IFDIR
|| plan.flags & libc::O_PATH != 0 && kind == libc::S_IFLNK
}
};
if !ok {
return Err(OpenFail::Policy(Refusal("not a regular file or directory")));
}
if prefix.access == Access::WriteOnly && kind != libc::S_IFREG {
return Err(OpenFail::Policy(Refusal(
"a directory under a write-only prefix",
)));
}
if plan.flags & libc::O_NONBLOCK == 0 {
sys::clear_nonblock(fd.as_raw_fd()).map_err(errno)?;
}
Ok(fd)
}
/// An `O_CREAT` open, as the broker: a file that exists is opened as it
/// is; a file the broker CREATES (`O_EXCL`, so it knows) is created under
/// the caller's fs uid/gid ([`sys::as_fs_ids`]) and so is the caller's, as if the caller had made it (a runner with a
/// host id range runs jobs as other ids than the broker's). A file that
/// already existed is never chowned.
fn open_or_create(
&self,
n: &Notif,
dir: RawFd,
rel: &Path,
how: &sys::OpenHow,
) -> std::io::Result<OwnedFd> {
let excl = how.flags & libc::O_EXCL as u64 != 0;
for _ in 0..8 {
if !excl {
let plain = sys::OpenHow {
flags: how.flags & !(libc::O_CREAT as u64),
mode: 0,
resolve: how.resolve,
};
match sys::openat2(dir, rel, &plain) {
Ok(fd) => return Ok(fd),
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => {}
Err(e) => return Err(e),
}
}
let create = sys::OpenHow {
flags: how.flags | libc::O_EXCL as u64,
..*how
};
// Created AS the caller (its fs uid/gid): a job-owned tmpfs cannot map host root.
let (uid, gid) = sys::fs_ids(n.pid)?;
match sys::as_fs_ids(uid, gid, || sys::openat2(dir, rel, &create)) {
Ok(fd) => {
sys::fchown_if_other(fd.as_raw_fd(), uid, gid)?;
return Ok(fd);
}
// Made by someone else in between: open it as it is.
Err(e) if e.raw_os_error() == Some(libc::EEXIST) && !excl => continue,
Err(e) => return Err(e),
}
}
Err(std::io::Error::from_raw_os_error(libc::EAGAIN))
}
/// Installs `fd` in the worker as the call's result.
fn inject(&self, n: &Notif, call: &str, mut args: Vec<String>, fd: OwnedFd, cloexec: bool) {
let send = self.support.addfd_send;
let how = AddFd {
target_fd: None,
cloexec,
send,
};
match self.listener.add_fd(n.id, fd.as_fd(), how) {
Ok(newfd) => {
if !send {
// Kernels 5.9–5.13: answer with the fd number ourselves.
self.answer(n.id, Response::Value(newfd as i64), &mut args);
}
args.push(format!("fd={newfd}"));
self.push(call, args, Decision::Allowed);
}
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => {
args.push(gone_note(&e));
self.push(call, args, Decision::Allowed);
}
Err(e) => {
// EMFILE (the worker's RLIMIT_NOFILE) and the like: the
// worker's call fails with it.
let err = e.raw_os_error().unwrap_or(libc::EIO);
args.push(format!("result={}", errno_name(err)));
self.answer_err_logged(n.id, call, args, Decision::Allowed, err);
}
}
}
fn on_socket(&mut self, n: &Notif, call: &str) {
let a = n.data.args;
let mut args = vec![
format!("pid={}", n.pid),
format!("domain={}", a[0]),
format!("type={:#x}", a[1]),
format!("protocol={}", a[2]),
];
self.hold();
if let Err(why) = self.policy.check_socket(a[0], a[1], a[2]) {
return self.refuse(n, call, args, why);
}
let ty = a[1] as u32 as i32;
if self.mode == Mode::CheckOnly {
// Value arguments only: letting the call proceed cannot race.
self.answer(n.id, Response::Continue, &mut args);
self.push(call, args, Decision::Allowed);
return;
}
// The broker's own socket, in the broker's network namespace: the
// worker's stays empty.
match sys::socket(a[0] as i32, ty | libc::SOCK_CLOEXEC, a[2] as i32) {
Ok(fd) => self.inject(n, call, args, fd, ty & libc::SOCK_CLOEXEC != 0),
Err(e) => {
let e = e.raw_os_error().unwrap_or(libc::EIO);
args.push(format!("result={}", errno_name(e)));
self.answer_err_logged(n.id, call, args, Decision::Allowed, e);
}
}
}
fn on_connect(&mut self, n: &Notif, call: &str) {
let a = n.data.args;
let wfd = a[0] as i32;
let len = (a[2] as u32 as usize).min(SOCKADDR_MAX);
let mut args = vec![format!("pid={}", n.pid), format!("fd={wfd}")];
let mut sa = vec![0u8; len];
if sys::read_mem(n.pid, a[1], &mut sa).is_err() || !self.valid(n.id) {
args.push("sockaddr unreadable".into());
self.answer_err_logged(n.id, call, args, Decision::Refused, libc::EFAULT);
return;
}
match parse_sockaddr(&sa) {
Ok(addr) => args.push(format!("addr={addr}")),
Err(_) => args.push(format!("sockaddr={}", hex(&sa))),
}
self.hold();
let addr = match self.policy.check_connect(&sa) {
Ok(a) => a,
Err(why) => return self.refuse(n, call, args, why),
};
if self.mode == Mode::CheckOnly {
args.push("check-only: the worker's own call proceeds".into());
self.answer(n.id, Response::Continue, &mut args);
self.push(call, args, Decision::Allowed);
return;
}
// Connect the worker's own socket, by the broker, to the address the
// broker checked (never the worker's pointer).
let Some(pfd) = self.caller_pidfd(n) else {
args.push("worker gone".into());
self.push(call, args, Decision::Refused);
return;
};
let sock = match sys::pidfd_getfd(&pfd, wfd) {
Ok(s) => s,
Err(e) => {
let e = e.raw_os_error().unwrap_or(libc::EBADF);
args.push(format!("result={}", errno_name(e)));
self.answer_err_logged(n.id, call, args, Decision::Allowed, e);
return;
}
};
if let Err(why) = check_stream_socket(sock.as_raw_fd(), addr) {
return self.refuse(n, call, args, why);
}
let listener = Arc::clone(&self.listener);
let log = Arc::clone(&self.log);
let id = n.id;
let call = call.to_string();
// A connect can take a while: answer it from its own thread so the
// broker keeps serving the worker's other threads.
let h = std::thread::spawn(move || {
let r = sys::connect_raw(sock.as_raw_fd(), &sockaddr_bytes(addr));
args.push(format!(
"result={}",
if r == 0 {
"ok".to_string()
} else {
errno_name(r)
}
));
let resp = if r == 0 {
Response::Value(0)
} else {
Response::Errno(r)
};
if let Err(e) = listener.respond(id, resp) {
args.push(gone_note(&e));
}
log.lock()
.unwrap_or_else(|e| e.into_inner())
.push(&call, args, Decision::Allowed);
});
self.pending.retain(|h| !h.is_finished());
self.pending.push(h);
}
fn on_exec(&mut self, n: &Notif, call: &str) {
let a = n.data.args;
let ptr = if call == "execveat" { a[1] } else { a[0] };
let mut args = vec![format!("pid={}", n.pid)];
if call == "execveat" {
args.push(format!("dirfd={}", a[0] as i32));
}
// For the log only: the decision below never reads the worker's
// memory, so CONTINUE cannot race it.
if let Ok(p) = sys::read_cstr(n.pid, ptr, PATH_MAX) {
args.push(format!("path={}", render_bytes(&p)));
}
let allow = match self.policy.exec {
ExecPolicy::Allow => true,
ExecPolicy::AllowFirst => {
let first = !self.exec_used && n.pid == self.initial_pid;
if first {
self.exec_used = true;
args.push("the launcher's first exec".into());
}
first
}
ExecPolicy::Deny => false,
};
if allow {
self.answer(n.id, Response::Continue, &mut args);
self.push(call, args, Decision::Allowed);
return;
}
self.hold();
self.refuse(n, call, args, Refusal("exec is not allowed"));
}
}
enum OpenFail {
Policy(Refusal),
Errno(i32),
}
fn check_stream_socket(fd: RawFd, addr: SocketAddr) -> Result<(), Refusal> {
if sys::fstat_mode(fd).ok() != Some(libc::S_IFSOCK) {
return Err(Refusal("not a socket"));
}
let want = match addr {
SocketAddr::V4(_) => libc::AF_INET,
SocketAddr::V6(_) => libc::AF_INET6,
};
if sys::sockopt_int(fd, libc::SOL_SOCKET, libc::SO_DOMAIN).ok() != Some(want) {
return Err(Refusal("socket family not allowed"));
}
if sys::sockopt_int(fd, libc::SOL_SOCKET, libc::SO_TYPE).ok() != Some(libc::SOCK_STREAM) {
return Err(Refusal("only TCP stream sockets"));
}
Ok(())
}
fn gone_note(e: &std::io::Error) -> String {
if e.raw_os_error() == Some(libc::ENOENT) {
"worker gone before the answer".into()
} else {
format!("answer failed: {e}")
}
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
/// `ENOENT` etc. for the log.
pub fn errno_name(e: i32) -> String {
let n = match e {
libc::ENOENT => "ENOENT",
libc::EACCES => "EACCES",
libc::EPERM => "EPERM",
libc::EEXIST => "EEXIST",
libc::EISDIR => "EISDIR",
libc::ENOTDIR => "ENOTDIR",
libc::EINVAL => "EINVAL",
libc::EBADF => "EBADF",
libc::EMFILE => "EMFILE",
libc::ECONNREFUSED => "ECONNREFUSED",
libc::EINPROGRESS => "EINPROGRESS",
libc::ETIMEDOUT => "ETIMEDOUT",
libc::ENETUNREACH => "ENETUNREACH",
libc::EAGAIN => "EAGAIN",
libc::EROFS => "EROFS",
libc::ENXIO => "ENXIO",
libc::EFAULT => "EFAULT",
_ => return format!("errno {e}"),
};
n.to_string()
}