aion-worker 0.26.0

//! What a declared command body actually does to a process.
//!
//! Its own file because its subjects are live process trees rather than
//! template shapes, and because [`super`] is already at the file-size law.

use std::collections::BTreeMap;

use aion_core::{ActivityId, RunId, WorkflowId};
use aion_package::{
    ArgvSlot, CommandParameterContract, DeclaredCommandContract, EnvBindingContract, FillPiece,
    FillTemplate,
};
use serde_json::json;

use super::DeclaredCommandAction;
use crate::activity::Classification;
use crate::context::{ActivityCancellationHandle, ActivityContext};

/// What a test returns. Every fallible step is carried rather than unwrapped,
/// because the workspace denies panicking accessors in test code as firmly as
/// in library code.
type TestResult = Result<(), Box<dyn std::error::Error>>;

fn context() -> (ActivityContext, ActivityCancellationHandle) {
    ActivityContext::new(
        WorkflowId::new_v4(),
        RunId::new_v4(),
        ActivityId::from_sequence_position(1),
        1,
    )
}

fn hole(parameter: &str) -> FillTemplate {
    FillTemplate {
        pieces: vec![FillPiece::Hole {
            parameter: parameter.to_owned(),
        }],
    }
}

fn slot(fill: FillTemplate, label: &str) -> ArgvSlot {
    ArgvSlot {
        fill,
        label: label.to_owned(),
        admits_leading_dash: true,
    }
}

fn contract(program: &[&str], args: Vec<ArgvSlot>) -> DeclaredCommandContract {
    DeclaredCommandContract {
        name: "probe".to_owned(),
        parameters: Vec::new(),
        program: program.iter().map(|word| (*word).to_owned()).collect(),
        args,
        env: Vec::new(),
        cwd: None,
        hardened_path: None,
        timeout_ms: None,
        timeout_owner: None,
    }
}

fn parameter(name: &str) -> CommandParameterContract {
    CommandParameterContract {
        name: name.to_owned(),
        list: false,
        default: None,
    }
}

fn arguments(pairs: &[(&str, serde_json::Value)]) -> BTreeMap<String, serde_json::Value> {
    pairs
        .iter()
        .map(|(name, value)| ((*name).to_owned(), value.clone()))
        .collect()
}

#[tokio::test]
async fn a_declared_command_runs_and_returns_its_output() -> TestResult {
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(&arguments(&[("who", json!("world"))]), &context)
        .await?;
    assert_eq!(outcome.exit_code, 0);
    assert_eq!(outcome.stdout, "world");
    Ok(())
}

/// THE PROPERTY THE WHOLE SURFACE EXISTS FOR: a value that would be four
/// commands under a shell is one inert argv word here, because no shell ever
/// sees it. If this regresses, every other guarantee is decoration.
#[tokio::test]
async fn a_hostile_fill_arrives_as_exactly_one_literal_argv_word() -> TestResult {
    let mut command = contract(&["printf"], vec![]);
    command.args = vec![
        slot(FillTemplate::literal("[%s]"), "format"),
        slot(hole("value"), "value"),
    ];
    command.parameters = vec![parameter("value")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(
            &arguments(&[("value", json!("$(boom); rm -rf / && echo PWNED"))]),
            &context,
        )
        .await?;
    assert_eq!(outcome.stdout, "[$(boom); rm -rf / && echo PWNED]");
    assert!(
        !outcome.stdout.contains("PWNED\n"),
        "the injected command must never have run"
    );
    Ok(())
}

#[tokio::test]
async fn a_surplus_action_parameter_is_not_a_refusal() -> TestResult {
    // An action may declare more than the command it serves needs; the
    // command's own names select from the input.
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(
            &arguments(&[("who", json!("world")), ("unused", json!("ignored"))]),
            &context,
        )
        .await?;
    assert_eq!(outcome.stdout, "world");
    Ok(())
}

#[tokio::test]
async fn a_declared_env_binding_reaches_the_child() -> TestResult {
    let mut command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal("echo \"[$DECLARED_GREETING]\""),
                "body",
            ),
        ],
    );
    command.env = vec![EnvBindingContract {
        name: "DECLARED_GREETING".to_owned(),
        value: hole("greeting"),
    }];
    command.parameters = vec![parameter("greeting")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action
        .run(&arguments(&[("greeting", json!("supplied"))]), &context)
        .await?;
    assert_eq!(outcome.stdout, "[supplied]");
    Ok(())
}

