crates/heimdall-probe/src/live.rs
// SPDX-License-Identifier: AGPL-3.0-only
// Copyright (C) 2026 Ignalina ApS and Vetra Automation AB
//! **The live probe** (`heimdall profile probe-exec`): learning mode around a
//! program that runs where it really runs — a daemon in a booted machine, a
//! plugin its host executes — instead of over a corpus. Heimdall is a
//! defensive sandbox: the program sees its own stdio, environment, working
//! directory and argv[0]; the probe only listens (seccomp USER_NOTIF, the same
//! learning filter as [`crate::learn`]) and adds what it saw to an
//! `<arch>-<libc>.probe.json`-shaped file, merged under a lock with whatever
//! other runs (other invocations of the same program) wrote there. The file is
//! rewritten whenever a new call shows up (at most every `flush`), so the
//! evidence is current while a daemon still runs. Signals the probe gets
//! (TERM, INT, HUP, QUIT, USR1, USR2) go on to the program; the probe ends
//! with the program's own exit.
use std::ffi::OsString;
use std::path::PathBuf;
use std::time::Duration;
use heimdall_seccomp::launch::Exit;
use heimdall_seccomp::Libc;
use crate::facts::fixed_rules;
use crate::learn::call_of;
use crate::observe::{Observations, RunInfo};
/// One live probe.
#[derive(Clone, Debug)]
pub struct LiveRun {
/// The workload the evidence belongs to (the file's `workload`).
pub workload: String,
/// The run name every call is recorded under.
pub run: String,
pub error_path: bool,
pub program: PathBuf,
pub arg0: Option<OsString>,
pub args: Vec<OsString>,
/// Added to the inherited environment.
pub env: Vec<(OsString, OsString)>,
pub libc: Libc,
/// The evidence file (merged into; created when missing).
pub out: PathBuf,
/// The shortest time between two rewrites of `out` while the program runs.
pub flush: Duration,
/// Record only the program's own process (its threads share the tid
/// space, so: calls whose tid is the program's pid, or a thread of it),
/// not the processes it starts — e.g. a container runtime's shim, whose
/// children are other workloads with their own profiles.
pub self_only: bool,
}
/// The pid signals go on to (0: none yet).
static CHILD: std::sync::atomic::AtomicI32 = std::sync::atomic::AtomicI32::new(0);
extern "C" fn forward(sig: libc::c_int) {
let pid = CHILD.load(std::sync::atomic::Ordering::SeqCst);
if pid > 0 {
// SAFETY: kill(2) is async-signal-safe.
unsafe { libc::kill(pid, sig) };
}
}
/// The signals the probe passes on to the program.
const FORWARDED: [libc::c_int; 6] = [
libc::SIGTERM,
libc::SIGINT,
libc::SIGHUP,
libc::SIGQUIT,
libc::SIGUSR1,
libc::SIGUSR2,
];
/// Where a live probe's evidence goes: the file, and what this run is called
/// in it. Shared by `probe-exec` ([`learn_live`]) and `probe-container`
/// (heimdall-container's learning runs, which hand over only a listener).
#[derive(Clone, Debug)]
pub struct Evidence {
pub workload: String,
pub run: String,
pub error_path: bool,
/// The evidence file (merged into; created when missing).
pub out: PathBuf,
/// The shortest time between two rewrites of `out` while the program runs.
pub flush: Duration,
}
impl LiveRun {
/// This run's [`Evidence`].
pub fn evidence(&self) -> Evidence {
Evidence {
workload: self.workload.clone(),
run: self.run.clone(),
error_path: self.error_path,
out: self.out.clone(),
flush: self.flush,
}
}
}
/// Record every notification on `listener` into `obs` (calls whose thread
/// `keep` refuses are let through unrecorded), answering each with
/// `CONTINUE`, and rewrite `ev.out` (marked "running") whenever a new call
/// showed up, at most every `ev.flush`. Returns when every task using the
/// filter is gone (hangup), or when `done` is set and no notification came
/// for half a second. The final rewrite, with the run's exit, is the
/// caller's ([`Evidence::write_merged`]).
pub fn record(
listener: &heimdall_seccomp::notif::Listener,
ev: &Evidence,
obs: &mut Observations,
keep: &dyn Fn(u32) -> bool,
done: &std::sync::atomic::AtomicBool,
) -> std::io::Result<()> {
use heimdall_seccomp::notif::{Readiness, Response};
use std::time::Instant;
let arch = obs.arch;
let mut dirty = false;
let mut last = Instant::now();
loop {
match listener.wait(Some(Duration::from_millis(500))) {
Ok(Readiness::Hangup) => return Ok(()),
Ok(Readiness::Timeout) => {
if done.load(std::sync::atomic::Ordering::SeqCst) {
return Ok(());
}
}
Ok(Readiness::Ready) => match listener.recv() {
Ok(n) if !keep(n.pid) => match listener.respond(n.id, Response::Continue) {
Ok(()) => {}
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => {}
Err(e) => return Err(e),
},
Ok(n) => {
match call_of(arch, &n) {
Ok(c) => {
if !obs.calls.contains_key(&c) {
dirty = true;
}
obs.record(c, &ev.run);
}
Err(u) => {
dirty |= obs.unknown.entry(u).or_default().insert(ev.run.clone());
}
}
match listener.respond(n.id, Response::Continue) {
Ok(()) => {}
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => {}
Err(e) => return Err(e),
}
}
Err(e) if e.raw_os_error() == Some(libc::ENOENT) => {}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
},
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
}
if dirty && last.elapsed() >= ev.flush {
// The evidence so far, while the program still runs.
ev.write_merged(obs, "running")?;
dirty = false;
last = Instant::now();
}
}
}
/// Run `r` under the learning filter; returns the program's exit.
pub fn learn_live(r: &LiveRun) -> std::io::Result<Exit> {
use heimdall_seccomp::launch::Launch;
use heimdall_seccomp::{Action, Arch};
let arch = Arch::native().ok_or_else(|| std::io::Error::other("this arch is not supported"))?;
let mut obs = Observations::new(&r.workload, arch, r.libc);
let mut l = Launch::new(&r.program)
.args(&r.args)
.rules(arch, fixed_rules(), Action::UserNotif)
.listener(true);
if let Some(a0) = &r.arg0 {
l = l.arg0(a0);
}
for (k, v) in &r.env {
l = l.env(k, v);
}
let mut child = l.spawn()?;
CHILD.store(child.pid(), std::sync::atomic::Ordering::SeqCst);
for sig in FORWARDED {
// SAFETY: a handler that only calls kill(2); installed after the spawn,
// so the program starts with its own (default) dispositions.
unsafe { libc::signal(sig, forward as *const () as libc::sighandler_t) };
}
let listener = child
.take_listener()
.ok_or_else(|| std::io::Error::other("no listener"))?;
let ev = r.evidence();
let pid = child.pid();
let self_only = r.self_only;
let keep = move |tid: u32| !self_only || own(pid, tid);
let never = std::sync::atomic::AtomicBool::new(false);
let result = record(&listener, &ev, &mut obs, &keep, &never);
let exit = child.wait();
CHILD.store(0, std::sync::atomic::Ordering::SeqCst);
let exit = exit?;
ev.write_merged(&obs, &exit.to_string())?;
result.map(|()| exit)
}
/// Whether thread `tid` belongs to process `pid` (its thread group).
fn own(pid: libc::pid_t, tid: u32) -> bool {
if tid as libc::pid_t == pid {
return true;
}
std::fs::read_to_string(format!("/proc/{tid}/status"))
.ok()
.and_then(|s| {
s.lines().find_map(|l| {
l.strip_prefix("Tgid:")
.map(|v| v.trim().parse::<libc::pid_t>().ok())
})
})
.flatten()
== Some(pid)
}
impl Evidence {
/// Merge `obs` (and this run, ended `how`) into the evidence file, under
/// a lock beside it; a "running" record of this run is replaced.
pub fn write_merged(&self, obs: &Observations, how: &str) -> std::io::Result<()> {
use std::os::fd::AsRawFd as _;
let lock_path = self.out.with_extension("json.lock");
let lock = std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.open(&lock_path)?;
// SAFETY: flock on a file this function owns; released when it is closed.
if unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX) } != 0 {
return Err(std::io::Error::last_os_error());
}
let mut all = match std::fs::read_to_string(&self.out) {
Ok(t) => Observations::from_json(&t)
.map_err(|e| std::io::Error::other(format!("{}: {e}", self.out.display())))?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
Observations::new(&self.workload, obs.arch, obs.libc)
}
Err(e) => return Err(e),
};
if (all.arch, all.libc) != (obs.arch, obs.libc) {
return Err(std::io::Error::other(format!(
"{}: evidence of {} {}, this run is {} {}",
self.out.display(),
all.arch,
all.libc.as_str(),
obs.arch,
obs.libc.as_str()
)));
}
all.merge(obs);
all.runs
.retain(|x| !(x.name == self.run && x.exit == "running"));
all.runs.insert(RunInfo {
name: self.run.clone(),
error_path: self.error_path,
exit: how.to_string(),
});
let tmp = self
.out
.with_extension(format!("json.{}.tmp", std::process::id()));
std::fs::write(&tmp, all.to_json())?;
std::fs::rename(&tmp, &self.out)
}
}