crates/heimdall-container/src/shim.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! The shim: one short-lived process per job that drives youki's
//! `libcontainer` (create, start, wait, kill on the wall clock, delete) with
//! Heimdall's executor, which applies Landlock inside the job right before
//! `execve` and reports what the job really got.
//!
//! Heimdall is a defensive sandbox. The runner never forks inside its own
//! (multi-threaded) process: it starts this shim with a JSON request on stdin
//! and reads a JSON response from stdout. The shim never receives the
//! per-boot signing key; the job's stdio is `/dev/null` (or the shim's stderr
//! with `HEIMDALL_CONTAINER_DEBUG=1`).
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use crate::brokered::BrokerRequest;
use crate::confine::{LandlockApplied, LandlockPlan};
use crate::genprof::GeneratedPlan;
use crate::learn::LearnPlan;
/// What the runner asks the shim to do.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ShimRequest {
/// Container id (unique per job).
pub id: String,
/// The bundle: `config.json` + `rootfs/`.
pub bundle: PathBuf,
/// libcontainer's state root.
pub state_root: PathBuf,
/// The wall-clock limit in milliseconds.
pub wall_ms: u64,
/// The job's cgroup v2 dir.
pub cgroup_dir: PathBuf,
/// Landlock to apply inside the job (`None`: no Landlock).
pub landlock: Option<LandlockPlan>,
/// The generated seccomp profile to install inside the job on top of
/// the fixed one (`None`: the fixed profile only).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub generated: Option<GeneratedPlan>,
/// A learning run (crate::learn): the executor installs the learning
/// filter instead of a profile and the shim hands its listener over.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub learning: Option<LearnPlan>,
/// The syscall broker for this job (`None`: no broker filter).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub broker: Option<BrokerRequest>,
/// cgroup v2 `io.weight` (1..=10000) the shim writes before the job
/// starts (`None`: no io controller, nothing written).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub io_weight: Option<u16>,
/// A host-folder workspace the shim watches (`None`: none, or a tmpfs,
/// which the kernel caps itself).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace: Option<WorkspaceWatch>,
}
/// A host-folder workspace: its folder and cap.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkspaceWatch {
pub dir: PathBuf,
pub cap: u64,
}
/// What the job's executor saw right before `execve`, inside the job.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct InsideReport {
/// `prctl(PR_GET_NO_NEW_PRIVS)`.
pub no_new_privs: bool,
/// `prctl(PR_GET_SECCOMP)`: 2 = filter.
pub seccomp_mode: i32,
/// Landlock as enforced, `None` when not asked for.
pub landlock: Option<LandlockApplied>,
/// The executor failed before or at `execve`; the job never ran.
pub error: Option<String>,
/// The digest of the generated seccomp profile, reported AFTER the
/// executor installed it (`None`: none was installed).
#[serde(skip_serializing_if = "Option::is_none")]
pub generated: Option<String>,
/// The learning filter is installed and its listener was handed over,
/// reported AFTER the install (crate::learn).
#[serde(skip_serializing_if = "std::ops::Not::not")]
pub learning: bool,
/// The broker filter mode installed, reported AFTER the install
/// (`None`: none was installed).
#[serde(skip_serializing_if = "Option::is_none")]
pub broker: Option<heimdall_broker::Mode>,
}
/// What happened.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct ShimResponse {
/// The container never ran: why.
pub error: Option<String>,
/// The job root was entered with `MS_MOVE` + `chroot`, not `pivot_root`
/// ([`heimdall_plain::rootfs::NO_PIVOT_NOTE`]): `/` is the initial rootfs.
#[serde(skip_serializing_if = "std::ops::Not::not")]
pub no_pivot: bool,
pub exit_code: Option<i32>,
pub signal: Option<i32>,
/// The shim killed the job at the wall-clock limit.
pub wall_killed: bool,
/// `oom_kill` count from the job's `memory.events`.
pub oom_kills: u64,
/// `max` count from the job's `pids.events`.
pub pids_max: u64,
/// Milliseconds the job ran.
pub ran_ms: u64,
/// A learning run: the shim killed the job after a stop signal and the grace.
pub term_killed: bool,
/// `memory.swap.max` as read back before the job started.
pub swap_max: Option<String>,
/// `io.max` as read back before the job started (when the file exists).
#[serde(skip_serializing_if = "Option::is_none")]
pub io_max: Option<String>,
/// The `default` line of `io.weight` as read back before the job started.
#[serde(skip_serializing_if = "Option::is_none")]
pub io_weight: Option<String>,
/// The shim killed the job: its host-folder workspace went over its cap.
pub workspace_killed: bool,
/// The most the host-folder workspace held (bytes on disk).
pub workspace_bytes: u64,
/// The executor's report from inside the job.
pub inside: Option<InsideReport>,
/// What the broker did (full and check-only modes).
#[serde(skip_serializing_if = "Option::is_none")]
pub broker: Option<heimdall_broker::BrokerReport>,
}
/// Parse the executor's report lines (the last complete report wins; a
/// later `error` line after a good report means `execve` failed).
pub fn parse_inside(text: &str) -> Option<InsideReport> {
let mut out: Option<InsideReport> = None;
for line in text.lines().filter(|l| !l.trim().is_empty()) {
match serde_json::from_str::<InsideReport>(line) {
Ok(r) => {
out = Some(match out {
Some(mut prev) if r.error.is_some() => {
prev.error = r.error;
prev
}
_ => r,
})
}
Err(e) => {
let mut prev = out.unwrap_or_default();
prev.error = Some(format!("bad executor report ({e})"));
out = Some(prev);
}
}
}
out
}
/// Call first thing in `main()` of a binary that should also serve as the
/// shim: when the first argument is [`crate::SHIM_ARG`], it runs the shim
/// and exits.
#[cfg(feature = "youki")]
pub fn run_if_shim() {
if std::env::args().nth(1).as_deref() == Some(crate::SHIM_ARG) {
std::process::exit(main());
}
}
/// The shim's main: request on stdin, response on stdout. Returns the exit code.
#[cfg(feature = "youki")]
pub fn main() -> i32 {
use std::io::Read;
let mut input = Vec::new();
if let Err(e) = std::io::stdin().take(1 << 20).read_to_end(&mut input) {
return respond(&ShimResponse {
error: Some(format!("reading the request: {e}")),
..Default::default()
});
}
let resp = match serde_json::from_slice::<ShimRequest>(&input) {
Ok(req) => youki::run(&req),
Err(e) => ShimResponse {
error: Some(format!("bad request: {e}")),
..Default::default()
},
};
respond(&resp)
}
#[cfg(feature = "youki")]
fn respond(r: &ShimResponse) -> i32 {
println!("{}", serde_json::to_string(r).expect("response serializes"));
0
}
#[cfg(feature = "youki")]
mod youki {
use std::fs::File;
use std::io::{Read, Write};
use std::os::fd::{AsFd, AsRawFd, OwnedFd};
use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::time::{Duration, Instant};
use libcontainer::container::builder::ContainerBuilder;
use libcontainer::syscall::syscall::SyscallType;
use libcontainer::workload::default::DefaultExecutor;
use libcontainer::workload::{Executor, ExecutorError, ExecutorValidationError};
use nix::sys::signal::Signal;
use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
use oci_spec::runtime::Spec;
use super::{parse_inside, InsideReport, ShimRequest, ShimResponse};
use crate::confine::{self, LandlockPlan};
use crate::genprof::{GeneratedPlan, REPORT_FD};
use crate::learn::{LearnPlan, FORWARDED, STOP_SIGNALS};
use heimdall_plain::shim::events_count;
/// Runs inside the job's init process, after pivot_root, capabilities
/// and seccomp, right before `execve`: applies Landlock and reports what
/// the job really got on the report pipe (close-on-exec, so the job
/// itself never holds it).
#[derive(Clone)]
struct HeimdallExecutor {
report_fd: i32,
landlock: Option<LandlockPlan>,
generated: Option<GeneratedPlan>,
/// A learning run: the init's end of the listener hand-off socket.
learning: Option<i32>,
/// A learning run's uid (crate::learn::LearnPlan::uid).
learn_uid: u32,
/// The broker filter to install and the worker end of the listener
/// hand-off socket.
broker: Option<(heimdall_broker::Mode, Option<Arc<OwnedFd>>)>,
}
fn line(r: &InsideReport) -> Vec<u8> {
let mut line = serde_json::to_vec(r).expect("report serializes");
line.push(b'\n');
line
}
/// The job's program as a path: argv\[0\] when it has a `/`, else the
/// first match in the job's `PATH` (inside the container).
fn program_path(args: &[String], env: &[String]) -> Result<std::path::PathBuf, String> {
let a0 = args.first().ok_or("argv is empty")?;
if a0.contains('/') {
return Ok(a0.into());
}
let path = env
.iter()
.find_map(|e| e.strip_prefix("PATH="))
.unwrap_or("/usr/local/bin:/usr/bin:/bin");
path.split(':')
.map(|d| std::path::Path::new(d).join(a0))
.find(|p| p.is_file())
.ok_or_else(|| format!("{a0} not found in PATH {path}"))
}
impl HeimdallExecutor {
fn send(&self, r: &InsideReport) {
let _ = crate::sys::write_fd(self.report_fd, &line(r));
}
/// Install the generated profile on top of the fixed one and start
/// the job's program with `execveat` (crate::genprof). Returns only
/// when it could not; after the install it never returns.
fn exec_generated(
&self,
spec: &Spec,
g: &GeneratedPlan,
mut rep: InsideReport,
) -> Result<(), ExecutorError> {
use std::ffi::OsString;
let fail = |rep: &mut InsideReport, e: String| {
rep.error = Some(e.clone());
self.send(rep);
Err(ExecutorError::Other(e))
};
let process = spec.process().as_ref();
let args: Vec<String> = process.and_then(|p| p.args().clone()).unwrap_or_default();
let env: Vec<String> = process.and_then(|p| p.env().clone()).unwrap_or_default();
let program = match program_path(&args, &env) {
Ok(p) => p,
Err(e) => return fail(&mut rep, format!("generated seccomp: {e}")),
};
if let Err(e) = crate::sys::dup_cloexec_to(self.report_fd, REPORT_FD) {
return fail(&mut rep, format!("generated seccomp: report fd: {e}"));
}
let argv: Vec<OsString> = args.iter().map(OsString::from).collect();
let envp: Vec<OsString> = env.iter().map(OsString::from).collect();
let plan = match heimdall_seccomp::launch::ExecSelf::new(
&program,
&argv,
&envp,
g.profile.arch,
vec![GeneratedPlan::report_rule()],
&g.profile.rules,
heimdall_seccomp::Action::KillProcess,
) {
Ok(p) => p,
Err(e) => return fail(&mut rep, format!("generated seccomp: {e}")),
};
// Everything that allocates happens before the install.
let applied = line(&InsideReport {
generated: Some(g.digest.clone()),
..rep.clone()
});
let exec_failed = line(&InsideReport {
error: Some(
"execveat of the job's program failed under the generated seccomp profile"
.into(),
),
..rep.clone()
});
self.send(&rep);
if let Err(e) = plan.install() {
return fail(&mut rep, format!("generated seccomp: install: {e}"));
}
let _ = crate::sys::write_fd(REPORT_FD, &applied);
let _ = plan.exec();
let _ = crate::sys::write_fd(REPORT_FD, &exec_failed);
crate::sys::exit_now(127)
}
/// Install the LEARNING filter (heimdall-probe's: default USER_NOTIF,
/// its fixed rules, the prelude with the hand-off `sendmsg`), hand the
/// listener to the shim and start the job's program with `execveat`
/// (crate::learn). Returns only when it could not; after the install
/// it never returns.
fn exec_learning(
&self,
spec: &Spec,
handoff: i32,
mut rep: InsideReport,
) -> Result<(), ExecutorError> {
use heimdall_seccomp::launch::{prelude, ExecSelf, HANDOFF_FD};
use std::ffi::OsString;
let fail = |rep: &mut InsideReport, e: String| {
rep.error = Some(e.clone());
self.send(rep);
Err(ExecutorError::Other(e))
};
let process = spec.process().as_ref();
let args: Vec<String> = process.and_then(|p| p.args().clone()).unwrap_or_default();
let env: Vec<String> = process.and_then(|p| p.env().clone()).unwrap_or_default();
let program = match program_path(&args, &env) {
Ok(p) => p,
Err(e) => return fail(&mut rep, format!("learning: {e}")),
};
let Some(arch) = heimdall_seccomp::Arch::native() else {
return fail(&mut rep, "learning: this arch has no seccomp tables".into());
};
if let Err(e) = crate::sys::dup_cloexec_to(self.report_fd, REPORT_FD) {
return fail(&mut rep, format!("learning: report fd: {e}"));
}
if let Err(e) = crate::sys::dup_cloexec_to(handoff, HANDOFF_FD) {
return fail(&mut rep, format!("learning: hand-off fd: {e}"));
}
let argv: Vec<OsString> = args.iter().map(OsString::from).collect();
let envp: Vec<OsString> = env.iter().map(OsString::from).collect();
// The prelude's hand-off `sendmsg` (ExecSelf adds its `execveat`)
// and the report `write`: the calls between install and execveat.
let mut extra: Vec<heimdall_seccomp::Rule> =
prelude(true).into_iter().skip(1).collect();
extra.push(GeneratedPlan::report_rule());
let plan = match ExecSelf::new(
&program,
&argv,
&envp,
arch,
extra,
&heimdall_probe::facts::fixed_rules(),
heimdall_seccomp::Action::UserNotif,
) {
Ok(p) => p,
Err(e) => return fail(&mut rep, format!("learning: {e}")),
};
// Everything that allocates happens before the install.
let applied = line(&InsideReport {
learning: true,
..rep.clone()
});
let handoff_failed = line(&InsideReport {
error: Some("learning: the listener could not be handed over".into()),
..rep.clone()
});
let exec_failed = line(&InsideReport {
error: Some(
"execveat of the job's program failed under the learning filter".into(),
),
learning: true,
..rep.clone()
});
self.send(&rep);
let listener = match plan.install_listener() {
Ok(l) => l,
Err(e) => return fail(&mut rep, format!("learning: install: {e}")),
};
if crate::sys::send_fd_at(HANDOFF_FD, listener.as_fd()).is_err() {
let _ = crate::sys::write_fd(REPORT_FD, &handoff_failed);
crate::sys::exit_now(125)
}
let _ = crate::sys::write_fd(REPORT_FD, &applied);
// Close-on-exec; closing it here would itself be a learned call.
std::mem::forget(listener);
let _ = plan.exec();
let _ = crate::sys::write_fd(REPORT_FD, &exec_failed);
crate::sys::exit_now(127)
}
}
/// libcontainer's `DefaultExecutor::exec` (execvp of the spec's args), with
/// the `execve` retried while the program is busy ([`crate::exec`]).
fn exec_spec(spec: &Spec) -> Result<(), ExecutorError> {
use std::ffi::CString;
let args = spec
.process()
.as_ref()
.and_then(|p| p.args().as_ref())
.filter(|a| !a.is_empty())
.ok_or(ExecutorError::InvalidArg)?;
let a: Vec<CString> = args
.iter()
.map(|s| CString::new(s.as_bytes()))
.collect::<Result<_, _>>()
.map_err(|_| ExecutorError::InvalidArg)?;
let errno = crate::exec::retry_text_busy(crate::exec::TRIES, crate::exec::PAUSE, || {
match nix::unistd::execvp(&a[0], &a) {
Err(e) => e as i32,
Ok(_) => 0,
}
});
Err(ExecutorError::Execution(
format!(
"error '{}' executing '{:?}' with args '{:?}'",
nix::errno::Errno::from_raw(errno),
args[0],
args
)
.into(),
))
}
impl Executor for HeimdallExecutor {
fn exec(&self, spec: &Spec) -> Result<(), ExecutorError> {
let mut rep = InsideReport {
no_new_privs: crate::sys::no_new_privs(),
seccomp_mode: crate::sys::seccomp_mode(),
landlock: None,
error: None,
generated: None,
learning: false,
broker: None,
};
if self.learning.is_some() && self.learn_uid != 0 {
// Before Landlock, which leaves /proc read-only.
if let Err(e) = crate::sys::become_uid(self.learn_uid) {
rep.error = Some(format!("learning: uid {}: {e}", self.learn_uid));
self.send(&rep);
return Err(ExecutorError::Other(format!("uid: {e}")));
}
}
if let Some(plan) = &self.landlock {
match confine::apply(plan) {
Ok(a) => rep.landlock = Some(a),
Err(e) => {
rep.error = Some(format!("Landlock: {e}"));
self.send(&rep);
return Err(ExecutorError::Other(format!("Landlock: {e}")));
}
}
}
// A learning run never has a broker: its filter holds the one
// USER_NOTIF listener the job may have.
if let Some(h) = self.learning {
return self.exec_learning(spec, h, rep);
}
// The broker filter: after Landlock, before a generated profile
// (which allows no seccomp() of its own), so it covers the
// program from its first instruction.
if let Some((mode, sock)) = &self.broker {
let fm = if *mode == heimdall_broker::Mode::DenyOnly {
heimdall_broker::FilterMode::Deny
} else {
heimdall_broker::FilterMode::Notify
};
let handoff = sock.as_ref().map(|s| s.as_fd());
if let Err(e) = heimdall_broker::install_worker_filter(
fm,
heimdall_broker::ExecPolicy::Allow,
handoff,
) {
rep.error = Some(format!("broker filter: {e}"));
self.send(&rep);
return Err(ExecutorError::Other(format!("broker filter: {e}")));
}
rep.broker = Some(*mode);
}
if let Some(g) = &self.generated {
return self.exec_generated(spec, g, rep);
}
self.send(&rep);
let r = exec_spec(spec);
if let Err(e) = &r {
self.send(&InsideReport {
error: Some(format!("execve: {}", super::error_chain(e))),
..rep
});
}
r
}
fn validate(&self, spec: &Spec) -> Result<(), ExecutorValidationError> {
DefaultExecutor {}.validate(spec)
}
}
fn stdio(debug: bool, log: Option<&std::path::Path>) -> std::io::Result<(File, File, File)> {
let null = || File::options().read(true).write(true).open("/dev/null");
if let Some(log) = log {
// A learning run's log: a regular host file the program appends to.
use std::os::unix::fs::OpenOptionsExt;
let f = File::options()
.create(true)
.append(true)
.custom_flags(libc::O_NOFOLLOW)
.open(log)?;
if !f.metadata()?.is_file() {
return Err(std::io::Error::other("the log is not a regular file"));
}
return Ok((null()?, f.try_clone()?, f));
}
let err = if debug {
File::from(std::io::stderr().as_fd().try_clone_to_owned()?)
} else {
null()?
};
Ok((null()?, null()?, err))
}
/// Hand the learning listener and the program's pid to the runner.
fn hand_over(l: &LearnPlan, listener: &std::os::fd::OwnedFd, pid: i32) -> std::io::Result<()> {
let mut c = {
use std::os::linux::net::SocketAddrExt;
let addr = std::os::unix::net::SocketAddr::from_abstract_name(l.handoff.as_bytes())?;
UnixStream::connect_addr(&addr)?
};
heimdall_seccomp::notif::send_fd(c.as_fd(), listener.as_fd())?;
writeln!(c, "{pid}")
}
fn err(e: impl std::fmt::Display) -> ShimResponse {
ShimResponse {
error: Some(e.to_string()),
..Default::default()
}
}
pub(super) fn run(req: &ShimRequest) -> ShimResponse {
let debug = std::env::var_os("HEIMDALL_CONTAINER_DEBUG").is_some();
let learning = req.learning.as_ref();
if learning.is_some() {
// Before anything else: a stop signal must never kill the shim
// and leave the job behind.
crate::sys::catch_signals(&FORWARDED);
}
let (i, o, e) = match stdio(debug, learning.and_then(|l| l.log.as_deref())) {
Ok(x) => x,
Err(e) => return err(format!("stdio: {e}")),
};
if let Err(e) = std::fs::create_dir_all(&req.state_root) {
return err(format!("state root: {e}"));
}
let (report_r, report_w) = match crate::sys::pipe_cloexec() {
Ok(p) => p,
Err(e) => return err(format!("report pipe: {e}")),
};
let (handoff_shim, handoff_init) = match learning {
Some(_) => match UnixStream::pair() {
Ok((a, b)) => (Some(a), Some(b)),
Err(e) => return err(format!("hand-off socket: {e}")),
},
None => (None, None),
};
// The broker: the listener comes over a socket pair from the job's
// init right before its execve; a thread serves it until the job is
// gone (deny-only needs no broker process).
// Probe what the kernel gives the broker now, before the container's
// processes exist (the probe forks a short-lived child, which no
// other waiter may reap first), and refuse a mode it cannot run.
if let Some(b) = &req.broker {
let support = heimdall_broker::Support::probe();
if !support.allows(b.mode) {
return err(format!(
"the broker cannot run in {:?} here ({support:?})",
b.mode
));
}
}
let mut broker_sock: Option<OwnedFd> = None;
let mut worker_sock: Option<Arc<OwnedFd>> = None;
if let Some(b) = &req.broker {
if b.mode != heimdall_broker::Mode::DenyOnly {
match heimdall_broker::handoff_pair() {
Ok((p, c)) => {
broker_sock = Some(p);
worker_sock = Some(Arc::new(c));
}
Err(e) => return err(format!("broker socket: {e}")),
}
}
}
let executor = HeimdallExecutor {
report_fd: report_w.as_raw_fd(),
landlock: req.landlock.clone(),
generated: req.generated.clone(),
learning: handoff_init.as_ref().map(|s| s.as_raw_fd()),
learn_uid: learning.map_or(0, |l| l.uid),
broker: req.broker.as_ref().map(|b| (b.mode, worker_sock.clone())),
};
// The initramfs root cannot be pivoted out of (EINVAL): MS_MOVE + chroot there, said aloud.
let no_pivot = heimdall_plain::rootfs::no_pivot();
if no_pivot {
eprintln!(
"heimdall-container-shim: {}",
heimdall_plain::rootfs::NO_PIVOT_NOTE
);
}
let builder = match ContainerBuilder::new(req.id.clone(), SyscallType::default())
.with_executor(executor)
.with_root_path(req.state_root.clone())
{
Ok(b) => b,
Err(e) => return err(format!("container root: {e}")),
};
let built = builder
.with_stdin(i)
.with_stdout(o)
.with_stderr(e)
.as_init(&req.bundle)
// The initramfs root cannot be pivoted out of (EINVAL): MS_MOVE + chroot there.
.with_no_pivot(no_pivot)
.with_systemd(false)
.with_detach(true)
.build();
// Only the job's init holds the write end now (until its execve).
drop(report_w);
drop(handoff_init);
drop(worker_sock);
let mut container = match built {
Ok(c) => c,
Err(e) => return err(format!("create: {e}")),
};
let Some(pid) = container.pid() else {
let _ = container.delete(true);
return err("create: no init pid");
};
let mut resp = ShimResponse {
no_pivot,
..Default::default()
};
let fail = |container: &mut libcontainer::container::Container, why: String| {
let _ = container.kill(Signal::SIGKILL, true);
let _ = waitpid(pid, None);
let _ = container.delete(true);
err(why)
};
// No swap beyond the memory limit. libcgroups 0.7.0 skips writing a
// computed `memory.swap.max` of 0 (OCI swap == limit), which leaves
// swap unlimited; write it here, before the job starts, and read it
// back for the runner.
let swap = req.cgroup_dir.join("memory.swap.max");
if swap.exists() {
if let Err(e) = std::fs::write(&swap, "0") {
return fail(&mut container, format!("memory.swap.max: {e}"));
}
resp.swap_max = std::fs::read_to_string(&swap)
.ok()
.map(|s| s.trim().to_string());
}
// io.weight, written here: libcgroups 0.7.0 overflows u16 converting
// an OCI blockIO.weight. Read back for the runner.
if let Some(w) = req.io_weight {
let f = req.cgroup_dir.join("io.weight");
if let Err(e) = std::fs::write(&f, format!("default {w}")) {
return fail(&mut container, format!("io.weight: {e}"));
}
resp.io_weight = std::fs::read_to_string(&f)
.ok()
.and_then(|s| s.lines().next().map(|l| l.trim().to_string()));
}
// The io limits as the kernel took them (libcgroups wrote them from
// the config's blockIO throttles); the runner checks them.
let io_max = req.cgroup_dir.join("io.max");
if io_max.exists() {
resp.io_max = std::fs::read_to_string(&io_max).ok();
}
let stop = match heimdall_broker::StopSignal::new() {
Ok(s) => Arc::new(s),
Err(e) => return fail(&mut container, format!("broker stop signal: {e}")),
};
// The network observer's events: JSON lines on stdout, written by a
// thread of their own from a bounded queue the broker never waits on.
let events = match &req.broker {
Some(b) if b.observe => Some(super::event_writer(b.policy.event_job_id())),
_ => None,
};
let broker_thread = match (&req.broker, broker_sock) {
(Some(b), Some(sock)) => {
let (mode, policy, stop) = (b.mode, b.policy.clone(), Arc::clone(&stop));
let pid = pid.as_raw() as u32;
let sink = events.as_ref().map(|(s, _, _)| s.clone());
Some(std::thread::spawn(move || {
heimdall_broker::serve_handoff(
sock.as_fd(),
stop.raw(),
Some(pid),
policy,
mode,
heimdall_broker::DebugKnobs::default(),
sink,
)
}))
}
_ => None,
};
let t0 = Instant::now();
if let Err(e) = container.start() {
stop.signal();
return fail(&mut container, format!("start: {e}"));
}
if let (Some(l), Some(sock)) = (learning, &handoff_shim) {
// The executor sends the listener right after the install, or
// its end closes (execve, or it failed): never wait forever.
let _ = sock.set_read_timeout(Some(Duration::from_secs(30)));
match heimdall_seccomp::notif::recv_fd(sock.as_fd()) {
Ok(listener) => {
if let Err(e) = hand_over(l, &listener, pid.as_raw()) {
return fail(&mut container, format!("handing the listener over: {e}"));
}
// Only the runner holds it now: when it goes, the job's
// learned calls fail instead of hanging.
drop(listener);
}
Err(e) => {
if debug {
eprintln!("heimdall-container-shim: no listener from the executor: {e}");
}
}
}
}
let grace = Duration::from_millis(learning.map_or(0, |l| l.term_grace_ms));
let mut stop_at: Option<Instant> = None;
let deadline = t0 + Duration::from_millis(req.wall_ms);
let mut next_ws = t0;
let status = loop {
match waitpid(pid, Some(WaitPidFlag::WNOHANG)) {
Ok(WaitStatus::StillAlive) => {
if learning.is_some() {
let pending = crate::sys::take_signals();
for sig in FORWARDED {
if pending & (1u64 << sig) == 0 {
continue;
}
if let Ok(s) = Signal::try_from(sig) {
let _ = nix::sys::signal::kill(pid, s);
}
if STOP_SIGNALS.contains(&sig) && stop_at.is_none() {
stop_at = Some(Instant::now());
}
}
if stop_at.is_some_and(|t| t.elapsed() >= grace) {
resp.term_killed = true;
let _ = container.kill(Signal::SIGKILL, true);
break waitpid(pid, None);
}
}
if Instant::now() >= deadline {
resp.wall_killed = true;
let _ = container.kill(Signal::SIGKILL, true);
break waitpid(pid, None);
}
// A host-folder workspace past its cap: measured every
// 200 ms (a tmpfs is capped by the kernel itself).
if let Some(w) = &req.workspace {
if next_ws <= Instant::now() {
next_ws = Instant::now() + Duration::from_millis(200);
let used = crate::workspace::usage(&w.dir);
resp.workspace_bytes = resp.workspace_bytes.max(used);
if used > w.cap {
resp.workspace_killed = true;
let _ = container.kill(Signal::SIGKILL, true);
break waitpid(pid, None);
}
}
}
std::thread::sleep(Duration::from_millis(5));
}
other => break other,
}
};
resp.ran_ms = t0.elapsed().as_millis() as u64;
// The last measure: a job that filled its workspace and exited
// between two looks is still over its cap.
if let Some(w) = &req.workspace {
let used = crate::workspace::usage(&w.dir);
resp.workspace_bytes = resp.workspace_bytes.max(used);
if used > w.cap {
resp.workspace_killed = true;
}
}
match status {
Ok(WaitStatus::Exited(_, c)) => resp.exit_code = Some(c),
Ok(WaitStatus::Signaled(_, s, _)) => resp.signal = Some(s as i32),
Ok(other) => resp.error = Some(format!("unexpected wait status {other:?}")),
Err(e) => resp.error = Some(format!("wait: {e}")),
}
let read = |f: &str| std::fs::read_to_string(req.cgroup_dir.join(f)).unwrap_or_default();
resp.oom_kills = events_count(&read("memory.events"), "oom_kill");
resp.pids_max = events_count(&read("pids.events"), "max");
if let Err(e) = container.delete(true) {
if debug {
eprintln!("heimdall-container-shim: delete: {e}");
}
}
stop.signal();
if let Some(h) = broker_thread {
match h.join() {
Ok(Ok(r)) => resp.broker = r,
Ok(Err(e)) => resp.error = Some(format!("broker: {e}")),
Err(_) => resp.error = Some("broker: the broker thread panicked".into()),
}
}
if let Some((sink, stop, h)) = events {
drop(sink);
stop.store(true, std::sync::atomic::Ordering::Release);
let _ = h.join();
}
// The init's copy closed at execve (or when it died): read what it
// wrote, without blocking on anything else that might hold it.
let mut buf = Vec::new();
let _ = crate::sys::set_nonblocking(report_r.as_raw_fd());
let _ = File::from(report_r).take(64 * 1024).read_to_end(&mut buf);
resp.inside = parse_inside(&String::from_utf8_lossy(&buf));
resp
}
}
/// The shim's event writer: a sink for the broker and a thread writing its
/// events, one JSON line each, to stdout (the runner relays them to its
/// observer). It stops when `stop` is set and the queue is empty.
#[cfg(feature = "youki")]
fn event_writer(
job_id: String,
) -> (
heimdall_broker::EventSink,
std::sync::Arc<std::sync::atomic::AtomicBool>,
std::thread::JoinHandle<()>,
) {
use heimdall_broker::observe::RecvError;
use std::io::Write;
let (sink, rx) =
heimdall_broker::EventSink::channel(job_id, heimdall_broker::observe::DEFAULT_QUEUE);
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let st = std::sync::Arc::clone(&stop);
let h = std::thread::spawn(move || loop {
match rx.recv_timeout(std::time::Duration::from_millis(50)) {
Ok(ev) => {
let mut out = std::io::stdout().lock();
let _ = writeln!(out, "{}", ev.to_json_line());
let _ = out.flush();
}
Err(RecvError::Timeout) if st.load(std::sync::atomic::Ordering::Acquire) => break,
Err(RecvError::Timeout) => {}
Err(RecvError::Closed) => break,
}
});
(sink, stop, h)
}
/// `e` and every source below it, joined by ": " (libcontainer's
/// `ExecutorError::Execution` shows only "failed to execute workload"; the
/// errno is in its source).
#[cfg(feature = "youki")]
fn error_chain(e: &dyn std::error::Error) -> String {
let mut s = e.to_string();
let mut cur = e.source();
while let Some(c) = cur {
s.push_str(": ");
s.push_str(&c.to_string());
cur = c.source();
}
s
}