Something went wrong. Try again.
Spindle fork for filtered Stream ingestion and bounded PDS reconciliation on a home CI runner
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228use crate::command::{self, OutKind, Spec};use crate::protocol::{self, Message, v1};use nix::unistd::{Group, User};use std::ffi::OsString;use std::path::PathBuf;use std::time::Duration;use tokio::sync::mpsc::Sender;use tracing::{info, warn};
const DEFAULT_USER: &str = "spindle-workflow";
pub async fn run(id: String, req: v1::ExecStart, out: Sender<Message>) { let send_exit = async |exit_code: i32, error: Option<String>, timed_out: bool| { let msg = Message { id: id.clone(), exec_exit: Some(v1::ExecExit { exit_code, error: protocol::error_or_empty(error), timed_out, }), ..Default::default() }; let _ = out.send(msg).await; };
if req.argv.is_empty() { send_exit(127, Some("missing argv".to_owned()), false).await; return; }
let user = if req.user.is_empty() { DEFAULT_USER } else { req.user.as_str() }; let run_as = match resolve_user(user) { Ok(run_as) => run_as, Err(err) => { send_exit(127, Some(err), false).await; return; } };
let mut env = run_as.login_env(); let runtime_dir = run_as.runtime_dir(); match runtime_dir.try_exists() { Ok(true) => env.push(( OsString::from("XDG_RUNTIME_DIR"), runtime_dir.into_os_string(), )), Ok(false) => {} Err(err) => warn!(error = %err, "could not stat XDG_RUNTIME_DIR for workflow user"), }
let mut spec = Spec::new(req.argv[0].clone()) .args(req.argv[1..].iter().cloned()) .envs(env) .envs(parse_env(&req.env)) .run_as(run_as.uid, run_as.gid); if !req.cwd.is_empty() { spec = spec.cwd(req.cwd.clone()); } let timeout = (req.timeout_seconds > 0).then(|| Duration::from_secs(u64::from(req.timeout_seconds))); if let Some(timeout) = timeout { spec = spec.timeout(timeout); }
info!( %id, user = %run_as.name, uid = run_as.uid, gid = run_as.gid, argv = ?req.argv, cwd = ?req.cwd, "starting exec" );
let cmd = match command::spawn_streaming(spec) { Ok(cmd) => cmd, Err(err) => { send_exit(127, Some(err.to_string()), false).await; return; } }; let (mut events, exit_task) = cmd.into_parts(); while let Some(event) = events.recv().await { let data = String::from_utf8_lossy(&event.data).into_owned(); let output = match event.kind { OutKind::Stdout => Message { id: id.clone(), exec_stdout: Some(v1::ExecStdout { data }), ..Default::default() }, OutKind::Stderr => Message { id: id.clone(), exec_stderr: Some(v1::ExecStderr { data }), ..Default::default() }, }; let _ = out.send(output).await; } let exit = match exit_task .await .unwrap_or_else(|error| Err(anyhow::anyhow!("command supervisor failed: {error}"))) { Ok(exit) => exit, Err(err) => { send_exit(127, Some(err.to_string()), false).await; return; } };
send_exit(exit.exit_code, exit.error, exit.timed_out).await}
#[derive(Clone, Debug)]struct ResolvedUser { name: String, uid: u32, gid: u32, home: OsString, shell: OsString,}
impl ResolvedUser { fn login_env(&self) -> Vec<(OsString, OsString)> { let xdg_cache_home = PathBuf::from(&self.home).join(".cache"); vec![ (OsString::from("USER"), OsString::from(&self.name)), (OsString::from("LOGNAME"), OsString::from(&self.name)), (OsString::from("HOME"), self.home.clone()), ( OsString::from("XDG_CACHE_HOME"), xdg_cache_home.into_os_string(), ), (OsString::from("SHELL"), self.shell.clone()), ] }
fn runtime_dir(&self) -> PathBuf { PathBuf::from(format!("/run/user/{}", self.uid)) }}
fn resolve_user(spec: &str) -> Result<ResolvedUser, String> { let spec = spec.trim(); if spec.is_empty() { return resolve_user(DEFAULT_USER); }
let (user_part, group_part) = spec .split_once(':') .map(|(user, group)| (user, Some(group))) .unwrap_or((spec, None));
let mut user = lookup_user(user_part)?; if let Some(group) = group_part.filter(|group| !group.is_empty()) { user.gid = lookup_group(group)?; }
if user.uid == 0 || user.gid == 0 { return Err(format!("refusing to run exec as privileged user {spec:?}")); }
Ok(user)}
fn lookup_user(name: &str) -> Result<ResolvedUser, String> { match User::from_name(name) { Ok(Some(user)) => Ok(ResolvedUser { name: name.to_owned(), uid: user.uid.as_raw(), gid: user.gid.as_raw(), home: user.dir.into_os_string(), shell: user.shell.into_os_string(), }), Ok(None) => { let uid = name .parse::<u32>() .map_err(|_| format!("workflow user {name:?} was not found"))?; Ok(ResolvedUser { name: name.to_owned(), uid, gid: uid, home: OsString::from("/"), shell: OsString::from("/bin/sh"), }) } Err(error) => Err(format!("lookup workflow user {name:?}: {error}")), }}
fn lookup_group(name: &str) -> Result<u32, String> { match Group::from_name(name) { Ok(Some(group)) => Ok(group.gid.as_raw()), Ok(None) => name .parse::<u32>() .map_err(|_| format!("workflow group {name:?} was not found")), Err(error) => Err(format!("lookup workflow group {name:?}: {error}")), }}
fn parse_env(values: &[String]) -> Vec<(OsString, OsString)> { values .iter() .filter_map(|value| value.split_once('=')) .map(|(key, value)| (OsString::from(key), OsString::from(value))) .collect()}
#[cfg(test)]mod tests { use super::*;
#[test] fn refuses_root_exec_user() { let err = resolve_user("root").unwrap_err(); assert!(err.contains("refusing to run exec as privileged user")); }
#[test] fn refuses_root_exec_group() { let err = resolve_user("65534:0").unwrap_err(); assert!(err.contains("refusing to run exec as privileged user")); }}