liminal-server 0.25.4

//! Starting and stopping a service's template process (SRV-028 R4).
//!
//! The process gets only the template's args, the environment names the
//! template passes through, and four variables of its own: `LIMINAL_SERVER`,
//! `LIMINAL_SERVICE`, `LIMINAL_SUBJECTS` (comma-separated) and
//! `LIMINAL_TOKEN`, a credential made for this process alone, kept only as
//! its digest in memory, that admits registering its subjects and nothing
//! else. Read-only configuration templates retain their existing log. Approved
//! record templates discard process output so payloads cannot enter a log. A
//! watcher thread waits on it and reports its exit to the registry.

use std::io::Read;
use std::os::unix::process::CommandExt;
use std::process::{Command, Stdio};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

use super::{Lifecycle, LifecycleError, ServiceRecord};
use crate::config::types::ServiceTemplateConfig;

/// The system's kill program; std signals a child only with SIGKILL, and the
/// crate forbids `unsafe`.
const KILL: &str = "/bin/kill";

/// How long a stop waits for the exit after SIGKILL.
const KILL_WAIT: Duration = Duration::from_secs(5);

/// A started process.
#[derive(Debug)]
pub(super) struct Running {
    /// Its process id.
    pid: u32,
    /// The digest of its credential.
    pub(super) digest: [u8; 32],
    /// The holder that started it.
    pub(super) starter: String,
    /// Set by the watcher when the process has exited.
    exited: Arc<(Mutex<bool>, Condvar)>,
}

/// Starts `record`'s `template` for `starter`; answers the process and its
/// credential's digest.
pub(super) fn start(
    lifecycle: &Arc<Lifecycle>,
    record: &ServiceRecord,
    template: &ServiceTemplateConfig,
    approved: Option<&super::templates::TemplateRecord>,
    starter: &str,
) -> Result<(Running, [u8; 32]), LifecycleError> {
    let staged = approved
        .map(|approved| lifecycle.stage_template(approved))
        .transpose()
        .map_err(|error| LifecycleError::Process(error.to_string()))?;
    let server = lifecycle
        .server_address()
        .ok_or_else(|| LifecycleError::Process("the server is not listening yet".to_owned()))?;
    let token = credential()?;
    let (output, errors) = if approved.is_some() {
        (Stdio::null(), Stdio::null())
    } else {
        let log = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(lifecycle.log_path(&record.id))
            .map_err(|error| LifecycleError::Process(format!("log: {}", error.kind())))?;
        let error_log = log
            .try_clone()
            .map_err(|error| LifecycleError::Process(format!("log: {}", error.kind())))?;
        (Stdio::from(log), Stdio::from(error_log))
    };
    let mut command = Command::new(staged.as_ref().map_or(
        template.command.as_path(),
        super::templates::StagedCommand::command,
    ));
    if staged.is_some() {
        command.arg0(&template.command);
    }
    command
        .args(&template.args)
        .env_clear()
        .stdin(Stdio::null())
        .stdout(output)
        .stderr(errors);
    for name in &template.env {
        if let Some(value) = std::env::var_os(name) {
            command.env(name, value);
        }
    }
    command
        .env("LIMINAL_SERVER", server)
        .env("LIMINAL_SERVICE", &record.id)
        .env("LIMINAL_SUBJECTS", record.subjects.join(","))
        .env("LIMINAL_TOKEN", token.as_str());
    let (pid, exited) = spawn_watched(&mut command, lifecycle, &record.id, staged)?;
    drop(command);
    let digest = *blake3::hash(token.as_bytes()).as_bytes();
    tracing::info!(service = %record.id, pid, "service process started");
    Ok((
        Running {
            pid,
            digest,
            starter: starter.to_owned(),
            exited,
        },
        digest,
    ))
}

type ExitSignal = Arc<(Mutex<bool>, Condvar)>;

fn spawn_watched(
    command: &mut Command,
    lifecycle: &Arc<Lifecycle>,
    id: &str,
    staged: Option<super::templates::StagedCommand>,
) -> Result<(u32, ExitSignal), LifecycleError> {
    // One ownership handoff for one child; no child exists unless its watcher
    // was created. A failed spawn closes and joins that waiting watcher.
    let (send, receive) = std::sync::mpsc::sync_channel::<(
        std::process::Child,
        Option<super::templates::StagedCommand>,
    )>(1);
    let mut child_signals = signal_hook::iterator::Signals::new([signal_hook::consts::SIGCHLD])
        .map_err(|error| LifecycleError::Process(format!("child_exit_signal: {}", error.kind())))?;
    let exited = Arc::new((Mutex::new(false), Condvar::new()));
    let seen = Arc::clone(&exited);
    let watcher = Arc::clone(lifecycle);
    let id = id.to_owned();
    let thread = std::thread::Builder::new()
        .name(format!("service-{id}"))
        .spawn(move || {
            let Ok((mut child, staged)) = receive.recv() else {
                return;
            };
            let status = observe_exit(&mut child, &mut child_signals, &seen)
                .unwrap_or_else(std::convert::identity);
            drop(staged);
            seen.1.notify_all();
            watcher.exited(&id, &status, now_ms());
        })
        .map_err(|error| LifecycleError::Process(format!("watcher: {}", error.kind())))?;
    let child = match command.spawn() {
        Ok(child) => child,
        Err(error) => {
            drop(send);
            thread
                .join()
                .map_err(|_| LifecycleError::Process("watcher_join_failed".into()))?;
            return Err(LifecycleError::Process(format!(
                "command could not start: {}",
                error.kind()
            )));
        }
    };
    let pid = child.id();
    if let Err(error) = send.send((child, staged)) {
        let (mut child, staged) = error.0;
        let killed = child.kill();
        let joined = child.wait();
        drop(staged);
        thread
            .join()
            .map_err(|_| LifecycleError::Process("watcher_join_failed".into()))?;
        killed.map_err(|error| {
            LifecycleError::Process(format!("unwatched_child_signal_failed: {}", error.kind()))
        })?;
        joined.map_err(|error| {
            LifecycleError::Process(format!("unwatched_child_join_failed: {}", error.kind()))
        })?;
        return Err(LifecycleError::Process("watcher_handoff_failed".into()));
    }
    Ok((pid, exited))
}

