use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::{mpsc, Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use crate::delegation::{
self, CompletionAddress, DelegationOutcome, DelegationStatus, Reporter, Spawner, SpawnerHandle,
SPAWNER_ENV,
};
use crate::errors::RuntimeError;
use crate::mac_token;
use crate::spawn_credential::SpawnApproval;
pub const MUR_BINARY_ENV: &str = "MURMUR_MUR_BINARY";
const CHILD_READY_TIMEOUT: Duration = Duration::from_secs(180);
#[cfg(unix)]
const CHILD_DIR_MODE: u32 = 0o700;
const CHILD_STDERR_TAIL_LINES: usize = 20;
const CHILD_STDERR_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
const CHILD_WATCH_INTERVAL: Duration = Duration::from_millis(100);
#[derive(Debug)]
pub struct ChildLaunchRequest {
pub parent_accessible_workdir: PathBuf,
pub capsule_name: String,
pub capsule_version: String,
pub grant: SpawnApproval,
pub child_env_allow: Vec<String>,
pub roost_url: String,
pub spawner: Option<Spawner>,
pub completion_deadline: Option<Duration>,
}
enum Ending {
Exited(Option<ExitStatus>),
Deliberate,
}
struct ChildProcess {
child: Option<Child>,
deliberate: bool,
status: Option<ExitStatus>,
}
impl ChildProcess {
fn poll(&mut self) -> Option<Ending> {
if self.deliberate {
return Some(Ending::Deliberate);
}
let Some(child) = self.child.as_mut() else {
return Some(Ending::Exited(self.status));
};
match child.try_wait() {
Ok(Some(status)) => {
self.status = Some(status);
Some(Ending::Exited(Some(status)))
}
Ok(None) => None,
Err(_) => Some(Ending::Exited(self.status)),
}
}
fn end(&mut self) -> Result<(), std::io::Error> {
let Some(mut child) = self.child.take() else {
return Ok(());
};
let _ = child.kill();
let status = child.wait()?;
self.status = Some(status);
Ok(())
}
}
pub struct LaunchedChild {
pub workdir: PathBuf,
pub session_id: String,
pub capsule_url: String,
pub argv: Vec<String>,
pub env: Vec<(String, String)>,
pub delegation_id: Option<String>,
pub door_token: Option<crate::door_auth::DoorToken>,
process: Arc<Mutex<ChildProcess>>,
stderr_tail: Arc<StderrTail>,
started: Instant,
released: bool,
}
impl std::fmt::Debug for LaunchedChild {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LaunchedChild")
.field("workdir", &self.workdir)
.field("session_id", &self.session_id)
.field("capsule_url", &self.capsule_url)
.field("delegation_id", &self.delegation_id)
.field(
"door_token",
&self.door_token.as_ref().map(|_| "<redacted>"),
)
.field("pid", &self.pid())
.finish()
}
}
impl LaunchedChild {
pub fn pid(&self) -> u32 {
lock(&self.process)
.child
.as_ref()
.map(Child::id)
.unwrap_or(0)
}
pub fn stderr_tail(&self) -> Vec<String> {
self.stderr_tail.lines()
}
pub fn shutdown(&mut self) -> Result<(), RuntimeError> {
let mut process = lock(&self.process);
process.deliberate = true;
process.end().map_err(|error| {
RuntimeError::Runtime(format!("failed to reap child capsule: {error}"))
})
}
pub fn release(mut self) {
self.released = true;
}
}
impl Drop for LaunchedChild {
fn drop(&mut self) {
if self.released {
return;
}
let mut process = lock(&self.process);
process.deliberate = true;
let _ = process.end();
}
}
pub(crate) fn workdir_relative_to(workdir: &Path, root: &Path) -> String {
workdir
.strip_prefix(root)
.unwrap_or(workdir)
.to_string_lossy()
.replace('\\', "/")
}
fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
mutex
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
pub fn child_workdir_for(parent_accessible_workdir: &Path, capsule_name: &str) -> PathBuf {
let suffix = mac_token::random_hex(8).unwrap_or_else(|_| {
format!(
"{:016x}",
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|elapsed| elapsed.as_nanos() as u64)
.unwrap_or(0)
)
});
parent_accessible_workdir
.join(".murmur")
.join("children")
.join(format!("{capsule_name}-{suffix}"))
}
pub fn launch_child_capsule(request: ChildLaunchRequest) -> Result<LaunchedChild, RuntimeError> {
let workdir = child_workdir_for(&request.parent_accessible_workdir, &request.capsule_name);
create_child_dir(&workdir)?;
let binary = mur_binary()?;
let argv: Vec<String> = vec![
binary.display().to_string(),
"run".to_string(),
"--capsule".to_string(),
request.capsule_name.clone(),
"--capsule-version".to_string(),
request.capsule_version.clone(),
"--workdir".to_string(),
workdir.display().to_string(),
"--json".to_string(),
"--no-env-file".to_string(),
"--spawn-grant-stdin".to_string(),
];
let handle = request
.spawner
.as_ref()
.map(|spawner| SpawnerHandle::for_delegation(spawner, delegation::new_delegation_id()));
let env = child_environment(&request, handle.as_ref());
let started = Instant::now();
let mut command = Command::new(&binary);
command
.args(&argv[1..])
.current_dir(&workdir)
.env_clear()
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
for (key, value) in &env {
command.env(key, value);
}
let mut child = command.spawn().map_err(|error| {
RuntimeError::Runtime(format!(
"failed to start the child capsule process '{}': {error}",
binary.display()
))
})?;
let write_result = child
.stdin
.take()
.ok_or_else(|| "the child process exposed no standard input".to_string())
.and_then(|mut stdin| {
writeln!(stdin, "{}", request.grant.expose())
.map_err(|error| format!("failed to hand the child its launch grant: {error}"))
});
let stderr_tail = drain_stderr(&mut child);
let mut launched = LaunchedChild {
workdir,
session_id: String::new(),
capsule_url: String::new(),
argv,
env,
delegation_id: handle.as_ref().map(|handle| handle.delegation_id.clone()),
door_token: None,
process: Arc::new(Mutex::new(ChildProcess {
child: Some(child),
deliberate: false,
status: None,
})),
stderr_tail: Arc::clone(&stderr_tail),
started,
released: false,
};
if let Err(reason) = write_result {
return Err(RuntimeError::Runtime(reason));
}
let line = first_json_line(&mut launched, &stderr_tail)?;
let report: serde_json::Value = serde_json::from_str(&line).map_err(|error| {
RuntimeError::Runtime(format!(
"the child capsule's first --json line did not parse: {error}"
))
})?;
let session_id = report
.get("session_id")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| {
RuntimeError::Runtime(
"the child capsule's --json line carried no session_id".to_string(),
)
})?;
let url = report
.get("url")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
launched.session_id = session_id.to_string();
launched.door_token = readiness_door_token(&report);
launched.capsule_url = if url.is_empty() {
String::new()
} else {
format!("http://{url}")
};
if let Some(handle) = handle {
if let Some(address) = handle.report_to.clone() {
watch_for_completion(
&launched,
handle,
address,
&request.capsule_name,
&request.capsule_version,
request.completion_deadline,
);
}
}
Ok(launched)
}
fn readiness_door_token(report: &serde_json::Value) -> Option<crate::door_auth::DoorToken> {
report
.pointer("/tokens/operator")
.and_then(serde_json::Value::as_str)
.filter(|token| !token.is_empty())
.map(|token| crate::door_auth::DoorToken::new(token.to_string()))
}
fn watch_for_completion(
launched: &LaunchedChild,
handle: SpawnerHandle,
address: CompletionAddress,
capsule_name: &str,
capsule_version: &str,
deadline: Option<Duration>,
) {
let process = Arc::clone(&launched.process);
let stderr_tail = Arc::clone(&launched.stderr_tail);
let workdir = launched.workdir.clone();
let session_id = launched.session_id.clone();
let capsule_name = capsule_name.to_string();
let capsule_version = capsule_version.to_string();
let started = launched.started;
let watching_from = Instant::now();
std::thread::spawn(move || {
let expires_at = deadline.map(|bound| watching_from + bound);
let mut expired = false;
let ending = loop {
if expires_at.is_some_and(|at| Instant::now() >= at) {
let mut process = lock(&process);
process.deliberate = true;
let _ = process.end();
expired = true;
break Ending::Deliberate;
}
if let Some(ending) = lock(&process).poll() {
break ending;
}
std::thread::sleep(CHILD_WATCH_INTERVAL);
};
let duration_ms = started.elapsed().as_millis().try_into().unwrap_or(u64::MAX);
if let Some(recorded) = delegation::read_completion(&workdir) {
if recorded.delivered {
return;
}
let retried = delegation::report_completion(&handle, &address, recorded, &workdir);
if !retried.delivered {
crate::runtime_err!(
"[capsule-runtime] delegation {}: the child's completion is undelivered after one retry",
handle.delegation_id
);
}
return;
}
let outcome = |status: DelegationStatus, detail: String| DelegationOutcome {
delegation_id: handle.delegation_id.clone(),
capsule_name: capsule_name.clone(),
capsule_version: capsule_version.clone(),
session_id: session_id.clone(),
status,
result_path: None,
workdir: workdir.display().to_string(),
duration_ms,
detail: Some(detail),
reported_by: Reporter::Launcher,
delivered: false,
delivery_error: None,
};
match ending {
Ending::Deliberate if expired => {
let bound = deadline.map(|bound| bound.as_secs()).unwrap_or_default();
delegation::report_completion(
&handle,
&address,
outcome(
DelegationStatus::Terminated,
format!(
"the capsule was still running after {bound}s and was ended at the \
delegation deadline"
),
),
&workdir,
);
}
Ending::Deliberate => {
delegation::record_terminated(
&workdir,
outcome(
DelegationStatus::Terminated,
"the parent ended this delegation".to_string(),
),
);
}
Ending::Exited(status) => {
let status_text = match status {
Some(status) => status.to_string(),
None => "unknown exit status".to_string(),
};
let tail = stderr_tail.lines_at_end().join("\n");
let detail = if tail.is_empty() {
format!(
"the child process ended without recording a completion ({status_text})"
)
} else {
format!(
"the child process ended without recording a completion ({status_text}); \
its last stderr lines were:\n{tail}"
)
};
delegation::report_completion(
&handle,
&address,
outcome(DelegationStatus::Crashed, detail),
&workdir,
);
}
}
});
}
fn child_environment(
request: &ChildLaunchRequest,
handle: Option<&SpawnerHandle>,
) -> Vec<(String, String)> {
let mut env: Vec<(String, String)> = Vec::new();
for key in &request.child_env_allow {
if let Ok(value) = std::env::var(key) {
env.push((key.clone(), value));
}
}
env.retain(|(key, _)| {
!matches!(key.as_str(), "PATH" | "HOME" | "MURMUR_ROOST_URL") && key != SPAWNER_ENV
});
if let Ok(path) = std::env::var("PATH") {
env.push(("PATH".to_string(), path));
}
if let Ok(home) = std::env::var("HOME") {
env.push(("HOME".to_string(), home));
}
env.push(("MURMUR_ROOST_URL".to_string(), request.roost_url.clone()));
if let Some(handle) = handle {
env.push((SPAWNER_ENV.to_string(), handle.to_env_value()));
}
env
}
fn mur_binary() -> Result<PathBuf, RuntimeError> {
if let Some(path) = std::env::var_os(MUR_BINARY_ENV) {
if !path.is_empty() {
return Ok(PathBuf::from(path));
}
}
std::env::current_exe().map_err(|error| {
RuntimeError::Runtime(format!(
"cannot locate the mur binary to launch a child capsule with: {error}"
))
})
}
fn create_child_dir(workdir: &Path) -> Result<(), RuntimeError> {
std::fs::create_dir_all(workdir).map_err(|source| RuntimeError::CreateWorkdir {
path: workdir.display().to_string(),
source,
})?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(workdir, std::fs::Permissions::from_mode(CHILD_DIR_MODE))
.map_err(|source| RuntimeError::CreateWorkdir {
path: workdir.display().to_string(),
source,
})?;
}
Ok(())
}
struct StderrTail {
state: Mutex<StderrState>,
drained: Condvar,
}
#[derive(Default)]
struct StderrState {
lines: Vec<String>,
at_eof: bool,
}
impl StderrTail {
fn lines(&self) -> Vec<String> {
lock(&self.state).lines.clone()
}
fn lines_at_end(&self) -> Vec<String> {
let (state, _) = self
.drained
.wait_timeout_while(lock(&self.state), CHILD_STDERR_DRAIN_TIMEOUT, |state| {
!state.at_eof
})
.unwrap_or_else(|poisoned| poisoned.into_inner());
state.lines.clone()
}
fn finish(&self) {
lock(&self.state).at_eof = true;
self.drained.notify_all();
}
}
fn drain_stderr(child: &mut Child) -> Arc<StderrTail> {
let tail = Arc::new(StderrTail {
state: Mutex::new(StderrState::default()),
drained: Condvar::new(),
});
let Some(stderr) = child.stderr.take() else {
tail.finish();
return tail;
};
let collector = Arc::clone(&tail);
std::thread::spawn(move || {
for line in BufReader::new(stderr).lines().map_while(Result::ok) {
crate::runtime_err!("{line}");
let mut state = lock(&collector.state);
if state.lines.len() == CHILD_STDERR_TAIL_LINES {
state.lines.remove(0);
}
state.lines.push(line);
}
collector.finish();
});
tail
}
fn first_json_line(
launched: &mut LaunchedChild,
stderr_tail: &StderrTail,
) -> Result<String, RuntimeError> {
let stdout = lock(&launched.process)
.child
.as_mut()
.and_then(|child| child.stdout.take())
.ok_or_else(|| {
RuntimeError::Runtime("the child process exposed no standard output".to_string())
})?;
let (tx, rx) = mpsc::channel::<Option<String>>();
std::thread::spawn(move || {
let mut reader = BufReader::new(stdout);
let mut line = String::new();
let first = match reader.read_line(&mut line) {
Ok(0) | Err(_) => None,
Ok(_) => Some(line.trim_end().to_string()),
};
let _ = tx.send(first);
let mut rest = String::new();
while let Ok(read) = reader.read_line(&mut rest) {
if read == 0 {
break;
}
crate::diagnostic::raw_to_stdout(&rest);
rest.clear();
}
});
match rx.recv_timeout(CHILD_READY_TIMEOUT) {
Ok(Some(line)) => Ok(line),
Ok(None) => {
let status = lock(&launched.process)
.child
.as_mut()
.and_then(|child| child.wait().ok())
.map(|status| status.to_string())
.unwrap_or_else(|| "unknown".to_string());
let reason = stderr_tail.lines_at_end().join("\n");
Err(RuntimeError::Runtime(format!(
"the child capsule '{}' exited without reporting a launch line ({status}): {reason}",
launched.workdir.display()
)))
}
Err(_) => Err(RuntimeError::Runtime(format!(
"the child capsule '{}' did not report a launch line within {}s",
launched.workdir.display(),
CHILD_READY_TIMEOUT.as_secs()
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_child_directory_is_unique_per_delegation() {
let parent = PathBuf::from("/tmp/parent");
let first = child_workdir_for(&parent, "worker");
let second = child_workdir_for(&parent, "worker");
assert_ne!(first, second);
assert_eq!(
first.parent(),
Some(parent.join(".murmur").join("children").as_path())
);
assert!(first
.file_name()
.unwrap()
.to_string_lossy()
.starts_with("worker-"));
assert_eq!(
first.file_name().unwrap().to_string_lossy().len(),
"worker-".len() + 16
);
}
fn request(env_allow: &[&str], spawner: Option<Spawner>) -> ChildLaunchRequest {
ChildLaunchRequest {
parent_accessible_workdir: PathBuf::from("/tmp/parent"),
capsule_name: "worker".to_string(),
capsule_version: "0.1.0".to_string(),
grant: SpawnApproval::new("msa1.token".to_string()),
child_env_allow: env_allow.iter().map(|name| name.to_string()).collect(),
roost_url: "http://127.0.0.1:7700".to_string(),
spawner,
completion_deadline: None,
}
}
fn decoy() -> String {
SpawnerHandle {
session_id: "ses_decoy".to_string(),
context_id: "ctx_decoy".to_string(),
delegation_id: "dlg_decoy".to_string(),
report_to: None,
}
.to_env_value()
}
fn spawner() -> Spawner {
Spawner {
session_id: "ses_parent".to_string(),
context_id: "ctx_parent".to_string(),
report_to: Some(CompletionAddress {
url: "http://127.0.0.1:7000".to_string(),
trust: crate::origin::TrustClass::Trusted,
}),
}
}
#[test]
fn a_childs_environment_is_built_from_its_own_declaration() {
std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_A", "a");
std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_B", "b");
let env = child_environment(&request(&["MURMUR_CHILD_LAUNCH_TEST_A"], None), None);
let names: Vec<&str> = env.iter().map(|(key, _)| key.as_str()).collect();
assert!(names.contains(&"MURMUR_CHILD_LAUNCH_TEST_A"));
assert!(!names.contains(&"MURMUR_CHILD_LAUNCH_TEST_B"));
assert!(names.contains(&"MURMUR_ROOST_URL"));
assert!(!env.iter().any(|(_, value)| value.contains("msa1.token")));
}
#[test]
fn a_runtime_owned_name_cannot_be_displaced_by_a_declaration() {
std::env::set_var("MURMUR_ROOST_URL", "http://127.0.0.1:1");
let env = child_environment(&request(&["MURMUR_ROOST_URL"], None), None);
let urls: Vec<&String> = env
.iter()
.filter(|(key, _)| key == "MURMUR_ROOST_URL")
.map(|(_, value)| value)
.collect();
assert_eq!(urls, vec!["http://127.0.0.1:7700"]);
}
#[test]
fn the_spawner_handle_is_injected_last_and_cannot_be_displaced() {
std::env::set_var(SPAWNER_ENV, decoy());
let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
let env = child_environment(&request(&[SPAWNER_ENV], Some(spawner())), Some(&handle));
let injected: Vec<&String> = env
.iter()
.filter(|(key, _)| key == SPAWNER_ENV)
.map(|(_, value)| value)
.collect();
assert_eq!(injected.len(), 1, "{env:?}");
assert_eq!(
SpawnerHandle::parse(injected[0]).expect("the injected value is a handle"),
handle
);
assert_eq!(
env.last().map(|(key, _)| key.as_str()),
Some(SPAWNER_ENV),
"the handle is applied in the runtime-owned tail: {env:?}"
);
std::env::remove_var(SPAWNER_ENV);
}
#[test]
fn a_launch_without_a_spawner_injects_no_handle() {
std::env::set_var(SPAWNER_ENV, decoy());
let env = child_environment(&request(&[SPAWNER_ENV], None), None);
assert!(!env.iter().any(|(key, _)| key == SPAWNER_ENV), "{env:?}");
std::env::remove_var(SPAWNER_ENV);
}
#[test]
fn a_tail_read_after_the_end_waits_for_the_drain_to_catch_up() {
let tail = Arc::new(StderrTail {
state: Mutex::new(StderrState::default()),
drained: Condvar::new(),
});
let writer = Arc::clone(&tail);
std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
lock(&writer.state)
.lines
.push("error[E-CAP-003]: not achievable on this host".to_string());
writer.finish();
});
assert!(tail.lines().is_empty());
assert_eq!(
tail.lines_at_end(),
vec!["error[E-CAP-003]: not achievable on this host".to_string()]
);
}
#[test]
fn a_tail_with_no_pipe_behind_it_is_at_its_end_already() {
let tail = StderrTail {
state: Mutex::new(StderrState::default()),
drained: Condvar::new(),
};
tail.finish();
let started = Instant::now();
assert!(tail.lines_at_end().is_empty());
assert!(
started.elapsed() < CHILD_STDERR_DRAIN_TIMEOUT,
"{started:?}"
);
}
#[test]
fn child_launch_reads_the_operator_token_from_the_readiness_line() {
let report = serde_json::json!({
"url": "localhost:1",
"session_id": "ses_x",
"tokens": {"operator": "mdt1.a.b", "watcher": "mdt1.c.d"},
});
assert_eq!(
readiness_door_token(&report).map(|token| token.expose().to_string()),
Some("mdt1.a.b".to_string())
);
let public = serde_json::json!({"url": "localhost:1", "session_id": "ses_x"});
assert_eq!(readiness_door_token(&public), None);
}
#[test]
fn child_launch_debug_prints_no_door_token() {
let child = LaunchedChild {
workdir: PathBuf::from("/tmp/child"),
session_id: "ses_child".to_string(),
capsule_url: "http://localhost:1".to_string(),
argv: Vec::new(),
env: Vec::new(),
delegation_id: None,
door_token: Some(crate::door_auth::DoorToken::new(
"mdt1.secretpayload.secretmac".to_string(),
)),
process: Arc::new(Mutex::new(ChildProcess {
child: None,
deliberate: false,
status: None,
})),
stderr_tail: Arc::new(StderrTail {
state: Mutex::new(StderrState::default()),
drained: Condvar::new(),
}),
started: Instant::now(),
released: true,
};
let debug = format!("{child:?}");
assert!(!debug.contains("secretpayload"), "{debug}");
assert!(debug.contains("door_token"), "{debug}");
}
}