server/sandbox/src/remote.rs
//! **The `Remote` engine: the scan worker as a SIDECAR** (lane B4, A31 §7.3).
//!
//! A hardened Kubernetes pod holds no container runtime, so neither podman nor
//! crun can start the worker there. Instead the worker runs in its own
//! container beside the server (a sidecar, its own image, its own limits), and
//! the two share one directory (an `emptyDir`) and nothing else:
//!
//! 1. The server writes the job directory (`artifact`, `job.json`, and a copy of
//! every extra read-only input such as the YARA rules) into the shared
//! directory (`work_dir`), exactly where the podman engine would have
//! mounted them from.
//! 2. It asks the sidecar over a unix socket in that directory (`socket`):
//! ONE JSON line, an [`Ask`].
//! 3. The sidecar ([`serve`], `holger-scan-worker serve`) runs the worker over
//! that directory in a fresh subprocess and answers ONE JSON line, an
//! [`Answer`], followed by exactly `report_len` bytes of the report.
//!
//! | What | Where it holds |
//! |---|---|
//! | No network | the sidecar's image holds no client, it listens only on the unix socket, and the pod's NetworkPolicy denies egress |
//! | The artifact, read-only | written `0444` by the server; the sidecar's root filesystem is read-only |
//! | No access to the data | the sidecar mounts the shared directory only, never the PVC |
//! | CPU, memory, files | `RLIMIT_CPU`, `RLIMIT_AS`, `RLIMIT_FSIZE`, `RLIMIT_NOFILE` per job, under the sidecar's own ceiling, and the container's own limits around all jobs |
//! | Processes (pids) | a cgroup per job with `pids.max` = the job's `pids` when the sidecar holds a delegated cgroup v2, the job sealed in its own cgroup namespace so it can neither leave its cgroup nor raise its bound (proved once by a probe job when the sidecar starts); else the container's own `pids.max`, shared by every job; with neither, a release build refuses every ping and job by name (`holger-sandbox-13`) and a dev build warns and runs. The ping names which ([`crate::pids`], RA5-27). Never `RLIMIT_NPROC`: it counts the uid's processes node-wide |
//! | A wall-clock timeout | the job is killed at the deadline: [`Refusal::Timeout`] |
//! | A capped report | more than `report_max_bytes` kills the job: [`Refusal::ReportTooLarge`] |
//! | No privilege gain, no core | `PR_SET_NO_NEW_PRIVS`, `RLIMIT_CORE` 0, its own session |
//! | A scratch space | a fresh directory per job (`HOME`, `TMPDIR`), removed after |
//! | Trusts nothing it is told | the job directory must be one segment under the share; the only command is the worker's own `probe` |
//!
//! Fail closed: a sidecar that does not answer is [`Refusal::NoEngine`], and
//! readiness asks it again every time (a pod's containers start together, so the
//! server may be up before its sidecar).
use std::io::{BufRead as _, BufReader, Read as _, Write as _};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize};
use crate::pids::PidsBound;
use crate::{
command_of, drive, mounts_of, valid_segment, Drive, Engine, Limits, Output, Refusal,
SandboxConfig, ARTIFACT_PATH, EXTRA_DIR, JOB_PATH,
};
/// The wire's schema, on every ask and answer.
pub const SCHEMA: &str = "holger.scan-remote/v1";
/// Where the sidecar listens by default: in the shared directory.
pub const DEFAULT_SOCKET: &str = "/run/holger-scan/worker.sock";
/// The shared directory by default (the pod's `emptyDir`).
pub const DEFAULT_SHARE: &str = "/run/holger-scan";
/// The longest ask or answer line.
const LINE_MAX: u64 = 64 << 10;
/// How long past the job's own wall clock the server waits for the answer.
const GRACE: Duration = Duration::from_secs(10);
/// How long a ping may take.
const PING: Duration = Duration::from_secs(5);
// ── the wire ────────────────────────────────────────────────────────────────
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Op {
/// Is the sidecar there, with its worker?
Ping,
/// Run the worker over one job directory.
Run,
}
/// What the server asks.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Ask {
pub schema: String,
pub op: Op,
/// The job directory's name in the shared directory: one path segment.
#[serde(default)]
pub dir: String,
/// The worker's argv: empty (a scan) or `probe …`.
#[serde(default)]
pub command: Vec<String>,
/// The limits the server asks for; the sidecar's ceiling bounds them.
#[serde(default)]
pub limits: Limits,
}
/// Why the sidecar did not hand back a report.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum Refused {
Timeout {
secs: u64,
},
Killed {
exit: Option<i32>,
stderr: String,
},
TooLarge {
cap: u64,
},
WorkerFailed {
exit: Option<i32>,
stderr: String,
},
Config {
why: String,
},
Io {
why: String,
},
/// Every job slot is taken, or the worker program is not there.
Unavailable {
why: String,
},
/// Nothing bounds a job's processes, and this is a release build (RA5-27).
NoPidsLimit {
why: String,
},
}
impl Refused {
/// The server's refusal, by the same code a local engine would give.
pub fn into_refusal(self) -> Refusal {
match self {
Refused::Timeout { secs } => Refusal::Timeout { secs },
Refused::Killed { exit, stderr } => Refusal::Killed {
exit,
oom: false,
stderr,
},
Refused::TooLarge { cap } => Refusal::ReportTooLarge { cap },
Refused::WorkerFailed { exit, stderr } => Refusal::WorkerFailed { exit, stderr },
Refused::Config { why } => {
Refusal::Config(format!("the scan sidecar refused the job: {why}"))
}
Refused::Io { why } => Refusal::Io(format!("the scan sidecar: {why}")),
Refused::Unavailable { why } => Refusal::NoEngine(format!("the scan sidecar: {why}")),
Refused::NoPidsLimit { why } => Refusal::NoPidsLimit(why),
}
}
}
/// What the sidecar answers; `report_len` bytes of report follow the line.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Answer {
pub schema: String,
#[serde(default)]
pub refused: Option<Refused>,
#[serde(default)]
pub report_len: u64,
#[serde(default)]
pub elapsed_ms: u64,
/// The sidecar's words for itself (on a ping).
#[serde(default)]
pub worker: String,
}
impl Answer {
fn refused(r: Refused) -> Self {
Self {
schema: SCHEMA.into(),
refused: Some(r),
report_len: 0,
elapsed_ms: 0,
worker: String::new(),
}
}
}
// ── the server's side: the engine ───────────────────────────────────────────
/// Ask the sidecar at `socket` once: the answer and the report's bytes (at most
/// `cap`). Waits at most `wait` for the answer.
pub fn ask(
socket: &Path,
ask: &Ask,
wait: Duration,
cap: u64,
) -> Result<(Answer, Vec<u8>), Refusal> {
let gone = |e: std::io::Error| {
Refusal::NoEngine(format!("the scan sidecar at {}: {e}", socket.display()))
};
let mut s = UnixStream::connect(socket).map_err(gone)?;
s.set_read_timeout(Some(wait)).map_err(gone)?;
s.set_write_timeout(Some(PING)).map_err(gone)?;
let mut line = serde_json::to_vec(ask).map_err(|e| Refusal::Io(format!("the ask: {e}")))?;
line.push(b'\n');
s.write_all(&line).map_err(gone)?;
let _ = s.shutdown(std::net::Shutdown::Write);
let mut r = BufReader::new(s);
let mut head = Vec::new();
let timed_out = |e: &std::io::Error| {
matches!(
e.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
)
};
if let Err(e) = (&mut r).take(LINE_MAX).read_until(b'\n', &mut head) {
return Err(if timed_out(&e) {
Refusal::Timeout {
secs: ask_secs(ask, wait),
}
} else {
gone(e)
});
}
if !head.ends_with(b"\n") {
return Err(Refusal::NoEngine(format!(
"the scan sidecar at {} hung up without an answer",
socket.display()
)));
}
let a: Answer = serde_json::from_slice(&head)
.map_err(|e| Refusal::ReportInvalid(format!("the scan sidecar's answer: {e}")))?;
if a.schema != SCHEMA {
return Err(Refusal::ReportInvalid(format!(
"the scan sidecar answers {:?}, not {SCHEMA}",
a.schema
)));
}
if let Some(refused) = a.refused {
return Err(refused.into_refusal());
}
if a.report_len > cap {
return Err(Refusal::ReportTooLarge { cap });
}
let mut report = Vec::with_capacity(a.report_len as usize);
if let Err(e) = (&mut r).take(a.report_len).read_to_end(&mut report) {
return Err(if timed_out(&e) {
Refusal::Timeout {
secs: ask_secs(ask, wait),
}
} else {
gone(e)
});
}
if report.len() as u64 != a.report_len {
return Err(Refusal::ReportInvalid(format!(
"the report was cut short: {} of {} bytes",
report.len(),
a.report_len
)));
}
Ok((a, report))
}
fn ask_secs(ask: &Ask, wait: Duration) -> u64 {
if ask.op == Op::Run {
ask.limits.timeout_secs
} else {
wait.as_secs()
}
}
/// **The engine**: the server's half of the sidecar.
#[derive(Debug, Clone)]
pub struct Remote {
socket: PathBuf,
}
impl Remote {
pub fn new(cfg: &SandboxConfig) -> Self {
Self {
socket: PathBuf::from(cfg.socket.as_deref().unwrap_or(DEFAULT_SOCKET)),
}
}
}
impl Engine for Remote {
fn ready(&self, _cfg: &SandboxConfig) -> Result<(), Refusal> {
let ping = Ask {
schema: SCHEMA.into(),
op: Op::Ping,
dir: String::new(),
command: Vec::new(),
limits: Limits::default(),
};
ask(&self.socket, &ping, PING, 0).map(|_| ())
}
fn run(&self, cfg: &SandboxConfig, _name: &str, args: &[String]) -> Result<Output, Refusal> {
let started = Instant::now();
let share = cfg.work_dir.as_deref().ok_or_else(|| {
Refusal::Config(
"engine `remote` needs `work_dir`: the directory shared with the sidecar".into(),
)
})?;
let mounts = mounts_of(args);
let artifact = mounts
.iter()
.find(|(_, c)| c == ARTIFACT_PATH)
.map(|(h, _)| PathBuf::from(h))
.ok_or_else(|| Refusal::Io("the job has no artifact".into()))?;
let job_dir = artifact
.parent()
.ok_or_else(|| Refusal::Io("the artifact has no job directory".into()))?;
let dir = job_dir
.strip_prefix(share)
.ok()
.and_then(|d| d.to_str())
.filter(|d| valid_segment(d))
.ok_or_else(|| {
Refusal::Config(format!(
"the job directory {} is not one segment under the shared directory {share}",
job_dir.display()
))
})?
.to_string();
// Every other read-only input is copied in: the sidecar sees the shared
// directory only, never the server's own paths.
for (host, container) in &mounts {
if container == ARTIFACT_PATH || container == JOB_PATH {
continue;
}
let name = container
.strip_prefix(EXTRA_DIR)
.and_then(|n| n.strip_prefix('/'))
.filter(|n| valid_segment(n))
.ok_or_else(|| {
Refusal::Config(format!(
"mount {container:?} is not one segment under {EXTRA_DIR}"
))
})?;
copy_ro(Path::new(host), &job_dir.join(name))
.map_err(|e| Refusal::Io(format!("copying {host} into the job: {e}")))?;
}
let a = Ask {
schema: SCHEMA.into(),
op: Op::Run,
dir,
command: command_of(cfg, args),
limits: cfg.limits.clone(),
};
let (_, report) = ask(
&self.socket,
&a,
Duration::from_secs(cfg.limits.timeout_secs) + GRACE,
cfg.limits.report_max_bytes,
)?;
Ok(Output {
report,
elapsed: started.elapsed(),
})
}
}
/// Copy a file or a tree, readable by all and writable by none. Kubernetes'
/// own `..data` links of a ConfigMap volume are skipped (their content is
/// reached through the plain names).
fn copy_ro(from: &Path, to: &Path) -> std::io::Result<()> {
use std::os::unix::fs::PermissionsExt as _;
let meta = std::fs::metadata(from)?;
if meta.is_dir() {
std::fs::create_dir(to)?;
for e in std::fs::read_dir(from)? {
let e = e?;
if e.file_name().to_string_lossy().starts_with("..") {
continue;
}
copy_ro(&e.path(), &to.join(e.file_name()))?;
}
// Owner-writable, so the job directory is removed after the scan.
std::fs::set_permissions(to, std::fs::Permissions::from_mode(0o755))
} else {
std::fs::copy(from, to)?;
std::fs::set_permissions(to, std::fs::Permissions::from_mode(0o444))
}
}
// ── the sidecar's side ──────────────────────────────────────────────────────
/// The sidecar (`holger-scan-worker serve`).
#[derive(Debug, Clone)]
pub struct ServeConfig {
/// The unix socket it listens on (in the shared directory).
pub socket: PathBuf,
/// The shared directory, as the sidecar sees it.
pub share: PathBuf,
/// The worker program each job runs (the sidecar's own binary).
pub worker: PathBuf,
/// Where each job's scratch directory is made.
pub scratch: PathBuf,
/// The most any job gets, whatever it asks.
pub ceiling: Limits,
/// Jobs at once; one more is refused as unavailable.
pub max_jobs: usize,
/// The cgroup mount the sidecar bounds its jobs' processes under (RA5-27,
/// [`crate::pids`]): `/sys/fs/cgroup`. `None`: no cgroup is read, so nothing bounds them.
pub cgroup_fs: Option<PathBuf>,
/// What a sidecar with no pids bound does: refuse every ping and job by name (`true`,
/// a release build) or warn and run (`false`, a dev build).
pub refuse_unbounded: bool,
/// The pod's pids limit as the operator states it (`--pod-pids-limit`): the kubelet's
/// `podPidsLimit` is on the pod's cgroup, out of the sidecar's sight. Taken only when
/// neither a delegated cgroup nor the container's own `pids.max` bounds the jobs.
pub pod_pids_limit: Option<u64>,
}
impl ServeConfig {
/// The sidecar's defaults: the shared directory and socket of the chart,
/// `/tmp` for scratch, this program as the worker.
pub fn defaults(worker: PathBuf) -> Self {
Self {
socket: PathBuf::from(DEFAULT_SOCKET),
share: PathBuf::from(DEFAULT_SHARE),
worker,
scratch: std::env::temp_dir(),
ceiling: Limits::default(),
max_jobs: 2,
cgroup_fs: Some(PathBuf::from(crate::pids::CGROUP_FS)),
refuse_unbounded: !cfg!(debug_assertions),
pod_pids_limit: None,
}
}
}
/// Bind the socket (a stale one from an earlier run is replaced), readable and
/// writable by the owner and the group the pod shares (`fsGroup`).
pub fn bind(cfg: &ServeConfig) -> std::io::Result<UnixListener> {
use std::os::unix::fs::{FileTypeExt as _, PermissionsExt as _};
if let Ok(m) = std::fs::symlink_metadata(&cfg.socket) {
if m.file_type().is_socket() {
std::fs::remove_file(&cfg.socket)?;
}
}
let l = UnixListener::bind(&cfg.socket)?;
std::fs::set_permissions(&cfg.socket, std::fs::Permissions::from_mode(0o660))?;
Ok(l)
}
/// Answer asks until the listener fails. Each connection is one ask. The pids bound is
/// decided first, from `cfg.cgroup_fs` and `/proc/self/cgroup` ([`crate::pids::establish`]).
pub fn serve(listener: UnixListener, cfg: ServeConfig) -> std::io::Result<()> {
let bound = match &cfg.cgroup_fs {
Some(fs) => crate::pids::establish(
fs,
&std::fs::read_to_string("/proc/self/cgroup").unwrap_or_default(),
),
None => PidsBound::Unbounded,
};
let bound = match (bound, cfg.pod_pids_limit) {
(PidsBound::Unbounded, Some(max)) if max > 0 => PidsBound::Pod { max },
(bound, _) => bound,
};
if bound == PidsBound::Unbounded {
match cfg.refuse_unbounded {
true => eprintln!("holger-scan-worker: [holger-sandbox-13] {}; every ping and job is refused", crate::pids::UNBOUNDED_FIX),
false => eprintln!("holger-scan-worker: WARNING: {}; a dev build runs the jobs anyway, a release build refuses them", crate::pids::UNBOUNDED_FIX),
}
}
serve_bounded(listener, cfg, bound)
}
/// [`serve`] under a bound already decided.
pub fn serve_bounded(
listener: UnixListener,
cfg: ServeConfig,
bound: PidsBound,
) -> std::io::Result<()> {
let cfg = Arc::new((cfg, bound));
let running = Arc::new(AtomicUsize::new(0));
for conn in listener.incoming() {
let Ok(stream) = conn else { continue };
let (cfg, running) = (Arc::clone(&cfg), Arc::clone(&running));
// A32-14, stays a thread: the sidecar's accept loop, one job per connection on its own thread; the sidecar is the scan worker's own process (no server, no jera there), and `running` caps the jobs.
std::thread::spawn(move || handle(stream, &cfg.0, &cfg.1, &running));
}
Ok(())
}
/// A release sidecar with nothing bounding a job's processes refuses (RA5-27).
fn unbounded_refusal(cfg: &ServeConfig, bound: &PidsBound) -> Option<Refused> {
(cfg.refuse_unbounded && *bound == PidsBound::Unbounded).then(|| Refused::NoPidsLimit {
why: crate::pids::UNBOUNDED_FIX.into(),
})
}
fn handle(mut stream: UnixStream, cfg: &ServeConfig, bound: &PidsBound, running: &AtomicUsize) {
let _ = stream.set_read_timeout(Some(PING));
let mut line = Vec::new();
let read = BufReader::new(&stream)
.take(LINE_MAX)
.read_until(b'\n', &mut line);
let answer = match read
.ok()
.and_then(|_| serde_json::from_slice::<Ask>(&line).ok())
{
None => (
Answer::refused(Refused::Config {
why: "the ask is not one JSON line".into(),
}),
Vec::new(),
),
Some(a) if a.schema != SCHEMA => (
Answer::refused(Refused::Config {
why: format!("schema {:?}, this sidecar speaks {SCHEMA}", a.schema),
}),
Vec::new(),
),
Some(a) if a.op == Op::Ping => (ping(cfg, bound), Vec::new()),
Some(a) => {
if let Some(refused) = unbounded_refusal(cfg, bound) {
(Answer::refused(refused), Vec::new())
} else if running.fetch_add(1, Ordering::SeqCst) >= cfg.max_jobs {
running.fetch_sub(1, Ordering::SeqCst);
(
Answer::refused(Refused::Unavailable {
why: format!("all {} job slots are taken", cfg.max_jobs),
}),
Vec::new(),
)
} else {
let r = run_job(cfg, bound, &a);
running.fetch_sub(1, Ordering::SeqCst);
match r {
Ok((report, elapsed)) => (
Answer {
schema: SCHEMA.into(),
refused: None,
report_len: report.len() as u64,
elapsed_ms: elapsed.as_millis() as u64,
worker: String::new(),
},
report,
),
Err(refused) => (Answer::refused(refused), Vec::new()),
}
}
}
};
let (head, report) = answer;
let mut out = serde_json::to_vec(&head).unwrap_or_default();
out.push(b'\n');
let _ = stream
.write_all(&out)
.and_then(|_| stream.write_all(&report))
.and_then(|_| stream.flush());
}
fn ping(cfg: &ServeConfig, bound: &PidsBound) -> Answer {
use std::os::unix::fs::PermissionsExt as _;
let worker_ok = std::fs::metadata(&cfg.worker)
.map(|m| m.is_file() && m.permissions().mode() & 0o111 != 0)
.unwrap_or(false);
if !worker_ok {
return Answer::refused(Refused::Unavailable {
why: format!("the worker program {} is not there", cfg.worker.display()),
});
}
if !cfg.share.is_dir() {
return Answer::refused(Refused::Unavailable {
why: format!("the shared directory {} is not there", cfg.share.display()),
});
}
if let Some(refused) = unbounded_refusal(cfg, bound) {
return Answer::refused(refused);
}
Answer {
schema: SCHEMA.into(),
refused: None,
report_len: 0,
elapsed_ms: 0,
worker: format!(
"holger-scan-worker sidecar, {} job slots; {}",
cfg.max_jobs,
bound.words()
),
}
}
/// The ask's limits under the ceiling, field by field.
pub fn bounded(ask: &Limits, ceiling: &Limits) -> Limits {
Limits {
cpu_millis: ask.cpu_millis.min(ceiling.cpu_millis).max(1),
memory_mib: ask.memory_mib.min(ceiling.memory_mib).max(64),
pids: ask.pids.min(ceiling.pids).max(8),
scratch_mib: ask.scratch_mib.min(ceiling.scratch_mib).max(1),
timeout_secs: ask.timeout_secs.min(ceiling.timeout_secs).max(1),
report_max_bytes: ask.report_max_bytes.min(ceiling.report_max_bytes).max(1024),
}
}
/// Run the worker over one job directory.
fn run_job(
cfg: &ServeConfig,
bound: &PidsBound,
ask: &Ask,
) -> Result<(Vec<u8>, Duration), Refused> {
let l = bounded(&ask.limits, &cfg.ceiling);
let mut argv: Vec<String> = Vec::new();
let mut artifact_len = 0u64;
if ask.command.is_empty() {
if !valid_segment(&ask.dir) {
return Err(Refused::Config {
why: format!("the job directory {:?} is not one path segment", ask.dir),
});
}
let dir = cfg.share.join(&ask.dir);
let meta = std::fs::symlink_metadata(&dir).map_err(|e| Refused::Config {
why: format!("the job directory {:?}: {e}", ask.dir),
})?;
if !meta.is_dir() {
return Err(Refused::Config {
why: format!("the job directory {:?} is not a directory", ask.dir),
});
}
artifact_len = std::fs::metadata(dir.join("artifact"))
.map(|m| m.len())
.unwrap_or(0);
argv.push(dir.display().to_string());
} else if ask.command[0] == "probe"
&& ask.command.iter().all(|a| {
a.len() <= 64
&& a.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"-_.".contains(&b))
})
{
argv.extend(ask.command.iter().cloned());
} else {
return Err(Refused::Config {
why: "the only command a job may name is the worker's own `probe`".into(),
});
}
let scratch = tempfile::Builder::new()
.prefix("holger-scan-scratch-")
.tempdir_in(&cfg.scratch)
.map_err(|e| Refused::Io {
why: format!("the job's scratch under {}: {e}", cfg.scratch.display()),
})?;
// The address space: the memory limit, the mapped artifact and room for
// the program itself.
let address_space = (l.memory_mib + 64) * (1 << 20) + artifact_len;
let cpu_secs = l.timeout_secs * l.cpu_millis.div_ceil(1000).max(1) + 1;
let fsize = l.scratch_mib * (1 << 20);
// RA5-27: the job's own cgroup, `pids.max` its bound, when the sidecar holds one. A
// delegated sidecar that cannot make it refuses the job: it never runs unbounded.
let cgroup = match bound {
PidsBound::PerJob { own } => Some(crate::pids::JobCgroup::create(own, l.pids).map_err(
|e| Refused::Io {
why: format!("the job's cgroup under {}: {e}", own.display()),
},
)?),
_ => None,
};
let procs_fd = cgroup.as_ref().map(|c| c.procs_fd());
// The seal's maps, made before the fork (RA5-27): the job's own cgroup namespace.
let seal = cgroup.as_ref().map(|_| crate::pids::Seal::new());
// A32-14, a process by design: the sidecar runs the scan worker it is configured with, one fresh subprocess per job; the isolation is the feature.
let mut cmd = Command::new(&cfg.worker);
cmd.args(&argv)
.env_clear()
.env("HOME", scratch.path())
.env("TMPDIR", scratch.path())
.current_dir(scratch.path())
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(p) = std::env::var_os("PATH") {
cmd.env("PATH", p);
}
{
use std::os::unix::process::CommandExt as _;
// SAFETY: between fork and exec only async-signal-safe system calls are
// made (write, unshare, open, close, setsid, setrlimit, prctl), with no allocation.
unsafe {
cmd.pre_exec(move || {
use rustix::process::{setrlimit, Resource, Rlimit};
let lim = |v: u64| Rlimit {
current: Some(v),
maximum: Some(v),
};
// Into the job's cgroup first, so everything it forks counts against its pids.max,
// and sealed there: the job's cgroup is the root of its own cgroup namespace, so
// it cannot move out of it nor write its own pids.max (RA5-27).
if let Some(fd) = procs_fd {
crate::pids::join_in_child(fd)?;
}
if let Some(seal) = &seal {
crate::pids::seal_in_child(seal)?;
}
rustix::process::setsid()?;
setrlimit(Resource::As, lim(address_space))?;
setrlimit(Resource::Cpu, lim(cpu_secs))?;
setrlimit(Resource::Fsize, lim(fsize))?;
setrlimit(Resource::Nofile, lim(256))?;
setrlimit(Resource::Core, lim(0))?;
rustix::thread::set_no_new_privs(true)?;
Ok(())
});
}
}
let started = Instant::now();
// A32-14, a process by design: the scan worker's subprocess (see its command above).
let child = cmd.spawn().map_err(|e| Refused::Unavailable {
why: format!("{}: {e}", cfg.worker.display()),
})?;
let out = match drive(
child,
started + Duration::from_secs(l.timeout_secs),
l.report_max_bytes,
) {
Drive::Done {
status,
stdout,
stderr,
} => {
let code = status.code();
if status.success() {
Ok((stdout, started.elapsed()))
} else if code.is_none() || code == Some(137) || code == Some(139) {
Err(Refused::Killed { exit: code, stderr })
} else {
Err(Refused::WorkerFailed { exit: code, stderr })
}
}
Drive::TimedOut => Err(Refused::Timeout {
secs: l.timeout_secs,
}),
Drive::TooLarge => Err(Refused::TooLarge {
cap: l.report_max_bytes,
}),
Drive::Io(why) => Err(Refused::Io { why }),
};
drop(scratch);
// Whatever the job left running is killed with its cgroup.
drop(cgroup);
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_ask_and_the_answer_read_back_as_written() {
let a = Ask {
schema: SCHEMA.into(),
op: Op::Run,
dir: "holger-scan-x".into(),
command: vec![],
limits: Limits {
timeout_secs: 9,
..Default::default()
},
};
let s = serde_json::to_string(&a).unwrap();
assert!(!s.contains('\n'), "one line: {s}");
assert_eq!(serde_json::from_str::<Ask>(&s).unwrap(), a);
let r = Answer::refused(Refused::Timeout { secs: 9 });
assert_eq!(
serde_json::from_str::<Answer>(&serde_json::to_string(&r).unwrap()).unwrap(),
r
);
assert!(
serde_json::from_str::<Ask>(r#"{"schema":"x","op":"run","extra":1}"#).is_err(),
"an unknown field is refused"
);
}
#[test]
fn each_sidecar_refusal_is_the_servers_refusal_by_the_same_code() {
for (r, code) in [
(Refused::Timeout { secs: 1 }, "holger-sandbox-2"),
(
Refused::Killed {
exit: None,
stderr: String::new(),
},
"holger-sandbox-3",
),
(Refused::TooLarge { cap: 1 }, "holger-sandbox-4"),
(
Refused::WorkerFailed {
exit: Some(2),
stderr: String::new(),
},
"holger-sandbox-9",
),
(Refused::Config { why: String::new() }, "holger-sandbox-10"),
(Refused::Io { why: String::new() }, "holger-sandbox-11"),
(
Refused::NoPidsLimit { why: String::new() },
"holger-sandbox-13",
),
(
Refused::Unavailable { why: String::new() },
"holger-sandbox-1",
),
] {
assert_eq!(r.clone().into_refusal().code().code, code, "{r:?}");
}
}
#[test]
fn the_ceiling_bounds_every_limit_and_the_floor_holds() {
let ceiling = Limits {
memory_mib: 512,
timeout_secs: 60,
..Default::default()
};
let asked = Limits {
memory_mib: 8192,
timeout_secs: 3600,
cpu_millis: 0,
..Default::default()
};
let b = bounded(&asked, &ceiling);
assert_eq!((b.memory_mib, b.timeout_secs, b.cpu_millis), (512, 60, 1));
assert_eq!(
bounded(
&Limits {
memory_mib: 128,
..Default::default()
},
&ceiling
)
.memory_mib,
128,
"less than the ceiling is kept"
);
}
#[test]
fn a_copied_input_is_read_only_and_skips_the_configmap_links() {
use std::os::unix::fs::PermissionsExt as _;
let d = tempfile::tempdir().unwrap();
let from = d.path().join("rules");
std::fs::create_dir_all(from.join("..2026_10_26")).unwrap();
std::fs::write(from.join("a.yar"), "rule a").unwrap();
let to = d.path().join("job-d-yara");
copy_ro(&from, &to).unwrap();
assert_eq!(std::fs::read_to_string(to.join("a.yar")).unwrap(), "rule a");
assert!(
!to.join("..2026_10_26").exists(),
"Kubernetes' own links are not copied"
);
assert_eq!(
std::fs::metadata(to.join("a.yar"))
.unwrap()
.permissions()
.mode()
& 0o777,
0o444
);
}
}