fn observe_exit(
    child: &mut std::process::Child,
    signals: &mut signal_hook::iterator::Signals,
    seen: &ExitSignal,
) -> Result<String, String> {
    // Reap under the same hold as stop's signal. Publishing the exit flag after
    // an unlocked wait would expose a reaped, potentially reused process id.
    for _ in signals.forever() {
        let Ok(mut done) = seen.0.lock() else {
            cleanup_child(child)?;
            return Err("child_exit_signal_poisoned".into());
        };
        match child.try_wait() {
            Ok(Some(status)) => {
                *done = true;
                return Ok(status.to_string());
            }
            Ok(None) => {}
            Err(error) => {
                let killed = child.kill();
                let joined = child.wait();
                *done = true;
                drop(done);
                killed.map_err(|error| format!("child_cleanup_signal_failed: {}", error.kind()))?;
                joined.map_err(|error| format!("child_cleanup_join_failed: {}", error.kind()))?;
                return Err(format!("child_wait_failed: {}", error.kind()));
            }
        }
    }
    cleanup_child(child)?;
    Err("child_exit_signal_closed".into())
}

fn cleanup_child(child: &mut std::process::Child) -> Result<(), String> {
    if child
        .try_wait()
        .map_err(|error| format!("child_cleanup_state_failed: {}", error.kind()))?
        .is_none()
    {
        child
            .kill()
            .map_err(|error| format!("child_cleanup_signal_failed: {}", error.kind()))?;
        child
            .wait()
            .map_err(|error| format!("child_cleanup_join_failed: {}", error.kind()))?;
    }
    Ok(())
}

/// Sends SIGTERM, waits `timeout_ms`, then SIGKILL and waits again; a
/// process that already exited is never signalled.
pub(super) fn stop(process: &Running, timeout_ms: u64) -> Result<(), String> {
    if signal(process, "TERM")? {
        return Ok(());
    }
    if wait(process, Duration::from_millis(timeout_ms)) {
        return Ok(());
    }
    if signal(process, "KILL")? {
        return Ok(());
    }
    if wait(process, KILL_WAIT) {
        Ok(())
    } else {
        Err(format!(
            "process {} did not exit after SIGKILL",
            process.pid
        ))
    }
}

/// Signals the process with `kill -<name>` unless it has exited; `Ok(true)`
/// when it had. The exit flag is held across the signal, so the pid is still
/// our unreaped child and cannot have been reused.
fn signal(process: &Running, name: &str) -> Result<bool, String> {
    let (flag, _) = &*process.exited;
    let done = flag
        .lock()
        .map_err(|_| "the exit flag is poisoned".to_owned())?;
    if *done {
        return Ok(true);
    }
    let sent = Command::new(KILL)
        .arg(format!("-{name}"))
        .arg(process.pid.to_string())
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
    drop(done);
    match sent {
        Ok(status) if status.success() => Ok(false),
        Ok(status) => Err(format!("kill -{name} {} answered {status}", process.pid)),
        Err(error) => Err(format!(
            "kill -{name} {} could not run: {}",
            process.pid,
            error.kind()
        )),
    }
}

/// Waits up to `limit` for the exit; whether it came.
fn wait(process: &Running, limit: Duration) -> bool {
    let (flag, signal) = &*process.exited;
    let deadline = Instant::now() + limit;
    let Ok(mut done) = flag.lock() else {
        return false;
    };
    while !*done {
        let left = deadline.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return false;
        }
        match signal.wait_timeout(done, left) {
            Ok((guard, _)) => done = guard,
            Err(_) => return false,
        }
    }
    true
}

/// 32 random bytes as 64 hex digits.
fn credential() -> Result<zeroize::Zeroizing<String>, LifecycleError> {
    let mut bytes = zeroize::Zeroizing::new([0_u8; 32]);
    std::fs::File::open("/dev/urandom")
        .and_then(|mut source| source.read_exact(&mut *bytes))
        .map_err(|error| {
            LifecycleError::Process(format!("no random credential: {}", error.kind()))
        })?;
    Ok(zeroize::Zeroizing::new(hex::encode(*bytes)))
}

/// Milliseconds since the epoch.
pub(super) fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |since| {
            u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
        })
}