#[tokio::test]
async fn the_hosts_environment_does_not_cross_into_a_declared_command() -> TestResult {
    // A host process legitimately holds credentials. A deployed package must
    // not be able to read them by declaring a command that prints them. Reads
    // a variable the host ALREADY has rather than setting one, because
    // `std::env::set_var` is unsafe and the workspace denies unsafe outright.
    let Some(present) = std::env::vars_os()
        .filter_map(|(name, _)| name.into_string().ok())
        .find(|name| name != "PATH" && !name.is_empty() && !name.contains('='))
    else {
        tracing::info!(
            "skipping: the host has no environment variable besides PATH to prove \
             non-inheritance with"
        );
        return Ok(());
    };
    let command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(
                FillTemplate::literal(format!("echo \"[${{{present}:-absent}}]\"")),
                "body",
            ),
        ],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(
        outcome.stdout, "[absent]",
        "the host's `{present}` leaked into a declared command"
    );
    Ok(())
}

#[tokio::test]
async fn a_hardened_path_wins_over_an_env_binding_of_the_same_name() -> TestResult {
    let mut command = contract(
        &["/bin/sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo \"$PATH\""), "body"),
        ],
    );
    command.env = vec![EnvBindingContract {
        name: "PATH".to_owned(),
        value: FillTemplate::literal("/should/not/win"),
    }];
    command.hardened_path = Some(FillTemplate::literal("/usr/bin:/bin"));
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "/usr/bin:/bin");
    Ok(())
}

#[tokio::test]
async fn a_command_runs_in_the_working_directory_it_was_given() -> TestResult {
    let command = contract(&["pwd"], Vec::new());
    let action = DeclaredCommandAction::new(command).with_working_directory("/");
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "/");
    Ok(())
}

#[test]
fn the_declared_working_directory_is_handed_back_unexpanded() {
    let mut command = contract(&["pwd"], Vec::new());
    command.cwd = Some("{workspace_root}/clones".to_owned());
    let action = DeclaredCommandAction::new(command);
    assert_eq!(
        action.declared_working_directory(),
        Some("{workspace_root}/clones"),
        "resolving the placeholder is the executing host's act, not this crate's"
    );
}

#[tokio::test]
async fn a_nonzero_exit_is_retryable_and_carries_the_exit_code_and_stderr() -> TestResult {
    let command = contract(
        &["sh"],
        vec![
            slot(FillTemplate::literal("-c"), "-c"),
            slot(FillTemplate::literal("echo trouble >&2; exit 3"), "body"),
        ],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a non-zero exit must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Retryable);
    assert!(
        failure.message().contains("trouble"),
        "stderr must ride the failure: {}",
        failure.message()
    );
    assert!(
        failure.message().contains('3'),
        "the exit code is reported: {}",
        failure.message()
    );
    Ok(())
}

/// A millisecond count no duration can hold is REFUSED, not read as "no
/// ceiling". The direction of that failure is a command running unbounded on
/// the operator's machine while the declaration, the `--check` report and the
/// named owner all say a bound was set — so the swallow would be silent in the
/// worst possible direction.
#[tokio::test]
async fn a_timeout_that_is_not_a_duration_refuses_rather_than_lifting_the_ceiling() -> TestResult {
    let mut command = contract(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    );
    command.timeout_ms = Some(-1);
    command.timeout_owner = Some("release".to_owned());
    let action = DeclaredCommandAction::new(command);
    let Err(failure) = action.declared_timeout() else {
        return Err("a negative millisecond count is not a duration".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("release"),
        "the refusal must name the owner whose ceiling would have been lifted: {}",
        failure.message()
    );

    // And the run refuses before spawning, rather than running unbounded.
    let (context, _handle) = context();
    let started = std::time::Instant::now();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a defective ceiling must not run the command unbounded".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        started.elapsed() < std::time::Duration::from_secs(5),
        "the refusal must precede the spawn"
    );
    Ok(())
}

