crates/heimdall-broker/src/policy.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! The per-job broker policy and its pure decisions (spec §4).
//!
//! Heimdall is a defensive sandbox. Everything here is a pure function of the
//! policy and the bytes the broker copied out of the worker once: no system
//! calls, so the decisions are unit- and property-tested without a kernel.
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::path::{Component, Path, PathBuf};
use heimdall_core::JobId;
use serde::{Deserialize, Serialize};
/// What a prefix allows.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Access {
/// Open for reading only (`O_RDONLY` or `O_PATH`; no `O_CREAT`,
/// `O_TRUNC`, `O_TMPFILE`).
Read,
/// Read, write and create (the job's /tmp, `policy.write`).
ReadWrite,
/// Write and create, never read back, never list (the job's /out):
/// `O_WRONLY` only, no `O_PATH`, no `O_DIRECTORY`, no `O_TMPFILE`, and
/// never the prefix directory itself.
WriteOnly,
}
/// How the broker resolves a path below a prefix when it opens it. Every
/// kind refuses magic links (`/proc/self/fd/N`, `/proc/<pid>/root`) and,
/// except [`Resolve::Device`], crossing into another mount.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Resolve {
/// Every symbolic link refuses the open (`RESOLVE_NO_SYMLINKS`).
NoSymlinks,
/// Symbolic links are followed but may not leave the prefix
/// (`RESOLVE_BENEATH`; an absolute link target is refused).
Beneath,
/// The worker's root image: absolute links resolve against the worker's
/// root (`RESOLVE_IN_ROOT`), never into another mount (`/in`, `/out`,
/// `/tmp`, `/proc`, `/dev` are mounts of their own).
InRoot,
/// The worker's `/proc`: `self`, `thread-self`, `net` and `mounts` are
/// rewritten to the caller's own pid in the worker's pid namespace, then
/// resolved beneath, magic links refused.
Proc,
/// One device node at exactly this path (`/dev/null`, `/dev/urandom`):
/// only a character device opens.
Device,
/// One regular file at exactly this path (`/etc/ld.so.cache`), resolved
/// from the worker's root, magic links refused.
File,
}
impl Resolve {
/// Does this prefix name one exact path (not a directory tree)?
pub fn exact(self) -> bool {
matches!(self, Resolve::Device | Resolve::File)
}
}
/// One allowed path prefix: an absolute path as the WORKER sees it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Prefix {
pub path: PathBuf,
pub access: Access,
pub resolve: Resolve,
}
/// What a refused call gets.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OnRefuse {
/// The call fails with `EACCES`; the worker goes on.
Eacces,
/// The worker's whole process is killed with `SIGKILL`.
Kill,
}
/// `execve` / `execveat`. Exec cannot be done by the broker on the worker's
/// behalf, so its decision never depends on the worker's memory: the path is
/// read for the log only.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecPolicy {
/// Every exec is refused.
Deny,
/// The launcher's own first exec of the worker binary (by the first
/// process, once) proceeds; every later exec is refused.
AllowFirst,
/// Every exec proceeds and is logged; what may be executed is decided by
/// the layers below (Landlock: only the image is executable).
Allow,
}
/// The policy of one job (spec §4): allowed path prefixes, the declared
/// address:port peers of a fetch job, and nothing else.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BrokerPolicy {
pub job_id: JobId,
pub prefixes: Vec<Prefix>,
/// A fetch job's declared TCP peers. Empty = no sockets at all.
#[serde(default)]
pub fetch: Vec<SocketAddr>,
pub on_refuse: OnRefuse,
pub exec: ExecPolicy,
/// At most this many events are kept; later ones are counted only.
pub max_events: usize,
}
/// Default cap on logged events per job.
pub const DEFAULT_MAX_EVENTS: usize = 10_000;
impl BrokerPolicy {
/// A policy with no prefixes, no network, `EACCES` on refusal and the
/// launcher's first exec only.
pub fn empty(job_id: JobId) -> Self {
Self {
job_id,
prefixes: Vec::new(),
fetch: Vec::new(),
on_refuse: OnRefuse::Eacces,
exec: ExecPolicy::AllowFirst,
max_events: DEFAULT_MAX_EVENTS,
}
}
/// The usual job: `input_dir` read-only, `output_dir` write-only, no
/// network, refused calls get `EACCES`, the launcher's first exec only.
pub fn job(
job_id: JobId,
input_dir: impl Into<PathBuf>,
output_dir: impl Into<PathBuf>,
) -> Self {
Self::empty(job_id)
.with_prefix(input_dir, Access::Read, Resolve::Beneath)
.with_prefix(output_dir, Access::WriteOnly, Resolve::NoSymlinks)
}
pub fn with_prefix(
mut self,
path: impl Into<PathBuf>,
access: Access,
resolve: Resolve,
) -> Self {
self.prefixes.push(Prefix {
path: path.into(),
access,
resolve,
});
self
}
/// Allow TCP to exactly `addr` too (a fetch job's peer).
pub fn with_fetch(mut self, addr: SocketAddr) -> Self {
self.fetch.push(addr);
self
}
/// Add a read-only system directory (symlinks followed beneath it).
pub fn with_system_read(self, dir: impl Into<PathBuf>) -> Self {
self.with_prefix(dir, Access::Read, Resolve::Beneath)
}
pub fn with_on_refuse(mut self, on_refuse: OnRefuse) -> Self {
self.on_refuse = on_refuse;
self
}
pub fn with_exec(mut self, exec: ExecPolicy) -> Self {
self.exec = exec;
self
}
/// Checks the policy itself: absolute, normalised prefixes.
pub fn validate(&self) -> Result<(), String> {
for p in &self.prefixes {
if !p.path.is_absolute() {
return Err(format!("prefix {} is not absolute", p.path.display()));
}
if p.path
.components()
.any(|c| matches!(c, Component::ParentDir | Component::CurDir))
{
return Err(format!("prefix {} is not normalised", p.path.display()));
}
if p.resolve.exact() && p.access == Access::WriteOnly {
return Err(format!(
"exact path {} cannot be write-only",
p.path.display()
));
}
}
Ok(())
}
}
/// Why a call was refused (a stable short text, logged in the event).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Refusal(pub &'static str);
/// The kernel's `O_LARGEFILE` (the libc crate says 0 for 64-bit glibc, but
/// musl and others pass the kernel bit).
#[cfg(target_arch = "aarch64")]
pub const O_LARGEFILE_KERNEL: i32 = 0o400000;
#[cfg(not(target_arch = "aarch64"))]
pub const O_LARGEFILE_KERNEL: i32 = 0o100000;
/// Open flags the broker understands; any other bit refuses the open.
const KNOWN_OPEN_FLAGS: i32 = libc::O_ACCMODE
| libc::O_CREAT
| libc::O_EXCL
| libc::O_NOCTTY
| libc::O_TRUNC
| libc::O_APPEND
| libc::O_NONBLOCK
| libc::O_DSYNC
| libc::O_SYNC
| libc::O_DIRECTORY
| libc::O_NOFOLLOW
| libc::O_CLOEXEC
| libc::O_LARGEFILE
| O_LARGEFILE_KERNEL
| libc::O_NOATIME
| libc::O_DIRECT
| libc::O_PATH
| libc::O_TMPFILE;
/// What the broker will open on the worker's behalf.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OpenPlan {
/// Index into [`BrokerPolicy::prefixes`].
pub prefix: usize,
/// The path below the prefix (`"."` for the prefix itself).
pub rel: PathBuf,
/// The worker's flags, checked (`O_CLOEXEC` decides the injected fd's flag).
pub flags: i32,
/// Creation mode, `& 0o7777`; 0 unless `O_CREAT`/`O_TMPFILE`.
pub mode: u32,
}
impl OpenPlan {
pub fn cloexec(&self) -> bool {
self.flags & libc::O_CLOEXEC != 0
}
}
fn is_tmpfile(flags: i32) -> bool {
flags & libc::O_TMPFILE == libc::O_TMPFILE
}
/// Does `flags` ask for anything beyond reading?
pub fn open_writes(flags: i32) -> bool {
if flags & libc::O_PATH != 0 {
// O_PATH ignores every flag but O_DIRECTORY, O_NOFOLLOW, O_CLOEXEC.
return false;
}
flags & libc::O_ACCMODE != libc::O_RDONLY
|| flags & (libc::O_CREAT | libc::O_TRUNC) != 0
|| is_tmpfile(flags)
}
/// The normal, absolute form of `path`, or a refusal. `..` anywhere refuses
/// (no lexical games), `.` and repeated slashes are dropped.
pub fn normalise(path: &[u8]) -> Result<PathBuf, Refusal> {
use std::os::unix::ffi::OsStrExt;
if path.is_empty() {
return Err(Refusal("empty path"));
}
if path[0] != b'/' {
return Err(Refusal("relative path"));
}
let p = Path::new(std::ffi::OsStr::from_bytes(path));
let mut out = PathBuf::from("/");
for c in p.components() {
match c {
Component::RootDir | Component::CurDir => {}
Component::ParentDir => return Err(Refusal("`..` in path")),
Component::Normal(n) => out.push(n),
Component::Prefix(_) => return Err(Refusal("bad path")),
}
}
Ok(out)
}
/// Joins a relative `path` onto `base` (an absolute directory as the worker
/// sees it, from `/proc/<tid>/cwd` or `/proc/<tid>/fd/<dirfd>`).
pub fn join_relative(base: &[u8], path: &[u8]) -> Result<Vec<u8>, Refusal> {
if base.first() != Some(&b'/') {
return Err(Refusal(
"the directory fd is not a directory in the worker's tree",
));
}
let mut v = base.to_vec();
v.push(b'/');
v.extend_from_slice(path);
Ok(v)
}
impl BrokerPolicy {
/// The decision for an `openat`/`openat2`/`open`/`creat` of the absolute
/// `path` (as copied from the worker) with `flags` and `mode`.
pub fn check_open(&self, path: &[u8], flags: u64, mode: u64) -> Result<OpenPlan, Refusal> {
if flags > u32::MAX as u64 {
return Err(Refusal("unknown open flags"));
}
let flags = flags as u32 as i32;
if flags & !KNOWN_OPEN_FLAGS != 0 {
return Err(Refusal("unknown open flags"));
}
if flags & libc::O_ACCMODE == libc::O_ACCMODE {
return Err(Refusal("bad access mode"));
}
let norm = normalise(path)?;
// The longest matching prefix wins (/in inside the image's /, say); a
// device prefix matches its exact path only.
let mut best: Option<(usize, usize)> = None;
for (i, p) in self.prefixes.iter().enumerate() {
let hit = if p.resolve.exact() {
norm == p.path
} else {
norm.starts_with(&p.path)
};
if hit {
let depth = p.path.components().count();
if best.is_none_or(|(_, d)| depth > d) {
best = Some((i, depth));
}
}
}
let Some((idx, _)) = best else {
return Err(Refusal("outside policy"));
};
let prefix = &self.prefixes[idx];
let rel = norm
.strip_prefix(&prefix.path)
.map_err(|_| Refusal("outside policy"))?;
let rel = if rel.as_os_str().is_empty() {
PathBuf::from(".")
} else {
rel.to_path_buf()
};
match prefix.access {
Access::Read if open_writes(flags) => {
return Err(Refusal("write under a read-only prefix"))
}
Access::WriteOnly => {
if flags & libc::O_PATH != 0 {
return Err(Refusal("O_PATH under a write-only prefix"));
}
if flags & libc::O_ACCMODE != libc::O_WRONLY {
return Err(Refusal("read under a write-only prefix"));
}
if is_tmpfile(flags) || flags & libc::O_DIRECTORY != 0 {
return Err(Refusal("a directory under a write-only prefix"));
}
if rel == Path::new(".") {
return Err(Refusal("the write-only directory itself"));
}
}
_ => {}
}
if prefix.resolve == Resolve::InRoot && rel == Path::new(".") {
// Landlock gives the image's top-level entries, never `/`
// itself: the broker never widens what Landlock allows.
return Err(Refusal("the root directory itself"));
}
// `2>/dev/null` opens with O_CREAT|O_TRUNC: harmless on an existing
// device; O_EXCL (create a new one) is not.
if prefix.resolve.exact()
&& (flags & (libc::O_EXCL | libc::O_DIRECTORY | libc::O_PATH) != 0
|| is_tmpfile(flags)
|| (prefix.resolve == Resolve::File && flags & libc::O_CREAT != 0))
{
return Err(Refusal("an exact path opens for read or write only"));
}
let creates = flags & libc::O_CREAT != 0 || is_tmpfile(flags);
let mode = if creates { (mode & 0o7777) as u32 } else { 0 };
Ok(OpenPlan {
prefix: idx,
rel,
flags,
mode,
})
}
/// The decision for `socket(domain, type, protocol)`.
pub fn check_socket(&self, domain: u64, ty: u64, protocol: u64) -> Result<(), Refusal> {
if self.fetch.is_empty() {
return Err(Refusal("no network for this job"));
}
let family_ok = self.fetch.iter().any(|f| {
let want = match f {
SocketAddr::V4(_) => libc::AF_INET,
SocketAddr::V6(_) => libc::AF_INET6,
};
domain == want as u64
});
if !family_ok {
return Err(Refusal("socket family not allowed"));
}
let ty = ty as u32 as i32;
if ty & !(libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC | 0xf) != 0
|| ty & 0xf != libc::SOCK_STREAM
{
return Err(Refusal("only TCP stream sockets"));
}
if protocol != 0 && protocol != libc::IPPROTO_TCP as u64 {
return Err(Refusal("only TCP stream sockets"));
}
Ok(())
}
/// The decision for `connect` to the sockaddr bytes copied from the worker.
pub fn check_connect(&self, sockaddr: &[u8]) -> Result<SocketAddr, Refusal> {
if self.fetch.is_empty() {
return Err(Refusal("no network for this job"));
}
let addr = parse_sockaddr(sockaddr)?;
if !self.fetch.iter().any(|f| same_peer(addr, *f)) {
return Err(Refusal("address not allowed"));
}
Ok(addr)
}
}
fn same_peer(a: SocketAddr, b: SocketAddr) -> bool {
a.ip() == b.ip() && a.port() == b.port()
}
/// Parses an `AF_INET` / `AF_INET6` sockaddr; anything else is refused.
pub fn parse_sockaddr(b: &[u8]) -> Result<SocketAddr, Refusal> {
if b.len() < 2 {
return Err(Refusal("short sockaddr"));
}
let family = u16::from_ne_bytes([b[0], b[1]]) as i32;
match family {
libc::AF_INET => {
if b.len() < 16 {
return Err(Refusal("short sockaddr"));
}
let port = u16::from_be_bytes([b[2], b[3]]);
let ip = Ipv4Addr::new(b[4], b[5], b[6], b[7]);
Ok(SocketAddr::new(IpAddr::V4(ip), port))
}
libc::AF_INET6 => {
if b.len() < 28 {
return Err(Refusal("short sockaddr"));
}
let port = u16::from_be_bytes([b[2], b[3]]);
let mut o = [0u8; 16];
o.copy_from_slice(&b[8..24]);
Ok(SocketAddr::new(IpAddr::V6(Ipv6Addr::from(o)), port))
}
libc::AF_UNIX => Err(Refusal("unix sockets not allowed")),
_ => Err(Refusal("address family not allowed")),
}
}
/// The sockaddr bytes for `addr` (what the broker passes to its own connect).
pub fn sockaddr_bytes(addr: SocketAddr) -> Vec<u8> {
match addr {
SocketAddr::V4(a) => {
let mut b = vec![0u8; std::mem::size_of::<libc::sockaddr_in>()];
b[0..2].copy_from_slice(&(libc::AF_INET as u16).to_ne_bytes());
b[2..4].copy_from_slice(&a.port().to_be_bytes());
b[4..8].copy_from_slice(&a.ip().octets());
b
}
SocketAddr::V6(a) => {
let mut b = vec![0u8; std::mem::size_of::<libc::sockaddr_in6>()];
b[0..2].copy_from_slice(&(libc::AF_INET6 as u16).to_ne_bytes());
b[2..4].copy_from_slice(&a.port().to_be_bytes());
b[4..8].copy_from_slice(&a.flowinfo().to_ne_bytes());
b[8..24].copy_from_slice(&a.ip().octets());
b[24..28].copy_from_slice(&a.scope_id().to_ne_bytes());
b
}
}
}
/// The path below a [`Resolve::Proc`] prefix with `self`, `thread-self`,
/// `net` and `mounts` replaced by the caller's ids in the worker's pid
/// namespace (the broker is not in it, so the kernel's `self` would not
/// resolve for it).
pub fn proc_rewrite(rel: &Path, tgid: u32, tid: u32) -> PathBuf {
let mut it = rel.components();
let first = it.next();
let rest: PathBuf = it.collect();
let head: PathBuf = match first {
Some(Component::Normal(n)) if n == "self" => PathBuf::from(tgid.to_string()),
Some(Component::Normal(n)) if n == "thread-self" => {
PathBuf::from(format!("{tgid}/task/{tid}"))
}
Some(Component::Normal(n)) if n == "net" || n == "mounts" => {
PathBuf::from(tgid.to_string()).join(n)
}
Some(c) => PathBuf::from(c.as_os_str()),
None => return rel.to_path_buf(),
};
if rest.as_os_str().is_empty() {
head
} else {
head.join(rest)
}
}
/// Renders open flags for the log, e.g. `O_WRONLY|O_CREAT|O_CLOEXEC`.
pub fn render_open_flags(flags: u64) -> String {
let f = flags as u32 as i32;
let mut parts = vec![match f & libc::O_ACCMODE {
libc::O_RDONLY => "O_RDONLY",
libc::O_WRONLY => "O_WRONLY",
libc::O_RDWR => "O_RDWR",
_ => "O_ACCMODE?",
}
.to_string()];
let named = [
(libc::O_CREAT, "O_CREAT"),
(libc::O_EXCL, "O_EXCL"),
(libc::O_TRUNC, "O_TRUNC"),
(libc::O_APPEND, "O_APPEND"),
(libc::O_NONBLOCK, "O_NONBLOCK"),
(libc::O_DIRECTORY, "O_DIRECTORY"),
(libc::O_NOFOLLOW, "O_NOFOLLOW"),
(libc::O_CLOEXEC, "O_CLOEXEC"),
(libc::O_PATH, "O_PATH"),
];
let mut rest = f & !libc::O_ACCMODE & !libc::O_LARGEFILE & !O_LARGEFILE_KERNEL;
if is_tmpfile(f) {
parts.push("O_TMPFILE".into());
rest &= !libc::O_TMPFILE;
}
for (bit, name) in named {
if rest & bit != 0 {
parts.push(name.into());
rest &= !bit;
}
}
if rest != 0 || flags > u32::MAX as u64 {
parts.push(format!("{flags:#x}"));
}
parts.join("|")
}
/// Renders bytes copied from the worker for the log: lossy UTF-8, control
/// characters escaped, at most 512 characters.
pub fn render_bytes(b: &[u8]) -> String {
let s: String = String::from_utf8_lossy(b).escape_debug().collect();
if s.chars().count() > 512 {
let mut t: String = s.chars().take(512).collect();
t.push('…');
t
} else {
s
}
}