clawft_kernel/workload_runtime/
native.rs

1//! Native-process adapter (ADR-100 section 3): the verified binary is
2//! written under the adapter's data root and run unprivileged under
3//! `[console]` / `[resources]` limits by [`super::supervise`].
4
5use std::collections::HashMap;
6use std::path::PathBuf;
7use std::time::Duration;
8
9use async_trait::async_trait;
10use clawft_types::placement::{AttrValue, Capability, CapabilityId, Provenance};
11use tokio::sync::Mutex;
12
13use super::cog_spec::CogSpec;
14use super::evidence::RunEvidence;
15use super::host_contract::HostContract;
16use super::supervise::{LaunchSpec, ProcLimits, Supervised};
17use super::types::{
18    Admission, ControlMode, InstanceHandle, InstanceState, InstanceStatus, RuntimeError,
19    SignedPayload, VerifiedWorkload, WorkloadConfig, WorkloadRuntime,
20};
21use crate::workload_governance::NetworkPolicy;
22
23/// Adapter id.
24pub const NATIVE_ID: &str = "native";
25
26/// Package arch name for this host, if cogs are built for it.
27pub fn host_arch() -> Option<&'static str> {
28    match std::env::consts::ARCH {
29        "aarch64" => Some("aarch64"),
30        "arm" => Some("armv7"),
31        "x86_64" => Some("x86_64"),
32        _ => None,
33    }
34}
35
36/// ELF `e_machine` for a package arch.
37pub fn elf_machine(arch: &str) -> Option<u16> {
38    match arch {
39        "aarch64" => Some(183),
40        "armv7" => Some(40),
41        "x86_64" => Some(62),
42        _ => None,
43    }
44}
45
46/// What kind of executable a payload is.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum PayloadKind {
49    /// ELF with this `e_machine`.
50    Elf(u16),
51    /// `#!` script.
52    Script,
53    /// Anything else.
54    Unknown,
55}
56
57/// Classify payload bytes.
58pub fn classify(bytes: &[u8]) -> PayloadKind {
59    if bytes.len() >= 20 && bytes[..4] == *b"\x7fELF" {
60        let m = [bytes[18], bytes[19]];
61        let machine = if bytes[5] == 2 {
62            u16::from_be_bytes(m)
63        } else {
64            u16::from_le_bytes(m)
65        };
66        PayloadKind::Elf(machine)
67    } else if bytes.starts_with(b"#!") {
68        PayloadKind::Script
69    } else {
70        PayloadKind::Unknown
71    }
72}
73
74/// Native adapter configuration.
75#[derive(Debug, Clone)]
76pub struct NativeConfig {
77    /// Root under which instance directories are created.
78    pub root: PathBuf,
79    /// `(uid, gid)` to drop to when the host runs as root.
80    pub run_as: Option<(u32, u32)>,
81    /// Accept signed `#!` script payloads (development and tests only;
82    /// production packages carry ELF binaries).
83    pub allow_interpreted: bool,
84}
85
86struct Instance {
87    dir: PathBuf,
88    program: PathBuf,
89    spec: CogSpec,
90    args: Vec<String>,
91    host: HostContract,
92    proc: Option<Supervised>,
93    last: Option<RunEvidence>,
94}
95
96/// Native-process runtime.
97pub struct NativeRuntime {
98    cfg: NativeConfig,
99    instances: Mutex<HashMap<String, Instance>>,
100}
101
102impl NativeRuntime {
103    /// New adapter rooted at `cfg.root`.
104    pub fn new(cfg: NativeConfig) -> Self {
105        Self {
106            cfg,
107            instances: Mutex::new(HashMap::new()),
108        }
109    }
110
111    fn pick<'a>(&self, p: &'a SignedPayload) -> Result<(String, &'a [u8], bool), RuntimeError> {
112        let host = host_arch()
113            .ok_or_else(|| RuntimeError::AdmissionRefused("unsupported host arch".into()))?;
114        if let Some(b) = p.binaries.get(host) {
115            return Ok((host.to_string(), b.bytes.as_slice(), false));
116        }
117        Err(RuntimeError::AdmissionRefused(format!(
118            "package has no {host} binary (has: {}); native never emulates",
119            p.binaries.keys().cloned().collect::<Vec<_>>().join(",")
120        )))
121    }
122
123    fn check_payload(&self, arch: &str, bytes: &[u8]) -> Result<String, RuntimeError> {
124        let refuse = |m: String| Err(RuntimeError::AdmissionRefused(m));
125        match classify(bytes) {
126            PayloadKind::Elf(m) => {
127                if std::env::consts::OS != "linux" {
128                    return refuse(format!(
129                        "native on {} cannot run Linux ELF binaries",
130                        std::env::consts::OS
131                    ));
132                }
133                if Some(m) != elf_machine(arch) {
134                    return refuse(format!("ELF machine {m} is not {arch}"));
135                }
136                Ok(format!("ELF {arch} on linux"))
137            }
138            PayloadKind::Script if self.cfg.allow_interpreted => Ok("interpreted payload".into()),
139            PayloadKind::Script => refuse("script payloads are disabled".into()),
140            PayloadKind::Unknown => refuse("payload is not an executable".into()),
141        }
142    }
143
144    fn launch(
145        &self,
146        inst: &Instance,
147        args: Vec<String>,
148        cpu_secs: Option<u64>,
149    ) -> Result<Supervised, RuntimeError> {
150        let data = inst.dir.join("data");
151        let mut limits = ProcLimits::for_ram_mb(inst.spec.resources.ram_mb);
152        limits.cpu_secs = cpu_secs;
153        Supervised::spawn(LaunchSpec {
154            program: inst.program.clone(),
155            args,
156            env: inst.host.env(&data.to_string_lossy()),
157            cwd: data,
158            limits,
159            output_limit: inst.spec.console.output_limit_bytes,
160            run_as: self.cfg.run_as,
161        })
162    }
163
164    fn stamp(&self, h: &InstanceHandle, mut ev: RunEvidence) -> RunEvidence {
165        ev.runtime = NATIVE_ID.into();
166        ev.instance_id = h.instance_id.clone();
167        ev
168    }
169}
170
171/// Instance id: `<cog>-<8 hex of config hash>-<node>`.
172pub fn instance_id(w: &VerifiedWorkload, cfg: &WorkloadConfig) -> Result<String, RuntimeError> {
173    if !crate::workload_pkg::manifest::valid_token(&cfg.node_id, 64) {
174        return Err(RuntimeError::InvalidConfig(
175            "node id must be a plain token".into(),
176        ));
177    }
178    let mut h = blake3::Hasher::new();
179    h.update(w.id.as_bytes());
180    h.update(w.version.as_bytes());
181    if let Ok(p) = w.signed("") {
182        h.update(p.package_id.as_bytes());
183    }
184    h.update(
185        serde_json::to_string(&cfg.mode)
186            .unwrap_or_default()
187            .as_bytes(),
188    );
189    h.update(cfg.args.join("\u{1f}").as_bytes());
190    h.update(cfg.host.csi_bind.to_string().as_bytes());
191    Ok(format!(
192        "{}-{}-{}",
193        w.id,
194        &h.finalize().to_hex()[..8],
195        cfg.node_id
196    ))
197}
198
199#[async_trait]
200impl WorkloadRuntime for NativeRuntime {
201    fn id(&self) -> &str {
202        NATIVE_ID
203    }
204
205    fn provides(&self) -> Vec<Capability> {
206        let arches: Vec<AttrValue> = host_arch().into_iter().map(AttrValue::from).collect();
207        let Ok(id) = CapabilityId::new("runtime.native") else {
208            return Vec::new();
209        };
210        vec![
211            Capability::new(id, Provenance::Probed)
212                .with_attr("arches_native", AttrValue::List(arches))
213                .with_attr("arches_emulated", AttrValue::List(Vec::new()))
214                .with_attr("os", std::env::consts::OS),
215        ]
216    }
217
218    async fn admit(&self, w: &VerifiedWorkload) -> Result<Admission, RuntimeError> {
219        let p = w.signed(NATIVE_ID)?;
220        let (arch, bytes, emulated) = self.pick(p)?;
221        let mut notes = vec![self.check_payload(&arch, bytes)?];
222        if super::supervise::running_as_root() && self.cfg.run_as.is_none() {
223            return Err(RuntimeError::AdmissionRefused(
224                "host runs as root and no unprivileged run_as user is configured".into(),
225            ));
226        }
227        notes.push(format!(
228            "limits: ram {} MiB, console {} s, output {} B",
229            p.spec.resources.ram_mb,
230            p.spec.console.max_runtime_secs,
231            p.spec.console.output_limit_bytes
232        ));
233        Ok(Admission {
234            runtime: NATIVE_ID.into(),
235            arch,
236            emulated,
237            notes,
238        })
239    }
240
241    async fn load(
242        &self,
243        w: &VerifiedWorkload,
244        cfg: &WorkloadConfig,
245    ) -> Result<InstanceHandle, RuntimeError> {
246        // Checked before admission: a config error on any host.
247        if cfg
248            .host
249            .ingest_upstream
250            .is_some_and(|u| u.to_string() != super::container_relay::COG_INGEST_ADDR)
251        {
252            return Err(RuntimeError::InvalidConfig(
253                "a native cog posts to this node's 127.0.0.1:80; run the ingest bridge there"
254                    .into(),
255            ));
256        }
257        let adm = self.admit(w).await?;
258        let p = w.signed(NATIVE_ID)?;
259        cfg.mode.validate()?;
260        cfg.host.validate()?;
261        p.spec.validate_args(&cfg.args)?;
262        let iid = instance_id(w, cfg)?;
263        let mut map = self.instances.lock().await;
264        if map.contains_key(&iid) {
265            return Err(RuntimeError::InvalidState(format!(
266                "{iid} is already loaded"
267            )));
268        }
269        let dir = self.cfg.root.join(&iid);
270        let program = dir.join(format!("cog-{}", w.id));
271        let bytes = p.binaries[&adm.arch].bytes.clone();
272        let (d, prog) = (dir.clone(), program.clone());
273        let run_as = self.cfg.run_as;
274        tokio::task::spawn_blocking(move || stage(&d, &prog, &bytes, run_as))
275            .await
276            .map_err(|e| RuntimeError::Backend(e.to_string()))??;
277        let mut args = cfg.mode.args();
278        args.extend(cfg.args.iter().cloned());
279        map.insert(
280            iid.clone(),
281            Instance {
282                dir,
283                program,
284                spec: p.spec.clone(),
285                args,
286                host: cfg.host.clone(),
287                proc: None,
288                last: None,
289            },
290        );
291        Ok(InstanceHandle {
292            runtime: NATIVE_ID.into(),
293            instance_id: iid,
294            workload_id: w.id.clone(),
295            store_installed: false,
296        })
297    }
298
299    async fn staged_payload(&self, h: &InstanceHandle) -> Option<std::path::PathBuf> {
300        self.instances.lock().await.get(&h.instance_id).map(|i| i.program.clone())
301    }
302
303    async fn start(&self, h: &InstanceHandle) -> Result<(), RuntimeError> {
304        let mut map = self.instances.lock().await;
305        let inst = map
306            .get_mut(&h.instance_id)
307            .ok_or_else(|| RuntimeError::UnknownInstance(h.instance_id.clone()))?;
308        if let Some(p) = inst.proc.as_mut()
309            && p.try_exit().is_none()
310        {
311            return Err(RuntimeError::InvalidState("already running".into()));
312        }
313        let once = inst.args.first().map(String::as_str) == Some("--once");
314        let cpu = once.then_some(inst.spec.console.max_runtime_secs + 1);
315        let proc = self.launch(inst, inst.args.clone(), cpu)?;
316        inst.proc = Some(proc);
317        Ok(())
318    }
319
320    async fn stop(&self, h: &InstanceHandle, grace: Duration) -> Result<RunEvidence, RuntimeError> {
321        let mut map = self.instances.lock().await;
322        let inst = map
323            .get_mut(&h.instance_id)
324            .ok_or_else(|| RuntimeError::UnknownInstance(h.instance_id.clone()))?;
325        let Some(proc) = inst.proc.take() else {
326            return inst
327                .last
328                .clone()
329                .ok_or_else(|| RuntimeError::InvalidState("not started".into()));
330        };
331        let ev = self.stamp(h, proc.terminate(grace).await);
332        inst.last = Some(ev.clone());
333        Ok(ev)
334    }
335
336    async fn unload(&self, h: InstanceHandle) -> Result<(), RuntimeError> {
337        let inst = self
338            .instances
339            .lock()
340            .await
341            .remove(&h.instance_id)
342            .ok_or_else(|| RuntimeError::UnknownInstance(h.instance_id.clone()))?;
343        if let Some(p) = inst.proc {
344            p.terminate(Duration::from_secs(2)).await;
345        }
346        tokio::fs::remove_dir_all(&inst.dir)
347            .await
348            .map_err(|e| RuntimeError::Backend(format!("remove instance dir: {e}")))
349    }
350
351    async fn status(&self, h: &InstanceHandle) -> InstanceStatus {
352        let mut map = self.instances.lock().await;
353        let Some(inst) = map.get_mut(&h.instance_id) else {
354            return InstanceStatus::of(InstanceState::Unknown);
355        };
356        match inst.proc.as_mut() {
357            None if inst.last.is_some() => InstanceStatus {
358                state: InstanceState::Exited,
359                exit_code: inst.last.as_ref().and_then(|e| e.exit_code),
360                detail: None,
361            },
362            None => InstanceStatus::of(InstanceState::Loaded),
363            Some(p) => match p.try_exit() {
364                None => InstanceStatus::of(InstanceState::Running),
365                Some(s) => InstanceStatus {
366                    state: InstanceState::Exited,
367                    exit_code: s.code(),
368                    detail: None,
369                },
370            },
371        }
372    }
373
374    fn control_mode(&self) -> ControlMode {
375        ControlMode::Managed
376    }
377
378    /// A native process has the host's network: nothing here restricts
379    /// egress yet (nftables / landlock are deferred, ADR-100 section 4),
380    /// so the gate is told `egress`.
381    fn network_exposure(&self) -> NetworkPolicy {
382        NetworkPolicy::Egress
383    }
384
385    async fn console(
386        &self,
387        h: &InstanceHandle,
388        command: &str,
389    ) -> Result<RunEvidence, RuntimeError> {
390        let proc = {
391            let mut map = self.instances.lock().await;
392            let inst = map
393                .get_mut(&h.instance_id)
394                .ok_or_else(|| RuntimeError::UnknownInstance(h.instance_id.clone()))?;
395            if let Some(p) = inst.proc.as_mut()
396                && p.try_exit().is_none()
397            {
398                return Err(RuntimeError::InvalidState(
399                    "stop the running instance before a console run (sensor feed contention)"
400                        .into(),
401                ));
402            }
403            let args = inst.spec.console_args(command)?;
404            let max = inst.spec.console.max_runtime_secs;
405            (self.launch(inst, args, Some(max + 1))?, max)
406        };
407        let ev = proc.0.wait_for(Duration::from_secs(proc.1)).await;
408        Ok(self.stamp(h, ev))
409    }
410}
411
412fn stage(
413    dir: &std::path::Path,
414    program: &std::path::Path,
415    bytes: &[u8],
416    run_as: Option<(u32, u32)>,
417) -> Result<(), RuntimeError> {
418    use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
419    let io = |e: std::io::Error| RuntimeError::Backend(format!("stage {}: {e}", dir.display()));
420    std::fs::DirBuilder::new()
421        .recursive(true)
422        .mode(0o755)
423        .create(dir)
424        .map_err(io)?;
425    let data = dir.join("data");
426    std::fs::DirBuilder::new()
427        .mode(0o700)
428        .create(&data)
429        .map_err(io)?;
430    std::fs::write(program, bytes).map_err(io)?;
431    std::fs::set_permissions(program, std::fs::Permissions::from_mode(0o555)).map_err(io)?;
432    if let Some((uid, gid)) = run_as {
433        std::os::unix::fs::chown(&data, Some(uid), Some(gid)).map_err(io)?;
434    }
435    Ok(())
436}
clawft_kernel/workload_runtime/native.rs

clawft_kernel/workload_runtime/
native.rs