/// A cancel and the declared bound can become ready in the same poll. The
/// select is BIASED so the cancel wins: both outcomes are terminal, so the
/// only thing at stake is which one the failure names, and sending an operator
/// to the ceiling's owner over a stop somebody else asked for is a wrong
/// answer to the question they are actually holding.
#[tokio::test]
async fn a_cancel_racing_the_declared_bound_is_reported_as_a_cancel() -> TestResult {
    for _ in 0..8 {
        let mut command = contract(
            &["sleep"],
            vec![slot(FillTemplate::literal("30"), "seconds")],
        );
        command.timeout_ms = Some(200);
        command.timeout_owner = Some("release".to_owned());
        let action = DeclaredCommandAction::new(command);
        let (context, handle) = context();
        let run = tokio::spawn(async move { action.run(&BTreeMap::new(), &context).await });
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        handle.cancel();
        let Err(failure) = run.await? else {
            return Err("a stopped command must fail the activity".into());
        };
        assert_eq!(failure.classification(), &Classification::Terminal);
        assert!(
            !failure.message().contains("declared timeout")
                || !failure.message().contains("cancelled"),
            "one cause, not both: {}",
            failure.message()
        );
    }
    Ok(())
}

#[tokio::test]
async fn a_declared_timeout_stops_the_command_and_names_its_owner() -> TestResult {
    let mut command = contract(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    );
    command.timeout_ms = Some(250);
    command.timeout_owner = Some("release".to_owned());
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let started = std::time::Instant::now();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a command outliving its declared timeout must fail".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("declared timeout"),
        "the failure must name the bound that fired: {}",
        failure.message()
    );
    assert!(
        failure.message().contains("release"),
        "a declared ceiling is a number somebody chose: {}",
        failure.message()
    );
    assert!(
        started.elapsed() < std::time::Duration::from_secs(25),
        "the bound must be enforced where the process is, not left to the caller"
    );
    Ok(())
}

#[tokio::test]
async fn a_command_with_no_declared_timeout_is_not_given_one() -> TestResult {
    // The absence of a ceiling is a legal declared state, and nothing here
    // substitutes one. A short-lived command must simply run to its own end.
    let command = contract(&["echo"], vec![slot(FillTemplate::literal("done"), "word")]);
    let action = DeclaredCommandAction::new(command);
    assert_eq!(action.declared_timeout()?, None);
    let (context, _handle) = context();
    let outcome = action.run(&BTreeMap::new(), &context).await?;
    assert_eq!(outcome.stdout, "done");
    Ok(())
}

#[tokio::test]
async fn cancellation_stops_the_command_and_fails_terminally() -> TestResult {
    let command = contract(
        &["sleep"],
        vec![slot(FillTemplate::literal("30"), "seconds")],
    );
    let action = DeclaredCommandAction::new(command);
    let (context, handle) = context();
    let run = tokio::spawn(async move { action.run(&BTreeMap::new(), &context).await });
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    handle.cancel();
    let Err(failure) = run.await? else {
        return Err("a cancelled command must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(
        failure.message().contains("cancelled"),
        "a cancellation must not be reported as a timeout: {}",
        failure.message()
    );
    Ok(())
}

#[tokio::test]
async fn a_missing_parameter_fails_terminally_before_anything_runs() -> TestResult {
    let mut command = contract(&["echo"], vec![slot(hole("who"), "who")]);
    command.parameters = vec![parameter("who")];
    let action = DeclaredCommandAction::new(command);
    let (context, _handle) = context();
    let Err(failure) = action.run(&BTreeMap::new(), &context).await else {
        return Err("a missing parameter must fail the activity".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(failure.message().contains("who"), "{}", failure.message());
    Ok(())
}

#[test]
fn a_capture_decides_the_result_and_only_the_result() -> TestResult {
    use aion_package::contract::CommandBodyCapture;

    let outcome = super::ShellOutcome {
        exit_code: 0,
        stdout: "{\"name\":\"world\"}".to_owned(),
        stderr: String::new(),
    };
    assert_eq!(
        super::shape_command_result("greet", CommandBodyCapture::Text, outcome.clone())?,
        json!("{\"name\":\"world\"}")
    );
    assert_eq!(
        super::shape_command_result("greet", CommandBodyCapture::Json, outcome)?,
        json!({ "name": "world" })
    );
    Ok(())
}

#[test]
fn a_json_capture_over_output_that_is_not_json_refuses_terminally() -> TestResult {
    use aion_package::contract::CommandBodyCapture;

    let outcome = super::ShellOutcome {
        exit_code: 0,
        stdout: "not json".to_owned(),
        stderr: String::new(),
    };
    let Err(failure) = super::shape_command_result("greet", CommandBodyCapture::Json, outcome)
    else {
        return Err("output that is not JSON must refuse a `json` capture".into());
    };
    assert_eq!(failure.classification(), &Classification::Terminal);
    assert!(failure.message().contains("greet"), "{}", failure.message());
    Ok(())
}