clawft_kernel/workload_runtime/
container_cmd.rs

1//! Container engine command lines (pure; unit-tested without an engine).
2//!
3//! The image is a minimal layer over an operator-pinned base (by digest)
4//! holding only the one verified binary, built locally: no registry pull
5//! other than the pinned base, which must already be present.
6
7use std::path::Path;
8use std::time::Duration;
9
10use async_trait::async_trait;
11
12use super::evidence::RunEvidence;
13use super::types::RuntimeError;
14
15/// A container engine.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum Engine {
18    /// Apple `container` (per-container VM).
19    Apple,
20    /// Docker-compatible engine (OrbStack, Docker Engine / Desktop).
21    Docker,
22    /// Podman.
23    Podman,
24}
25
26impl Engine {
27    /// CLI program.
28    pub fn cli(self) -> &'static str {
29        match self {
30            Engine::Apple => "container",
31            Engine::Docker => "docker",
32            Engine::Podman => "podman",
33        }
34    }
35
36    /// Adapter id.
37    pub fn runtime_id(self) -> &'static str {
38        match self {
39            Engine::Apple => "container.apple",
40            Engine::Docker => "container.docker",
41            Engine::Podman => "container.podman",
42        }
43    }
44
45    /// Capability id this adapter advertises.
46    pub fn capability_id(self) -> &'static str {
47        match self {
48            Engine::Apple => "runtime.container.apple",
49            Engine::Docker => "runtime.container.docker",
50            Engine::Podman => "runtime.container.podman",
51        }
52    }
53
54    /// Platform argument for a package arch.
55    pub fn platform_args(self, arch: &str) -> Result<Vec<String>, RuntimeError> {
56        let (docker, apple) = match arch {
57            "aarch64" => ("linux/arm64", "arm64"),
58            "armv7" => ("linux/arm/v7", "arm"),
59            "x86_64" => ("linux/amd64", "amd64"),
60            _ => {
61                return Err(RuntimeError::InvalidConfig(format!(
62                    "no platform for {arch}"
63                )));
64            }
65        };
66        Ok(match self {
67            Engine::Apple => vec!["--arch".into(), apple.into()],
68            _ => vec!["--platform".into(), docker.into()],
69        })
70    }
71}
72
73/// Unprivileged uid:gid inside the container (`nobody`).
74pub const CONTAINER_USER: &str = "65534:65534";
75/// Task (process + thread) cap for a cog container.
76pub const PIDS_LIMIT: u32 = 64;
77/// Where the binary lives in the image.
78pub const COG_PATH: &str = "/cog";
79/// Writable tmpfs for `COGNITUM_COG_DATA_DIR`.
80pub const DATA_DIR: &str = "/data";
81
82/// Validate an operator-pinned base image reference (`name@sha256:<64 hex>`).
83pub fn validate_base_image(r: &str) -> Result<(), RuntimeError> {
84    let ok = r.split_once("@sha256:").is_some_and(|(name, digest)| {
85        !name.is_empty()
86            && name.len() <= 200
87            && name
88                .bytes()
89                .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b"./:_-".contains(&b))
90            && crate::workload_pkg::codec::is_lower_hex(digest, 64)
91    });
92    if ok {
93        Ok(())
94    } else {
95        Err(RuntimeError::InvalidConfig(
96            "base image must be pinned by digest (name@sha256:<64 hex>)".into(),
97        ))
98    }
99}
100
101/// Dockerfile for the one-binary layer.
102pub fn dockerfile(base_image: &str) -> String {
103    format!(
104        "FROM {base_image}\nCOPY cog {COG_PATH}\nUSER {CONTAINER_USER}\nWORKDIR /\nENTRYPOINT [\"{COG_PATH}\"]\n"
105    )
106}
107
108/// Image tag for a binary: `weftos-cog/<id>:<16 hex of its BLAKE3>`.
109pub fn image_tag(cog_id: &str, blake3: &str) -> String {
110    format!(
111        "{}{cog_id}:{}",
112        crate::container::COG_IMAGE_PREFIX,
113        &blake3[..16.min(blake3.len())]
114    )
115}
116
117/// `build` command.
118pub fn build_cmd(
119    engine: Engine,
120    arch: &str,
121    tag: &str,
122    ctx: &Path,
123) -> Result<Vec<String>, RuntimeError> {
124    let mut v = vec!["build".to_string()];
125    v.extend(engine.platform_args(arch)?);
126    v.extend(["-t".into(), tag.into(), ctx.to_string_lossy().into_owned()]);
127    Ok(v)
128}
129
130/// Resource and isolation flags for a run.
131#[derive(Debug, Clone)]
132pub struct RunLimits {
133    /// Memory cap in MiB.
134    pub ram_mb: u32,
135    /// CPU share in percent of one core.
136    pub cpu_pct: u32,
137}
138
139/// Everything a `run` needs.
140#[derive(Debug, Clone)]
141pub struct RunSpec<'a> {
142    /// Container name.
143    pub name: &'a str,
144    /// Image tag.
145    pub tag: &'a str,
146    /// Package arch.
147    pub arch: &'a str,
148    /// Limits.
149    pub limits: RunLimits,
150    /// Env file (0600, holds the token).
151    pub env_file: &'a Path,
152    /// `(host address, host UDP port, container feed port)` to publish.
153    pub feed_publish: Option<(std::net::IpAddr, u16, u16)>,
154    /// Network to attach, if not the engine default.
155    pub network: Option<&'a str>,
156    /// Detached (instances) or foreground (console runs).
157    pub detach: bool,
158    /// The image runs the ingest relay as entrypoint (starts as root with
159    /// only [`super::container_relay::RELAY_CAPS`]; the cog drops to
160    /// `nobody`).
161    pub ingest_relay: bool,
162    /// Cog arguments.
163    pub args: &'a [String],
164}
165
166/// `run` command.
167pub fn run_cmd(engine: Engine, s: &RunSpec<'_>) -> Result<Vec<String>, RuntimeError> {
168    let mut v = vec!["run".to_string()];
169    v.push(if s.detach { "-d" } else { "--rm" }.into());
170    v.extend(["--name".into(), s.name.into()]);
171    v.extend(engine.platform_args(s.arch)?);
172    v.extend([
173        "--memory".into(),
174        format!("{}M", s.limits.ram_mb),
175        "--cpus".into(),
176        match engine {
177            // Apple allocates whole vCPUs.
178            Engine::Apple => s.limits.cpu_pct.div_ceil(100).max(1).to_string(),
179            _ => format!("{:.2}", f64::from(s.limits.cpu_pct) / 100.0),
180        },
181        "--read-only".into(),
182        "--cap-drop".into(),
183        "ALL".into(),
184    ]);
185    if s.ingest_relay {
186        for c in super::container_relay::RELAY_CAPS {
187            v.extend(["--cap-add".into(), c.into()]);
188        }
189    }
190    v.extend([
191        "--tmpfs".into(),
192        DATA_DIR.into(),
193        "--env-file".into(),
194        s.env_file.to_string_lossy().into_owned(),
195        "--label".into(),
196        "weftos.workload=cog".into(),
197    ]);
198    match engine {
199        // Apple `container` has neither `--pids-limit` nor
200        // `--security-opt`. The task cap is RLIMIT_NPROC instead: the cog
201        // runs as a non-root user alone in its VM, so the per-user limit
202        // caps its processes and threads. no_new_privs comes from the relay
203        // (when present); otherwise the cog runs as `nobody` with every
204        // capability dropped.
205        Engine::Apple => v.extend([
206            "--ulimit".into(),
207            format!("nproc={PIDS_LIMIT}:{PIDS_LIMIT}"),
208        ]),
209        _ => v.extend([
210            "--pids-limit".into(),
211            PIDS_LIMIT.to_string(),
212            "--security-opt".into(),
213            "no-new-privileges".into(),
214        ]),
215    }
216    if let Some((ip, host, inner)) = s.feed_publish {
217        let addr = match ip {
218            std::net::IpAddr::V4(a) => a.to_string(),
219            std::net::IpAddr::V6(a) => format!("[{a}]"),
220        };
221        v.extend(["-p".into(), format!("{addr}:{host}:{inner}/udp")]);
222    }
223    if let Some(n) = s.network {
224        v.extend(["--network".into(), n.into()]);
225    }
226    v.push(s.tag.into());
227    v.extend(s.args.iter().cloned());
228    Ok(v)
229}
230
231/// `stop` command.
232pub fn stop_cmd(name: &str, grace: Duration) -> Vec<String> {
233    vec![
234        "stop".into(),
235        "-t".into(),
236        grace.as_secs().max(1).to_string(),
237        name.into(),
238    ]
239}
240
241/// Force-remove command.
242pub fn rm_cmd(engine: Engine, name: &str) -> Vec<String> {
243    match engine {
244        Engine::Apple => vec!["delete".into(), "--force".into(), name.into()],
245        _ => vec!["rm".into(), "-f".into(), name.into()],
246    }
247}
248
249/// Remove-image command.
250pub fn rmi_cmd(engine: Engine, tag: &str) -> Vec<String> {
251    match engine {
252        Engine::Apple => vec!["image".into(), "delete".into(), tag.into()],
253        _ => vec!["rmi".into(), tag.into()],
254    }
255}
256
257/// Parse `(running, exit_code)` from `inspect` output.
258pub fn parse_inspect(engine: Engine, out: &str) -> Option<(String, Option<i32>)> {
259    let v: serde_json::Value = serde_json::from_str(out.trim()).ok()?;
260    let first = v.as_array().and_then(|a| a.first()).unwrap_or(&v);
261    match engine {
262        Engine::Apple => {
263            // container 1.0: {"status": {"state": "running", ...}};
264            // earlier releases: {"status": "running"}.
265            let st = first.get("status")?;
266            let status = st.get("state").unwrap_or(st).as_str()?.to_string();
267            Some((status, None))
268        }
269        _ => {
270            let st = first.get("State")?;
271            let status = st.get("Status")?.as_str()?.to_string();
272            let code = st
273                .get("ExitCode")
274                .and_then(|c| c.as_i64())
275                .map(|c| c as i32);
276            Some((status, code))
277        }
278    }
279}
280
281/// Output of one engine CLI call.
282#[derive(Debug, Clone, Default)]
283pub struct CmdOutput {
284    /// Exit status (`None` when killed on timeout).
285    pub status: Option<i32>,
286    /// Stdout (capped by the runner).
287    pub stdout: String,
288    /// Stderr (capped by the runner).
289    pub stderr: String,
290    /// Total stdout bytes.
291    pub stdout_bytes: u64,
292    /// Total stderr bytes.
293    pub stderr_bytes: u64,
294    /// Whether the call hit its timeout.
295    pub timed_out: bool,
296}
297
298/// Runs engine CLI commands. Mocked in unit tests; [`SystemRunner`] in use.
299#[async_trait]
300pub trait CommandRunner: Send + Sync {
301    /// Run `program args`, capturing at most `limit` bytes per stream.
302    async fn run(
303        &self,
304        program: &str,
305        args: &[String],
306        timeout: Duration,
307        limit: usize,
308    ) -> Result<CmdOutput, RuntimeError>;
309}
310
311/// Runs commands with `tokio::process`.
312pub struct SystemRunner;
313
314#[async_trait]
315impl CommandRunner for SystemRunner {
316    async fn run(
317        &self,
318        program: &str,
319        args: &[String],
320        timeout: Duration,
321        limit: usize,
322    ) -> Result<CmdOutput, RuntimeError> {
323        let mut child = tokio::process::Command::new(program)
324            .args(args)
325            .stdin(std::process::Stdio::null())
326            .stdout(std::process::Stdio::piped())
327            .stderr(std::process::Stdio::piped())
328            .kill_on_drop(true)
329            .spawn()
330            .map_err(|e| RuntimeError::Backend(format!("{program}: {e}")))?;
331        let (o, e) = (child.stdout.take(), child.stderr.take());
332        let work = async { tokio::join!(capture(o, limit), capture(e, limit), child.wait()) };
333        match tokio::time::timeout(timeout, work).await {
334            Ok((out, err, st)) => Ok(CmdOutput {
335                status: st.ok().and_then(|s| s.code()),
336                stdout: out.text(),
337                stderr: err.text(),
338                stdout_bytes: out.total(),
339                stderr_bytes: err.total(),
340                timed_out: false,
341            }),
342            Err(_) => Ok(CmdOutput {
343                timed_out: true,
344                ..CmdOutput::default()
345            }),
346        }
347    }
348}
349
350async fn capture<R: tokio::io::AsyncRead + Unpin>(
351    s: Option<R>,
352    limit: usize,
353) -> super::evidence::Capture {
354    use tokio::io::AsyncReadExt;
355    let mut cap = super::evidence::Capture::new(limit);
356    if let Some(mut s) = s {
357        let mut buf = [0u8; 8192];
358        while let Ok(n) = s.read(&mut buf).await {
359            if n == 0 {
360                break;
361            }
362            cap.push(&buf[..n]);
363        }
364    }
365    cap
366}
367
368/// Container-name-safe form of an instance id.
369pub fn container_name(iid: &str) -> String {
370    let s: String = iid
371        .chars()
372        .map(|c| {
373            if c.is_ascii_alphanumeric() || c == '-' {
374                c.to_ascii_lowercase()
375            } else {
376                '-'
377            }
378        })
379        .collect();
380    format!("weftos-{s}")
381}
382
383/// Allowance on top of `max_runtime_secs` for container creation / VM boot.
384pub const ENGINE_STARTUP_SECS: u64 = 20;
385
386pub(super) fn evidence_from(
387    out: &CmdOutput,
388    args: &[String],
389    exit: Option<i32>,
390    t0: std::time::Instant,
391) -> RunEvidence {
392    let limit_hit =
393        out.stdout_bytes > out.stdout.len() as u64 || out.stderr_bytes > out.stderr.len() as u64;
394    RunEvidence {
395        args: args.to_vec(),
396        exit_code: exit,
397        elapsed_ms: t0.elapsed().as_millis() as u64,
398        stdout: out.stdout.clone(),
399        stderr: out.stderr.clone(),
400        stdout_bytes: out.stdout_bytes,
401        stderr_bytes: out.stderr_bytes,
402        truncated: limit_hit,
403        ..RunEvidence::default()
404    }
405}
406
407pub(super) fn write_context(
408    ctx: &std::path::Path,
409    env_file: &std::path::Path,
410    bytes: &[u8],
411    dockerfile: &str,
412    env: &[(String, String)],
413) -> Result<(), RuntimeError> {
414    use std::io::Write;
415    use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
416    let io = |e: std::io::Error| RuntimeError::Backend(format!("build context: {e}"));
417    std::fs::create_dir_all(ctx).map_err(io)?;
418    std::fs::write(ctx.join("Dockerfile"), dockerfile).map_err(io)?;
419    let bin = ctx.join("cog");
420    std::fs::write(&bin, bytes).map_err(io)?;
421    std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o555)).map_err(io)?;
422    let mut f = std::fs::OpenOptions::new()
423        .create(true)
424        .truncate(true)
425        .write(true)
426        .mode(0o600)
427        .open(env_file)
428        .map_err(io)?;
429    for (k, v) in env {
430        if v.contains('\n') {
431            return Err(RuntimeError::InvalidConfig(format!(
432                "{k} contains a newline"
433            )));
434        }
435        writeln!(f, "{k}={v}").map_err(io)?;
436    }
437    Ok(())
438}
clawft_kernel/workload_runtime/container_cmd.rs

clawft_kernel/workload_runtime/
container_cmd.rs
