server/sandbox/src/api.rs
//! **The engine over podman's API socket** (`url: Some("unix:///…/podman.sock")`):
//! the same seam the Build Thing's container engine drives, and what a Holger
//! that itself runs in a container uses (the host's rootless podman socket is
//! mounted in; no podman program is needed beside the server).
//!
//! Plain HTTP/1.1 over the unix socket, written here so nothing async and no
//! client crate is pulled into the server for it. The worker's stdout is read
//! through `attach` (hijacked before `start`, the engine's multiplexed frames),
//! under the same cap and deadline as the program engine, and the container's
//! log driver is `none`, so nothing the worker prints is stored on the host.
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use crate::{mounts_of, Engine, Output, Refusal, SandboxConfig};
const API: &str = "/v4.0.0/libpod";
/// How long past its deadline a worker's recorded end may come after its output closed.
const ENDED_GRACE: Duration = Duration::from_secs(10);
/// The most bytes a control reply (not the worker's stdout) may carry.
const MAX_REPLY: usize = 4 << 20;
/// The libpod REST engine.
#[derive(Debug, Clone)]
pub struct PodmanApi {
socket: PathBuf,
}
impl PodmanApi {
/// `unix:///run/user/1000/podman/podman.sock`, or a bare path.
pub fn new(url: &str) -> Result<Self, Refusal> {
let p = url.strip_prefix("unix://").unwrap_or(url);
if !p.starts_with('/') {
return Err(Refusal::Config(format!(
"the engine url {url:?} is not a unix socket path"
)));
}
Ok(Self {
socket: PathBuf::from(p),
})
}
fn connect(&self, timeout: Duration) -> Result<UnixStream, Refusal> {
let s = UnixStream::connect(&self.socket).map_err(|e| {
Refusal::NoEngine(format!("podman's socket {}: {e}", self.socket.display()))
})?;
s.set_read_timeout(Some(timeout)).ok();
s.set_write_timeout(Some(timeout)).ok();
Ok(s)
}
fn call(
&self,
method: &str,
path: &str,
body: Option<&[u8]>,
) -> Result<(u16, Vec<u8>), Refusal> {
let mut s = self.connect(Duration::from_secs(60))?;
send(&mut s, method, path, body, false)
.map_err(|e| Refusal::NoEngine(format!("{method} {path}: {e}")))?;
let (status, headers, rest) =
read_head(&mut s).map_err(|e| Refusal::NoEngine(format!("{method} {path}: {e}")))?;
let body = read_body(&mut s, &headers, rest)
.map_err(|e| Refusal::NoEngine(format!("{method} {path}: {e}")))?;
Ok((status, body))
}
fn remove(&self, name: &str) {
let _ = self.call(
"DELETE",
&format!("{API}/containers/{name}?force=true&t=0&ignore=true"),
None,
);
}
fn state(&self, name: &str) -> Option<Ended> {
match self.call("GET", &format!("{API}/containers/{name}/json"), None) {
Ok((200, b)) => {
let v: serde_json::Value = serde_json::from_slice(&b).unwrap_or_default();
Some(parse_state(&v))
}
_ => None,
}
}
/// The container's end, once podman has recorded it. The attach stream can close
/// before that (the worker's stdout is shut while the kernel is still killing it), so
/// this asks again until the inspect says exited, within `limit`.
fn ended(&self, name: &str, limit: Instant) -> Result<Ended, Refusal> {
loop {
match self.state(name) {
Some(e) if e.finished => return Ok(e),
Some(_) if Instant::now() < limit => std::thread::sleep(Duration::from_millis(50)),
Some(e) => {
return Err(Refusal::Io(format!(
"{name}: its output ended but podman recorded no exit in time (it last said exit {:?})",
e.code
)))
}
None => return Err(Refusal::NoEngine(format!("inspect {name}: podman did not answer"))),
}
}
}
}
/// What a container's inspect says about its end.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Ended {
/// podman has recorded the exit: the code and the OOM flag below are its.
pub finished: bool,
pub code: Option<i32>,
pub oom: bool,
}
/// Pure: the end a container's inspect JSON records.
pub(crate) fn parse_state(v: &serde_json::Value) -> Ended {
let code = v
.pointer("/State/ExitCode")
.and_then(|c| c.as_i64())
.map(|c| c as i32);
let oom = v
.pointer("/State/OOMKilled")
.and_then(|c| c.as_bool())
.unwrap_or(false);
// `exited` is podman's word for an end it has recorded (`stopped` too); every other
// status, and an inspect with none, is a container whose exit is not known yet.
let finished = matches!(
v.pointer("/State/Status").and_then(|s| s.as_str()),
Some("exited" | "stopped")
);
Ended {
finished,
code,
oom,
}
}
fn send(
s: &mut UnixStream,
method: &str,
path: &str,
body: Option<&[u8]>,
upgrade: bool,
) -> std::io::Result<()> {
let mut req = format!("{method} {path} HTTP/1.1\r\nHost: d\r\nUser-Agent: holger-sandbox\r\n");
if upgrade {
req.push_str("Connection: Upgrade\r\nUpgrade: tcp\r\n");
} else {
req.push_str("Connection: close\r\n");
}
match body {
Some(b) => req.push_str(&format!(
"Content-Type: application/json\r\nContent-Length: {}\r\n\r\n",
b.len()
)),
None => req.push_str("Content-Length: 0\r\n\r\n"),
}
s.write_all(req.as_bytes())?;
if let Some(b) = body {
s.write_all(b)?;
}
s.flush()
}
/// A response head: the status, the lower-cased headers, and the body bytes read with them.
type Head = (u16, Vec<(String, String)>, Vec<u8>);
/// The status, the lower-cased headers, and whatever body bytes came with them.
fn read_head(s: &mut UnixStream) -> std::io::Result<Head> {
let mut buf = Vec::new();
let mut chunk = [0u8; 8192];
let end = loop {
if let Some(i) = find(&buf, b"\r\n\r\n") {
break i;
}
if buf.len() > 64 << 10 {
return Err(std::io::Error::other("the reply's headers are too long"));
}
let n = s.read(&mut chunk)?;
if n == 0 {
return Err(std::io::Error::other("the engine closed the connection"));
}
buf.extend_from_slice(&chunk[..n]);
};
let head = String::from_utf8_lossy(&buf[..end]).to_string();
let mut lines = head.split("\r\n");
let status: u16 = lines
.next()
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|c| c.parse().ok())
.ok_or_else(|| std::io::Error::other("not an HTTP reply"))?;
let headers = lines
.filter_map(|l| l.split_once(':'))
.map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
.collect();
Ok((status, headers, buf[end + 4..].to_vec()))
}
fn header<'a>(h: &'a [(String, String)], k: &str) -> Option<&'a str> {
h.iter().find(|(n, _)| n == k).map(|(_, v)| v.as_str())
}
fn read_body(
s: &mut UnixStream,
h: &[(String, String)],
mut have: Vec<u8>,
) -> std::io::Result<Vec<u8>> {
let mut chunk = [0u8; 8192];
if header(h, "transfer-encoding").is_some_and(|t| t.eq_ignore_ascii_case("chunked")) {
// Read to the end (Connection: close), then decode.
loop {
let n = s.read(&mut chunk)?;
if n == 0 {
break;
}
have.extend_from_slice(&chunk[..n]);
if have.len() > MAX_REPLY {
return Err(std::io::Error::other("the reply is too large"));
}
if have.ends_with(b"0\r\n\r\n") {
break;
}
}
return dechunk(&have);
}
if let Some(len) = header(h, "content-length").and_then(|l| l.parse::<usize>().ok()) {
if len > MAX_REPLY {
return Err(std::io::Error::other("the reply is too large"));
}
while have.len() < len {
let n = s.read(&mut chunk)?;
if n == 0 {
break;
}
have.extend_from_slice(&chunk[..n]);
}
have.truncate(len);
return Ok(have);
}
loop {
let n = s.read(&mut chunk)?;
if n == 0 {
return Ok(have);
}
have.extend_from_slice(&chunk[..n]);
if have.len() > MAX_REPLY {
return Err(std::io::Error::other("the reply is too large"));
}
}
}
fn dechunk(b: &[u8]) -> std::io::Result<Vec<u8>> {
let mut out = Vec::new();
let mut i = 0;
loop {
let eol =
find(&b[i..], b"\r\n").ok_or_else(|| std::io::Error::other("a broken chunk"))? + i;
let size = usize::from_str_radix(
String::from_utf8_lossy(&b[i..eol])
.split(';')
.next()
.unwrap_or("")
.trim(),
16,
)
.map_err(|_| std::io::Error::other("a broken chunk size"))?;
i = eol + 2;
if size == 0 {
return Ok(out);
}
if i + size > b.len() {
return Err(std::io::Error::other("a short chunk"));
}
out.extend_from_slice(&b[i..i + size]);
i += size + 2;
}
}
fn find(h: &[u8], n: &[u8]) -> Option<usize> {
h.windows(n.len()).position(|w| w == n)
}
/// **The container as a libpod create spec.** Pure, so every hardening field
/// is tested without an engine; the same container [`crate::run_args`] names.
pub fn create_spec(
cfg: &SandboxConfig,
name: &str,
mounts: &[(String, String)],
) -> serde_json::Value {
let l = &cfg.limits;
let mem = (l.memory_mib << 20) as i64;
let mut m = vec![serde_json::json!({
"destination": crate::SCRATCH_PATH, "type": "tmpfs", "source": "tmpfs",
"options": ["rw", format!("size={}m", l.scratch_mib), "mode=1777", "noexec", "nosuid", "nodev"],
})];
for (host, ctr) in mounts {
m.push(serde_json::json!({ "destination": ctr, "type": "bind", "source": host, "options": ["ro", "rbind"] }));
}
serde_json::json!({
"name": name,
"image": cfg.image,
"user": cfg.user,
"read_only_filesystem": true,
"read_write_tmpfs": false,
"cap_drop": ["ALL"],
"no_new_privileges": true,
"netns": { "nsmode": "none" },
"resource_limits": {
"memory": { "limit": mem, "swap": mem },
"cpu": { "quota": (l.cpu_millis as i64) * 100, "period": 100_000u64 },
"pids": { "limit": l.pids as i64 },
},
"mounts": m,
"env": { "HOME": crate::SCRATCH_PATH, "TMPDIR": crate::SCRATCH_PATH },
"log_configuration": { "driver": "none" },
"stop_timeout": 1,
"stdin": false,
"terminal": false,
})
}
/// [`create_spec`] from the `run` arguments themselves: the read-only mounts, and a
/// job's writable output directory at [`crate::OUT_PATH`].
pub fn create_spec_for(cfg: &SandboxConfig, name: &str, args: &[String]) -> serde_json::Value {
let mut v = create_spec(cfg, name, &mounts_of(args));
if let (Some(out), Some(m)) = (crate::out_of(args), v["mounts"].as_array_mut()) {
m.push(serde_json::json!({ "destination": crate::OUT_PATH, "type": "bind", "source": out, "options": ["rw", "rbind", "nosuid", "nodev", "noexec"] }));
}
v
}
impl Engine for PodmanApi {
fn ready(&self, cfg: &SandboxConfig) -> Result<(), Refusal> {
let (status, body) = self.call("GET", &format!("{API}/info"), None)?;
if status != 200 {
return Err(Refusal::NoEngine(format!(
"podman's socket answered {status} to info"
)));
}
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
if cfg.require_rootless
&& v.pointer("/host/security/rootless")
.and_then(|r| r.as_bool())
!= Some(true)
{
return Err(Refusal::NotRootless);
}
let (status, _) = self.call(
"GET",
&format!("{API}/images/{}/exists", enc(&cfg.image)),
None,
)?;
if status != 204 {
return Err(Refusal::ImageMissing(cfg.image.clone()));
}
Ok(())
}
fn pin(&self, cfg: &SandboxConfig) -> Result<String, Refusal> {
let (status, body) = self.call(
"GET",
&format!("{API}/images/{}/json", enc(&cfg.image)),
None,
)?;
if status != 200 {
return Err(Refusal::ImageMissing(cfg.image.clone()));
}
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
crate::pinned_ref(
&cfg.image,
v.get("Digest").and_then(|d| d.as_str()).unwrap_or(""),
)
}
fn run(&self, cfg: &SandboxConfig, name: &str, args: &[String]) -> Result<Output, Refusal> {
let started = Instant::now();
let mut spec = create_spec_for(cfg, name, args);
// Anything after the image is the worker's own argv.
if let Some(i) = args.iter().position(|a| a == &cfg.image) {
let cmd: Vec<&String> = args[i + 1..].iter().collect();
if !cmd.is_empty() {
spec["command"] = serde_json::json!(cmd);
}
}
let body = serde_json::to_vec(&spec).map_err(|e| Refusal::Io(e.to_string()))?;
let (status, reply) =
self.call("POST", &format!("{API}/containers/create"), Some(&body))?;
if status != 201 {
// The engine is there and said no: a path it cannot see, an image it
// cannot read. That is the sandbox's configuration, not a missing engine.
return Err(Refusal::Config(format!(
"podman refused the scan container ({status}): {}",
String::from_utf8_lossy(&reply)
.chars()
.take(400)
.collect::<String>()
)));
}
let out = self.drive(cfg, name, started);
let out = match out {
Ok(stdout) => {
let limit = started + Duration::from_secs(cfg.limits.timeout_secs) + ENDED_GRACE;
let Ended { code, oom, .. } = match self.ended(name, limit) {
Ok(e) => e,
Err(e) => {
self.remove(name);
return Err(e);
}
};
match code {
Some(0) if !oom => Ok(Output {
report: stdout,
elapsed: started.elapsed(),
}),
Some(137) | Some(139) => Err(Refusal::Killed {
exit: code,
oom,
stderr: String::new(),
}),
_ if oom => Err(Refusal::Killed {
exit: code,
oom,
stderr: String::new(),
}),
_ => Err(Refusal::WorkerFailed {
exit: code,
stderr: String::new(),
}),
}
}
Err(e) => Err(e),
};
self.remove(name);
out
}
}
impl PodmanApi {
/// Attach, start, read stdout under the cap and the deadline.
fn drive(&self, cfg: &SandboxConfig, name: &str, started: Instant) -> Result<Vec<u8>, Refusal> {
let deadline = started + Duration::from_secs(cfg.limits.timeout_secs);
let mut s = self.connect(Duration::from_secs(cfg.limits.timeout_secs.max(1)))?;
send(
&mut s,
"POST",
&format!("{API}/containers/{name}/attach?stream=true&stdout=true&stderr=true"),
None,
true,
)
.map_err(|e| Refusal::NoEngine(format!("attach: {e}")))?;
let (status, _h, mut rest) =
read_head(&mut s).map_err(|e| Refusal::NoEngine(format!("attach: {e}")))?;
if status != 101 && status != 200 {
return Err(Refusal::NoEngine(format!("attach answered {status}")));
}
let (st, reply) = self.call("POST", &format!("{API}/containers/{name}/start"), None)?;
if st != 204 && st != 304 {
return Err(Refusal::NoEngine(format!(
"start answered {st}: {}",
String::from_utf8_lossy(&reply)
)));
}
let cap = cfg.limits.report_max_bytes as usize;
let mut stdout = Vec::new();
let mut frame: Vec<u8> = std::mem::take(&mut rest);
let mut chunk = vec![0u8; 64 << 10];
loop {
// Whole frames: [stream, 0, 0, 0, len u32 BE] + payload.
while frame.len() >= 8 {
let len = u32::from_be_bytes([frame[4], frame[5], frame[6], frame[7]]) as usize;
if frame.len() < 8 + len {
break;
}
if frame[0] == 1 {
if stdout.len() + len > cap {
return Err(Refusal::ReportTooLarge {
cap: cfg.limits.report_max_bytes,
});
}
stdout.extend_from_slice(&frame[8..8 + len]);
}
frame.drain(..8 + len);
}
let now = Instant::now();
if now >= deadline {
return Err(Refusal::Timeout {
secs: cfg.limits.timeout_secs,
});
}
s.set_read_timeout(Some(deadline - now)).ok();
match s.read(&mut chunk) {
Ok(0) => return Ok(stdout),
Ok(n) => {
frame.extend_from_slice(&chunk[..n]);
if frame.len() > cap + (1 << 20) {
return Err(Refusal::ReportTooLarge {
cap: cfg.limits.report_max_bytes,
});
}
}
Err(e)
if matches!(
e.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
) =>
{
return Err(Refusal::Timeout {
secs: cfg.limits.timeout_secs,
})
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(Refusal::NoEngine(format!("attach: {e}"))),
}
}
}
}
fn enc(s: &str) -> String {
s.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-._~".contains(&b) {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect()
}
/// Is `path` a socket that exists? For the error's words.
pub fn socket_exists(path: &Path) -> bool {
use std::os::unix::fs::FileTypeExt as _;
std::fs::metadata(path)
.map(|m| m.file_type().is_socket())
.unwrap_or(false)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_api_spec_has_every_hardening_field() {
let cfg = SandboxConfig::default();
let v = create_spec(
&cfg,
"holger-scan-x",
&[("/w/a".into(), "/in/artifact".into())],
);
assert_eq!(v["netns"]["nsmode"], "none");
assert_eq!(v["read_only_filesystem"], true);
assert_eq!(v["read_write_tmpfs"], false);
assert_eq!(v["cap_drop"][0], "ALL");
assert_eq!(v["no_new_privileges"], true);
assert_eq!(v["user"], "65532:65532");
assert_eq!(
v["log_configuration"]["driver"], "none",
"nothing the worker prints is stored on the host"
);
assert_eq!(v["resource_limits"]["memory"]["limit"], 1024i64 << 20);
assert_eq!(v["resource_limits"]["memory"]["swap"], 1024i64 << 20);
assert_eq!(v["resource_limits"]["pids"]["limit"], 64);
assert_eq!(v["resource_limits"]["cpu"]["quota"], 100_000);
assert_eq!(v["mounts"][0]["type"], "tmpfs");
assert_eq!(v["mounts"][1]["options"][0], "ro");
assert!(
v.get("privileged").is_none()
&& v.get("cap_add").is_none()
&& v.get("seccomp_policy").is_none()
);
}
#[test]
fn a_jobs_output_is_its_one_writable_mount() {
let cfg = SandboxConfig::default();
let args = crate::run_args_out(
&cfg,
"n",
std::path::Path::new("/w/a"),
std::path::Path::new("/w/j"),
&[],
Some(std::path::Path::new("/w/out")),
);
let v = create_spec_for(&cfg, "n", &args);
let m = v["mounts"].as_array().unwrap();
let out = m
.iter()
.find(|x| x["destination"] == "/out")
.expect("the job's /out");
assert_eq!(out["source"], "/w/out");
assert_eq!(out["options"][0], "rw");
assert_eq!(
m.iter()
.filter(|x| x["options"][0] == "rw" && x["type"] == "bind")
.count(),
1
);
assert_eq!(v["read_only_filesystem"], true);
assert_eq!(v["netns"]["nsmode"], "none");
}
/// ★ The attach stream can end before podman has recorded the exit (ODEN-CI round 4,
/// oden under load: `a_worker_past_its_memory_is_killed_and_refused_by_name` got an empty
/// report and no refusal). An inspect taken in that window says ExitCode 0 and
/// OOMKilled false for a worker the kernel is killing: it is NOT an end.
#[test]
fn an_inspect_before_podman_records_the_exit_is_not_an_end() {
for status in ["running", "stopping", "configured", "created"] {
let v = serde_json::json!({"State": {"Status": status, "Running": status == "running", "ExitCode": 0, "OOMKilled": false}});
assert!(!parse_state(&v).finished, "{status} is not an end: {v}");
}
let oom = serde_json::json!({"State": {"Status": "exited", "Running": false, "ExitCode": 137, "OOMKilled": true}});
assert_eq!(
parse_state(&oom),
Ended {
finished: true,
code: Some(137),
oom: true
}
);
let ok = serde_json::json!({"State": {"Status": "exited", "Running": false, "ExitCode": 0, "OOMKilled": false}});
assert_eq!(
parse_state(&ok),
Ended {
finished: true,
code: Some(0),
oom: false
}
);
assert!(
parse_state(&serde_json::json!({"State": {"Status": "stopped", "ExitCode": 1}}))
.finished
);
assert!(
!parse_state(&serde_json::json!({})).finished,
"no state is no end"
);
}
#[test]
fn chunked_replies_are_decoded() {
assert_eq!(
dechunk(b"3\r\nabc\r\n2\r\nde\r\n0\r\n\r\n").unwrap(),
b"abcde"
);
}
#[test]
fn no_socket_is_no_engine() {
let e = PodmanApi::new("unix:///nonexistent/holger-test/podman.sock")
.unwrap()
.ready(&SandboxConfig::default());
assert!(matches!(e, Err(Refusal::NoEngine(_))), "{e:?}");
assert!(PodmanApi::new("tcp://x").is_err());
}
}