diff --git a/Cargo.lock b/Cargo.lock index a4815cf0..0a842b00 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1112,6 +1112,7 @@ dependencies = [ "command-group", "nix 0.30.1", "tempfile", + "tinybox-bus", "tinybox-core", "tinybox-jail", "tokio", diff --git a/crates/tinybox-bus/src/lib.rs b/crates/tinybox-bus/src/lib.rs index c4394bd7..8a10f1db 100644 --- a/crates/tinybox-bus/src/lib.rs +++ b/crates/tinybox-bus/src/lib.rs @@ -21,6 +21,11 @@ pub use grants::{LocalGrantConfig, LocalGrants, ProtectedPaths, ResolveLocalGran mod jail; pub use jail::{JailEnforcement, NativeJailConfig}; +mod shell_execution; +mod shell_selection; +pub use shell_execution::ShellExecRequest; +pub use shell_selection::{SelectShellRequest, ShellFlavor, ShellInvocation}; + mod shell; pub use shell::{CommandAnalysis, CommandClass, CommandSegment}; mod files; @@ -42,8 +47,10 @@ pub use files::{ /// bounded, module-owned one-shot Docker operation. 1.9 adds bounded Docker /// daemon status and label-scoped cleanup operations. 1.10 adds a bounded /// Docker command that does not mount a host workspace. 1.11 adds local grant -/// discovery. -pub const CONTRACT_VERSION: (u32, u32) = (1, 11); +/// discovery. 1.12 adds platform shell selection. 1.13 adds supervised native +/// shell execution on local passthrough resources. +/// 1.14 adds native shell execution with owned scratch, timeout and optional jail. +pub const CONTRACT_VERSION: (u32, u32) = (1, 14); /// Whether this host vocabulary can bind to a module's advertised version. /// The major must match and the module's minor must include every host member. @@ -117,6 +124,9 @@ pub const METHODS: &[&str] = &[ "CleanupDocker", "RunDockerWithoutWorkspace", "ResolveLocalGrants", + "SelectShell", + "StartShellExec", + "StartNativeShellExec", ]; /// Facts about the detected native directory-jail backend. @@ -554,3 +564,6 @@ mod tests; mod output; pub use output::{ExecutionFailure, ExecutionState, OutputBatch, OutputChunk, OutputStream}; + +/// Invalid native shell execution inputs. +pub const INVALID_SHELL_EXEC: &str = "ai.tinyhumans.tinybox.Error.InvalidShellExec"; diff --git a/crates/tinybox-bus/src/lib_tests.rs b/crates/tinybox-bus/src/lib_tests.rs index cb5d7473..68328cdd 100644 --- a/crates/tinybox-bus/src/lib_tests.rs +++ b/crates/tinybox-bus/src/lib_tests.rs @@ -46,15 +46,12 @@ fn docker_run_vocabulary_is_typed_bounded_and_additive() -> Result<(), serde_jso label: Some("openhuman.sandbox=true".into()), timeout_ms: 120_000, }; - assert!(METHODS.contains(&"RunDocker")); - assert_eq!(METHODS.get(METHODS.len() - 3), Some(&"CleanupDocker")); - assert_eq!( - METHODS.get(METHODS.len() - 2), - Some(&"RunDockerWithoutWorkspace") - ); - assert_eq!(METHODS.last(), Some(&"ResolveLocalGrants")); - assert!(METHODS.contains(&"RunDockerWithoutWorkspace")); - assert_eq!(CONTRACT_VERSION, (1, 11)); + assert_eq!(METHODS[27], "RunDocker"); + assert_eq!(METHODS[28], "DockerStatus"); + assert_eq!(METHODS[29], "CleanupDocker"); + assert_eq!(METHODS[30], "RunDockerWithoutWorkspace"); + assert_eq!(METHODS.last(), Some(&"StartNativeShellExec")); + assert_eq!(CONTRACT_VERSION, (1, 14)); assert_eq!(MAX_DOCKER_RUN_TIMEOUT_MS, 86_400_000); let serialized = serde_json::to_value(&request)?; assert_eq!(serialized["read_only_rootfs"], false); @@ -132,7 +129,7 @@ fn host_gateway_and_docker_facts_are_additive_and_default_safely() -> Result<(), #[test] fn version_and_capability_wire_snapshot_preserve_contract_compatibility() -> Result<(), serde_json::Error> { - assert_eq!(CONTRACT_VERSION, (1, 11)); + assert_eq!(CONTRACT_VERSION, (1, 14)); assert!(is_compatible(CONTRACT_VERSION)); assert!(!is_compatible((1, 8))); assert!(!is_compatible((1, 7))); @@ -150,7 +147,7 @@ fn version_and_capability_wire_snapshot_preserve_contract_compatibility() exec_backends: Vec::new(), spawn_backends: Vec::new(), }; - let snapshot = serde_json::json!({"contract_version":[1,11],"create_backends":["passthrough"],"exec_backends":[],"spawn_backends":[]}); + let snapshot = serde_json::json!({"contract_version":[1,14],"create_backends":["passthrough"],"exec_backends":[],"spawn_backends":[]}); assert_eq!(serde_json::to_value(&capabilities)?, snapshot); assert_eq!( serde_json::from_value::(snapshot)?, @@ -161,10 +158,7 @@ fn version_and_capability_wire_snapshot_preserve_contract_compatibility() #[test] fn no_workspace_docker_operation_is_present_in_the_contract() { - assert_eq!( - METHODS.get(METHODS.len() - 2), - Some(&"RunDockerWithoutWorkspace") - ); + assert_eq!(METHODS.get(30), Some(&"RunDockerWithoutWorkspace")); assert_eq!( METHODS .iter() @@ -172,7 +166,7 @@ fn no_workspace_docker_operation_is_present_in_the_contract() { .count(), 1 ); - assert_eq!(CONTRACT_VERSION, (1, 11)); + assert_eq!(CONTRACT_VERSION, (1, 14)); } #[test] @@ -278,3 +272,60 @@ fn local_grant_configuration_defaults_and_floor_round_trip() -> Result<(), serde ); Ok(()) } + +#[test] +fn shell_selection_wire_defaults_preserve_safe_pipeline_behavior() -> Result<(), serde_json::Error> +{ + let request: SelectShellRequest = + serde_json::from_value(serde_json::json!({"command":"false | true"}))?; + assert!(!request.preserve_pipeline_status); + let invocation = ShellInvocation { + flavor: ShellFlavor::Cmd, + program: "cmd".into(), + arguments: vec!["/C".into()], + raw_arguments: Some(r#"echo "quoted payload""#.into()), + bootstrap_environment: vec!["SystemRoot".into()], + remove_environment: Vec::new(), + }; + let value = serde_json::to_value(&invocation)?; + assert_eq!(value["flavor"], "cmd"); + assert_eq!(value["raw_arguments"], r#"echo "quoted payload""#); + assert_eq!( + serde_json::from_value::(value)?, + invocation + ); + assert_eq!(METHODS[31], "ResolveLocalGrants"); + assert_eq!(METHODS[32], "SelectShell"); + assert!(!is_compatible((1, 9))); + Ok(()) +} + +#[test] +fn older_shell_invocations_default_the_startup_removal_vocabulary() -> Result<(), serde_json::Error> +{ + let invocation: ShellInvocation = serde_json::from_value(serde_json::json!({ + "flavor":"posix", "program":"sh", "arguments":["-c","true"], + "raw_arguments":null, "bootstrap_environment":[] + }))?; + assert_eq!(invocation.remove_environment, Vec::::new()); + Ok(()) +} + +#[test] +fn native_shell_wire_defaults_close_stdin_and_keep_a_bounded_runtime() +-> Result<(), serde_json::Error> { + let request: ShellExecRequest = serde_json::from_value(serde_json::json!({ + "process":"process", "resource":"resource", "shell":{"command":"true"}, + "state_dir":"/state" + }))?; + assert_eq!(request.timeout_ms, 30_000); + assert_eq!(request.stdin, None); + assert_eq!(request.cwd, None); + assert_eq!(request.jail, None); + assert_eq!(request.extra_env, std::collections::BTreeMap::new()); + assert_eq!(METHODS[33], "StartShellExec"); + assert_eq!(METHODS[34], "StartNativeShellExec"); + assert!(!is_compatible((1, 13))); + assert!(!is_compatible((1, 10))); + Ok(()) +} diff --git a/crates/tinybox-bus/src/output.rs b/crates/tinybox-bus/src/output.rs index 1218cb5b..562fc5e8 100644 --- a/crates/tinybox-bus/src/output.rs +++ b/crates/tinybox-bus/src/output.rs @@ -27,6 +27,8 @@ pub struct OutputChunk { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ExecutionFailure { + /// The runtime budget elapsed; supervised native cleanup is acknowledged. + TimedOut, /// The native operation could not start. StartFailed, /// Combined output exceeded the execution budget. diff --git a/crates/tinybox-bus/src/shell_execution.rs b/crates/tinybox-bus/src/shell_execution.rs new file mode 100644 index 00000000..0e1fa6a6 --- /dev/null +++ b/crates/tinybox-bus/src/shell_execution.rs @@ -0,0 +1,35 @@ +//! Reserved native shell execution; authorization remains caller-owned. +use crate::{NativeJailConfig, ResourceId, SelectShellRequest}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +/// Start a local shell under module-owned supervision and output replay. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ShellExecRequest { + /// Caller-known process reservation. + pub process: ResourceId, + /// Existing local passthrough resource. + pub resource: ResourceId, + /// Shell payload and pipeline-status convention. + pub shell: SelectShellRequest, + /// Working directory inside the resource; absent means its root. + #[serde(default)] + pub cwd: Option, + /// Approved overrides on the module's safe parent environment snapshot. + #[serde(default)] + pub extra_env: BTreeMap, + /// Approved existing absolute private state directory for owned scratch. + pub state_dir: String, + /// Positive runtime budget, default thirty seconds, at most twenty-four hours. + #[serde(default = "default_timeout")] + pub timeout_ms: u64, + /// Binary input; absent closes stdin. + #[serde(default)] + pub stdin: Option>, + /// Explicit native confinement; absent is ordinary local execution. + #[serde(default)] + pub jail: Option, +} +const fn default_timeout() -> u64 { + 30_000 +} diff --git a/crates/tinybox-bus/src/shell_selection.rs b/crates/tinybox-bus/src/shell_selection.rs new file mode 100644 index 00000000..27ba1153 --- /dev/null +++ b/crates/tinybox-bus/src/shell_selection.rs @@ -0,0 +1,45 @@ +//! Platform shell invocation vocabulary; selecting a shell grants no execution access. + +use serde::{Deserialize, Serialize}; + +/// Command text and the desired pipeline exit-status convention. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SelectShellRequest { + /// Text interpreted by the platform shell. + pub command: String, + /// Preserve the shell's ordinary last-stage pipeline status rather than + /// enabling Bash pipefail. Windows cmd has no pipefail mode. + #[serde(default)] + pub preserve_pipeline_status: bool, +} + +/// Shell syntax used by the module machine. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ShellFlavor { + /// Windows cmd syntax, including percent-variable expansion. + Cmd, + /// POSIX shell syntax, using Bash when available. + Posix, +} + +/// Selected invocation, returned for discovery and consumed inside `TinyBox`. +/// Hosts execute commands through the module rather than spawning this plan. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ShellInvocation { + /// Syntax understood by the selected shell. + pub flavor: ShellFlavor, + /// Shell executable on the module machine. + pub program: String, + /// Ordinary process arguments, excluding the executable. + pub arguments: Vec, + /// Windows command payload appended verbatim after ordinary arguments. + /// Rust's ordinary argument escaping is incompatible with cmd quotes. + pub raw_arguments: Option, + /// Parent environment names needed to bootstrap Windows child tools. + /// Values are never returned by shell discovery. + pub bootstrap_environment: Vec, + /// Remove these startup-hook variables after applying the approved environment. + #[serde(default)] + pub remove_environment: Vec, +} diff --git a/crates/tinybox-host/Cargo.toml b/crates/tinybox-host/Cargo.toml index 5f011804..9c022f2a 100644 --- a/crates/tinybox-host/Cargo.toml +++ b/crates/tinybox-host/Cargo.toml @@ -14,6 +14,8 @@ publish = false [dependencies] tinybox-core.workspace = true +# Shell selection shares the module's serialized invocation vocabulary. +tinybox-bus.workspace = true tinybox-jail.workspace = true async-trait.workspace = true # `Dir` keeps workspace file handles beneath the approved root without @@ -25,8 +27,8 @@ cap-std = "4" tokio = { workspace = true, features = ["process", "io-util", "sync"] } [target.'cfg(unix)'.dependencies] -# Safe process-group signalling keeps shell descendants within supervised cleanup. -nix = { version = "0.30", default-features = false, features = ["signal", "process"] } +# Safe process-group signalling supervises descendants; access checks select executable shells. +nix = { version = "0.30", default-features = false, features = ["signal", "process", "fs"] } [target.'cfg(windows)'.dependencies] # Job Objects keep supervised Windows descendants inside one killable process tree. diff --git a/crates/tinybox-host/src/lib.rs b/crates/tinybox-host/src/lib.rs index d7a21858..7fa6699b 100644 --- a/crates/tinybox-host/src/lib.rs +++ b/crates/tinybox-host/src/lib.rs @@ -48,3 +48,6 @@ mod local; pub use limited::{JailLocalHost, LimitedLocalHost, ManagedProcess}; pub use local::{LocalHost, NAME as LOCAL}; + +/// Platform shell discovery and command construction owned by the local host. +pub use local::shell::{build_shell_command, select_shell}; diff --git a/crates/tinybox-host/src/limited/jailed.rs b/crates/tinybox-host/src/limited/jailed.rs index b4a18f1a..a4f805f7 100644 --- a/crates/tinybox-host/src/limited/jailed.rs +++ b/crates/tinybox-host/src/limited/jailed.rs @@ -171,6 +171,34 @@ impl JailLocalHost { let mut command = crate::LocalHost::command(request)?.into_std(); // The host explicitly supplies the approved environment for jailed code. command.env_clear().envs(&request.env); + self.collect_command(command, request.stdin.clone(), observer) + .await + } + + /// Collect a prepared shell inside this host's explicit native jail. + /// The caller clears and supplies the approved environment before invocation. + /// # Errors + /// Reports unavailable spawn, output bounds and retained kill/reap failures. + pub async fn run_command_observed( + &self, + mut command: std::process::Command, + payload: Option>, + observer: Arc, + ) -> Result { + command.stdin(if payload.is_some() { + std::process::Stdio::piped() + } else { + std::process::Stdio::null() + }); + self.collect_command(command, payload, Some(observer)).await + } + + async fn collect_command( + &self, + mut command: std::process::Command, + payload: Option>, + observer: Option>, + ) -> Result { #[cfg(unix)] { use std::os::unix::process::CommandExt; @@ -181,7 +209,6 @@ impl JailLocalHost { .stderr(std::process::Stdio::piped()); let jail = self.jail.clone(); let backend = self.backend.clone(); - let payload = request.stdin.clone(); let state = self.state.clone(); let limit = self.limit; let runtime = tokio::runtime::Handle::try_current() diff --git a/crates/tinybox-host/src/limited/mod.rs b/crates/tinybox-host/src/limited/mod.rs index 6ece7977..b0556eb3 100644 --- a/crates/tinybox-host/src/limited/mod.rs +++ b/crates/tinybox-host/src/limited/mod.rs @@ -147,12 +147,75 @@ impl LimitedLocalHost { } impl LimitedLocalHost { + /// Execute platform shell text with the ordinary output and cleanup supervisor. + /// The request supplies cwd, environment and stdin; its argv is unused. + /// + /// # Errors + /// Returns native startup, output-limit and cleanup errors just like observed argv execution. + pub async fn run_shell_observed( + &self, + request: &ExecRequest, + shell: &tinybox_bus::SelectShellRequest, + observer: Arc, + ) -> Result { + let mut command = crate::build_shell_command(shell); + command.envs(&request.env); + if let Some(cwd) = &request.cwd { + command.current_dir(cwd); + } + command.stdin(if request.stdin.is_some() { + std::process::Stdio::piped() + } else { + std::process::Stdio::null() + }); + // Apply startup removals after provider/caller environment overrides. + for name in crate::select_shell(shell).remove_environment { + command.env_remove(name); + } + let mut command = tokio::process::Command::from(command); + command.kill_on_drop(true); + self.collect_command(command, request.stdin.clone(), Some(observer)) + .await + } + async fn collect( &self, request: &ExecRequest, observer: Option>, ) -> Result { - let mut command = LocalHost::command(request)?; + self.collect_command( + LocalHost::command(request)?, + request.stdin.clone(), + observer, + ) + .await + } + + /// Collect a prepared native command with the same bounded ownership as argv execution. + /// The provider supplies its approved environment and working directory. + /// # Errors + /// Reports spawn, pipe, output-budget and retained native cleanup failures. + pub async fn run_command_observed( + &self, + command: std::process::Command, + payload: Option>, + observer: Arc, + ) -> Result { + let mut command = tokio::process::Command::from(command); + command.kill_on_drop(true).stdin(if payload.is_some() { + std::process::Stdio::piped() + } else { + std::process::Stdio::null() + }); + self.collect_command(command, payload, Some(observer)).await + } + + async fn collect_command( + &self, + mut command: tokio::process::Command, + payload: Option>, + observer: Option>, + ) -> Result { process::prepare(&mut command)?; command .stdout(std::process::Stdio::piped()) @@ -169,7 +232,6 @@ impl LimitedLocalHost { let stdout = pipe(process::take_stdout(&mut child), "stdout")?; let stderr = pipe(process::take_stderr(&mut child), "stderr")?; let stdin = process::take_stdin(&mut child); - let payload = request.stdin.clone(); let budget = Arc::new(AtomicUsize::new(0)); let limit = self.limit; let state = self.state.clone(); diff --git a/crates/tinybox-host/src/limited/observed_tests.rs b/crates/tinybox-host/src/limited/observed_tests.rs index 82bc3ffb..077a08fd 100644 --- a/crates/tinybox-host/src/limited/observed_tests.rs +++ b/crates/tinybox-host/src/limited/observed_tests.rs @@ -272,3 +272,35 @@ async fn cleanup_retry_does_not_wait_for_or_cancel_an_active_local_execution() - ); Ok(()) } + +#[tokio::test] +async fn observed_shell_keeps_cwd_environment_stdin_and_output_limits() -> Result<()> { + let root = tempfile::tempdir().map_err(|error| Error::io("create fixture", &error))?; + let host = LimitedLocalHost::new(64); + let request = ExecRequest::new(Vec::::new()) + .with_cwd(root.path()) + .with_env("TINYBOX_SHELL_VALUE", "fixture") + .with_env("SHELLOPTS", "errexit") + .with_stdin("input"); + let shell = tinybox_bus::SelectShellRequest { + command: "false; printf '%s:' \"$TINYBOX_SHELL_VALUE\"; cat; printf file > result.txt" + .into(), + preserve_pipeline_status: false, + }; + let output = host + .run_shell_observed(&request, &shell, Arc::new(Observer::default())) + .await?; + assert!(output.succeeded()); + assert_eq!(output.stdout, b"fixture:input"); + assert!(root.path().join("result.txt").is_file()); + let error = LimitedLocalHost::new(1) + .run_shell_observed(&request, &shell, Arc::new(Observer::default())) + .await + .err(); + assert!(matches!( + error, + Some(Error::OutputLimitExceeded { limit: 1 }) + )); + host.drain_checked().await?; + Ok(()) +} diff --git a/crates/tinybox-host/src/local/mod.rs b/crates/tinybox-host/src/local/mod.rs index cc16c6d8..b8b80231 100644 --- a/crates/tinybox-host/src/local/mod.rs +++ b/crates/tinybox-host/src/local/mod.rs @@ -1,5 +1,7 @@ //! Running commands on the machine tinybox is running on. +pub(crate) mod shell; + use async_trait::async_trait; use std::path::Path; use tinybox_core::{ diff --git a/crates/tinybox-host/src/local/shell.rs b/crates/tinybox-host/src/local/shell.rs new file mode 100644 index 00000000..9a7bec2d --- /dev/null +++ b/crates/tinybox-host/src/local/shell.rs @@ -0,0 +1,147 @@ +//! Platform shell selection and quote-preserving native command construction. + +use tinybox_bus::{SelectShellRequest, ShellFlavor, ShellInvocation}; + +const WINDOWS_BOOTSTRAP_ENVIRONMENT: &[&str] = &[ + "SystemRoot", + "WINDIR", + "COMSPEC", + "PATHEXT", + "TEMP", + "TMP", + "USERPROFILE", + "APPDATA", + "LOCALAPPDATA", + "ProgramFiles", + "ProgramFiles(x86)", + "ProgramW6432", +]; + +/// Select this machine's shell without starting a process or returning secrets. +/// Unix uses Bash at a standard path with pipefail unless preservation was +/// requested; minimal hosts fall back to sh. Windows uses cmd with raw payload. +#[must_use] +pub fn select_shell(request: &SelectShellRequest) -> ShellInvocation { + let flavor = if cfg!(windows) { + ShellFlavor::Cmd + } else { + ShellFlavor::Posix + }; + selection_for(request, flavor, bash_path()) +} + +/// Build the selected native command without spawning it. The execution +/// provider remains responsible for environment, confinement, pipes and lifecycle. +#[must_use] +pub fn build_shell_command(request: &SelectShellRequest) -> std::process::Command { + command_for(select_shell(request)) +} + +fn command_for(selection: ShellInvocation) -> std::process::Command { + let mut command = std::process::Command::new(selection.program); + command.args(selection.arguments); + for name in selection.remove_environment { + command.env_remove(name); + } + #[cfg(windows)] + if let Some(raw) = selection.raw_arguments { + use std::os::windows::process::CommandExt; + command.raw_arg(raw); + } + command +} + +fn selection_for( + request: &SelectShellRequest, + flavor: ShellFlavor, + bash: Option<&str>, +) -> ShellInvocation { + match flavor { + ShellFlavor::Cmd => ShellInvocation { + flavor, + program: cmd_path(), + // Disable registry AutoRun and make cmd strip exactly our outer + // wrapper, preserving quoted executable/argument pairs inside. + arguments: vec![ + "/D".into(), + "/E:ON".into(), + "/V:OFF".into(), + "/S".into(), + "/C".into(), + ], + raw_arguments: Some(format!("\"{}\"", request.command)), + bootstrap_environment: WINDOWS_BOOTSTRAP_ENVIRONMENT + .iter() + .map(|name| (*name).into()) + .collect(), + remove_environment: Vec::new(), + }, + ShellFlavor::Posix => ShellInvocation { + flavor, + program: bash.unwrap_or("/bin/sh").into(), + arguments: { + let mut arguments = Vec::new(); + if bash.is_some() { + // Privileged startup ignores exported functions even when + // environment overrides are applied after command construction. + arguments.push("-p".into()); + } + arguments.extend([ + "-c".into(), + if bash.is_some() && !request.preserve_pipeline_status { + format!("set -o pipefail\n{}", request.command) + } else { + request.command.clone() + }, + ]); + arguments + }, + raw_arguments: None, + bootstrap_environment: Vec::new(), + remove_environment: vec![ + "BASH_ENV".into(), + "ENV".into(), + "SHELLOPTS".into(), + "BASHOPTS".into(), + "POSIXLY_CORRECT".into(), + ], + }, + } +} + +// Resolve before any payload environment overrides are applied. COMSPEC and +// the child's PATH must not choose a different command processor. +fn cmd_path() -> String { + let root = std::env::var_os("SystemRoot").map_or_else( + || std::path::PathBuf::from(r"C:\Windows"), + std::path::PathBuf::from, + ); + root.join("System32") + .join("cmd.exe") + .to_string_lossy() + .into_owned() +} + +fn bash_path() -> Option<&'static str> { + bash_path_in(&["/usr/bin/bash", "/bin/bash"]) +} + +fn bash_path_in<'a>(paths: &[&'a str]) -> Option<&'a str> { + paths.iter().copied().find(|path| { + if !std::path::Path::new(path).is_file() { + return false; + } + #[cfg(unix)] + { + nix::unistd::access(*path, nix::unistd::AccessFlags::X_OK).is_ok() + } + #[cfg(not(unix))] + { + true + } + }) +} + +#[cfg(test)] +#[path = "shell_tests.rs"] +mod tests; diff --git a/crates/tinybox-host/src/local/shell_tests.rs b/crates/tinybox-host/src/local/shell_tests.rs new file mode 100644 index 00000000..17654682 --- /dev/null +++ b/crates/tinybox-host/src/local/shell_tests.rs @@ -0,0 +1,325 @@ +//! Shell startup isolation, quoting and pipeline-status regressions. +use super::*; + +fn request(preserve_pipeline_status: bool) -> SelectShellRequest { + SelectShellRequest { + command: "false | true".into(), + preserve_pipeline_status, + } +} + +#[test] +fn bash_surfaces_pipe_failures_and_preservation_keeps_the_original_payload() { + let selected = selection_for(&request(false), ShellFlavor::Posix, Some("/bin/bash")); + assert_eq!( + selected.arguments, + ["-p", "-c", "set -o pipefail\nfalse | true"] + ); + let preserved = selection_for(&request(true), ShellFlavor::Posix, Some("/bin/bash")); + assert_eq!(preserved.arguments, ["-p", "-c", "false | true"]); +} + +#[test] +fn sh_fallback_never_receives_bash_only_flags() { + let selected = selection_for(&request(false), ShellFlavor::Posix, None); + assert_eq!(selected.program, "/bin/sh"); + assert_eq!(selected.arguments, ["-c", "false | true"]); + assert_eq!(selected.raw_arguments, None); + assert_eq!(selected.bootstrap_environment, Vec::::new()); +} + +#[test] +fn cmd_preserves_quoted_payload_and_lists_bootstrap_names_without_values() { + let payload = r#"echo %USERPROFILE% > "C:\Program Files (x86)\out.txt""#; + let selected = selection_for( + &SelectShellRequest { + command: payload.into(), + preserve_pipeline_status: false, + }, + ShellFlavor::Cmd, + None, + ); + assert_eq!(selected.program, cmd_path()); + assert_eq!(selected.arguments, ["/D", "/E:ON", "/V:OFF", "/S", "/C"]); + assert_eq!(selected.raw_arguments, Some(format!("\"{payload}\""))); + assert_eq!( + selected.bootstrap_environment, + WINDOWS_BOOTSTRAP_ENVIRONMENT + ); +} + +#[test] +#[cfg(windows)] +fn actual_cmd_raw_payload_keeps_a_quoted_executable_and_argument() +-> Result<(), Box> { + let root = tempfile::tempdir()?; + let executable = root.path().join("tool with spaces.exe"); + std::fs::copy(std::env::current_exe()?, &executable)?; + let output = root.path().join("quoted executable output.txt"); + let requested = SelectShellRequest { + command: format!( + "\"{}\" \"--list\" > \"{}\"", + executable.display(), + output.display() + ), + preserve_pipeline_status: false, + }; + assert!(build_shell_command(&requested).status()?.success()); + assert!( + std::fs::read_to_string(output)? + .contains("actual_cmd_raw_payload_keeps_a_quoted_executable_and_argument") + ); + Ok(()) +} + +#[test] +fn selected_shell_and_native_builder_agree() { + let requested = request(false); + let selected = select_shell(&requested); + let command = build_shell_command(&requested); + assert_eq!( + command.get_program(), + std::ffi::OsStr::new(&selected.program) + ); +} + +#[test] +#[cfg(unix)] +fn actual_bash_pipeline_failure_is_visible_unless_preservation_is_requested() +-> Result<(), Box> { + // The fallback is tested above; execution only asserts Bash semantics on + // hosts whose selected shell actually supports them. + if bash_path().is_none() { + return Ok(()); + } + assert!(!build_shell_command(&request(false)).status()?.success()); + assert!(build_shell_command(&request(true)).status()?.success()); + Ok(()) +} + +#[test] +#[cfg(windows)] +fn actual_cmd_raw_payload_keeps_quoted_redirection_paths() -> Result<(), Box> +{ + let root = tempfile::tempdir()?; + let output = root.path().join("output with spaces (fixture).txt"); + let requested = SelectShellRequest { + command: format!("echo fixture > \"{}\"", output.display()), + preserve_pipeline_status: false, + }; + assert!(build_shell_command(&requested).status()?.success()); + assert_eq!(std::fs::read_to_string(output)?.trim(), "fixture"); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn bash_startup_environment_cannot_replace_the_requested_payload() +-> Result<(), Box> { + const CHILD: &str = "TINYBOX_SHELL_ENV_REGRESSION_CHILD"; + if std::env::var_os(CHILD).is_some() { + let request = SelectShellRequest { + command: "exit 0".into(), + preserve_pipeline_status: false, + }; + assert!(build_shell_command(&request).status()?.success()); + return Ok(()); + } + let fixture = tempfile::tempdir()?; + let startup = fixture.path().join("startup.sh"); + std::fs::write(&startup, "exit 77\n")?; + let status = std::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "local::shell::tests::bash_startup_environment_cannot_replace_the_requested_payload", + "--nocapture", + ]) + .env(CHILD, "1") + .env("BASH_ENV", &startup) + .status()?; + assert!( + status.success(), + "startup environment changed the child payload result" + ); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn inherited_bash_options_cannot_abort_the_requested_payload() +-> Result<(), Box> { + const CHILD: &str = "TINYBOX_SHELL_OPTIONS_REGRESSION_CHILD"; + if std::env::var_os(CHILD).is_some() { + let request = SelectShellRequest { + command: "false; echo reached".into(), + preserve_pipeline_status: false, + }; + let output = build_shell_command(&request).output()?; + assert!(output.status.success()); + assert_eq!(output.stdout, b"reached\n"); + return Ok(()); + } + let status = std::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "local::shell::tests::inherited_bash_options_cannot_abort_the_requested_payload", + "--nocapture", + ]) + .env(CHILD, "1") + .env("SHELLOPTS", "errexit:nounset") + .env("BASHOPTS", "failglob") + .status()?; + assert!( + status.success(), + "inherited Bash options changed payload semantics" + ); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn exported_bash_functions_cannot_hide_pipeline_failures() -> Result<(), Box> +{ + const CHILD: &str = "TINYBOX_SHELL_FUNCTION_REGRESSION_CHILD"; + if bash_path().is_none() { + return Ok(()); + } + if std::env::var_os(CHILD).is_some() { + assert!(!build_shell_command(&request(false)).status()?.success()); + return Ok(()); + } + let status = std::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "local::shell::tests::exported_bash_functions_cannot_hide_pipeline_failures", + "--nocapture", + ]) + .env(CHILD, "1") + .env("BASH_FUNC_false%%", "() { return 0; }") + .status()?; + assert!( + status.success(), + "exported function hid the failing pipeline stage" + ); + Ok(()) +} + +#[test] +fn bash_discovery_rechecks_availability() -> Result<(), Box> { + // A changing candidate list represents installation/removal without altering + // global system paths or relying on the machine's installed shell. + let fixture = tempfile::NamedTempFile::new()?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(fixture.path(), std::fs::Permissions::from_mode(0o700))?; + } + let candidate_string = fixture.path().to_string_lossy().into_owned(); + let candidate = candidate_string.as_str(); + assert_eq!(bash_path_in(&[candidate]), Some(candidate)); + fixture.close()?; + assert_eq!(bash_path_in(&[candidate]), None); + std::fs::write(candidate, "fixture")?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(candidate, std::fs::Permissions::from_mode(0o700))?; + } + let discovered = bash_path_in(&[candidate]); + std::fs::remove_file(candidate)?; + assert_eq!(discovered, Some(candidate)); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn fallback_shell_survives_an_overridden_payload_path() -> Result<(), Box> { + let selected = selection_for( + &SelectShellRequest { + command: "exit 0".into(), + preserve_pipeline_status: false, + }, + ShellFlavor::Posix, + None, + ); + let status = command_for(selected).env("PATH", "").status()?; + assert!(status.success()); + Ok(()) +} + +#[test] +fn cmd_selection_uses_the_parent_system_directory() { + let selected = selection_for(&request(false), ShellFlavor::Cmd, None); + let root = std::env::var_os("SystemRoot").map_or_else( + || std::path::PathBuf::from(r"C:\Windows"), + std::path::PathBuf::from, + ); + assert_eq!( + std::path::Path::new(&selected.program), + root.join("System32").join("cmd.exe") + ); +} + +#[test] +#[cfg(windows)] +fn cmd_startup_ignores_the_payload_path() -> Result<(), Box> { + let fixture = tempfile::tempdir()?; + // A test runner is a valid executable but cannot interpret cmd's switches. + std::fs::copy(std::env::current_exe()?, fixture.path().join("cmd.exe"))?; + let requested = SelectShellRequest { + command: "echo %PATH%".into(), + preserve_pipeline_status: false, + }; + let output = build_shell_command(&requested) + .env("PATH", fixture.path()) + .output()?; + assert!(output.status.success()); + assert_eq!( + String::from_utf8(output.stdout)?.trim(), + fixture.path().to_string_lossy() + ); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn inherited_posix_mode_cannot_change_bash_command_substitution() +-> Result<(), Box> { + let Some(bash) = bash_path() else { + return Ok(()); + }; + let selected = selection_for( + &SelectShellRequest { + command: "set -e; x=$(false; echo reached); echo \"$x\"".into(), + preserve_pipeline_status: true, + }, + ShellFlavor::Posix, + Some(bash), + ); + // Supply the ambient value before applying the invocation's removal list. + let mut command = std::process::Command::new(&selected.program); + command + .args(&selected.arguments) + .env("POSIXLY_CORRECT", "1"); + for name in selected.remove_environment { + command.env_remove(name); + } + let output = command.output()?; + assert!(output.status.success()); + assert_eq!(output.stdout, b"reached\n"); + Ok(()) +} + +#[test] +#[cfg(unix)] +fn bash_discovery_skips_non_executable_files() -> Result<(), Box> { + use std::os::unix::fs::PermissionsExt; + let first = tempfile::NamedTempFile::new()?; + let second = tempfile::NamedTempFile::new()?; + std::fs::set_permissions(second.path(), std::fs::Permissions::from_mode(0o700))?; + let first = first.path().to_str().ok_or("non utf8 fixture")?; + let second = second.path().to_str().ok_or("non utf8 fixture")?; + assert_eq!(bash_path_in(&[first, second]), Some(second)); + assert_eq!(bash_path_in(&[first]), None); + Ok(()) +} diff --git a/crates/tinybox-module/Cargo.toml b/crates/tinybox-module/Cargo.toml index 5509873a..0a60adf9 100644 --- a/crates/tinybox-module/Cargo.toml +++ b/crates/tinybox-module/Cargo.toml @@ -41,9 +41,10 @@ tinybox-microvm.workspace = true tinybus.workspace = true tinybus-module.workspace = true -[dev-dependencies] -# Native cancellation verification uses isolated readiness files, not wall-clock sleeps. +# Private shell scratch is owned until supervised process cleanup completes. tempfile = "3" + +[dev-dependencies] # Deterministic delayed-host fixtures implement the existing object-safe host trait. async-trait.workspace = true # Module integration tests exercise the real asynchronous in-memory TinyBus. diff --git a/crates/tinybox-module/src/tinybus_module/README.md b/crates/tinybox-module/src/tinybus_module/README.md index 43148143..812117c6 100644 --- a/crates/tinybox-module/src/tinybus_module/README.md +++ b/crates/tinybox-module/src/tinybus_module/README.md @@ -45,8 +45,11 @@ The interface is `ai.tinyhumans.tinybox.Box` at | FinishFileWrite | FinishFileWriteRequest | FileWriteProgress | | AbortFileWrite | AbortFileWriteRequest | unit | | ResolveLocalGrants | ResolveLocalGrantsRequest | LocalGrants | +| SelectShell | SelectShellRequest | ShellInvocation | +| StartShellExec | SpawnRequest, SelectShellRequest | ProcessRef | +| StartNativeShellExec | ShellExecRequest | ProcessRef | -Capabilities advertises contract version 1.11; 1.0 denotes the original +Capabilities advertises contract version 1.14; 1.0 denotes the original discovery-only surface. Hosts require equal majors and a module minor at least as new as their vocabulary, using `tinybox_bus::is_compatible`. Version 1.1 added reserved resource/process ownership and terminal shutdown; 1.2 adds host @@ -320,3 +323,45 @@ It does not install a kernel deny rule for filenames created later. Git include aliases retain their logical parent. Empty XDG_CONFIG_HOME is unset; unsupported ~user/ and %(prefix)/ expansion rejects complete discovery. [Contract specification](../../../../docs/specs/local-grants.md). + +### Platform shell discovery + +Contract 1.12 adds `SelectShell`. The request carries command text and an optional +pipeline-status preservation flag. Unix selects Bash at a standard path with +`set -o pipefail` by default, or /bin/sh without that Bash-only flag when Bash is +absent. Unix uses non-login `-c` to avoid loading login profiles before the +payload. Windows selects `SystemRoot/System32/cmd.exe` from the parent +environment independently of payload PATH and COMSPEC overrides. Its +`/D /E:ON /V:OFF /S /C` switches disable registry AutoRun hooks and strip one +explicit outer wrapper around the raw command-line payload. +Quoted executable/argument pairs, redirection paths and percent expansion remain +inside that wrapper. +The result includes Windows bootstrap environment names, never their values. +Selection does not execute a command, authorize access or provide confinement; +hosts continue to execute through the module. Native providers can use the +shared quote-preserving TinyBox command builder. + +Windows explicitly enables extensions and disables delayed expansion. +Unix builders remove BASH_ENV, ENV, SHELLOPTS, BASHOPTS and POSIXLY_CORRECT; +ShellInvocation returns those removal names. Bash uses -p to ignore exported functions. +Discovery checks standard Bash paths on every call. Providers apply the +removals after approved environment overrides. + +Contract 1.13 adds `StartShellExec(SpawnRequest, SelectShellRequest)`. Supply an +empty argv in SpawnRequest; its resource, cwd, environment and stdin retain +their usual meaning. This operation supports local passthrough resources only, +because selection describes the module machine's shell. It uses the native +builder, including Windows raw arguments, and the existing supervised bounded +output, cancellation and cleanup lifecycle. Read and release its journal with +ReadOutput and ReleaseOutput. Retries must preserve both requests exactly. + +## Native shell execution + +Contract 1.14 adds `StartNativeShellExec(ShellExecRequest) -> ProcessRef` for local +passthrough resources. It composes platform shell selection with safe parent +environment snapshots, approved overrides, private scratch, bounded pipes and +owned timeout/cancellation cleanup. `ReadOutput` retains scratch until +`ReleaseOutput` or resource close acknowledges cleanup. Explicit native jail +requests refuse unavailable enforcement; remote/Docker providers refuse this +operation. State directory authorization and Windows private parent ACLs belong +to the caller. See [the native shell contract](../../../docs/specs/native-shell-execution.md). diff --git a/crates/tinybox-module/src/tinybus_module/docker_tests.rs b/crates/tinybox-module/src/tinybus_module/docker_tests.rs index 677ad192..fa3e220d 100644 --- a/crates/tinybox-module/src/tinybus_module/docker_tests.rs +++ b/crates/tinybox-module/src/tinybus_module/docker_tests.rs @@ -10,6 +10,63 @@ use tinybox_docker::DockerCli; use super::*; +/// Publish a fake CLI and wait out write handles inherited by concurrent forks. +fn write_cli_fixture(path: &Path, script: impl AsRef) -> std::io::Result<()> { + let script = script.as_ref().replacen( + "#!/bin/sh\n", + "#!/bin/sh\n[ \"$1\" = --tinybox-probe ] && exit 0\n", + 1, + ); + std::fs::write(path, script)?; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755))?; + probe_cli_fixture(|| { + std::process::Command::new(path) + .arg("--tinybox-probe") + .status() + }) +} + +/// Confirm the fixture is executable without recording a Docker invocation. +fn probe_cli_fixture( + mut probe: impl FnMut() -> std::io::Result, +) -> std::io::Result<()> { + // Forked children can retain a writer until exec. Successful execution + // proves the writer is gone; ETXTBSY alone is retryable, without a deadline. + loop { + match probe() { + Err(error) if error.raw_os_error() == Some(26) => { + std::thread::yield_now(); + } + result => { + let status = result?; + return if status.success() { + Ok(()) + } else { + Err(std::io::Error::other("fake Docker CLI probe failed")) + }; + } + } + } +} + +#[test] +fn fixture_probe_waits_for_inherited_writers_without_an_attempt_limit() -> std::io::Result<()> { + use std::os::unix::process::ExitStatusExt; + let mut busy = 300; + probe_cli_fixture(|| { + if busy > 0 { + busy -= 1; + Err(std::io::Error::from_raw_os_error(26)) + } else { + Ok(std::process::ExitStatus::from_raw(0)) + } + })?; + assert_eq!(busy, 0); + assert!(probe_cli_fixture(|| Err(std::io::Error::from_raw_os_error(2))).is_err()); + assert!(probe_cli_fixture(|| Ok(std::process::ExitStatus::from_raw(256))).is_err()); + Ok(()) +} + fn canonical(path: &Path) -> std::io::Result { path.canonicalize().map(|path| path.display().to_string()) } @@ -25,16 +82,13 @@ async fn docker_status_uses_the_module_owned_cli_probe() let temp = tempfile::tempdir()?; let calls = temp.path().join("calls"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\n[ \"$1\" = info ] && exit 0\nexit 1\n", calls.display() ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let status = status_with_cli( &DockerCli::with_program(&script_path), @@ -55,10 +109,7 @@ async fn docker_status_reports_an_unavailable_daemon_without_failing_the_probe() -> std::result::Result<(), Box> { let temp = tempfile::tempdir()?; let script_path = temp.path().join("docker"); - std::fs::write(&script_path, "#!/bin/sh\nexit 1\n")?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; + write_cli_fixture(&script_path, "#!/bin/sh\nexit 1\n")?; let status = status_with_cli( &DockerCli::with_program(&script_path), @@ -76,16 +127,13 @@ async fn docker_cleanup_only_kills_ids_returned_for_the_requested_label() let temp = tempfile::tempdir()?; let calls = temp.path().join("calls"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\nif [ \"$1\" = ps ]; then printf 'box-a\\nbox-b\\n'; exit 0; fi\n[ \"$1\" = kill ] && exit 0\nexit 1\n", calls.display() ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let result = cleanup_with_cli( &DockerCli::with_program(&script_path), @@ -109,13 +157,10 @@ async fn docker_cleanup_reports_failed_kill_instead_of_claiming_stopped() -> std::result::Result<(), Box> { let temp = tempfile::tempdir()?; let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, "#!/bin/sh\n[ \"$1\" = ps ] && { printf 'box-a\\n'; exit 0; }\n[ \"$1\" = kill ] && exit 1\nexit 2\n", )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let Err(error) = cleanup_with_cli( &DockerCli::with_program(&script_path), @@ -141,16 +186,13 @@ async fn docker_cleanup_does_not_kill_ids_from_a_failed_listing() let temp = tempfile::tempdir()?; let calls = temp.path().join("calls"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\n[ \"$1\" = ps ] && {{ printf 'box-a\\n'; exit 1; }}\nexit 0\n", calls.display() ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let Err(error) = cleanup_with_cli( &DockerCli::with_program(&script_path), @@ -180,16 +222,13 @@ async fn docker_cleanup_with_no_matching_containers_does_not_kill() let temp = tempfile::tempdir()?; let calls = temp.path().join("calls"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\n[ \"$1\" = ps ] && exit 0\nexit 2\n", calls.display() ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let result = cleanup_with_cli( &DockerCli::with_program(&script_path), @@ -214,16 +253,13 @@ async fn docker_control_rejects_invalid_deadline_and_label_before_cli_call() let temp = tempfile::tempdir()?; let calls = temp.path().join("calls"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\nprintf called >> '{}'\nexit 0\n", calls.display() ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let cli = DockerCli::with_program(&script_path); let Err(error) = status_with_cli(&cli, DockerStatusRequest { timeout_ms: 0 }).await else { @@ -269,7 +305,7 @@ async fn run_docker_preserves_one_shot_settings_output_and_timeout() std::fs::create_dir_all(&readonly)?; let argv_file = temp.path().join("argv.bin"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\n: > '{}'\nfor a in \"$@\"; do printf '%s\\0' \"$a\" >> '{}'; done\nprintf out\nprintf err >&2\nexit 3\n", @@ -277,9 +313,6 @@ async fn run_docker_preserves_one_shot_settings_output_and_timeout() argv_file.display(), ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let request = DockerRunRequest { workspace: workspace.display().to_string(), @@ -335,7 +368,7 @@ async fn run_without_workspace_preserves_docker_settings_without_host_mounts() let temp = tempfile::tempdir()?; let argv_file = temp.path().join("argv.bin"); let script_path = temp.path().join("docker"); - std::fs::write( + write_cli_fixture( &script_path, format!( "#!/bin/sh\n: > '{}'\nfor a in \"$@\"; do printf '%s\\0' \"$a\" >> '{}'; done\nprintf out\nprintf err >&2\nexit 0\n", @@ -343,9 +376,6 @@ async fn run_without_workspace_preserves_docker_settings_without_host_mounts() argv_file.display(), ), )?; - let mut permissions = std::fs::metadata(&script_path)?.permissions(); - permissions.set_mode(0o755); - std::fs::set_permissions(&script_path, permissions)?; let request = DockerRunRequest { workspace: String::new(), diff --git a/crates/tinybox-module/src/tinybus_module/mod.rs b/crates/tinybox-module/src/tinybus_module/mod.rs index 145ee576..b71fe027 100644 --- a/crates/tinybox-module/src/tinybus_module/mod.rs +++ b/crates/tinybox-module/src/tinybus_module/mod.rs @@ -18,6 +18,7 @@ use tinybox_bus::{ mod docker; mod native_jail; +mod native_shell; mod output; mod resources; @@ -70,6 +71,13 @@ impl BoxService { ) -> TinyBusResult { std::future::ready(Ok(tinybox_core::shell::analysis::analyze_command(&command))).await } + /// Discover the module machine's shell syntax and invocation plan. + async fn select_shell( + &self, + request: tinybox_bus::SelectShellRequest, + ) -> TinyBusResult { + std::future::ready(Ok(tinybox_host::select_shell(&request))).await + } /// Discover canonical grants using the caller's protected-path floor. async fn resolve_local_grants( &self, @@ -261,6 +269,31 @@ impl BoxService { .await } + /// Execute platform shell text using the original two-argument contract. + async fn start_shell_exec( + &self, + request: SpawnRequest, + shell: tinybox_bus::SelectShellRequest, + ) -> TinyBusResult { + let resources = self.resources.clone(); + finish_operation(tokio::spawn(async move { + resources.start_platform_shell_exec(request, shell).await + })) + .await + } + + /// Start a shell with module-owned environment, scratch and supervision. + async fn start_native_shell_exec( + &self, + request: tinybox_bus::ShellExecRequest, + ) -> TinyBusResult { + let resources = self.resources.clone(); + finish_operation(tokio::spawn(async move { + resources.start_shell_exec(request).await + })) + .await + } + /// Run one Docker-isolated shell command with bounded output and runtime. async fn run_docker( &self, @@ -474,7 +507,7 @@ tinybus_module::module_export_optional_static! { setup = setup, worker_threads = 1, provides = ["ai.tinyhumans.tinybox.Box"], - methods = ["Describe", "Create", "Exec", "Inspect", "Close", "Spawn", "IsRunning", "Cancel", "AnalyzeShell", "Reserve", "Capabilities", "Shutdown", "Forward", "CloseForward", "JailStatus", "AnalyzeCommand", "BeginFileRead", "ReadFileChunk", "FinishFileRead", "BeginFileWrite", "WriteFileChunk", "FinishFileWrite", "AbortFileWrite", "StartExec", "ReadOutput", "ReleaseOutput", "StartJailedExec", "RunDocker", "DockerStatus", "CleanupDocker", "RunDockerWithoutWorkspace", "ResolveLocalGrants"], + methods = ["Describe", "Create", "Exec", "Inspect", "Close", "Spawn", "IsRunning", "Cancel", "AnalyzeShell", "Reserve", "Capabilities", "Shutdown", "Forward", "CloseForward", "JailStatus", "AnalyzeCommand", "BeginFileRead", "ReadFileChunk", "FinishFileRead", "BeginFileWrite", "WriteFileChunk", "FinishFileWrite", "AbortFileWrite", "StartExec", "ReadOutput", "ReleaseOutput", "StartJailedExec", "RunDocker", "DockerStatus", "CleanupDocker", "RunDockerWithoutWorkspace", "ResolveLocalGrants", "SelectShell", "StartShellExec", "StartNativeShellExec"], signals = [], requires = [], optional = [], diff --git a/crates/tinybox-module/src/tinybus_module/mod_tests.rs b/crates/tinybox-module/src/tinybus_module/mod_tests.rs index b3dd2f40..cc62be72 100644 --- a/crates/tinybox-module/src/tinybus_module/mod_tests.rs +++ b/crates/tinybox-module/src/tinybus_module/mod_tests.rs @@ -1045,3 +1045,210 @@ async fn unsupported_git_include_expansions_refuse_the_whole_discovery() } Ok(()) } + +#[tokio::test] +async fn native_shell_selection_and_execution_roundtrip_through_the_bus_codec() +-> Result<(), Box> { + let state = tempfile::tempdir()?; + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let service = Connection::connect(bus.connect().await?).await?; + setup(service.clone()).await?; + let client = Connection::connect(bus.connect().await?).await?; + let proxy = client.proxy(INTERFACE, OBJECT_PATH, INTERFACE)?; + let shell = tinybox_bus::SelectShellRequest { + command: "echo native".into(), + preserve_pipeline_status: false, + }; + let selection: tinybox_bus::ShellInvocation = + proxy.call("SelectShell", (shell.clone(),)).await?; + assert_eq!(selection, tinybox_host::select_shell(&shell)); + let resource: tinybox_bus::ResourceId = proxy + .call("Reserve", (tinybox_bus::ReserveRequest::Resource,)) + .await?; + proxy + .call::( + "Create", + (tinybox_bus::CreateRequest { + resource: resource.clone(), + backend: "passthrough".into(), + workspace: tinybox_bus::Workspace::Directory(".".into()), + ..Default::default() + },), + ) + .await?; + let process = proxy + .call( + "Reserve", + (tinybox_bus::ReserveRequest::Process(resource.clone()),), + ) + .await?; + let process: tinybox_bus::ProcessRef = proxy + .call( + "StartNativeShellExec", + (tinybox_bus::ShellExecRequest { + process, + resource: resource.clone(), + shell, + cwd: None, + extra_env: std::collections::BTreeMap::new(), + state_dir: state.path().to_str().ok_or("non-UTF8 state")?.into(), + timeout_ms: 5000, + stdin: None, + jail: None, + },), + ) + .await?; + let output = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let batch: tinybox_bus::OutputBatch = + proxy.call("ReadOutput", (process.clone(), 0_u64)).await?; + if batch.state != tinybox_bus::ExecutionState::Running { + break Ok::<_, tinybus::Error>(batch); + } + tokio::task::yield_now().await; + } + }) + .await??; + assert_eq!( + output.state, + tinybox_bus::ExecutionState::Exited { exit_code: 0 } + ); + let bytes: Vec = output + .chunks + .iter() + .flat_map(|chunk| chunk.bytes.clone()) + .collect(); + assert_eq!(String::from_utf8_lossy(&bytes).trim(), "native"); + proxy.call::<()>("ReleaseOutput", (process,)).await?; + assert_eq!( + std::fs::read_dir(state.path().join("artifacts/sandbox-scratch"))?.count(), + 0 + ); + proxy.call::<()>("Close", (resource,)).await?; + Ok(()) +} + +#[tokio::test] +async fn existing_shell_method_keeps_its_two_argument_bus_codec() +-> Result<(), Box> { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let service = Connection::connect(bus.connect().await?).await?; + setup(service.clone()).await?; + let client = Connection::connect(bus.connect().await?).await?; + let proxy = client.proxy(INTERFACE, OBJECT_PATH, INTERFACE)?; + let shell = tinybox_bus::SelectShellRequest { + command: "echo native".into(), + preserve_pipeline_status: false, + }; + let selection: tinybox_bus::ShellInvocation = + proxy.call("SelectShell", (shell.clone(),)).await?; + assert_eq!(selection, tinybox_host::select_shell(&shell)); + let resource: tinybox_bus::ResourceId = proxy + .call("Reserve", (tinybox_bus::ReserveRequest::Resource,)) + .await?; + proxy + .call::( + "Create", + (tinybox_bus::CreateRequest { + resource: resource.clone(), + backend: "passthrough".into(), + workspace: tinybox_bus::Workspace::Directory(".".into()), + ..Default::default() + },), + ) + .await?; + let process = proxy + .call( + "Reserve", + (tinybox_bus::ReserveRequest::Process(resource.clone()),), + ) + .await?; + let process: tinybox_bus::ProcessRef = proxy + .call( + "StartShellExec", + ( + tinybox_bus::SpawnRequest { + process, + command: tinybox_bus::ExecRequest { + resource: resource.clone(), + argv: Vec::new(), + cwd: None, + env: std::collections::BTreeMap::new(), + stdin: None, + }, + }, + shell, + ), + ) + .await?; + let output = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let batch: tinybox_bus::OutputBatch = + proxy.call("ReadOutput", (process.clone(), 0_u64)).await?; + if batch.state != tinybox_bus::ExecutionState::Running { + break Ok::<_, tinybus::Error>(batch); + } + tokio::task::yield_now().await; + } + }) + .await??; + assert_eq!( + output.state, + tinybox_bus::ExecutionState::Exited { exit_code: 0 } + ); + let bytes: Vec = output + .chunks + .iter() + .flat_map(|chunk| chunk.bytes.clone()) + .collect(); + assert_eq!(String::from_utf8_lossy(&bytes).trim(), "native"); + proxy.call::<()>("ReleaseOutput", (process,)).await?; + proxy.call::<()>("Close", (resource,)).await?; + Ok(()) +} + +#[tokio::test] +async fn shell_discovery_preserves_the_requested_pipeline_status() -> tinybus::Result<()> { + let service = BoxService::default(); + let selected = service + .select_shell(tinybox_bus::SelectShellRequest { + command: "false | true".into(), + preserve_pipeline_status: true, + }) + .await?; + if cfg!(windows) { + assert_eq!(selected.flavor, tinybox_bus::ShellFlavor::Cmd); + assert_eq!(selected.raw_arguments.as_deref(), Some("\"false | true\"")); + } else { + assert_eq!(selected.flavor, tinybox_bus::ShellFlavor::Posix); + assert_eq!( + selected.arguments.last().map(String::as_str), + Some("false | true") + ); + } + Ok(()) +} + +#[tokio::test] +async fn docker_control_dispatch_rejects_unbounded_deadlines() { + let service = BoxService::default(); + let status = service + .docker_status(tinybox_bus::DockerStatusRequest { timeout_ms: 0 }) + .await; + assert!( + matches!(status, Err(tinybus::Error::MethodFailed { name, .. }) + if name == tinybox_bus::INVALID_DOCKER_CONTROL) + ); + let cleanup = service + .cleanup_docker(tinybox_bus::CleanupDockerRequest { + label: "openhuman.sandbox=true".into(), + timeout_ms: tinybox_bus::MAX_DOCKER_CONTROL_TIMEOUT_MS + 1, + }) + .await; + assert!( + matches!(cleanup, Err(tinybus::Error::MethodFailed { name, .. }) + if name == tinybox_bus::INVALID_DOCKER_CONTROL) + ); +} diff --git a/crates/tinybox-module/src/tinybus_module/native_shell/README.md b/crates/tinybox-module/src/tinybus_module/native_shell/README.md new file mode 100644 index 00000000..33e1b9a2 --- /dev/null +++ b/crates/tinybox-module/src/tinybus_module/native_shell/README.md @@ -0,0 +1,34 @@ +# Native shell execution + +This module prepares caller-approved local shell requests. The retained context +owns the safe environment snapshot, private scratch and a supervised native +collector. It rejects invalid requests before consuming a process reservation. +Resource streams own replay, output budgets, timeout publication and cleanup +barriers. Explicit confinement uses the native jail contract and refuses +unavailable enforcement. + +See [the contract](../../../../../docs/specs/native-shell-execution.md) and +[its sequence](../../../../../docs/plans/native-shell-execution.md). + +Output release and resource close explicitly remove private scratch on a +blocking worker after draining the native collector. Timeout and cancellation +drain the process but retain scratch for as long as its journal is retained. A deletion failure is returned to the caller and retains the context +and scratch path for a later cleanup retry. + +Preparation runs on Tokio’s blocking pool so filesystem and jail preflight I/O +do not stall the module’s async worker. + +Stream admission snapshots shell inputs under the resource lock, releases it +before preparation, and rechecks resource ownership after preparation. Close +and Shutdown can finish while preparation is stalled; late completion cannot +publish an execution for a closed resource. + +A prepared shell that loses admission is explicitly disposed on the blocking +pool. Failed deletion retains its context in the resource registry; a later +Close (including for an already closed resource) or Shutdown retries cleanup. + +Preparation validates the process reservation and claims a bounded pending slot +before scheduling blocking work. Pending preparations and retained journals +share the module admission budget. Close fences the resource and stops owned +work before reporting failed scratch deletion. The sanitized child environment +includes resource variables, with explicit shell-request overrides applied last. diff --git a/crates/tinybox-module/src/tinybus_module/native_shell/mod.rs b/crates/tinybox-module/src/tinybus_module/native_shell/mod.rs new file mode 100644 index 00000000..610a4823 --- /dev/null +++ b/crates/tinybox-module/src/tinybus_module/native_shell/mod.rs @@ -0,0 +1,253 @@ +//! Native shell preparation; the retained context owns private scratch and cleanup. +use std::{ + path::{Component, Path}, + sync::Arc, +}; +use tinybox_bus::ShellExecRequest; +use tinybus::Result; + +mod types; +pub(super) use types::ShellContext; +use types::ShellHost; +impl ShellContext { + pub(super) async fn prepare_owned( + request: ShellExecRequest, + root: std::path::PathBuf, + resolved: tinybox_core::ExecRequest, + ) -> Result> { + Self::prepare_owned_with(move || Self::prepare(request, &root, &resolved)).await + } + async fn prepare_owned_with( + prepare: impl FnOnce() -> Result> + Send + 'static, + ) -> Result> { + tokio::task::spawn_blocking(prepare).await.map_err(|_| { + super::resources::failure( + tinybox_bus::BACKEND_ERROR, + "native shell preparation worker failed", + ) + })? + } + pub(super) fn prepare( + request: ShellExecRequest, + root: &Path, + resolved: &tinybox_core::ExecRequest, + ) -> Result> { + Self::prepare_with_backend(request, root, resolved, tinybox_jail::default_backend()) + } + fn prepare_with_backend( + request: ShellExecRequest, + root: &Path, + resolved: &tinybox_core::ExecRequest, + backend: Arc, + ) -> Result> { + validate_runtime(&request)?; + let state = Path::new(&request.state_dir); + if request.state_dir.len() > tinybox_bus::MAX_FILE_PATH_BYTES + || request.state_dir.contains('\0') + || !state.is_absolute() + || state + .components() + .any(|c| matches!(c, Component::ParentDir)) + { + return Err(invalid()); + } + let state = state.canonicalize().map_err(|_| invalid())?; + if !state.is_dir() { + return Err(invalid()); + } + let scratch_root = state.join("artifacts/sandbox-scratch"); + std::fs::create_dir_all(&scratch_root).map_err(|_| invalid())?; + let scratch = tempfile::Builder::new() + .prefix("shell-") + .tempdir_in(scratch_root) + .map_err(|_| invalid())?; + let selection = tinybox_host::select_shell(&request.shell); + let mut environment = std::collections::BTreeMap::new(); + for name in [ + "PATH", "HOME", "TERM", "LANG", "LC_ALL", "LC_CTYPE", "USER", "SHELL", "TMPDIR", + ] + .into_iter() + .chain(selection.bootstrap_environment.iter().map(String::as_str)) + { + if let Ok(value) = std::env::var(name) { + if value.is_empty() { + return Err(invalid()); + } + environment.insert(name.to_owned(), value); + } + } + for (name, value) in resolved.env.iter().chain(&request.extra_env) { + if name.is_empty() || name.contains(['=', '\0']) || value.contains('\0') { + return Err(invalid()); + } + if cfg!(windows) { + environment.retain(|key, _| !key.eq_ignore_ascii_case(name)); + } + environment.insert(name.clone(), value.clone()); + } + for name in ["TMPDIR", "TEMP", "TMP"] { + if !resolved + .env + .keys() + .chain(request.extra_env.keys()) + .any(|key| environment_name_matches(key, name)) + { + environment.insert( + name.into(), + scratch.path().to_str().ok_or_else(invalid)?.into(), + ); + } + } + for name in selection.remove_environment { + environment.remove(&name); + } + let host = if let Some(config) = &request.jail { + let mut jail = super::native_jail::prepare(root, config, resolved, backend.as_ref())?; + jail.read_write.push(scratch.path().to_path_buf()); + if config.enforcement == tinybox_bus::JailEnforcement::Enforced { + backend.require(&jail).map_err(|_| invalid())?; + } else if backend + .plan_check(&jail) + .constraints + .iter() + .any(|(_, support)| *support == tinybox_core::Enforcement::Unsupported) + { + return Err(invalid()); + } + ShellHost::Jailed( + tinybox_host::JailLocalHost::with_backend( + jail, + backend, + tinybox_bus::MAX_OUTPUT_BYTES, + ) + .map_err(|_| invalid())?, + ) + } else { + ShellHost::Local(tinybox_host::LimitedLocalHost::new( + tinybox_bus::MAX_OUTPUT_BYTES, + )) + }; + Ok(Arc::new(Self { + request, + environment, + scratch: Arc::new(std::sync::Mutex::new(Some(scratch))), + scratch_cleanup_failed: Arc::new(std::sync::atomic::AtomicBool::new(false)), + host, + #[cfg(test)] + scratch_deletion_blocked: std::sync::atomic::AtomicBool::new(false), + })) + } + pub(super) async fn run( + &self, + resolved: &tinybox_core::ExecRequest, + observer: Arc, + ) -> tinybox_core::Result { + let mut command = tinybox_host::build_shell_command(&self.request.shell); + command.env_clear().envs(&self.environment); + if let Some(cwd) = &resolved.cwd { + command.current_dir(cwd); + } + match &self.host { + ShellHost::Local(host) => { + host.run_command_observed(command, self.request.stdin.clone(), observer) + .await + } + ShellHost::Jailed(host) => { + host.run_command_observed(command, self.request.stdin.clone(), observer) + .await + } + } + } + pub(super) fn pending(&self) -> bool { + if self + .scratch_cleanup_failed + .load(std::sync::atomic::Ordering::SeqCst) + { + return true; + } + match &self.host { + ShellHost::Local(host) => host.has_pending_cleanup(), + ShellHost::Jailed(host) => host.has_pending_cleanup(), + } + } + pub(super) async fn drain(&self) -> tinybox_core::Result<()> { + match &self.host { + ShellHost::Local(host) => host.drain_checked().await?, + ShellHost::Jailed(host) => host.drain_checked().await?, + } + Ok(()) + } + pub(super) async fn dispose_scratch(&self) -> tinybox_core::Result<()> { + #[cfg(test)] + if self + .scratch_deletion_blocked + .load(std::sync::atomic::Ordering::SeqCst) + { + return self + .dispose_scratch_with(|_| { + Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "injected scratch deletion failure", + )) + }) + .await; + } + self.dispose_scratch_with(|path| std::fs::remove_dir_all(path)) + .await + } + async fn dispose_scratch_with( + &self, + remove: impl FnOnce(&Path) -> std::io::Result<()> + Send + 'static, + ) -> tinybox_core::Result<()> { + let scratch = self.scratch.clone(); + let failed = self.scratch_cleanup_failed.clone(); + tokio::task::spawn_blocking(move || { + let mut scratch = scratch + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(directory) = scratch.as_ref() { + // Keep the TempDir and its path until deletion succeeds; close() consumes + // ownership even on failure and would make cleanup impossible to retry. + if let Err(error) = remove(directory.path()) + && error.kind() != std::io::ErrorKind::NotFound + { + failed.store(true, std::sync::atomic::Ordering::SeqCst); + return Err(tinybox_core::Error::io("remove shell scratch", &error)); + } + scratch.take(); + } + failed.store(false, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }) + .await + .map_err(|error| { + tinybox_core::Error::io("join shell scratch cleanup", &std::io::Error::other(error)) + })? + } +} +fn environment_name_matches(left: &str, right: &str) -> bool { + if cfg!(windows) { + left.eq_ignore_ascii_case(right) + } else { + left == right + } +} +fn validate_runtime(request: &ShellExecRequest) -> Result<()> { + if request.timeout_ms == 0 + || request.timeout_ms > 86_400_000 + || request.shell.command.contains('\0') + { + return Err(invalid()); + } + Ok(()) +} +fn invalid() -> tinybus::Error { + super::resources::failure( + tinybox_bus::INVALID_SHELL_EXEC, + "invalid native shell environment, state directory or runtime budget", + ) +} + +#[cfg(test)] +#[path = "mod_tests.rs"] +mod tests; diff --git a/crates/tinybox-module/src/tinybus_module/native_shell/mod_tests.rs b/crates/tinybox-module/src/tinybus_module/native_shell/mod_tests.rs new file mode 100644 index 00000000..84ecccd3 --- /dev/null +++ b/crates/tinybox-module/src/tinybus_module/native_shell/mod_tests.rs @@ -0,0 +1,268 @@ +//! Validation and environment behavior before process admission. +use super::*; +fn request(state: &Path) -> ShellExecRequest { + ShellExecRequest { + process: tinybox_bus::ResourceId("p".into()), + resource: tinybox_bus::ResourceId("r".into()), + shell: tinybox_bus::SelectShellRequest { + command: "echo accepted".into(), + preserve_pipeline_status: false, + }, + cwd: None, + extra_env: std::collections::BTreeMap::new(), + state_dir: state.to_string_lossy().into_owned(), + timeout_ms: 1000, + stdin: None, + jail: None, + } +} +#[test] +fn invalid_shell_state_budget_and_environment_are_refused_before_execution() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let resolved = tinybox_core::ExecRequest::new(["unused"]); + let original = request(state.path()); + for timeout in [0, 86_400_001] { + let mut invalid = original.clone(); + invalid.timeout_ms = timeout; + assert!(ShellContext::prepare(invalid, state.path(), &resolved).is_err()); + } + for directory in ["relative", "absent", "nul\0"] { + let mut invalid = original.clone(); + invalid.state_dir = directory.into(); + assert!(ShellContext::prepare(invalid, state.path(), &resolved).is_err()); + } + let file = state.path().join("file"); + std::fs::write(&file, "data")?; + let mut invalid = original.clone(); + invalid.state_dir = file.to_string_lossy().into_owned(); + assert!(ShellContext::prepare(invalid, state.path(), &resolved).is_err()); + let mut invalid = original.clone(); + invalid.shell.command = "nul\0".into(); + assert!(ShellContext::prepare(invalid, state.path(), &resolved).is_err()); + for (name, value) in [("", "value"), ("NAME=bad", "value"), ("NAME", "nul\0")] { + let mut invalid = original.clone(); + invalid.extra_env.insert(name.into(), value.into()); + assert!(ShellContext::prepare(invalid, state.path(), &resolved).is_err()); + } + Ok(()) +} +#[test] +fn approved_environment_overrides_preserve_custom_temporary_directory_and_remove_startup_hooks() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let resolved = tinybox_core::ExecRequest::new(["unused"]); + let mut request = request(state.path()); + request + .extra_env + .insert("TMPDIR".into(), "approved-temporary-location".into()); + request + .extra_env + .insert("SHELLOPTS".into(), "errexit".into()); + let shell = ShellContext::prepare(request, state.path(), &resolved)?; + assert_eq!( + shell.environment.get("TMPDIR").map(String::as_str), + Some("approved-temporary-location") + ); + #[cfg(unix)] + assert_eq!(shell.environment.get("SHELLOPTS"), None); + assert!(!shell.pending()); + Ok(()) +} + +#[cfg(unix)] +struct Backend { + full: bool, +} +#[cfg(unix)] +impl tinybox_jail::JailBackend for Backend { + fn name(&self) -> &'static str { + "native-shell-preflight-fixture" + } + fn is_available(&self) -> bool { + true + } + fn isolation(&self) -> tinybox_core::IsolationLevel { + tinybox_core::IsolationLevel::None + } + fn is_suitable_for_untrusted_code(&self) -> bool { + true + } + fn constraint_support(&self) -> tinybox_core::ConstraintSupport { + use tinybox_core::{Constraint, ConstraintSupport, Enforcement}; + if self.full { + ConstraintSupport::NONE + .with(Constraint::Filesystem, Enforcement::Enforced) + .with(Constraint::Network, Enforcement::Enforced) + .with(Constraint::Subprocess, Enforcement::Enforced) + } else { + ConstraintSupport::NONE.with(Constraint::Filesystem, Enforcement::BestEffort) + } + } + fn spawn( + &self, + jail: &tinybox_jail::Jail, + mut command: std::process::Command, + ) -> std::io::Result { + assert!(jail.read_write.iter().any(|path| { + path.components() + .any(|component| component.as_os_str() == "sandbox-scratch") + })); + command.spawn() + } +} +#[cfg(unix)] +#[tokio::test] +async fn native_shell_jail_adds_owned_scratch_and_refuses_unsupported_constraints() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let mut resolved = tinybox_core::ExecRequest::new(["unused"]); + resolved.cwd = Some(state.path().to_path_buf()); + let mut request = request(state.path()); + request.jail = Some(tinybox_bus::NativeJailConfig { + read_only: vec![], + read_write: vec![], + allow_net: false, + allow_subprocess: false, + enforcement: tinybox_bus::JailEnforcement::Enforced, + require_process_isolation: false, + }); + let shell = ShellContext::prepare_with_backend( + request.clone(), + state.path(), + &resolved, + Arc::new(Backend { full: true }), + )?; + assert_eq!( + shell + .run( + &resolved, + Arc::new(super::super::output::OutputJournal::default()) + ) + .await? + .exit_code, + 0 + ); + assert!(!shell.pending()); + shell.drain().await?; + let config = request.jail.as_mut().ok_or("missing jail fixture")?; + config.enforcement = tinybox_bus::JailEnforcement::Declared; + let shell = ShellContext::prepare_with_backend( + request.clone(), + state.path(), + &resolved, + Arc::new(Backend { full: true }), + )?; + shell.drain().await?; + assert!( + ShellContext::prepare_with_backend( + request, + state.path(), + &resolved, + Arc::new(Backend { full: false }) + ) + .is_err() + ); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn scratch_deletion_runs_outside_the_async_worker() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let resolved = tinybox_core::ExecRequest::new(["unused"]); + let shell = ShellContext::prepare(request(state.path()), state.path(), &resolved)?; + let observed = Arc::new(std::sync::Mutex::new(None)); + let recorded = observed.clone(); + let worker = std::thread::current().id(); + shell + .dispose_scratch_with(move |path| { + *recorded + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(std::thread::current().id()); + std::fs::remove_dir_all(path) + }) + .await?; + assert_ne!( + *observed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + Some(worker), + "recursive filesystem deletion must not run on the sole async worker" + ); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn scratch_preparation_runs_outside_the_async_worker() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let root = state.path().to_path_buf(); + let request = request(&root); + let resolved = tinybox_core::ExecRequest::new(["unused"]); + let observed = Arc::new(std::sync::Mutex::new(None)); + let recorded = observed.clone(); + let worker = std::thread::current().id(); + let shell = ShellContext::prepare_owned_with(move || { + *recorded + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(std::thread::current().id()); + ShellContext::prepare(request, &root, &resolved) + }) + .await?; + assert_ne!( + *observed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + Some(worker), + "filesystem preparation must not run on the sole async worker" + ); + shell.dispose_scratch().await?; + Ok(()) +} + +#[test] +fn missing_state_and_blocked_scratch_paths_are_refused() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let resolved = tinybox_core::ExecRequest::new(["unused"]); + for root in [state.path().join("missing"), state.path().to_path_buf()] { + std::fs::write(state.path().join("artifacts"), "blocks scratch creation")?; + assert!(matches!( + ShellContext::prepare(request(&root), &root, &resolved), + Err(tinybus::Error::MethodFailed { name, .. }) if name == tinybox_bus::INVALID_SHELL_EXEC + )); + } + Ok(()) +} + +#[test] +fn native_shell_preserves_resource_environment_and_request_overrides() +-> std::result::Result<(), Box> { + let state = tempfile::tempdir()?; + let mut resolved = tinybox_core::ExecRequest::new(["unused"]); + resolved + .env + .insert("RESOURCE_CONFIG".into(), "preserved".into()); + resolved.env.insert("TMPDIR".into(), "resource-temp".into()); + resolved.env.insert("OVERRIDE".into(), "resource".into()); + let mut request = request(state.path()); + request + .extra_env + .insert("OVERRIDE".into(), "request".into()); + let shell = ShellContext::prepare(request, state.path(), &resolved)?; + assert_eq!( + shell.environment.get("RESOURCE_CONFIG").map(String::as_str), + Some("preserved") + ); + assert_eq!( + shell.environment.get("TMPDIR").map(String::as_str), + Some("resource-temp") + ); + assert_eq!( + shell.environment.get("OVERRIDE").map(String::as_str), + Some("request") + ); + Ok(()) +} diff --git a/crates/tinybox-module/src/tinybus_module/native_shell/types.rs b/crates/tinybox-module/src/tinybus_module/native_shell/types.rs new file mode 100644 index 00000000..e2c6c408 --- /dev/null +++ b/crates/tinybox-module/src/tinybus_module/native_shell/types.rs @@ -0,0 +1,15 @@ +//! Retained native collector and scratch ownership. +use tinybox_bus::ShellExecRequest; +pub(in crate::tinybus_module) struct ShellContext { + pub(super) request: ShellExecRequest, + pub(super) environment: std::collections::BTreeMap, + pub(super) scratch: std::sync::Arc>>, + pub(super) scratch_cleanup_failed: std::sync::Arc, + pub(super) host: ShellHost, + #[cfg(test)] + pub(in crate::tinybus_module) scratch_deletion_blocked: std::sync::atomic::AtomicBool, +} +pub(super) enum ShellHost { + Local(tinybox_host::LimitedLocalHost), + Jailed(tinybox_host::JailLocalHost), +} diff --git a/crates/tinybox-module/src/tinybus_module/resources.rs b/crates/tinybox-module/src/tinybus_module/resources.rs index 4cb8918f..c4683e06 100644 --- a/crates/tinybox-module/src/tinybus_module/resources.rs +++ b/crates/tinybox-module/src/tinybus_module/resources.rs @@ -129,6 +129,8 @@ pub(super) struct Resources { state: Mutex, executions: std::sync::Mutex, streams: std::sync::Mutex>>, + rejected_shells: Mutex)>>, + preparations: tokio::sync::Semaphore, } #[derive(Default)] @@ -155,6 +157,8 @@ impl Default for Resources { state: Mutex::default(), executions: std::sync::Mutex::default(), streams: std::sync::Mutex::default(), + rejected_shells: Mutex::default(), + preparations: tokio::sync::Semaphore::new(stream::MAX_STREAM_EXECUTIONS), } } } @@ -1192,7 +1196,7 @@ impl Resources { pub(super) async fn close(&self, resource: &ResourceId) -> Result<()> { validate_id(resource)?; let Some(slot) = self.begin_close(resource).await? else { - return Ok(()); + return self.dispose_rejected_shells(Some(resource)).await; }; let mut slot = slot.lock().await; let mut failure = self.close_streams(resource).await.err(); @@ -1262,6 +1266,9 @@ impl Resources { entry.sandbox_destroyed = true; } } + if let Err(error) = self.dispose_rejected_shells(Some(resource)).await { + failure.get_or_insert(error); + } *slot = None; self.state.lock().await.entries.remove(resource); let mut executions = self @@ -1367,6 +1374,9 @@ impl Resources { } finished.await; } + if let Err(error) = self.dispose_rejected_shells(None).await { + failure.get_or_insert(error); + } failure.map_or(Ok(()), Err) } } diff --git a/crates/tinybox-module/src/tinybus_module/resources_stream.rs b/crates/tinybox-module/src/tinybus_module/resources_stream.rs index 0a17023c..c1b8e99c 100644 --- a/crates/tinybox-module/src/tinybus_module/resources_stream.rs +++ b/crates/tinybox-module/src/tinybus_module/resources_stream.rs @@ -6,17 +6,25 @@ use super::{ use crate::tinybus_module::output::OutputJournal; use tinybox_bus::{ExecutionFailure, OutputBatch}; -const MAX_STREAM_EXECUTIONS: usize = 32; +pub(super) const MAX_STREAM_EXECUTIONS: usize = 32; pub(super) struct StreamExecution { request: SpawnRequest, + shell: Option, + platform_shell: Option, jail: Option, journal: Arc, task: Mutex>>, collector: Collector, + timed_out: Arc, } #[derive(Clone)] enum Collector { + Shell(Arc), + PlatformShell( + Arc, + tinybox_bus::SelectShellRequest, + ), Host( Arc, Option>, @@ -28,6 +36,32 @@ enum Collector { Option>, ), } +enum CollectorPreparation { + Ready(Collector), + Shell( + Box<( + tinybox_bus::ShellExecRequest, + std::path::PathBuf, + tinybox_core::ExecRequest, + )>, + ), +} +impl CollectorPreparation { + async fn finish(self) -> Result { + match self { + Self::Ready(collector) => Ok(collector), + Self::Shell(preparation) => { + let (request, root, resolved) = *preparation; + Ok(Collector::Shell( + super::super::native_shell::ShellContext::prepare_owned( + request, root, resolved, + ) + .await?, + )) + } + } + } +} impl Collector { async fn run( &self, @@ -35,6 +69,10 @@ impl Collector { observer: Arc, ) -> tinybox_core::Result { match self { + Self::Shell(shell) => shell.run(command, observer).await, + Self::PlatformShell(host, shell) => { + host.run_shell_observed(command, shell, observer).await + } Self::Host(host, _) => host.run_observed(command, observer).await, Self::Jailed(host) => host.run_observed(command, observer).await, Self::Sandbox(sandbox, id, _) => sandbox.exec_observed(id, command, observer).await, @@ -42,6 +80,8 @@ impl Collector { } fn pending(&self) -> bool { match self { + Self::Shell(shell) => shell.pending(), + Self::PlatformShell(host, _) => host.has_pending_cleanup(), Self::Host(host, native) => { host.has_pending_observed_cleanup() || native @@ -66,6 +106,8 @@ impl Collector { } async fn drain(&self) -> tinybox_core::Result<()> { match self { + Self::Shell(shell) => shell.drain().await, + Self::PlatformShell(host, _) => host.drain_checked().await, Self::Host(host, native) => { let remote = host.retry_observed_cleanup().await; if let Some(native) = native { @@ -121,46 +163,184 @@ fn select_collector( } }) } +fn validate_execution_ids(request: &SpawnRequest) -> Result<()> { + validate_id(&request.process)?; + validate_id(&request.command.resource) +} +fn require_owned_collector(entry: &super::Resource) -> Result<()> { + if entry.collector.is_none() { + return Err(failure( + tinybox_bus::UNSUPPORTED_OPERATION, + "owned native collector unavailable", + )); + } + Ok(()) +} +fn select_execution_collector( + entry: &super::Resource, + jail: Option<&tinybox_bus::NativeJailConfig>, + shell: Option<&tinybox_bus::ShellExecRequest>, + resolved: &tinybox_core::ExecRequest, +) -> Result { + Ok(if let Some(shell) = shell { + if !entry.native_host || entry.local_sandbox.is_none() { + return Err(failure( + tinybox_bus::UNSUPPORTED_OPERATION, + "native shell requires local passthrough execution", + )); + } + let root = entry.workspace_root.as_ref().ok_or_else(|| { + failure( + tinybox_bus::UNSUPPORTED_OPERATION, + "native shell requires directory resource", + ) + })?; + CollectorPreparation::Shell(Box::new((shell.clone(), root.clone(), resolved.clone()))) + } else { + CollectorPreparation::Ready(select_collector(entry, jail, resolved)?) + }) +} +fn prepare_execution( + entry: &super::Resource, + request: &SpawnRequest, + jail: Option<&tinybox_bus::NativeJailConfig>, + shell: Option<&tinybox_bus::ShellExecRequest>, + platform_shell: Option<&tinybox_bus::SelectShellRequest>, +) -> Result<(tinybox_core::ExecRequest, CollectorPreparation)> { + if !entry.execution_supported + || (entry.local_sandbox.is_none() && entry.docker_sandbox.is_none()) + { + return Err(failure( + tinybox_bus::UNSUPPORTED_OPERATION, + "live output is unavailable for this provider", + )); + } + require_owned_collector(entry)?; + // Each execution owns its cleanup collector. Cancelling one stream must + // not drain an unrelated running supervisor or inherit its failure. + // Resolve before consuming admission, and fence native start against Close. + if platform_shell.is_some() && (!entry.native_host || entry.local_sandbox.is_none()) { + return Err(failure( + tinybox_bus::UNSUPPORTED_OPERATION, + "platform shell execution requires local passthrough", + )); + } + let mut native_request = command(request.command.clone()); + if platform_shell.is_some() { + // Resolve resource cwd and environment; the shell builder supplies argv. + native_request.argv = vec!["shell".into()]; + } + let resolved = if let Some(sandbox) = &entry.local_sandbox { + sandbox + .resolve_command(&entry.id, &native_request) + .map_err(|error| backend_error(&error))? + } else { + let command = native_request; + if command.argv.is_empty() { + return Err(failure(tinybox_bus::INVALID_ID, "empty execution command")); + } + command + }; + let collector = if let Some(shell) = platform_shell { + CollectorPreparation::Ready(Collector::PlatformShell( + Arc::new(tinybox_host::LimitedLocalHost::new( + tinybox_bus::MAX_OUTPUT_BYTES, + )), + shell.clone(), + )) + } else { + select_execution_collector(entry, jail, shell, &resolved)? + }; + Ok((resolved, collector)) +} impl StreamExecution { + fn check_replay( + &self, + request: &SpawnRequest, + jail: Option<&tinybox_bus::NativeJailConfig>, + shell: Option<&tinybox_bus::ShellExecRequest>, + platform_shell: Option<&tinybox_bus::SelectShellRequest>, + ) -> Result<()> { + // Native shell identity is its original wire request; synthesized argv + // can change when platform discovery changes between delivery retries. + let same_request = shell.is_some() || self.request == *request; + if same_request + && self.jail.as_ref() == jail + && self.shell.as_ref() == shell + && self.platform_shell.as_ref() == platform_shell + { + Ok(()) + } else { + Err(failure( + tinybox_bus::INVALID_ID, + "stream identifier belongs to another command", + )) + } + } fn start( request: SpawnRequest, resolved: tinybox_core::ExecRequest, collector: Collector, jail: Option, + shell: Option, + platform_shell: Option, ) -> Self { let journal = Arc::new(OutputJournal::default()); let observer = journal.clone(); let runner = collector.clone(); + let timeout = shell.as_ref().map(|request| request.timeout_ms); + let timed_out = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let timeout_outcome = timed_out.clone(); let task = tokio::spawn(async move { let _completion = CompletionGuard(observer.clone()); - let result = runner.run(&resolved, observer.clone()).await; + let result = if let Some(timeout) = timeout { + if let Ok(result) = tokio::time::timeout( + std::time::Duration::from_millis(timeout), + runner.run(&resolved, observer.clone()), + ) + .await + { + Some(result) + } else { + timeout_outcome.store(true, std::sync::atomic::Ordering::SeqCst); + let _cleanup = runner.drain().await; + None + } + } else { + Some(runner.run(&resolved, observer.clone()).await) + }; if runner.pending() { observer.finish_failed(ExecutionFailure::CleanupFailed); } else if observer.cancel_requested() { observer.finish_exit(0); + } else if result.is_none() { + observer.finish_failed(ExecutionFailure::TimedOut); } else { match result { - Ok(output) => observer.finish_exit(output.exit_code), - Err(tinybox_core::Error::OutputLimitExceeded { .. }) => { + Some(Ok(output)) => observer.finish_exit(output.exit_code), + Some(Err(tinybox_core::Error::OutputLimitExceeded { .. })) => { observer.finish_failed(ExecutionFailure::OutputLimit); } - Err(_) => observer.finish_failed(ExecutionFailure::BackendFailed), + Some(Err(_)) | None => observer.finish_failed(ExecutionFailure::BackendFailed), } } }); Self { request, + platform_shell, + shell, jail, journal, task: Mutex::new(Some(task)), collector, + timed_out, } } pub(super) fn running(&self) -> bool { !self.journal.cleanup_acknowledged() } pub(super) fn cleanup_acknowledged(&self) -> bool { - self.journal.cleanup_acknowledged() + self.journal.cleanup_acknowledged() && !self.collector.pending() } pub(super) async fn stop(&self) -> Result<()> { self.journal.cancel(); @@ -178,7 +358,24 @@ impl StreamExecution { ) })?; if !self.journal.cleanup_acknowledged() { - self.journal.finish_failed(ExecutionFailure::BackendFailed); + let failure = if self.timed_out.load(std::sync::atomic::Ordering::SeqCst) { + ExecutionFailure::TimedOut + } else { + ExecutionFailure::BackendFailed + }; + self.journal.finish_failed(failure); + } + Ok(()) + } + async fn dispose(&self) -> Result<()> { + self.stop().await?; + if let Collector::Shell(shell) = &self.collector { + shell.dispose_scratch().await.map_err(|_| { + failure( + tinybox_bus::BACKEND_ERROR, + "shell scratch cleanup remains pending", + ) + })?; } Ok(()) } @@ -200,118 +397,255 @@ impl Resources { &self, request: SpawnRequest, ) -> Result { - self.start_with_jail(request, None).await + self.start_with_jail(request, None, None, None).await } pub(in crate::tinybus_module) async fn start_jailed_exec( &self, request: SpawnRequest, config: tinybox_bus::NativeJailConfig, ) -> Result { - self.start_with_jail(request, Some(config)).await + self.start_with_jail(request, Some(config), None, None) + .await + } + pub(in crate::tinybus_module) async fn start_shell_exec( + &self, + shell: tinybox_bus::ShellExecRequest, + ) -> Result { + let request = SpawnRequest { + process: shell.process.clone(), + command: tinybox_bus::ExecRequest { + resource: shell.resource.clone(), + argv: vec![tinybox_host::select_shell(&shell.shell).program], + cwd: shell.cwd.clone(), + env: shell.extra_env.clone(), + stdin: shell.stdin.clone(), + }, + }; + self.start_with_jail(request, shell.jail.clone(), Some(shell), None) + .await + } + pub(in crate::tinybus_module) async fn start_platform_shell_exec( + &self, + request: SpawnRequest, + shell: tinybox_bus::SelectShellRequest, + ) -> Result { + if !request.command.argv.is_empty() { + return Err(failure( + tinybox_bus::INVALID_ID, + "shell execution requires empty argv", + )); + } + self.start_with_jail(request, None, None, Some(shell)).await } async fn start_with_jail( &self, request: SpawnRequest, jail: Option, + shell: Option, + platform_shell: Option, ) -> Result { - validate_id(&request.process)?; - validate_id(&request.command.resource)?; + self.start_with_preparation( + request, + jail, + shell, + platform_shell, + CollectorPreparation::finish, + ) + .await + } + async fn start_with_preparation( + &self, + request: SpawnRequest, + jail: Option, + shell: Option, + platform_shell: Option, + prepare: impl FnOnce(CollectorPreparation) -> Fut, + ) -> Result + where + Fut: std::future::Future>, + { + validate_execution_ids(&request)?; let process = ProcessRef { resource: request.command.resource.clone(), process: request.process.clone(), }; if let Some(existing) = self.stream(&process)? { - return if existing.request == request && existing.jail == jail { - Ok(process) - } else { - Err(failure( - tinybox_bus::INVALID_ID, - "stream identifier belongs to another command", - )) - }; + existing.check_replay( + &request, + jail.as_ref(), + shell.as_ref(), + platform_shell.as_ref(), + )?; + return Ok(process); } let slot = self.slot(&process.resource).await?; - let slot = slot.lock().await; - let entry = slot + let guard = slot.lock().await; + let entry = guard .as_ref() .ok_or_else(|| failure(tinybox_bus::UNKNOWN_RESOURCE, "closed resource"))?; - if !entry.execution_supported - || (entry.local_sandbox.is_none() && entry.docker_sandbox.is_none()) - { - return Err(failure( - tinybox_bus::UNSUPPORTED_OPERATION, - "live output is unavailable for this provider", - )); + let _preparation = self.claim_preparation(entry, &request).await?; + let (resolved, collector) = prepare_execution( + entry, + &request, + jail.as_ref(), + shell.as_ref(), + platform_shell.as_ref(), + )?; + // Preparation owns its snapshot; Close must not wait for filesystem I/O. + drop(guard); + let collector = prepare(collector).await?; + let admitted = async { + let guard = slot.lock().await; + let entry = guard + .as_ref() + .ok_or_else(|| failure(tinybox_bus::UNKNOWN_RESOURCE, "closed resource"))?; + let mut state = self.state.lock().await; + let executions = self + .executions + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + check_resource_open(&state, &executions, &process.resource)?; + let mut streams = self + .streams + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(existing) = streams.get(&process.process) { + existing.check_replay( + &request, + jail.as_ref(), + shell.as_ref(), + platform_shell.as_ref(), + )?; + return Ok(false); + } + if entry.processes.len() + + streams + .values() + .filter(|stream| stream.request.command.resource == process.resource) + .count() + >= tinybox_bus::MAX_PROCESSES_PER_RESOURCE + { + return Err(failure( + tinybox_bus::RESOURCE_LIMIT, + "resource process limit reached", + )); + } + if streams.len() >= MAX_STREAM_EXECUTIONS { + return Err(failure( + tinybox_bus::RESOURCE_LIMIT, + "live execution admission limit reached", + )); + } + self.consume( + &mut state, + &request.process, + &ReserveRequest::Process(process.resource.clone()), + )?; + streams.insert( + process.process.clone(), + Arc::new(StreamExecution::start( + request, + resolved, + collector.clone(), + jail, + shell, + platform_shell, + )), + ); + Ok(true) } - if entry.collector.is_none() { - return Err(failure( - tinybox_bus::UNSUPPORTED_OPERATION, - "owned native collector unavailable", - )); + .await; + if !matches!(admitted, Ok(true)) { + self.dispose_rejected_collector(&process.resource, collector) + .await?; } - // Each execution owns its cleanup collector. Cancelling one stream must - // not drain an unrelated running supervisor or inherit its failure. - // Resolve before consuming admission, and fence native start against Close. - let resolved = if let Some(sandbox) = &entry.local_sandbox { - sandbox - .resolve_command(&entry.id, &command(request.command.clone())) - .map_err(|error| backend_error(&error))? - } else { - let command = command(request.command.clone()); - if command.argv.is_empty() { - return Err(failure(tinybox_bus::INVALID_ID, "empty execution command")); - } - command - }; - let collector = select_collector(entry, jail.as_ref(), &resolved)?; + admitted?; + Ok(process) + } + + async fn claim_preparation( + &self, + entry: &super::Resource, + request: &SpawnRequest, + ) -> Result> { let mut state = self.state.lock().await; let executions = self .executions .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - check_resource_open(&state, &executions, &process.resource)?; - let mut streams = self + check_resource_open(&state, &executions, &request.command.resource)?; + self.check_reservation( + &mut state, + &request.process, + &ReserveRequest::Process(request.command.resource.clone()), + )?; + let streams = self .streams .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if let Some(existing) = streams.get(&process.process) { - return if existing.request == request && existing.jail == jail { - Ok(process) - } else { - Err(failure( - tinybox_bus::INVALID_ID, - "stream identifier belongs to another command", - )) - }; - } - if entry.processes.len() - + streams - .values() - .filter(|stream| stream.request.command.resource == process.resource) - .count() - >= tinybox_bus::MAX_PROCESSES_PER_RESOURCE + let pending = MAX_STREAM_EXECUTIONS - self.preparations.available_permits(); + if streams.len() + pending >= MAX_STREAM_EXECUTIONS + || entry.processes.len() + + streams + .values() + .filter(|stream| stream.request.command.resource == request.command.resource) + .count() + >= tinybox_bus::MAX_PROCESSES_PER_RESOURCE { return Err(failure( tinybox_bus::RESOURCE_LIMIT, - "resource process limit reached", + "execution preparation limit reached", )); } - if streams.len() >= MAX_STREAM_EXECUTIONS { - return Err(failure( + self.preparations.try_acquire().map_err(|_| { + failure( tinybox_bus::RESOURCE_LIMIT, - "live execution admission limit reached", - )); + "execution preparation limit reached", + ) + }) + } + + async fn dispose_rejected_collector( + &self, + resource: &ResourceId, + collector: Collector, + ) -> Result<()> { + if let Collector::Shell(shell) = collector { + // Retain ownership before awaiting cleanup, including if the caller cancels. + self.rejected_shells + .lock() + .await + .push((resource.clone(), shell)); + self.dispose_rejected_shells(Some(resource)).await?; } - self.consume( - &mut state, - &request.process, - &ReserveRequest::Process(process.resource.clone()), - )?; - streams.insert( - process.process.clone(), - Arc::new(StreamExecution::start(request, resolved, collector, jail)), - ); - Ok(process) + Ok(()) + } + + pub(super) async fn dispose_rejected_shells( + &self, + resource: Option<&ResourceId>, + ) -> Result<()> { + let mut shells = self.rejected_shells.lock().await; + let mut index = 0; + let mut error = None; + while index < shells.len() { + if resource.is_some_and(|resource| *resource != shells[index].0) { + index += 1; + continue; + } + if shells[index].1.dispose_scratch().await.is_ok() { + shells.remove(index); + } else { + error.get_or_insert_with(|| { + failure( + tinybox_bus::BACKEND_ERROR, + "rejected shell scratch cleanup remains pending", + ) + }); + index += 1; + } + } + error.map_or(Ok(()), Err) } pub(super) fn stream(&self, process: &ProcessRef) -> Result>> { @@ -354,7 +688,7 @@ impl Resources { pub(super) async fn close_streams(&self, resource: &ResourceId) -> Result<()> { let mut failure = None; for stream in self.streams_for(resource) { - if let Err(error) = stream.stop().await { + if let Err(error) = stream.dispose().await { failure.get_or_insert(error); } } @@ -401,13 +735,13 @@ impl Resources { process: &ProcessRef, ) -> Result<()> { if let Some(stream) = self.stream(process)? { - if !stream.cleanup_acknowledged() { + if !stream.journal.cleanup_acknowledged() { return Err(failure( tinybox_bus::UNSUPPORTED_OPERATION, "execution cleanup is not acknowledged", )); } - stream.stop().await?; + stream.dispose().await?; self.streams .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) diff --git a/crates/tinybox-module/src/tinybus_module/resources_stream_tests.rs b/crates/tinybox-module/src/tinybus_module/resources_stream_tests.rs index 7911179c..4ea16912 100644 --- a/crates/tinybox-module/src/tinybus_module/resources_stream_tests.rs +++ b/crates/tinybox-module/src/tinybus_module/resources_stream_tests.rs @@ -206,6 +206,9 @@ async fn a_panicked_job_is_joined_and_cleanup_failure_stays_owned_until_retry() }, }, jail: None, + shell: None, + platform_shell: None, + timed_out: Arc::new(std::sync::atomic::AtomicBool::new(false)), journal, task: Mutex::new(Some(task)), collector: Collector::Host(Arc::new(tinybox_host::LimitedLocalHost::new(64)), None), @@ -301,6 +304,7 @@ struct RemoteObservedFixture { pending: std::sync::atomic::AtomicBool, reject_retry: std::sync::atomic::AtomicBool, collected_calls: std::sync::atomic::AtomicUsize, + wait_for_timeout: bool, } #[async_trait::async_trait] impl tinybox_core::Host for RemoteObservedFixture { @@ -325,6 +329,9 @@ impl tinybox_core::Host for RemoteObservedFixture { observer.output(tinybox_core::OutputStream::Stdout, &[255, 0])?; self.pending .store(true, std::sync::atomic::Ordering::SeqCst); + if self.wait_for_timeout { + std::future::pending::<()>().await; + } Err(tinybox_core::Error::ObservedCleanupPending { provider: "fixture".into(), }) @@ -350,6 +357,7 @@ async fn remote_cleanup_failure_keeps_binary_output_and_owner_until_retry_ack() pending: std::sync::atomic::AtomicBool::new(false), reject_retry: std::sync::atomic::AtomicBool::new(true), collected_calls: std::sync::atomic::AtomicUsize::new(0), + wait_for_timeout: false, }); let resource = ResourceId("fixture-resource".into()); let command = tinybox_core::ExecRequest::new(["fixture"]); @@ -367,6 +375,8 @@ async fn remote_cleanup_failure_keeps_binary_output_and_owner_until_retry_ack() command, Collector::Host(host.clone(), None), None, + None, + None, ); assert!(stream.stop().await.is_err()); assert!(stream.running()); @@ -556,6 +566,8 @@ async fn confirmed_container_destroy_keeps_native_cleanup_retryable_without_dest tinybox_core::ExecRequest::new(["fixture"]), Collector::Sandbox(Arc::new(sandbox), entry.id.clone(), None), None, + None, + None, )); resources .streams @@ -617,3 +629,676 @@ async fn collector_drain_joins_its_owned_active_native_supervisor() -> Result<() ); Ok(()) } + +async fn shell_request( + resources: &Resources, + resource: &ResourceId, + state: &std::path::Path, + payload: &str, +) -> Result { + Ok(tinybox_bus::ShellExecRequest { + process: resources + .reserve(ReserveRequest::Process(resource.clone())) + .await?, + resource: resource.clone(), + shell: tinybox_bus::SelectShellRequest { + command: payload.into(), + preserve_pipeline_status: false, + }, + cwd: None, + extra_env: std::collections::BTreeMap::default(), + state_dir: state.to_string_lossy().into_owned(), + timeout_ms: 5000, + stdin: None, + jail: None, + }) +} +#[tokio::test] +async fn native_shell_replay_uses_original_request_and_retains_scratch_until_release() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let request = shell_request(&resources, &resource, state.path(), "echo owned").await?; + let process = resources.start_shell_exec(request.clone()).await?; + assert_eq!(resources.start_shell_exec(request.clone()).await?, process); + let mut changed = request; + changed.timeout_ms += 1; + assert!(resources.start_shell_exec(changed).await.is_err()); + completed(&resources, &process).await?; + assert_eq!( + resources.read_output(&process, 0).await?.state, + ExecutionState::Exited { exit_code: 0 } + ); + let scratch = state.path().join("artifacts/sandbox-scratch"); + assert_eq!(std::fs::read_dir(&scratch)?.count(), 1); + resources.release_output(&process).await?; + assert_eq!(std::fs::read_dir(&scratch)?.count(), 0); + resources.close(&resource).await?; + Ok(()) +} +#[tokio::test] +async fn native_shell_invalid_budget_does_not_consume_its_reservation() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let mut request = shell_request(&resources, &resource, state.path(), "echo accepted").await?; + request.timeout_ms = 0; + assert!(resources.start_shell_exec(request.clone()).await.is_err()); + request.timeout_ms = 5000; + let process = resources.start_shell_exec(request).await?; + completed(&resources, &process).await?; + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} +#[tokio::test] +async fn native_shell_timeout_acknowledges_child_cleanup_before_output_release() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + #[cfg(unix)] + let payload = "echo partial; sleep 60"; + #[cfg(windows)] + let payload = + "echo partial & powershell -NoProfile -NonInteractive -Command \"Start-Sleep -Seconds 60\""; + let mut request = shell_request(&resources, &resource, state.path(), payload).await?; + request.timeout_ms = 100; + let process = resources.start_shell_exec(request).await?; + completed(&resources, &process).await?; + let output = resources.read_output(&process, 0).await?; + assert_eq!( + output.state, + ExecutionState::Failed { + code: ExecutionFailure::TimedOut + } + ); + assert_eq!( + std::fs::read_dir(state.path().join("artifacts/sandbox-scratch"))?.count(), + 1 + ); + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +#[cfg(unix)] +async fn native_shell_clears_startup_hooks_after_approved_overrides_and_captures_binary_pipes() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let startup = state.path().join("startup.sh"); + std::fs::write(&startup, "exit 77\n")?; + let mut request = shell_request( + &resources, + &resource, + state.path(), + "cat; printf \\\\377 >&2; false; echo $APPROVED", + ) + .await?; + request + .extra_env + .insert("BASH_ENV".into(), startup.to_string_lossy().into_owned()); + request + .extra_env + .insert("SHELLOPTS".into(), "errexit".into()); + request.extra_env.insert("APPROVED".into(), "yes".into()); + request.stdin = Some(vec![0, 255, 1]); + let process = resources.start_shell_exec(request).await?; + completed(&resources, &process).await?; + let output = resources.read_output(&process, 0).await?; + assert_eq!(output.state, ExecutionState::Exited { exit_code: 0 }); + let stdout: Vec = output + .chunks + .iter() + .filter(|c| c.stream == tinybox_bus::OutputStream::Stdout) + .flat_map(|c| c.bytes.clone()) + .collect(); + assert_eq!(stdout, [0, 255, 1, b'y', b'e', b's', 10]); + let stderr: Vec = output + .chunks + .iter() + .filter(|c| c.stream == tinybox_bus::OutputStream::Stderr) + .flat_map(|c| c.bytes.clone()) + .collect(); + assert_eq!(stderr, [255]); + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +#[cfg(unix)] +async fn native_shell_output_limit_terminates_and_reaps_before_release() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let request = shell_request(&resources, &resource, state.path(), "yes output").await?; + let process = resources.start_shell_exec(request).await?; + completed(&resources, &process).await?; + let output = resources.read_output(&process, 0).await?; + assert_eq!( + output.state, + ExecutionState::Failed { + code: ExecutionFailure::OutputLimit + } + ); + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn failed_scratch_deletion_keeps_output_and_close_retryable() +-> std::result::Result<(), Box> { + for close in [false, true] { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let request = shell_request(&resources, &resource, state.path(), "echo owned").await?; + let process = resources.start_shell_exec(request).await?; + completed(&resources, &process).await?; + let scratch = std::fs::read_dir(state.path().join("artifacts/sandbox-scratch"))? + .next() + .ok_or("missing scratch")?? + .path(); + let retained = scratch.with_extension("retained"); + std::fs::rename(&scratch, &retained)?; + std::fs::write(&scratch, "blocks directory removal")?; + assert!(resources.release_output(&process).await.is_err()); + assert!(resources.stream(&process)?.is_some()); + if close { + assert!(resources.close(&resource).await.is_err()); + assert!(resources.stream(&process)?.is_some()); + } + std::fs::remove_file(&scratch)?; + std::fs::rename(&retained, &scratch)?; + if !close { + resources.release_output(&process).await?; + } + resources.close(&resource).await?; + assert!(!scratch.exists()); + assert!(resources.stream(&process)?.is_none()); + } + Ok(()) +} + +#[tokio::test] +async fn native_shell_refuses_nonlocal_or_nondirectory_collectors_before_admission() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let shell = shell_request(&resources, &resource, state.path(), "echo owned").await?; + let slot = resources.slot(&resource).await?; + { + let mut entry = slot.lock().await; + let entry = entry.as_mut().ok_or("missing resource")?; + entry.native_host = false; + } + let error = resources + .start_shell_exec(shell.clone()) + .await + .err() + .ok_or("accepted nonlocal shell")?; + assert_eq!(error.wire_name(), tinybox_bus::UNSUPPORTED_OPERATION); + { + let mut entry = slot.lock().await; + let entry = entry.as_mut().ok_or("missing resource")?; + entry.native_host = true; + entry.workspace_root = None; + } + let error = resources + .start_shell_exec(shell) + .await + .err() + .ok_or("accepted nondirectory shell")?; + assert_eq!(error.wire_name(), tinybox_bus::UNSUPPORTED_OPERATION); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn cancelled_native_shell_retains_scratch_until_output_release() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + #[cfg(unix)] + let payload = "echo ready; sleep 60"; + #[cfg(windows)] + let payload = + "echo ready & powershell -NoProfile -NonInteractive -Command \"Start-Sleep -Seconds 60\""; + let request = shell_request(&resources, &resource, state.path(), payload).await?; + let process = resources.start_shell_exec(request).await?; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while resources.read_output(&process, 0).await?.chunks.is_empty() { + tokio::task::yield_now().await; + } + Ok::<_, tinybus::Error>(()) + }) + .await??; + resources.cancel(&process).await?; + assert_eq!( + resources.read_output(&process, 0).await?.state, + ExecutionState::Cancelled + ); + let scratch = state.path().join("artifacts/sandbox-scratch"); + assert_eq!(std::fs::read_dir(&scratch)?.count(), 1); + resources.release_output(&process).await?; + assert_eq!(std::fs::read_dir(&scratch)?.count(), 0); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn native_shell_execution_preserves_admission_and_request_identity() -> Result<()> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let request = request(&resources, &resource, Vec::new()).await?; + let shell = tinybox_bus::SelectShellRequest { + command: "echo fixture".into(), + preserve_pipeline_status: false, + }; + let process = resources + .start_platform_shell_exec(request.clone(), shell.clone()) + .await?; + assert_eq!( + process, + resources + .start_platform_shell_exec(request.clone(), shell.clone()) + .await? + ); + let mut different = shell.clone(); + different.command = "echo different".into(); + assert!( + resources + .start_platform_shell_exec(request.clone(), different) + .await + .is_err() + ); + assert!(resources.start_exec(request.clone()).await.is_err()); + completed(&resources, &process).await?; + let batch = resources.read_output(&process, 0).await?; + assert_eq!(batch.state, ExecutionState::Exited { exit_code: 0 }); + let bytes: Vec<_> = batch + .chunks + .iter() + .flat_map(|chunk| chunk.bytes.iter().copied()) + .collect(); + assert_eq!(String::from_utf8_lossy(&bytes).trim(), "fixture"); + resources.release_output(&process).await?; + let mut nonempty = request; + nonempty.command.argv = true_command(); + assert!( + resources + .start_platform_shell_exec(nonempty, shell) + .await + .is_err() + ); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn shell_execution_rejects_remote_selection_without_consuming_admission() -> Result<()> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let request = request(&resources, &resource, Vec::new()).await?; + let shell = tinybox_bus::SelectShellRequest { + command: "echo fixture".into(), + preserve_pipeline_status: false, + }; + let slot = resources.slot(&resource).await?; + { + let mut guard = slot.lock().await; + let entry = guard + .as_mut() + .ok_or_else(|| Error::failed("fixture resource missing"))?; + entry.native_host = false; + } + assert!( + matches!(resources.start_platform_shell_exec(request.clone(), shell.clone()).await, + Err(tinybus::Error::MethodFailed { ref name, .. }) if name.as_str() == tinybox_bus::UNSUPPORTED_OPERATION) + ); + { + let mut guard = slot.lock().await; + let entry = guard + .as_mut() + .ok_or_else(|| Error::failed("fixture resource missing"))?; + entry.native_host = true; + } + let process = resources.start_platform_shell_exec(request, shell).await?; + completed(&resources, &process).await?; + resources.cancel(&process).await?; + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn timeout_status_survives_failed_cleanup_until_retry_acknowledges_it() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let mut shell = shell_request(&resources, &resource, state.path(), "unused").await?; + shell.timeout_ms = 1; + let host = Arc::new(RemoteObservedFixture { + pending: std::sync::atomic::AtomicBool::new(false), + reject_retry: std::sync::atomic::AtomicBool::new(true), + collected_calls: std::sync::atomic::AtomicUsize::new(0), + wait_for_timeout: true, + }); + let stream = StreamExecution::start( + request(&resources, &resource, true_command()).await?, + tinybox_core::ExecRequest::new(["unused"]), + Collector::Host(host.clone(), None), + None, + Some(shell), + None, + ); + { + let mut task = stream.task.lock().await; + tokio::time::timeout( + std::time::Duration::from_secs(5), + task.as_mut().ok_or("missing execution task")?, + ) + .await??; + task.take(); + } + assert!(stream.stop().await.is_err()); + assert_eq!( + stream.journal.read(0)?.state, + ExecutionState::Failed { + code: ExecutionFailure::CleanupFailed, + } + ); + assert!(!stream.cleanup_acknowledged()); + host.reject_retry + .store(false, std::sync::atomic::Ordering::SeqCst); + stream.stop().await?; + assert_eq!( + stream.journal.read(0)?.state, + ExecutionState::Failed { + code: ExecutionFailure::TimedOut, + } + ); + assert!(stream.cleanup_acknowledged()); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn close_during_shell_preparation_fences_admission_without_waiting_for_filesystem() +-> std::result::Result<(), Box> { + let resources = Arc::new(Resources::default()); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let shell = shell_request(&resources, &resource, state.path(), "echo must-not-run").await?; + let request = SpawnRequest { + process: shell.process.clone(), + command: ExecRequest { + resource: resource.clone(), + argv: vec!["shell".into()], + cwd: None, + env: std::collections::BTreeMap::default(), + stdin: None, + }, + }; + let entered = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let started = entered.clone(); + let resume = release.clone(); + let prepared = Arc::new(std::sync::Mutex::new(None)); + let observed = prepared.clone(); + let owner = resources.clone(); + let pending = tokio::spawn(async move { + owner + .start_with_preparation(request, None, Some(shell), None, |preparation| async move { + started.notify_one(); + resume.notified().await; + let collector = preparation.finish().await?; + *observed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(collector.clone()); + Ok(collector) + }) + .await + }); + entered.notified().await; + let closed = tokio::time::timeout( + std::time::Duration::from_secs(1), + resources.close(&resource), + ) + .await; + release.notify_one(); + let result = pending.await?; + assert!( + closed.is_ok(), + "Close waited for shell filesystem preparation" + ); + closed??; + assert!( + prepared + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_some() + ); + assert_eq!( + std::fs::read_dir(state.path().join("artifacts/sandbox-scratch"))?.count(), + 0, + "rejected preparation must explicitly dispose scratch while the context is still retained" + ); + assert!( + result.is_err(), + "late preparation admitted execution after Close" + ); + assert!( + resources + .streams + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() + ); + Ok(()) +} + +#[tokio::test] +async fn native_shell_replay_ignores_rediscovered_program() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let shell = shell_request(&resources, &resource, state.path(), "echo replay").await?; + let process = resources.start_shell_exec(shell.clone()).await?; + let stream = resources.stream(&process)?.ok_or("missing stream")?; + let mut rediscovered = stream.request.clone(); + rediscovered.command.argv = vec!["different-discovered-shell".into()]; + stream.check_replay(&rediscovered, None, Some(&shell), None)?; + completed(&resources, &process).await?; + resources.release_output(&process).await?; + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn rejected_shell_scratch_cleanup_is_retained_for_close_and_shutdown_retry() +-> std::result::Result<(), Box> { + for shutdown in [false, true] { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let shell = shell_request(&resources, &resource, state.path(), "echo must-not-run").await?; + let request = SpawnRequest { + process: shell.process.clone(), + command: ExecRequest { + resource: resource.clone(), + argv: vec!["shell".into()], + cwd: None, + env: std::collections::BTreeMap::default(), + stdin: None, + }, + }; + let scratch = state.path().join("artifacts/sandbox-scratch"); + let result = resources + .start_with_preparation(request, None, Some(shell), None, |preparation| async { + let collector = preparation.finish().await?; + resources.close(&resource).await?; + if let Collector::Shell(shell) = &collector { + shell + .scratch_deletion_blocked + .store(true, std::sync::atomic::Ordering::SeqCst); + } + Ok(collector) + }) + .await; + assert_eq!( + result.err().map(|error| error.wire_name().to_owned()), + Some(tinybox_bus::BACKEND_ERROR.to_owned()) + ); + assert_eq!(resources.rejected_shells.lock().await.len(), 1); + resources.rejected_shells.lock().await[0] + .1 + .scratch_deletion_blocked + .store(false, std::sync::atomic::Ordering::SeqCst); + if shutdown { + resources.shutdown().await?; + } else { + resources.close(&resource).await?; + } + assert!(resources.rejected_shells.lock().await.is_empty()); + assert_eq!(std::fs::read_dir(&scratch)?.count(), 0); + } + Ok(()) +} + +#[tokio::test] +async fn unreserved_shell_processes_are_rejected_before_preparation() -> Result<()> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let request = SpawnRequest { + process: ResourceId("unreserved-process".into()), + command: ExecRequest { + resource: resource.clone(), + argv: true_command(), + cwd: None, + env: std::collections::BTreeMap::default(), + stdin: None, + }, + }; + let called = std::sync::atomic::AtomicBool::new(false); + let result = resources + .start_with_preparation(request, None, None, None, |preparation| async { + called.store(true, std::sync::atomic::Ordering::SeqCst); + preparation.finish().await + }) + .await; + assert!(result.is_err()); + assert!( + !called.load(std::sync::atomic::Ordering::SeqCst), + "unreserved requests must not allocate scratch or run preflight" + ); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn rejected_scratch_failure_does_not_prevent_close_from_fencing_and_stopping_streams() +-> std::result::Result<(), Box> { + let resources = Resources::default(); + let resource = local_resource(&resources).await?; + let state = tempfile::tempdir()?; + let shell_request = shell_request(&resources, &resource, state.path(), "echo unused").await?; + let context = super::super::super::native_shell::ShellContext::prepare( + shell_request, + state.path(), + &tinybox_core::ExecRequest::new(["unused"]), + )?; + context + .scratch_deletion_blocked + .store(true, std::sync::atomic::Ordering::SeqCst); + resources + .rejected_shells + .lock() + .await + .push((resource.clone(), context.clone())); + let process = resources + .start_exec(request(&resources, &resource, true_command()).await?) + .await?; + let stream = resources.stream(&process)?.ok_or("missing stream")?; + assert!(resources.close(&resource).await.is_err()); + assert!( + stream.cleanup_acknowledged(), + "scratch failure must not bypass stream shutdown" + ); + assert!( + resources + .reserve(ReserveRequest::Process(resource.clone())) + .await + .is_err(), + "scratch failure must not leave the resource open" + ); + context + .scratch_deletion_blocked + .store(false, std::sync::atomic::Ordering::SeqCst); + resources.close(&resource).await?; + Ok(()) +} + +#[tokio::test] +async fn concurrent_preparations_share_a_bounded_admission_budget() +-> std::result::Result<(), Box> { + let resources = Arc::new(Resources::default()); + let resource = local_resource(&resources).await?; + let request = request(&resources, &resource, true_command()).await?; + let (entered, mut receiver) = tokio::sync::mpsc::channel(1); + let release = Arc::new(tokio::sync::Barrier::new(MAX_STREAM_EXECUTIONS + 1)); + let mut pending = Vec::new(); + for _ in 0..MAX_STREAM_EXECUTIONS { + let owner = resources.clone(); + let request = request.clone(); + let entered = entered.clone(); + let release = release.clone(); + pending.push(tokio::spawn(async move { + owner + .start_with_preparation(request, None, None, None, |preparation| async move { + entered + .send(()) + .await + .map_err(|error| Error::failed(error.to_string()))?; + release.wait().await; + preparation.finish().await + }) + .await + })); + receiver.recv().await.ok_or("preparation did not start")?; + } + let called = std::sync::atomic::AtomicBool::new(false); + let rejected = resources + .start_with_preparation(request, None, None, None, |preparation| async { + called.store(true, std::sync::atomic::Ordering::SeqCst); + preparation.finish().await + }) + .await; + assert_eq!( + rejected.err().map(|error| error.wire_name().to_owned()), + Some(tinybox_bus::RESOURCE_LIMIT.to_owned()) + ); + assert!(!called.load(std::sync::atomic::Ordering::SeqCst)); + release.wait().await; + for task in pending { + task.await??; + } + resources.close(&resource).await?; + assert_eq!( + resources.preparations.available_permits(), + MAX_STREAM_EXECUTIONS + ); + Ok(()) +} diff --git a/crates/tinybox-module/src/tinybus_module/resources_tests.rs b/crates/tinybox-module/src/tinybus_module/resources_tests.rs index 452afdcc..28a513ee 100644 --- a/crates/tinybox-module/src/tinybus_module/resources_tests.rs +++ b/crates/tinybox-module/src/tinybus_module/resources_tests.rs @@ -2285,7 +2285,7 @@ async fn native_cancel_and_close_terminate_descendants_and_reap_direct_children( .lines() .find(|line| line.starts_with("State:")) .is_some_and(|line| line.contains("Z (zombie)")), - "descendant still executes after native cleanup" + "descendant still executes after native cleanup ({mode}): {status}" ); } } diff --git a/crates/tinybox-module/tests/grants_manifest_e2e.rs b/crates/tinybox-module/tests/grants_manifest_e2e.rs index 22f66250..9b33d44e 100644 --- a/crates/tinybox-module/tests/grants_manifest_e2e.rs +++ b/crates/tinybox-module/tests/grants_manifest_e2e.rs @@ -4,7 +4,7 @@ use std::time::Duration; use tinybus::{Connection, broker::Broker, module::ModuleHost, transport::memory::MemoryBus}; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn compiled_manifest_admits_and_executes_local_grant_discovery() +async fn compiled_manifest_admits_and_executes_grant_and_shell_discovery() -> Result<(), Box> { let built_artifact = compiled_artifact()?; // Cargo output inherits the checkout ACL on Windows, which need not meet @@ -59,6 +59,7 @@ async fn compiled_manifest_admits_and_executes_local_grant_discovery() tinybox_bus::INTERFACE, )?; verify_discovery(&proxy).await?; + verify_native_shell(&proxy).await?; proxy.call::<()>("Shutdown", ()).await?; broker_task.abort(); Ok(()) @@ -92,6 +93,45 @@ async fn verify_discovery(proxy: &tinybus::Proxy) -> Result<(), Box::new()); + let selection: tinybox_bus::ShellInvocation = proxy + .call( + "SelectShell", + (tinybox_bus::SelectShellRequest { + command: "false | true".into(), + preserve_pipeline_status: false, + },), + ) + .await?; + if cfg!(windows) { + assert_eq!(selection.flavor, tinybox_bus::ShellFlavor::Cmd); + let system_root = std::env::var_os("SystemRoot").map_or_else( + || std::path::PathBuf::from(r"C:\Windows"), + std::path::PathBuf::from, + ); + assert_eq!( + std::path::Path::new(&selection.program), + system_root.join("System32").join("cmd.exe") + ); + assert_eq!(selection.arguments, ["/D", "/E:ON", "/V:OFF", "/S", "/C"]); + assert_eq!(selection.raw_arguments.as_deref(), Some("\"false | true\"")); + } else { + assert_eq!(selection.flavor, tinybox_bus::ShellFlavor::Posix); + assert!(selection.arguments.iter().any(|argument| argument == "-c")); + if selection.program.ends_with("bash") { + assert_eq!(selection.arguments[0], "-p"); + } + assert_eq!(selection.raw_arguments, None); + assert_eq!( + selection.remove_environment, + [ + "BASH_ENV", + "ENV", + "SHELLOPTS", + "BASHOPTS", + "POSIXLY_CORRECT" + ] + ); + } for include in ["~other/shared.conf", "%(prefix)/shared.conf"] { std::fs::write( home.path().join(".gitconfig"), @@ -115,24 +155,98 @@ async fn verify_discovery(proxy: &tinybus::Proxy) -> Result<(), Box Result<(), Box> { + let workspace = tempfile::tempdir()?; + let state_dir = tempfile::tempdir()?; + let resource: tinybox_bus::ResourceId = proxy + .call("Reserve", (tinybox_bus::ReserveRequest::Resource,)) + .await?; + let _: tinybox_bus::ResourceInfo = proxy + .call( + "Create", + (tinybox_bus::CreateRequest { + resource: resource.clone(), + backend: "passthrough".into(), + workspace: tinybox_bus::Workspace::Directory( + workspace.path().to_str().ok_or("non-UTF8 fixture")?.into(), + ), + ..Default::default() + },), + ) + .await?; + let process: tinybox_bus::ResourceId = proxy + .call( + "Reserve", + (tinybox_bus::ReserveRequest::Process(resource.clone()),), + ) + .await?; + #[cfg(windows)] + let payload = { + let executable = workspace.path().join("quoted fixture tool.exe"); + std::fs::copy(std::env::current_exe()?, &executable)?; + format!( + "\"{}\" \"--list\" > \"{}\" & echo native-fixture", + executable.display(), + workspace.path().join("output (fixture).txt").display() + ) + }; + #[cfg(not(windows))] + let payload = "printf fixture > 'output (fixture).txt'; echo native-fixture"; + let running: tinybox_bus::ProcessRef = proxy + .call( + "StartNativeShellExec", + (serde_json::json!({ + "process":process, "resource":resource, + "shell":{"command":payload}, + "state_dir":state_dir.path().to_str().ok_or("non-UTF8 fixture")?, + "timeout_ms":5000 + }),), + ) + .await?; + let batch = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let batch: tinybox_bus::OutputBatch = + proxy.call("ReadOutput", (running.clone(), 0_u64)).await?; + if batch.state != tinybox_bus::ExecutionState::Running { + return Ok::<_, tinybus::Error>(batch); + } + tokio::task::yield_now().await; + } + }) + .await??; + assert_eq!( + batch.state, + tinybox_bus::ExecutionState::Exited { exit_code: 0 } + ); + let stdout: Vec = batch + .chunks + .into_iter() + .filter(|chunk| chunk.stream == tinybox_bus::OutputStream::Stdout) + .flat_map(|chunk| chunk.bytes) + .collect(); + let stdout = String::from_utf8(stdout)?; + assert!(stdout.trim().ends_with("native-fixture")); + let redirected = std::fs::read_to_string(workspace.path().join("output (fixture).txt"))?; + #[cfg(windows)] + assert!(redirected.contains("compiled_manifest_admits_and_executes_grant_and_shell_discovery")); + #[cfg(not(windows))] + assert_eq!(redirected, "fixture"); + proxy.call::<()>("ReleaseOutput", (running,)).await?; + proxy.call::<()>("Close", (resource,)).await?; + Ok(()) +} + fn compiled_artifact() -> Result> { - let executable = std::env::current_exe()?; - let dependencies = executable - .parent() - .ok_or("integration test has no parent directory")?; let filename = format!( "{}tinybox{}", std::env::consts::DLL_PREFIX, std::env::consts::DLL_SUFFIX ); - // All-feature/static-link builds omit the dynamic ABI. Always build the - // dynamic variant separately: another validation command may replace the - // ordinary dependency artifact after Cargo releases its compilation lock. - let target = dependencies - .parent() - .and_then(std::path::Path::parent) - .ok_or("integration test has no target directory")? - .join("module-dynamic-proof"); + // All-feature/static-link builds omit the dynamic ABI. Build a dynamic + // variant under this checkout: other PRs share ordinary Cargo artifacts + // and must not replace the module between its build and admission. + let target = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/module-dynamic-proof"); let status = std::process::Command::new(env!("CARGO")) .args([ "build", diff --git a/docs/plans/README.md b/docs/plans/README.md index b0924f18..9a93dc14 100644 --- a/docs/plans/README.md +++ b/docs/plans/README.md @@ -21,4 +21,7 @@ into the plan. See [`example-retry-policy.md`](example-retry-policy.md) for a test-first sample. +Shell discovery: [shell-selection.md](shell-selection.md). Local grant discovery: [local-grants.md](local-grants.md). + +- [Native shell execution](native-shell-execution.md) diff --git a/docs/plans/native-shell-execution.md b/docs/plans/native-shell-execution.md new file mode 100644 index 00000000..0dbafc9a --- /dev/null +++ b/docs/plans/native-shell-execution.md @@ -0,0 +1,20 @@ +# Native shell execution implementation order + +[Specification](../specs/native-shell-execution.md). + +1. Reproduce the absent module consumer through the unconditional real-artifact + test in crates/tinybox-module/tests/grants_manifest_e2e.rs. +2. Add ShellExecRequest in crates/tinybox-bus/src/shell_execution.rs, append + StartNativeShellExec and contract 1.14, and add TimedOut output failure. +3. Extend the existing collectors in tinybox-host/src/limited/{mod,jailed}.rs + with observed standard-command bridges; retain owned kill/reap behavior. +4. Implement validation, environment and scratch ownership in the module's + native_shell directory. Reuse native_jail preparation for strict confinement. +5. Integrate shell collectors and exact request identity in resources_stream.rs. + Reuse existing reservations, output journals and resource cleanup. +6. Add behavioral tests beside each module and actual artifact execution. + Validate formatting, affected/full tests, Clippy, build, coverage and + cross-platform CI through the workflow runner. + +Implementation is complete; upstream review and cross-platform CI remain. No host gitlink or availability claim is +part of this upstream slice. diff --git a/docs/plans/shell-selection.md b/docs/plans/shell-selection.md new file mode 100644 index 00000000..80e930be --- /dev/null +++ b/docs/plans/shell-selection.md @@ -0,0 +1,47 @@ +# Shell selection implementation sequence + +[Specification](../specs/shell-selection.md). +This records the discovery and native execution slices and their verification order. + +1. Add SelectShellRequest, ShellFlavor and ShellInvocation in + crates/tinybox-bus/src/shell_selection.rs. Re-export from src/lib.rs, + append SelectShell to METHODS, advance the additive contract to 1.12, + and cover defaults, serialization and method ordering in lib_tests.rs. +2. Implement platform selection and the native quote-preserving builder in + crates/tinybox-host/src/local/shell.rs; centralize exports in src/lib.rs. + Portable regressions in local/shell_tests.rs pin non-login startup, + Bash pipefail, preservation, sh fallback and cmd /D /E:ON /V:OFF /S /C raw wrapping. + The startup regressions failed before the review corrections. +3. Add module dispatch and manifest declaration in + crates/tinybox-module/src/tinybus_module/mod.rs. Extend + tests/grants_manifest_e2e.rs to build the actual artifact with --locked + --offline, enforce production admission and invoke SelectShell. +4. Execute pipeline fixtures on Unix and quoted executable, quoted argument + and redirection fixtures on Windows. Keep fallible tests returning Result; + portable selection fixtures also cover unavailable-Bash fallback. +5. Update the module README's method table, contract version and startup + semantics. Run formatting, affected tests, full Clippy/build and the + hosted cross-platform matrix before claiming complete validation. + +Local verification: + +- cargo fmt --all -- --check +- cargo test --locked -p tinybox-bus -p tinybox-host -p tinybox-module --all-features +- cargo clippy --locked --workspace --all-targets --all-features -- -D warnings +- cargo build --locked --workspace --all-targets --all-features + +6. Contract 1.13 appends StartShellExec(SpawnRequest, SelectShellRequest). + Reuse the live supervisor with a native-shell collector, keep shell text in + the idempotency identity, resolve resource cwd/environment, and reject + non-local or non-passthrough resources. Empty argv distinguishes shell + execution from ordinary StartExec. Cover identity/admission in stream tests + and execute through the compiled artifact on Unix and Windows. + +Discovery and supervised native execution are implemented. Hosted validation +is reported on the pull request; Windows behavior requires its native matrix. + +Windows explicitly enables extensions and disables delayed expansion. +Unix builders remove BASH_ENV, ENV, SHELLOPTS and BASHOPTS; ShellInvocation +returns those removal names. Bash uses -p to ignore imported environment +functions. Each discovery probes the standard paths again so shell installation +or removal is reflected immediately. Providers apply them after approved environment overrides. diff --git a/docs/specs/README.md b/docs/specs/README.md index a5a46565..15872277 100644 --- a/docs/specs/README.md +++ b/docs/specs/README.md @@ -22,4 +22,7 @@ the contract; production code still belongs under `src/`. See [`example-retry-policy.md`](example-retry-policy.md) for a complete sample. +Shell discovery: [shell-selection.md](shell-selection.md). Local grant discovery: [local-grants.md](local-grants.md). + +- [Native shell execution](native-shell-execution.md) diff --git a/docs/specs/native-shell-execution.md b/docs/specs/native-shell-execution.md new file mode 100644 index 00000000..1f45941c --- /dev/null +++ b/docs/specs/native-shell-execution.md @@ -0,0 +1,38 @@ +# Native shell execution + +Status: implemented upstream; review and cross-platform CI pending. Owner: TinyBox. +Sequence: [native shell plan](../plans/native-shell-execution.md). + +Contract 1.14 appends StartNativeShellExec(ShellExecRequest) -> ProcessRef. +The caller supplies process/resource reservations, SelectShellRequest, optional +cwd, approved extra environment, stdin, native jail configuration, timeout +(default 30 seconds, positive, at most 24 hours), and approved existing absolute +state_dir. This is local passthrough execution only; other providers and +unavailable native jails refuse explicitly. + +The module owns shell selection and raw Windows quoting, snapshots only the +standard execution passthrough and Windows bootstrap names, rejects empty +present passthrough values, applies approved overrides, and finally removes +startup-hook variables. It clears the rest of the ambient environment. + +Each execution owns a private scratch directory under +state_dir/artifacts/sandbox-scratch. TMPDIR, TEMP and TMP default to it unless +explicitly overridden. Approved state directory ACLs must be private on Windows; +the host owns approval of state_dir. Native jails add this module-created +scratch as a writable grant after validating caller grants and enforcement. +Filesystem preparation and recursive cleanup run on blocking workers so slow +storage does not stall the module runtime. + +Capture uses owned pipes and the existing combined output budget, not shell +redirection. ReadOutput, Cancel, ReleaseOutput, Close and Shutdown retain their +cleanup barriers. Timeouts drop the waiter, wait for supervised kill/reap and +then publish TimedOut. Cleanup failures retain ownership and scratch for retry; +the pending timeout outcome survives until a later cleanup retry succeeds. +Scratch survives through journal retention and is reclaimed after acknowledged +release/close. Exact original request identity makes retry idempotent even if +the parent environment changed; mixing shell/argv reservations is refused. + +Acceptance proves actual compiled-module execution on Windows and Unix, +quoting, environment isolation, scratch retention/cleanup, timeout/cancellation, +bounded output, identity replay, admission refusal and unavailable confinement. +OpenHuman host adaptation waits for an available released module. diff --git a/docs/specs/shell-selection.md b/docs/specs/shell-selection.md new file mode 100644 index 00000000..4096c734 --- /dev/null +++ b/docs/specs/shell-selection.md @@ -0,0 +1,63 @@ +# Platform shell selection + +Status: implemented discovery and supervised native execution, under review. Owner: TinyBox. +Implementation sequence: [shell selection plan](../plans/shell-selection.md). + +## Problem and boundary + +A host must discover the module machine's shell syntax without linking TinyBox +implementation crates or returning environment values. Shell selection does +not execute code, approve access, or claim confinement. Native shell execution +uses the module-owned output and cleanup supervisor. + +## Contract + +Contract 1.12 appends SelectShell(SelectShellRequest) -> ShellInvocation. +The request carries command text and preserve_pipeline_status, default false. +The result carries flavor, program, ordinary arguments, optional Windows raw +arguments, and bootstrap environment variable names. + +Unix selects /usr/bin/bash or /bin/bash when present, otherwise /bin/sh. The fallback path is absolute so approved +PATH overrides affect payload tools without preventing shell startup. Both use +non-login -c; Bash also uses -p to ignore imported environment functions. Bash prepends "set -o pipefail" unless preservation is requested; +sh never receives that Bash-only prefix. For "false | true", Bash's default +selection reports the failing stage; preservation retains the last-stage status. + +Windows selects SystemRoot/System32/cmd.exe from the parent environment, +independently of payload PATH or COMSPEC overrides, with /D /E:ON /V:OFF /S /C. /D suppresses registry AutoRun. /S strips +exactly one supplied outer wrapper around the raw payload, leaving inner quoted +executable paths, argument pairs and redirection paths intact. Native builders +must use raw_arg for this payload rather than ordinary argument escaping. + +Bootstrap results list Windows environment names, including SystemRoot, +COMSPEC and USERPROFILE, without values. The execution provider owns its +approved environment, working directory, pipes and lifecycle. + +## Acceptance and constraints + +- Wire defaults and method ordering remain backward compatible. +- Selecting a shell never starts a process or returns environment secrets. +- Portable fixtures cover Bash, sh fallback, pipeline preservation and cmd + argument/wrapper semantics. +- Unix native fixtures exercise pipeline exit statuses. +- Windows native fixtures execute quoted redirection and a quoted executable + with a quoted argument. +- The actual compiled dynamic module passes production manifest admission and + executes SelectShell, including when the surrounding test build is static. +- Hosts execute behavior through the compiled module. Contract 1.13 appends + StartShellExec(SpawnRequest, SelectShellRequest) -> ProcessRef. The spawn + command supplies the resource, cwd, environment and stdin with empty argv. +- StartShellExec supports local passthrough resources only and rejects other + providers before consuming the process reservation. Selection describes the + module machine, so its plan cannot be applied to remote hosts or containers. +- It shares StartExec's bounded output journals, request-identity checks, + cancellation and cleanup barriers. The native builder preserves cmd raw + arguments through to the supervised child. +- Compiled-module fixtures execute shell text on Unix and quoted executable, + argument and redirection paths on Windows. + +Windows explicitly enables extensions and disables delayed expansion. +Unix builders remove BASH_ENV, ENV, SHELLOPTS, BASHOPTS and POSIXLY_CORRECT; ShellInvocation +returns those removal names. Bash uses -p to ignore imported environment +functions. Each discovery probes the standard paths again so shell installation +or removal is reflected immediately. Providers apply them after approved environment overrides.