diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6cbc487221..cb18c0e0e4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -90,6 +90,7 @@ jobs: # is the PR this job starts covering it. AARCH64_CRATES: >- -p litebox + -p litebox_broker_userland -p litebox_common_linux -p litebox_egress_proxy -p litebox_syscall_rewriter diff --git a/dev_bench/unixbench/README.md b/dev_bench/unixbench/README.md index 19304a6368..f5b89d0a2a 100644 --- a/dev_bench/unixbench/README.md +++ b/dev_bench/unixbench/README.md @@ -6,11 +6,12 @@ Run [byte-unixbench](https://github.com/kdlucas/byte-unixbench) benchmarks nativ - The UnixBench source tree at `byte-unixbench-6.0.0/UnixBench/` (extracted from `v6.0.0.zip`). - `gcc`, `make`, `ldd`, `tar` on the host. -- Pre-built LiteBox binaries (`litebox_runner_linux_userland` and `litebox_syscall_rewriter`). +- Pre-built LiteBox binaries (`litebox-broker-userland`, + `litebox_runner_linux_userland`, and `litebox_syscall_rewriter`). Build LiteBox (from workspace root): ```bash -cargo build --release -p litebox_runner_linux_userland -p litebox_syscall_rewriter +cargo build --release -p litebox_broker_userland -p litebox_runner_linux_userland -p litebox_syscall_rewriter ``` ## Quick Start @@ -129,8 +130,8 @@ This creates `dev_bench/unixbench/prepared/` containing: Build the Windows runner, then run benchmarks using the prepared artifacts: ```powershell -# Build the Windows runner -cargo build -p litebox_runner_linux_on_windows_userland --release +# Build the Windows runner and broker +cargo build -p litebox_runner_linux_on_windows_userland -p litebox_broker_userland --release # Run benchmarks python run_unixbench.py --mode litebox --windows --prepared-dir ./prepared --release diff --git a/dev_bench/unixbench/run_unixbench.py b/dev_bench/unixbench/run_unixbench.py index 8df54a87d3..0fe02d66c4 100644 --- a/dev_bench/unixbench/run_unixbench.py +++ b/dev_bench/unixbench/run_unixbench.py @@ -39,6 +39,7 @@ import json import os import re +import signal import shutil import subprocess import sys @@ -350,35 +351,73 @@ def _run_litebox_cmd( # Use a shorter timeout for alarm-based benchmarks under LiteBox, # since if SIGALRM isn't delivered the process will hang forever. timeout = duration * 3 + 30 if bench.uses_alarm else duration * 10 + 60 - try: - result = subprocess.run( - cmd, capture_output=True, timeout=timeout, + popen_options = {} + if sys.platform == "win32": + popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP + else: + popen_options["start_new_session"] = True + with tempfile.TemporaryFile() as stderr_file: + process = subprocess.Popen( + cmd, + stdout=subprocess.DEVNULL, + stderr=stderr_file, + **popen_options, ) - except subprocess.TimeoutExpired: - hint = " (this benchmark uses alarm/SIGALRM)" if bench.uses_alarm else "" - print(f" [TIMEOUT] {bench.name}{hint}") - return None + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + _terminate_process_tree(process) + hint = " (this benchmark uses alarm/SIGALRM)" if bench.uses_alarm else "" + print(f" [TIMEOUT] {bench.name}{hint}") + return None + except KeyboardInterrupt: + _terminate_process_tree(process) + raise + + stderr_file.seek(0) + stderr = stderr_file.read() elapsed = time.monotonic() - t0 - if result.returncode != 0: - stderr = result.stderr.decode("utf-8", errors="replace") - print(f" [FAIL] {bench.name} exited with {result.returncode}") - print(f" stderr (last 300 chars): ...{stderr[-300:]}") + if process.returncode != 0: + decoded_stderr = stderr.decode("utf-8", errors="replace") + print(f" [FAIL] {bench.name} exited with {process.returncode}") + print(f" stderr (last 300 chars): ...{decoded_stderr[-300:]}") return None - stderr = result.stderr.decode("utf-8", errors="replace") - parsed = parse_count_line(stderr) + decoded_stderr = stderr.decode("utf-8", errors="replace") + parsed = parse_count_line(decoded_stderr) if parsed is None: - print(f" [FAIL] {bench.name}: no COUNT line in stderr:\n{stderr[:500]}") + print(f" [FAIL] {bench.name}: no COUNT line in stderr:\n{decoded_stderr[:500]}") return None count, base, unit = parsed return BenchmarkResult( name=bench.name, count=count, base=base, unit=unit, - elapsed=elapsed, raw_stderr=stderr, + elapsed=elapsed, raw_stderr=decoded_stderr, ) +def _terminate_process_tree(process: subprocess.Popen) -> None: + """Terminate a timed-out broker and the runner process it launched.""" + if sys.platform == "win32": + result = subprocess.run( + ["taskkill", "/PID", str(process.pid), "/T", "/F"], + capture_output=True, + ) + if process.poll() is None: + if result.returncode != 0: + error = result.stderr.decode("utf-8", errors="replace").strip() + print(f" Warning: process-tree termination failed: {error}") + process.kill() + else: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + process.wait() + + def run_litebox( pgms_dir: Path, bench: BenchmarkDef, @@ -395,10 +434,11 @@ def run_litebox( return None tar_path, rewritten = prepared + broker_path = runner_path.with_name("litebox-broker-userland") cmd = [ - str(runner_path), - "--unstable", + str(broker_path), + "--runner", str(runner_path), "--env", "LD_LIBRARY_PATH=/lib64:/lib32:/lib", "--env", "HOME=/", ] @@ -448,8 +488,10 @@ def run_litebox_windows( print(f" [SKIP] {bench.name}: tar not found at {tar_path}") return None + broker_path = runner_path.with_name("litebox-broker-userland.exe") cmd = [ - str(runner_path), + str(broker_path), + "--runner", str(runner_path), "--env", "LD_LIBRARY_PATH=/lib64:/lib32:/lib", "--env", "HOME=/", ] @@ -533,7 +575,7 @@ def build_litebox_binaries( workspace_root: Path, release: bool, ) -> tuple[Path, Path]: """ - Build litebox_runner_linux_userland and litebox_packager via cargo. + Build the LiteBox runner, broker, and packager via cargo. Returns (runner_path, packager_path). """ @@ -541,6 +583,7 @@ def build_litebox_binaries( cmd = [ "cargo", "build", "-p", "litebox_runner_linux_userland", + "-p", "litebox_broker_userland", "-p", "litebox_packager", ] if release: @@ -554,8 +597,10 @@ def build_litebox_binaries( print("Build complete.") runner = workspace_root / "target" / build_type / "litebox_runner_linux_userland" + broker = workspace_root / "target" / build_type / "litebox-broker-userland" packager = workspace_root / "target" / build_type / "litebox_packager" assert runner.exists(), f"Runner not found at {runner}" + assert broker.exists(), f"Broker not found at {broker}" assert packager.exists(), f"Packager not found at {packager}" return runner, packager @@ -604,7 +649,8 @@ def main(): ) parser.add_argument( "--runner-path", type=str, default=None, - help="Path to litebox_runner_linux_userland binary (auto-detected if not given)", + help="Path to the LiteBox runner binary; litebox-broker-userland must be beside it " + "(auto-detected if not given)", ) parser.add_argument( "--packager-path", type=str, default=None, @@ -692,6 +738,7 @@ def main(): build_cmd = [ "cargo", "build", "-p", "litebox_runner_linux_on_windows_userland", + "-p", "litebox_broker_userland", ] if args.release: build_cmd.append("--release") @@ -725,6 +772,22 @@ def main(): workspace_root, args.release, ) + if not runner_path.exists(): + print(f"Error: LiteBox runner not found at {runner_path}") + sys.exit(1) + broker_name = ( + "litebox-broker-userland.exe" + if is_windows_mode + else "litebox-broker-userland" + ) + broker_path = runner_path.with_name(broker_name) + if not broker_path.exists(): + print(f"Error: LiteBox broker not found at {broker_path}") + print("Build it beside the runner:") + print(" cargo build -p litebox_broker_userland" + + (" --release" if args.release else "")) + sys.exit(1) + # Working directory if args.work_dir: work_dir = Path(args.work_dir) diff --git a/litebox/src/broker/mod.rs b/litebox/src/broker/mod.rs index 58c9ac599c..1aec97a3cc 100644 --- a/litebox/src/broker/mod.rs +++ b/litebox/src/broker/mod.rs @@ -21,6 +21,7 @@ use litebox_broker_protocol::socket::{ ReceiveFromSocketResponse, ReceiveSocketResponse, SendFlags as BrokerSendFlags, ShutdownMode, SocketConnectionStatus, SocketOutcome, SocketStatusResponse, TcpOptionName, TcpOptionValue, }; +use litebox_broker_protocol::stdio::{MAX_STDIO_TRANSFER_SIZE, StdioOutputStream}; use litebox_broker_transport::channel::LocalCallChannel; use crate::event::{Events, polling::Pollee}; @@ -44,6 +45,12 @@ use shared_buffer::{SlotAllocator, SlotLease}; pub(crate) trait BrokerControl: Send + Sync { fn fill_random(&self, output: &mut [u8]) -> core::result::Result<(), BrokerControlError>; + fn write_stdio( + &self, + stream: StdioOutputStream, + data: &[u8], + ) -> core::result::Result; + fn create_tcp_socket(&self) -> core::result::Result; fn create_udp_socket(&self) -> core::result::Result; @@ -313,6 +320,20 @@ where self.request(|local| local.fill_random(lease.descriptor(), output)) } + fn write_stdio( + &self, + stream: StdioOutputStream, + data: &[u8], + ) -> core::result::Result { + if data.len() > MAX_STDIO_TRANSFER_SIZE as usize { + return Err(BrokerControlError::Broker(ErrorCode::ResourceExhausted)); + } + let length = u32::try_from(data.len()) + .expect("validated shared stdio transfer length must fit in u32"); + let lease = self.acquire_shared_buffer(length)?; + self.request(|local| local.write_stdio(stream, lease.descriptor(), data)) + } + fn create_tcp_socket(&self) -> core::result::Result { self.request(BrokerLocal::create_tcp_socket) } diff --git a/litebox/src/event/counter.rs b/litebox/src/event/counter.rs index 56f533941c..a90b5e8c81 100644 --- a/litebox/src/event/counter.rs +++ b/litebox/src/event/counter.rs @@ -485,10 +485,7 @@ mod tests { BrokerOperation::CheckReadiness(_) => { BrokerResult::Readiness(ReadinessFlags::WRITE) } - request @ (BrokerOperation::Event(_) - | BrokerOperation::Pipe(_) - | BrokerOperation::Socket(_) - | BrokerOperation::FillRandom(_)) => { + request => { panic!("unexpected broker request: {request:?}") } }; diff --git a/litebox/src/fs/devices.rs b/litebox/src/fs/devices.rs index 0957ad9bc2..68d6ec2f67 100644 --- a/litebox/src/fs/devices.rs +++ b/litebox/src/fs/devices.rs @@ -11,6 +11,7 @@ use alloc::vec::Vec; use litebox_broker_protocol::random::MAX_RANDOM_TRANSFER_SIZE; use crate::LiteBox; +use crate::stdio::StdioOutputStream; use crate::sync::RawSyncPrimitivesProvider; use super::backend::{ @@ -291,8 +292,8 @@ where let h = h.get_typed::(); let stream = match h.device { Device::Stdin => return Err(WriteError::NotForWriting), - Device::Stdout => crate::platform::StdioOutStream::Stdout, - Device::Stderr => crate::platform::StdioOutStream::Stderr, + Device::Stdout => StdioOutputStream::Stdout, + Device::Stderr => StdioOutputStream::Stderr, Device::Null | Device::URandom => { // /dev/null discards data: report as if written fully // @@ -306,12 +307,8 @@ where } }; self.litebox - .x - .platform - .write_to(stream, buf) - .map_err(|e| match e { - crate::platform::StdioWriteError::Closed => WriteError::Io, - }) + .write_stdio(stream, buf) + .map_err(|_| WriteError::Io) } fn truncate(&self, _h: &FileHandle, _len: usize) -> Result<(), TruncateError> { diff --git a/litebox/src/fs/tests.rs b/litebox/src/fs/tests.rs index 6a5cbd7072..8c56ab3def 100644 --- a/litebox/src/fs/tests.rs +++ b/litebox/src/fs/tests.rs @@ -1946,6 +1946,7 @@ mod overlay { mod stdio { use crate::LiteBox; use crate::fs::devices::Devices; + use crate::fs::errors::WriteError; use crate::fs::resolver::Resolver; use crate::fs::{Mode, OFlags}; use crate::platform::mock::MockPlatform; @@ -1953,7 +1954,7 @@ mod stdio { extern crate std; #[test] - fn stdio_open_read_write() { + fn stdio_open_read_without_brokered_output() { let platform = MockPlatform::new(); let litebox = LiteBox::new(platform); let fs = Resolver::new( @@ -1968,23 +1969,25 @@ mod stdio { let fd_stdout = fs .open("/dev/stdout", OFlags::WRONLY, Mode::empty()) .expect("Failed to open /dev/stdout"); - let data = b"Hello, stdout!"; - fs.write(&fd_stdout, data, None) - .expect("Failed to write to /dev/stdout"); + assert!(matches!(fs.write(&fd_stdout, b"", None), Ok(0))); + assert!(matches!( + fs.write(&fd_stdout, b"Hello, stdout!", None), + Err(WriteError::Io) + )); fs.close(&fd_stdout).expect("Failed to close /dev/stdout"); - assert_eq!(platform.stdout_queue.read().unwrap().len(), 1); - assert_eq!(platform.stdout_queue.read().unwrap()[0], data); + assert!(platform.stdout_queue.read().unwrap().is_empty()); // Test opening and writing to /dev/stderr let fd_stderr = fs .open("/dev/stderr", OFlags::WRONLY, Mode::empty()) .expect("Failed to open /dev/stderr"); - let data = b"Hello, stderr!"; - fs.write(&fd_stderr, data, None) - .expect("Failed to write to /dev/stderr"); + assert!(matches!(fs.write(&fd_stderr, b"", None), Ok(0))); + assert!(matches!( + fs.write(&fd_stderr, b"Hello, stderr!", None), + Err(WriteError::Io) + )); fs.close(&fd_stderr).expect("Failed to close /dev/stderr"); - assert_eq!(platform.stderr_queue.read().unwrap().len(), 1); - assert_eq!(platform.stderr_queue.read().unwrap()[0], data); + assert!(platform.stderr_queue.read().unwrap().is_empty()); // Test opening and reading from /dev/stdin platform @@ -2030,6 +2033,7 @@ mod composed_stdio { use crate::LiteBox; use crate::fs::composer::Composer; use crate::fs::devices::Devices; + use crate::fs::errors::WriteError; use crate::fs::in_mem::{InMem, InitialNode}; use crate::fs::resolver::Resolver; use crate::fs::{Mode, OFlags, UserInfo}; @@ -2059,7 +2063,7 @@ mod composed_stdio { } #[test] - fn stdio_open_read_write() { + fn stdio_open_read_without_brokered_output() { let platform = MockPlatform::new(); let litebox = LiteBox::new(platform); let fs = composed_fs(&litebox); @@ -2068,23 +2072,25 @@ mod composed_stdio { let fd_stdout = fs .open("/dev/stdout", OFlags::WRONLY, Mode::empty()) .expect("Failed to open /dev/stdout"); - let data = b"Hello, composed stdout!"; - fs.write(&fd_stdout, data, None) - .expect("Failed to write to /dev/stdout"); + assert!(matches!(fs.write(&fd_stdout, b"", None), Ok(0))); + assert!(matches!( + fs.write(&fd_stdout, b"Hello, composed stdout!", None), + Err(WriteError::Io) + )); fs.close(&fd_stdout).expect("Failed to close /dev/stdout"); - assert_eq!(platform.stdout_queue.read().unwrap().len(), 1); - assert_eq!(platform.stdout_queue.read().unwrap()[0], data); + assert!(platform.stdout_queue.read().unwrap().is_empty()); // Test opening and writing to /dev/stderr let fd_stderr = fs .open("/dev/stderr", OFlags::WRONLY, Mode::empty()) .expect("Failed to open /dev/stderr"); - let data = b"Hello, composed stderr!"; - fs.write(&fd_stderr, data, None) - .expect("Failed to write to /dev/stderr"); + assert!(matches!(fs.write(&fd_stderr, b"", None), Ok(0))); + assert!(matches!( + fs.write(&fd_stderr, b"Hello, composed stderr!", None), + Err(WriteError::Io) + )); fs.close(&fd_stderr).expect("Failed to close /dev/stderr"); - assert_eq!(platform.stderr_queue.read().unwrap().len(), 1); - assert_eq!(platform.stderr_queue.read().unwrap()[0], data); + assert!(platform.stderr_queue.read().unwrap().is_empty()); // Test opening and reading from /dev/stdin platform diff --git a/litebox/src/lib.rs b/litebox/src/lib.rs index 6ff7d4f575..0d87da5c1b 100644 --- a/litebox/src/lib.rs +++ b/litebox/src/lib.rs @@ -26,6 +26,7 @@ pub mod pipes; pub mod platform; pub mod random; pub mod shim; +pub mod stdio; pub mod sync; pub mod tls; diff --git a/litebox/src/pipes.rs b/litebox/src/pipes.rs index 57d13fe7e8..8985832cfb 100644 --- a/litebox/src/pipes.rs +++ b/litebox/src/pipes.rs @@ -847,10 +847,7 @@ mod tests { BrokerOperation::CheckReadiness(_) => { BrokerResult::Readiness(ReadinessFlags::default()) } - request @ (BrokerOperation::Pipe(_) - | BrokerOperation::Event(_) - | BrokerOperation::Socket(_) - | BrokerOperation::FillRandom(_)) => { + request => { panic!("unexpected broker request: {request:?}") } }; diff --git a/litebox/src/stdio.rs b/litebox/src/stdio.rs new file mode 100644 index 0000000000..a705027db0 --- /dev/null +++ b/litebox/src/stdio.rs @@ -0,0 +1,36 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +//! Broker-provided standard output. + +use litebox_broker_protocol::stdio::MAX_STDIO_TRANSFER_SIZE; +pub use litebox_broker_protocol::stdio::StdioOutputStream; +use thiserror::Error; + +use crate::{LiteBox, sync::RawSyncPrimitivesProvider}; + +/// A broker could not write to a standard output stream. +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +#[error("brokered standard output is unavailable")] +pub struct WriteStdioError; + +impl LiteBox { + /// Writes bytes to the selected standard output stream. + /// + /// Empty writes succeed without a broker. Non-empty writes require a + /// negotiated broker and may write at most one broker transfer. + pub fn write_stdio( + &self, + stream: StdioOutputStream, + input: &[u8], + ) -> Result { + if input.is_empty() { + return Ok(0); + } + let broker = self.broker_control().ok_or(WriteStdioError)?; + let input = &input[..input.len().min(MAX_STDIO_TRANSFER_SIZE as usize)]; + broker + .write_stdio(stream, input) + .map_err(|_| WriteStdioError) + } +} diff --git a/litebox_broker_core/src/lib.rs b/litebox_broker_core/src/lib.rs index 2b043747fa..52a3a73d03 100644 --- a/litebox_broker_core/src/lib.rs +++ b/litebox_broker_core/src/lib.rs @@ -26,6 +26,7 @@ pub mod random; pub mod readiness; mod session; pub mod socket; +pub mod stdio; use alloc::sync::Arc; use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -43,6 +44,7 @@ use random::RandomProvider; use session::ObjectReference; pub use session::{BrokerSession, CallerCredential, ObjectRights, SessionId}; use socket::{BrokerSocketPorts, SocketProvider}; +use stdio::StdioProvider; /// BrokerCore result type. pub type Result = core::result::Result; @@ -119,6 +121,7 @@ pub struct BrokerCore { pub(crate) reserved_pipe_capacity: Arc, pub(crate) reserved_sockets: Arc, pub(crate) random_provider: Arc, + pub(crate) stdio_provider: Arc, pub(crate) socket_provider: Arc, pub(crate) socket_ports: BrokerSocketPorts, } @@ -126,26 +129,29 @@ pub struct BrokerCore { static BROKER_CORE_CREATED: AtomicBool = AtomicBool::new(false); impl BrokerCore { - /// Creates the broker core with a broker-wide platform socket provider. + /// Creates the broker core with broker-wide platform service providers. pub fn new( policy: PolicyEngine, socket_provider: Arc, random_provider: Arc, + stdio_provider: Arc, ) -> Result { Self::new_with_limits( policy, BrokerCoreLimits::DEFAULT, socket_provider, random_provider, + stdio_provider, ) } - /// Creates the broker core with explicit limits and a socket provider. + /// Creates the broker core with explicit limits and platform service providers. pub fn new_with_limits( policy: PolicyEngine, limits: BrokerCoreLimits, socket_provider: Arc, random_provider: Arc, + stdio_provider: Arc, ) -> Result { BROKER_CORE_CREATED .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) @@ -161,6 +167,7 @@ impl BrokerCore { reserved_pipe_capacity: Arc::new(AtomicUsize::new(0)), reserved_sockets: Arc::new(AtomicUsize::new(0)), random_provider, + stdio_provider, socket_provider, socket_ports: BrokerSocketPorts::default(), }) diff --git a/litebox_broker_core/src/session.rs b/litebox_broker_core/src/session.rs index e14c59d631..30ea3c9601 100644 --- a/litebox_broker_core/src/session.rs +++ b/litebox_broker_core/src/session.rs @@ -545,6 +545,7 @@ mod tests { BrokerCoreLimits::new_with_all_limits(2, 4, 2, 1), socket_provider.clone(), Arc::new(crate::random::TestRandomProvider), + Arc::new(crate::stdio::UnsupportedStdioProvider), ) .unwrap(); diff --git a/litebox_broker_core/src/socket/tests.rs b/litebox_broker_core/src/socket/tests.rs index 9b262e2e66..39afd9628a 100644 --- a/litebox_broker_core/src/socket/tests.rs +++ b/litebox_broker_core/src/socket/tests.rs @@ -1328,6 +1328,7 @@ fn test_broker_with_policy( reserved_pipe_capacity: Arc::new(AtomicUsize::new(0)), reserved_sockets: Arc::new(AtomicUsize::new(0)), random_provider: Arc::new(crate::random::TestRandomProvider), + stdio_provider: Arc::new(crate::stdio::UnsupportedStdioProvider), socket_provider, socket_ports: BrokerSocketPorts::default(), } diff --git a/litebox_broker_core/src/stdio.rs b/litebox_broker_core/src/stdio.rs new file mode 100644 index 0000000000..cae8e364d6 --- /dev/null +++ b/litebox_broker_core/src/stdio.rs @@ -0,0 +1,70 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +//! Broker-authoritative standard I/O. + +use litebox_broker_protocol::stdio::{MAX_STDIO_TRANSFER_SIZE, StdioOutputStream}; +use thiserror::Error; + +use crate::{BrokerError, BrokerSession, Result}; + +/// Failure reported by a trusted standard-I/O provider. +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +pub enum StdioProviderError { + /// The selected output stream is closed. + #[error("standard output stream is closed")] + Closed, + /// The host standard-I/O operation failed internally. + #[error("trusted standard-I/O provider failed")] + Failed, + /// This broker deployment does not provide standard I/O. + #[error("standard I/O is unsupported")] + Unsupported, +} + +/// Trusted provider of standard-I/O operations. +pub trait StdioProvider: Send + Sync { + /// Writes bytes to the selected standard output stream. + fn write( + &self, + stream: StdioOutputStream, + input: &[u8], + ) -> core::result::Result; +} + +/// Standard-I/O provider for deployments that do not expose standard streams. +pub struct UnsupportedStdioProvider; + +impl StdioProvider for UnsupportedStdioProvider { + fn write( + &self, + _stream: StdioOutputStream, + _input: &[u8], + ) -> core::result::Result { + Err(StdioProviderError::Unsupported) + } +} + +/// Writes `input` through the standard-I/O provider configured for this broker. +pub fn write(session: &BrokerSession, stream: StdioOutputStream, input: &[u8]) -> Result { + if input.len() > MAX_STDIO_TRANSFER_SIZE as usize { + return Err(BrokerError::ResourceExhausted); + } + if input.is_empty() { + return Ok(0); + } + let written = + session + .core + .stdio_provider + .write(stream, input) + .map_err(|error| match error { + StdioProviderError::Closed => BrokerError::PeerClosed, + StdioProviderError::Failed => BrokerError::Internal, + StdioProviderError::Unsupported => BrokerError::UnsupportedOperation, + })?; + if written > input.len() { + return Err(BrokerError::Internal); + } + Ok(written) +} diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index a7889d5731..d9424b6034 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -30,6 +30,7 @@ use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse}; use litebox_broker_protocol::message::{ BrokerHandshakeResponse, BrokerOperation, BrokerRequest, BrokerResponse, BrokerResult, EventRequest, EventResponse, PipeRequest, PipeResponse, SocketRequest, SocketResponse, + StdioRequest, StdioResponse, }; use litebox_broker_protocol::pipe::{ CreatePipeResponse, MAX_PIPE_TRANSFER_SIZE, ReadPipeResponse, WritePipeResponse, @@ -44,6 +45,9 @@ use litebox_broker_protocol::socket::{ MAX_UDP_DATAGRAM_SIZE, ReceiveFlags, ReceiveFromSocketResponse, ReceiveSocketResponse, SendSocketResponse, SendToSocketResponse, SocketOutcome, }; +use litebox_broker_protocol::stdio::{ + MAX_STDIO_TRANSFER_SIZE, WriteStdioRequest, WriteStdioResponse, +}; use litebox_broker_protocol::{BROKER_PROTOCOL_VERSION, RequestId}; use litebox_broker_transport::channel::{HostReceive, HostSetupChannel, PeerCredential}; use litebox_broker_transport::shared_memory::{SharedBufferPool, SharedMemory}; @@ -98,6 +102,7 @@ impl BrokerHostAssociation<'_, Memory> { BrokerOperation::Socket(SocketRequest::SendTo(request)) => Some(request.buffer), BrokerOperation::Socket(SocketRequest::Receive(request)) => Some(request.buffer), BrokerOperation::Socket(SocketRequest::ReceiveFrom(request)) => Some(request.buffer), + BrokerOperation::Stdio(StdioRequest::Write(request)) => Some(request.buffer), BrokerOperation::CloseObject(_) | BrokerOperation::CheckReadiness(_) | BrokerOperation::Event(_) @@ -369,6 +374,36 @@ fn handle_request( .map_err(|_| RequestFailure::Abort(ErrorCode::Internal))?; Ok(BrokerResult::RandomFilled) } + BrokerOperation::Stdio(request) => { + handle_stdio_request(session, request, shared_buffers).map(BrokerResult::Stdio) + } + } +} + +fn handle_stdio_request( + session: &BrokerSession, + request: StdioRequest, + shared_buffers: &SharedBufferPool, +) -> RequestResult { + match request { + StdioRequest::Write(WriteStdioRequest { stream, buffer }) => { + if buffer.length > MAX_STDIO_TRANSFER_SIZE { + return Err(RequestFailure::Abort(ErrorCode::MalformedRequest)); + } + let mut data = Vec::new(); + data.try_reserve_exact(buffer.length as usize) + .map_err(|_| RequestFailure::Respond(ErrorCode::OutOfMemory))?; + data.resize(buffer.length as usize, 0); + shared_buffers + .read(buffer.slot_index, &mut data) + .map_err(|_| RequestFailure::Abort(ErrorCode::Internal))?; + let written = litebox_broker_core::stdio::write(session, stream, &data) + .map_err(RequestFailure::from)?; + Ok(StdioResponse::Write(WriteStdioResponse { + written: u32::try_from(written) + .expect("validated stdio write length must fit in u32"), + })) + } } } @@ -716,6 +751,7 @@ mod tests { AcceptedPlatformSocket, PlatformConnectError, PlatformDatagramReceive, PlatformSocket, PlatformSocketStatus, PlatformStreamReceive, SocketProvider, }; + use litebox_broker_core::stdio::{StdioProvider, StdioProviderError}; use litebox_broker_core::{ObjectRights, PolicyEngine, SessionId, SocketPolicy}; use litebox_broker_protocol::event::{ AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode, @@ -735,6 +771,7 @@ mod tests { SocketError, SocketStatusRequest, SocketStatusResponse, SocketType, TcpOptionName, TcpOptionValue, }; + use litebox_broker_protocol::stdio::StdioOutputStream; use litebox_broker_protocol::{ObjectHandle, ProtocolVersion, RequestId}; use litebox_broker_transport::shared_memory::{SharedBufferPool, SharedMemoryError}; use std::sync::{Arc, Condvar, Mutex, mpsc}; @@ -804,6 +841,22 @@ mod tests { } } + #[derive(Default)] + struct TestStdioProvider { + writes: Mutex)>>, + } + + impl StdioProvider for TestStdioProvider { + fn write( + &self, + stream: StdioOutputStream, + input: &[u8], + ) -> core::result::Result { + self.writes.lock().unwrap().push((stream, input.to_vec())); + Ok(input.len()) + } + } + struct TestPlatformSocket { readiness: ReadinessRegistration, create_request: CreateSocketRequest, @@ -953,11 +1006,13 @@ mod tests { #[test] fn host_request_handling_uses_one_broker_core() { + let stdio_provider = Arc::new(TestStdioProvider::default()); let broker = BrokerCore::new( PolicyEngine::with_unauthenticated_rights(ObjectRights::all()) .with_socket_policy(SocketPolicy::guest_network()), Arc::new(TestSocketProvider), Arc::new(TestRandomProvider), + Arc::clone(&stdio_provider) as Arc, ) .unwrap(); @@ -976,6 +1031,7 @@ mod tests { association_shared_buffer_descriptors_stage_pipe_data(&broker); association_shared_buffer_descriptors_stage_socket_data(&broker); association_shared_buffer_descriptor_stages_random_data(&broker); + association_shared_buffer_descriptor_stages_stdio_data(&broker, &stdio_provider); shared_buffer_usage_rejects_invalid_descriptors(); association_executes_distinct_slots_concurrently(&broker); association_allows_slot_reuse_during_response_emission(&broker); @@ -1034,6 +1090,47 @@ mod tests { assert_eq!(output, [0xa5; 2]); } + fn association_shared_buffer_descriptor_stages_stdio_data( + broker: &BrokerCore, + provider: &TestStdioProvider, + ) { + let session = broker + .create_session(CallerCredential::Unauthenticated) + .unwrap(); + let shared_buffers = test_shared_buffers(); + shared_buffers + .write(SharedBufferSlotIndex(7), b"error") + .unwrap(); + + assert_eq!( + handle_test_request_with_buffers( + &session, + BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stderr, + buffer: descriptor(7, 5), + })), + &shared_buffers, + ), + BrokerResult::Stdio(StdioResponse::Write(WriteStdioResponse { written: 5 })) + ); + assert_eq!( + provider.writes.lock().unwrap().as_slice(), + [(StdioOutputStream::Stderr, b"error".to_vec())] + ); + assert_eq!( + handle_request( + &session, + BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stdout, + buffer: descriptor(7, MAX_STDIO_TRANSFER_SIZE + 1), + })), + &shared_buffers, + &test_readiness_sink(), + ), + Err(RequestFailure::Abort(ErrorCode::MalformedRequest)) + ); + } + fn test_channel_negotiates_routes_one_request_and_returns_peer_closed(broker: &BrokerCore) { let mut channel = FakeHostControlChannel::new( std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest { diff --git a/litebox_broker_local/src/lib.rs b/litebox_broker_local/src/lib.rs index 9d09a47a52..25f34118dc 100644 --- a/litebox_broker_local/src/lib.rs +++ b/litebox_broker_local/src/lib.rs @@ -25,6 +25,7 @@ mod event; mod pipe; mod random; mod socket; +mod stdio; use alloc::sync::Arc; use core::sync::atomic::{AtomicU64, Ordering}; @@ -238,8 +239,12 @@ mod tests { use core::convert::Infallible; use litebox_broker_protocol::ObjectHandle; use litebox_broker_protocol::ProtocolVersion; - use litebox_broker_protocol::message::ReadinessNotification; + use litebox_broker_protocol::message::{ReadinessNotification, StdioRequest, StdioResponse}; use litebox_broker_protocol::readiness::ReadinessFlags; + use litebox_broker_protocol::shared_buffer::{SharedBufferDescriptor, SharedBufferSlotIndex}; + use litebox_broker_protocol::stdio::{ + StdioOutputStream, WriteStdioRequest, WriteStdioResponse, + }; use litebox_broker_transport::channel::LocalNotificationChannel; use std::sync::Mutex; @@ -292,6 +297,42 @@ mod tests { ); } + #[test] + fn write_stdio_stages_the_requested_stream_and_buffer() { + let descriptor = SharedBufferDescriptor { + slot_index: SharedBufferSlotIndex(2), + length: 3, + }; + let channel = FakeControlChannel::new( + None, + Some(BrokerResult::Stdio(StdioResponse::Write( + WriteStdioResponse { written: 2 }, + ))), + ); + let local = BrokerLocal { + channel, + shared_buffers: noop_shared_buffers(), + next_request_id: AtomicU64::new(0), + }; + + assert_eq!( + local + .write_stdio(StdioOutputStream::Stderr, descriptor, b"err") + .unwrap(), + 2 + ); + assert_eq!( + local.channel.sent_request.borrow().clone(), + Some(BrokerRequest { + request_id: RequestId(0), + operation: BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stderr, + buffer: descriptor, + })), + }) + ); + } + #[test] fn active_requests_use_monotonic_identifiers() { let handle = ObjectHandle(7); diff --git a/litebox_broker_local/src/stdio.rs b/litebox_broker_local/src/stdio.rs new file mode 100644 index 0000000000..295bf26625 --- /dev/null +++ b/litebox_broker_local/src/stdio.rs @@ -0,0 +1,60 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +use litebox_broker_protocol::message::{ + BrokerOperation, BrokerResult, StdioRequest, StdioResponse, +}; +use litebox_broker_protocol::shared_buffer::SharedBufferDescriptor; +use litebox_broker_protocol::stdio::{ + MAX_STDIO_TRANSFER_SIZE, StdioOutputStream, WriteStdioRequest, +}; +use litebox_broker_transport::channel::LocalCallChannel; + +use crate::{BrokerLocal, BrokerLocalError, Result}; + +impl BrokerLocal { + /// Writes bytes staged in an operation-scoped shared-buffer lease to a + /// standard output stream. + /// + /// The caller must retain exclusive ownership of the descriptor's slot + /// until this method returns. + /// + /// # Panics + /// + /// Panics if the broker reports an unrecoverable error, returns a response + /// for a different operation, or reports an oversized write. + pub fn write_stdio( + &self, + stream: StdioOutputStream, + buffer: SharedBufferDescriptor, + data: &[u8], + ) -> Result { + if buffer.length > MAX_STDIO_TRANSFER_SIZE { + return Err(BrokerLocalError::Broker( + litebox_broker_protocol::error::ErrorCode::ResourceExhausted, + )); + } + assert_eq!( + data.len(), + buffer.length as usize, + "shared stdio write data must match its descriptor" + ); + self.shared_buffers + .write(buffer.slot_index, data) + .expect("validated shared stdio write range must be accessible"); + match self.request(BrokerOperation::Stdio(StdioRequest::Write( + WriteStdioRequest { stream, buffer }, + )))? { + BrokerResult::Stdio(StdioResponse::Write(response)) => { + let written = response.written as usize; + assert!( + written <= data.len(), + "broker returned oversized shared stdio write" + ); + Ok(written) + } + BrokerResult::Error(error) => Err(BrokerLocalError::Broker(error)), + response => panic!("broker returned unexpected stdio response: {response:?}"), + } + } +} diff --git a/litebox_broker_platform_linux_userland/src/socket/tests/mod.rs b/litebox_broker_platform_linux_userland/src/socket/tests/mod.rs index 7b891db30c..8399c05936 100644 --- a/litebox_broker_platform_linux_userland/src/socket/tests/mod.rs +++ b/litebox_broker_platform_linux_userland/src/socket/tests/mod.rs @@ -11,6 +11,7 @@ use super::*; use litebox_broker_core::random::{RandomProvider, RandomProviderError}; use litebox_broker_core::readiness::ReadinessSink; use litebox_broker_core::socket::{GUEST_IPV4_ADDRESS, HOST_GATEWAY_IPV4_ADDRESS}; +use litebox_broker_core::stdio::UnsupportedStdioProvider; use litebox_broker_core::{ BrokerCore, BrokerCoreLimits, BrokerSession, CallerCredential, DestinationPortRange, DestinationRule, Ipv4Cidr, ObjectRights, PolicyEngine, SocketPolicy, @@ -48,6 +49,7 @@ fn test_broker_core( limits, socket_provider, Arc::new(TestRandomProvider), + Arc::new(UnsupportedStdioProvider), ) } diff --git a/litebox_broker_protocol/src/lib.rs b/litebox_broker_protocol/src/lib.rs index efb866a434..a14649219d 100644 --- a/litebox_broker_protocol/src/lib.rs +++ b/litebox_broker_protocol/src/lib.rs @@ -24,6 +24,7 @@ pub mod random; pub mod readiness; pub mod shared_buffer; pub mod socket; +pub mod stdio; pub mod wire; /// Opaque broker object reference handle. diff --git a/litebox_broker_protocol/src/message.rs b/litebox_broker_protocol/src/message.rs index 5af7b4d1c1..0dc9b725f1 100644 --- a/litebox_broker_protocol/src/message.rs +++ b/litebox_broker_protocol/src/message.rs @@ -21,6 +21,7 @@ use crate::socket::{ SendToSocketResponse, SetTcpOptionRequest, ShutdownSocketRequest, SocketError, SocketStatusRequest, SocketStatusResponse, }; +use crate::stdio::{WriteStdioRequest, WriteStdioResponse}; use crate::{ObjectHandle, ProtocolVersion, RequestId}; /// Broker handshake request sent before the control channel is active. @@ -45,6 +46,8 @@ pub enum BrokerOperation { Socket(SocketRequest), /// Fill a shared buffer with cryptographically secure random bytes. FillRandom(SharedBufferDescriptor), + /// Standard-I/O request family. + Stdio(StdioRequest), } /// Request sent over an active broker control channel. @@ -148,6 +151,8 @@ pub enum BrokerResult { Socket(SocketResponse), /// The requested shared buffer was filled with random bytes. RandomFilled, + /// Standard-I/O response family. + Stdio(StdioResponse), /// Operation failed with an ABI-neutral broker error. Error(ErrorCode), } @@ -223,6 +228,20 @@ pub enum SocketResponse { Failed(SocketError), } +/// Standard-I/O request. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum StdioRequest { + /// Write bytes to a standard output stream. + Write(WriteStdioRequest), +} + +/// Standard-I/O response. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum StdioResponse { + /// Standard output write response. + Write(WriteStdioResponse), +} + /// Broker-initiated asynchronous notification. /// /// Notifications are level-triggered snapshots and may be coalesced or diff --git a/litebox_broker_protocol/src/shared_buffer.rs b/litebox_broker_protocol/src/shared_buffer.rs index 8a7c8720a1..dfd1e74c4a 100644 --- a/litebox_broker_protocol/src/shared_buffer.rs +++ b/litebox_broker_protocol/src/shared_buffer.rs @@ -141,6 +141,7 @@ mod tests { assert_eq!(crate::pipe::MAX_PIPE_TRANSFER_SIZE, 32 * 1024); assert_eq!(crate::socket::MAX_SOCKET_TRANSFER_SIZE, 32 * 1024); assert_eq!(crate::socket::MAX_UDP_DATAGRAM_SIZE, 65_507); + assert_eq!(crate::stdio::MAX_STDIO_TRANSFER_SIZE, 32 * 1024); } #[test] diff --git a/litebox_broker_protocol/src/stdio.rs b/litebox_broker_protocol/src/stdio.rs new file mode 100644 index 0000000000..4b011199b6 --- /dev/null +++ b/litebox_broker_protocol/src/stdio.rs @@ -0,0 +1,34 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +use crate::shared_buffer::{SHARED_BUFFER_SLOT_SIZE, SharedBufferDescriptor}; + +/// Maximum standard-output bytes transferred by one broker request. +pub const MAX_STDIO_TRANSFER_SIZE: u32 = 32 * 1024; + +const _: () = assert!(MAX_STDIO_TRANSFER_SIZE <= SHARED_BUFFER_SLOT_SIZE); + +/// Standard output stream selected by a write request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum StdioOutputStream { + /// Process standard output. + Stdout, + /// Process standard error. + Stderr, +} + +/// Request to write bytes staged in shared memory to a standard output stream. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct WriteStdioRequest { + /// Destination standard output stream. + pub stream: StdioOutputStream, + /// Leased shared-buffer region containing the staged bytes. + pub buffer: SharedBufferDescriptor, +} + +/// Response describing a completed standard output write. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct WriteStdioResponse { + /// Number of bytes written to the selected stream. + pub written: u32, +} diff --git a/litebox_broker_protocol/src/wire.rs b/litebox_broker_protocol/src/wire.rs index 01e6f5caa0..4cb41e0d97 100644 --- a/litebox_broker_protocol/src/wire.rs +++ b/litebox_broker_protocol/src/wire.rs @@ -31,6 +31,7 @@ mod event; mod pipe; mod primitive; mod socket; +mod stdio; const REQUEST_TAG_NEGOTIATE: u8 = 0; const REQUEST_TAG_EVENT: u8 = 1; @@ -39,6 +40,7 @@ const REQUEST_TAG_PIPE: u8 = 3; const REQUEST_TAG_CHECK_READINESS: u8 = 4; const REQUEST_TAG_SOCKET: u8 = 5; const REQUEST_TAG_FILL_RANDOM: u8 = 6; +const REQUEST_TAG_STDIO: u8 = 7; // Paired request and successful-response tags intentionally share values. const RESPONSE_TAG_NEGOTIATED: u8 = 0; @@ -48,6 +50,7 @@ const RESPONSE_TAG_PIPE: u8 = 3; const RESPONSE_TAG_READINESS: u8 = 4; const RESPONSE_TAG_SOCKET: u8 = 5; const RESPONSE_TAG_RANDOM_FILLED: u8 = 6; +const RESPONSE_TAG_STDIO: u8 = 7; // Reserve the top of the tag space for responses without paired requests. const RESPONSE_TAG_ERROR: u8 = 253; @@ -102,7 +105,8 @@ pub fn decode_handshake_request(frame: &[u8]) -> Result { + | REQUEST_TAG_FILL_RANDOM + | REQUEST_TAG_STDIO => { return Err(WireError::WrongMessagePhase); } _ => return Err(WireError::InvalidTag), @@ -152,6 +156,11 @@ pub fn encode_request(request: BrokerRequest) -> Vec { encoder.request_id(request_id); encoder.shared_buffer_descriptor(buffer); } + BrokerOperation::Stdio(request) => { + encoder.u8(REQUEST_TAG_STDIO); + encoder.request_id(request_id); + stdio::encode_stdio_request(&mut encoder, request); + } } encoder.finish() } @@ -167,7 +176,8 @@ pub fn decode_request(frame: &[u8]) -> Result { | REQUEST_TAG_EVENT | REQUEST_TAG_PIPE | REQUEST_TAG_SOCKET - | REQUEST_TAG_FILL_RANDOM => {} + | REQUEST_TAG_FILL_RANDOM + | REQUEST_TAG_STDIO => {} _ => return Err(WireError::InvalidTag), } let request_id = decoder.request_id()?; @@ -178,6 +188,7 @@ pub fn decode_request(frame: &[u8]) -> Result { REQUEST_TAG_PIPE => BrokerOperation::Pipe(pipe::decode_pipe_request(&mut decoder)?), REQUEST_TAG_SOCKET => BrokerOperation::Socket(socket::decode_socket_request(&mut decoder)?), REQUEST_TAG_FILL_RANDOM => BrokerOperation::FillRandom(decoder.shared_buffer_descriptor()?), + REQUEST_TAG_STDIO => BrokerOperation::Stdio(stdio::decode_stdio_request(&mut decoder)?), _ => unreachable!("active request tag was validated"), }; decoder.finish()?; @@ -228,7 +239,8 @@ pub fn decode_handshake_response(frame: &[u8]) -> Result { + | RESPONSE_TAG_RANDOM_FILLED + | RESPONSE_TAG_STDIO => { return Err(WireError::WrongMessagePhase); } RESPONSE_TAG_VERSION_MISMATCH => BrokerHandshakeResponse::VersionMismatch { @@ -280,6 +292,11 @@ pub fn encode_response(response: BrokerResponse) -> Vec { encoder.u8(RESPONSE_TAG_RANDOM_FILLED); encoder.request_id(request_id); } + BrokerResult::Stdio(response) => { + encoder.u8(RESPONSE_TAG_STDIO); + encoder.request_id(request_id); + stdio::encode_stdio_response(&mut encoder, response); + } BrokerResult::Error(error) => { encoder.u8(RESPONSE_TAG_ERROR); encoder.request_id(request_id); @@ -303,7 +320,8 @@ pub fn decode_response(frame: &[u8]) -> Result { | RESPONSE_TAG_READINESS | RESPONSE_TAG_ERROR | RESPONSE_TAG_SOCKET - | RESPONSE_TAG_RANDOM_FILLED => {} + | RESPONSE_TAG_RANDOM_FILLED + | RESPONSE_TAG_STDIO => {} _ => return Err(WireError::InvalidTag), } let request_id = decoder.request_id()?; @@ -318,6 +336,7 @@ pub fn decode_response(frame: &[u8]) -> Result { RESPONSE_TAG_OBJECT_CLOSED => BrokerResult::ObjectClosed, RESPONSE_TAG_READINESS => BrokerResult::Readiness(ReadinessFlags(decoder.u32()?)), RESPONSE_TAG_RANDOM_FILLED => BrokerResult::RandomFilled, + RESPONSE_TAG_STDIO => BrokerResult::Stdio(stdio::decode_stdio_response(&mut decoder)?), _ => unreachable!("active response tag was validated"), }; decoder.finish()?; @@ -364,6 +383,7 @@ mod tests { }; use crate::message::{ EventRequest, EventResponse, PipeRequest, PipeResponse, SocketRequest, SocketResponse, + StdioRequest, StdioResponse, }; use crate::pipe::{ CreatePipeRequest, CreatePipeResponse, ReadPipeRequest, ReadPipeResponse, WritePipeRequest, @@ -381,6 +401,7 @@ mod tests { ShutdownSocketRequest, SocketConnectionStatus, SocketError, SocketStatusRequest, SocketStatusResponse, SocketType, TcpOptionName, TcpOptionValue, }; + use crate::stdio::{StdioOutputStream, WriteStdioRequest, WriteStdioResponse}; use crate::{ObjectHandle, ProtocolVersion, RequestId}; use core::net::{Ipv4Addr, SocketAddrV4}; @@ -405,6 +426,7 @@ mod tests { RESPONSE_TAG_READINESS, RESPONSE_TAG_SOCKET, RESPONSE_TAG_RANDOM_FILLED, + RESPONSE_TAG_STDIO, ], [ REQUEST_TAG_NEGOTIATE, @@ -414,6 +436,7 @@ mod tests { REQUEST_TAG_CHECK_READINESS, REQUEST_TAG_SOCKET, REQUEST_TAG_FILL_RANDOM, + REQUEST_TAG_STDIO, ] ); assert_eq!( @@ -483,6 +506,20 @@ mod tests { slot_index: SharedBufferSlotIndex(7), length: 256, }), + BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stdout, + buffer: SharedBufferDescriptor { + slot_index: SharedBufferSlotIndex(6), + length: 17, + }, + })), + BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stderr, + buffer: SharedBufferDescriptor { + slot_index: SharedBufferSlotIndex(5), + length: 23, + }, + })), BrokerOperation::Socket(SocketRequest::Create(CreateSocketRequest { address_family: AddressFamily::Ipv4, socket_type: SocketType::Stream, @@ -817,6 +854,7 @@ mod tests { ))), BrokerResult::Socket(SocketResponse::Failed(SocketError::ConnectionReset)), BrokerResult::RandomFilled, + BrokerResult::Stdio(StdioResponse::Write(WriteStdioResponse { written: 17 })), BrokerResult::Error(ErrorCode::PolicyDenied), BrokerResult::Error(ErrorCode::WouldBlock), BrokerResult::Error(ErrorCode::PeerClosed), @@ -947,6 +985,58 @@ mod tests { assert_eq!(decode_request(&frame), Err(WireError::TrailingBytes)); } + #[test] + fn decode_rejects_malformed_stdio_request_frames() { + let request = BrokerRequest { + request_id: TEST_REQUEST_ID, + operation: BrokerOperation::Stdio(StdioRequest::Write(WriteStdioRequest { + stream: StdioOutputStream::Stdout, + buffer: SharedBufferDescriptor { + slot_index: SharedBufferSlotIndex(1), + length: 9, + }, + })), + }; + + let mut unknown_operation = encode_request(request.clone()); + unknown_operation[9] = 0xff; + assert_eq!( + decode_request(&unknown_operation), + Err(WireError::InvalidTag) + ); + + let mut unknown_stream = encode_request(request.clone()); + unknown_stream[10] = 0xff; + assert_eq!(decode_request(&unknown_stream), Err(WireError::InvalidTag)); + + let frame = encode_request(request); + assert_eq!( + decode_request(&frame[..frame.len() - 1]), + Err(WireError::TruncatedFrame) + ); + } + + #[test] + fn decode_rejects_malformed_stdio_response_frames() { + let response = BrokerResponse { + request_id: TEST_REQUEST_ID, + result: BrokerResult::Stdio(StdioResponse::Write(WriteStdioResponse { written: 9 })), + }; + + let mut unknown_operation = encode_response(response.clone()); + unknown_operation[9] = 0xff; + assert_eq!( + decode_response(&unknown_operation), + Err(WireError::InvalidTag) + ); + + let frame = encode_response(response); + assert_eq!( + decode_response(&frame[..frame.len() - 1]), + Err(WireError::TruncatedFrame) + ); + } + #[test] fn decode_rejects_malformed_socket_request_frames() { let mut unknown_operation = Vec::from([REQUEST_TAG_SOCKET]); diff --git a/litebox_broker_protocol/src/wire/stdio.rs b/litebox_broker_protocol/src/wire/stdio.rs new file mode 100644 index 0000000000..3c14c6fde8 --- /dev/null +++ b/litebox_broker_protocol/src/wire/stdio.rs @@ -0,0 +1,69 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +use crate::message::{StdioRequest, StdioResponse}; +use crate::stdio::{StdioOutputStream, WriteStdioRequest, WriteStdioResponse}; + +use super::{ + WireError, + primitive::{Decoder, Encoder}, +}; + +const REQUEST_TAG_WRITE: u8 = 0; +const RESPONSE_TAG_WRITE: u8 = 0; + +const OUTPUT_STREAM_TAG_STDOUT: u8 = 0; +const OUTPUT_STREAM_TAG_STDERR: u8 = 1; + +pub(super) fn encode_stdio_request(encoder: &mut Encoder, request: StdioRequest) { + match request { + StdioRequest::Write(request) => { + encoder.u8(REQUEST_TAG_WRITE); + encode_output_stream(encoder, request.stream); + encoder.shared_buffer_descriptor(request.buffer); + } + } +} + +pub(super) fn decode_stdio_request(decoder: &mut Decoder<'_>) -> Result { + match decoder.u8()? { + REQUEST_TAG_WRITE => Ok(StdioRequest::Write(WriteStdioRequest { + stream: decode_output_stream(decoder)?, + buffer: decoder.shared_buffer_descriptor()?, + })), + _ => Err(WireError::InvalidTag), + } +} + +pub(super) fn encode_stdio_response(encoder: &mut Encoder, response: StdioResponse) { + match response { + StdioResponse::Write(response) => { + encoder.u8(RESPONSE_TAG_WRITE); + encoder.u32(response.written); + } + } +} + +pub(super) fn decode_stdio_response(decoder: &mut Decoder<'_>) -> Result { + match decoder.u8()? { + RESPONSE_TAG_WRITE => Ok(StdioResponse::Write(WriteStdioResponse { + written: decoder.u32()?, + })), + _ => Err(WireError::InvalidTag), + } +} + +fn encode_output_stream(encoder: &mut Encoder, stream: StdioOutputStream) { + encoder.u8(match stream { + StdioOutputStream::Stdout => OUTPUT_STREAM_TAG_STDOUT, + StdioOutputStream::Stderr => OUTPUT_STREAM_TAG_STDERR, + }); +} + +fn decode_output_stream(decoder: &mut Decoder<'_>) -> Result { + match decoder.u8()? { + OUTPUT_STREAM_TAG_STDOUT => Ok(StdioOutputStream::Stdout), + OUTPUT_STREAM_TAG_STDERR => Ok(StdioOutputStream::Stderr), + _ => Err(WireError::InvalidTag), + } +} diff --git a/litebox_broker_userland/src/linux.rs b/litebox_broker_userland/src/linux.rs index ea29e5aadd..10c6d6a902 100644 --- a/litebox_broker_userland/src/linux.rs +++ b/litebox_broker_userland/src/linux.rs @@ -80,6 +80,7 @@ pub(super) fn run(mut args: super::CliArgs) -> Result<(), Box> { limits.max_sockets_per_session, )?), Arc::new(super::UserlandRandomProvider), + Arc::new(super::UserlandStdioProvider), )?; crate::run_runner_process( diff --git a/litebox_broker_userland/src/main.rs b/litebox_broker_userland/src/main.rs index a40c854358..075d6e1816 100644 --- a/litebox_broker_userland/src/main.rs +++ b/litebox_broker_userland/src/main.rs @@ -17,6 +17,7 @@ use std::time::{Duration, Instant}; use clap::Parser; use litebox_broker_core::random::{RandomProvider, RandomProviderError}; +use litebox_broker_core::stdio::{StdioProvider, StdioProviderError}; use litebox_broker_core::{ BrokerCore, CallerCredential, DestinationPortRange, DestinationRule, Ipv4Cidr, SocketPolicy, SocketPolicyError, @@ -25,6 +26,7 @@ use litebox_broker_host::{BrokerHostAssociation, ConnectionTermination, setup_co use litebox_broker_protocol::message::{BrokerRequest, BrokerResponse}; use litebox_broker_protocol::shared_buffer::SHARED_BUFFER_LAYOUT; use litebox_broker_protocol::socket::{Ipv4Address, Port}; +use litebox_broker_protocol::stdio::StdioOutputStream; use litebox_broker_transport::channel::{HostNotificationChannel, HostReceive, HostSetupChannel}; use litebox_broker_transport::control_ring::ControlRing; use litebox_broker_transport::shared_memory::{ControlRingMemory, SharedBufferPool, SharedMemory}; @@ -49,6 +51,34 @@ impl RandomProvider for UserlandRandomProvider { } } +/// Routes output from the broker's single child runner to inherited streams. +/// +/// A broker serving multiple runners will need association-specific output +/// destinations instead of sharing process-wide standard streams. +struct UserlandStdioProvider; + +impl StdioProvider for UserlandStdioProvider { + fn write(&self, stream: StdioOutputStream, input: &[u8]) -> Result { + let result = match stream { + StdioOutputStream::Stdout => write_and_flush(std::io::stdout().lock(), input), + StdioOutputStream::Stderr => write_and_flush(std::io::stderr().lock(), input), + }; + result.map_err(|error| { + if error.kind() == ErrorKind::BrokenPipe { + StdioProviderError::Closed + } else { + StdioProviderError::Failed + } + }) + } +} + +fn write_and_flush(mut output: impl std::io::Write, input: &[u8]) -> IoResult { + let written = output.write(input)?; + output.flush()?; + Ok(written) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct AllowedDestination { destination: Ipv4Cidr, @@ -779,6 +809,7 @@ mod tests { PolicyEngine::with_host_guaranteed_rights(ObjectRights::all()), Arc::new(UnsupportedSocketProvider), Arc::new(UserlandRandomProvider), + Arc::new(UserlandStdioProvider), ) .unwrap(); let shared_memory = MemfdSharedMemory::create(SHARED_BUFFER_POOL_SIZE).unwrap(); diff --git a/litebox_broker_userland/src/windows.rs b/litebox_broker_userland/src/windows.rs index d7f144ab60..ff3fb5fca5 100644 --- a/litebox_broker_userland/src/windows.rs +++ b/litebox_broker_userland/src/windows.rs @@ -57,6 +57,7 @@ pub(super) fn run(args: super::CliArgs) -> Result<(), Box> { ), Arc::new(UnsupportedSocketProvider), Arc::new(super::UserlandRandomProvider), + Arc::new(super::UserlandStdioProvider), )?; crate::run_runner_process(&args, &control_pipe, None, |runner, runner_process_id| { diff --git a/litebox_broker_userland/tests/notification_runtime.rs b/litebox_broker_userland/tests/notification_runtime.rs index 1bbd999142..f52d1a950a 100644 --- a/litebox_broker_userland/tests/notification_runtime.rs +++ b/litebox_broker_userland/tests/notification_runtime.rs @@ -17,6 +17,7 @@ use std::time::Duration; use litebox_broker_core::random::{RandomProvider, RandomProviderError}; use litebox_broker_core::socket::UnsupportedSocketProvider; +use litebox_broker_core::stdio::UnsupportedStdioProvider; use litebox_broker_core::{BrokerCore, ObjectRights, PolicyEngine}; use litebox_broker_host::{ConnectionTermination, setup_connection}; use litebox_broker_local::{BrokerLocal, BrokerNotifications}; @@ -59,6 +60,7 @@ fn spawn_host( PolicyEngine::with_unauthenticated_rights(ObjectRights::all()), Arc::new(UnsupportedSocketProvider), Arc::new(FailingRandomProvider), + Arc::new(UnsupportedStdioProvider), ) .unwrap(); let shared_memory = MemfdSharedMemory::create(SHARED_BUFFER_POOL_SIZE).unwrap(); @@ -142,6 +144,7 @@ fn host_serves_control_requests_and_notifications_over_shared_rings() { PolicyEngine::with_unauthenticated_rights(ObjectRights::all()), Arc::new(UnsupportedSocketProvider), Arc::new(FailingRandomProvider), + Arc::new(UnsupportedStdioProvider), ) .unwrap(); let (local_control, host_control) = UnixStream::pair().unwrap(); diff --git a/litebox_runner_linux_userland/tests/rewritten_guests.rs b/litebox_runner_linux_userland/tests/rewritten_guests.rs index 530ee9ce9b..07f03ff66b 100644 --- a/litebox_runner_linux_userland/tests/rewritten_guests.rs +++ b/litebox_runner_linux_userland/tests/rewritten_guests.rs @@ -3,8 +3,7 @@ //! Tests for guests whose syscall sites the rewriter redirected. //! -//! Guests run under `--rewrite-syscalls` with no tar rootfs; broker-backed -//! guests live in `run.rs`. +//! Guests run under `--rewrite-syscalls` with no tar rootfs. #[allow(dead_code, reason = "shared with the other test binaries")] mod cache; @@ -16,8 +15,28 @@ fn run_rewritten_fixture(source: &str, unique_name: &str) -> std::process::Outpu let binary_path = std::env::var("NEXTEST_BIN_EXE_litebox_runner_linux_userland") .unwrap_or_else(|_| env!("CARGO_BIN_EXE_litebox_runner_linux_userland").to_string()); - std::process::Command::new(binary_path) - .args(["--unstable", "--rewrite-syscalls"]) + #[cfg(target_os = "linux")] + let mut command = { + let broker_path = + std::path::Path::new(&binary_path).with_file_name("litebox-broker-userland"); + assert!( + broker_path.is_file(), + "brokered runner tests require a workspace build producing {}", + broker_path.display() + ); + let mut command = std::process::Command::new(broker_path); + command.arg("--runner").arg(&binary_path); + command + }; + #[cfg(not(target_os = "linux"))] + let mut command = { + let mut command = std::process::Command::new(binary_path); + command.arg("--unstable"); + command + }; + + command + .args(["--rewrite-syscalls"]) .arg(target) .output() .expect("Failed to run litebox_runner_linux_userland") diff --git a/litebox_runner_linux_userland/tests/run.rs b/litebox_runner_linux_userland/tests/run.rs index 89362193cc..eb40c6b9c5 100644 --- a/litebox_runner_linux_userland/tests/run.rs +++ b/litebox_runner_linux_userland/tests/run.rs @@ -25,6 +25,28 @@ impl litebox_broker_core::random::RandomProvider for TestRandomProvider { Ok(()) } } + +#[cfg(all(target_arch = "x86_64", target_os = "linux"))] +struct CapturingStdioProvider { + stdout_tx: std::sync::mpsc::Sender>, +} + +#[cfg(all(target_arch = "x86_64", target_os = "linux"))] +impl litebox_broker_core::stdio::StdioProvider for CapturingStdioProvider { + fn write( + &self, + stream: litebox_broker_protocol::stdio::StdioOutputStream, + input: &[u8], + ) -> Result { + if stream == litebox_broker_protocol::stdio::StdioOutputStream::Stdout { + self.stdout_tx + .send(input.to_vec()) + .map_err(|_| litebox_broker_core::stdio::StdioProviderError::Failed)?; + } + Ok(input.len()) + } +} + // Dedicated fixtures build static binaries concurrently; exclude them to avoid // colliding with this sweep's dynamic `_rewriter` outputs. const DEDICATED_C_TESTS: &[&str] = &[ @@ -85,9 +107,9 @@ struct Runner { unique_name: String, cmd_path: PathBuf, cmd_args: Vec, - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] managed_proxy_hosts: Vec, - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] use_userland_broker: bool, has_run: bool, } @@ -145,10 +167,10 @@ impl Runner { tar_dir, cmd_path: target_guest_path, cmd_args: Vec::new(), - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] managed_proxy_hosts: Vec::new(), - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] - use_userland_broker: false, + #[cfg(target_os = "linux")] + use_userland_broker: true, has_run: false, unique_name: unique_name.to_owned(), } @@ -189,6 +211,7 @@ impl Runner { #[cfg(all(target_arch = "x86_64", target_os = "linux"))] fn broker_socket(&mut self, control_socket_path: &Path) -> &mut Self { + self.use_userland_broker = false; self.command .arg("--broker-control-channel") .arg(control_socket_path); @@ -227,7 +250,7 @@ impl Runner { .arg(&self.cmd_path) .args(&self.cmd_args); - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] if self.use_userland_broker || !self.managed_proxy_hosts.is_empty() { let runner = self.command.get_program().to_os_string(); let runner_arguments = self @@ -322,7 +345,7 @@ fn has_dedicated_c_test(path: &Path) -> bool { .is_some_and(|name| DEDICATED_C_TESTS.contains(&name)) } -#[cfg(all(target_arch = "x86_64", target_os = "linux"))] +#[cfg(target_os = "linux")] fn configure_pipe_broker(path: &Path, runner: &mut Runner) { if path.file_name().and_then(|name| name.to_str()) == Some("sendfile.c") { runner.use_userland_broker = true; @@ -343,7 +366,7 @@ fn test_dynamic_lib_with_rewriter() { let unique_name = format!("{stem}_rewriter"); let target = common::compile(path.to_str().unwrap(), &unique_name, false, false); let mut runner = Runner::new(&target, &unique_name); - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] configure_pipe_broker(&path, &mut runner); runner.run(); } @@ -362,7 +385,7 @@ fn test_static_exec_with_rewriter() { let unique_name = format!("{stem}_exec_rewriter"); let target = common::compile(path.to_str().unwrap(), &unique_name, true, false); let mut runner = Runner::new(&target, &unique_name); - #[cfg(all(target_arch = "x86_64", target_os = "linux"))] + #[cfg(target_os = "linux")] configure_pipe_broker(&path, &mut runner); runner.run(); } @@ -398,6 +421,8 @@ struct TestBroker { thread: Option>, done_rx: std::sync::mpsc::Receiver<()>, close_object_count_rx: std::sync::mpsc::Receiver, + stdout_rx: std::sync::mpsc::Receiver>, + pending_stdout: std::cell::RefCell>, control_socket_path: PathBuf, } @@ -409,7 +434,39 @@ impl TestBroker { .expect("broker test host did not report close-object count") } - fn join(mut self) { + fn next_stdout_line(&self) -> String { + let deadline = std::time::Instant::now() + BROKER_HELPER_TIMEOUT; + loop { + let newline = self + .pending_stdout + .borrow() + .iter() + .position(|byte| *byte == b'\n'); + if let Some(newline) = newline { + let mut line = self + .pending_stdout + .borrow_mut() + .drain(..=newline) + .collect::>(); + line.pop(); + if line.last() == Some(&b'\r') { + line.pop(); + } + return String::from_utf8(line).expect("guest stdout was not UTF-8"); + } + + let remaining = deadline + .checked_duration_since(std::time::Instant::now()) + .unwrap_or_default(); + let chunk = self + .stdout_rx + .recv_timeout(remaining) + .expect("timed out waiting for brokered guest stdout"); + self.pending_stdout.borrow_mut().extend_from_slice(&chunk); + } + } + + fn finish(&mut self) { self.done_rx .recv_timeout(BROKER_HELPER_TIMEOUT) .expect("broker test host did not finish"); @@ -418,7 +475,19 @@ impl TestBroker { .expect("broker test host thread missing") .join() .expect("broker test host panicked"); - let _ = std::fs::remove_file(&self.control_socket_path); + } + + fn join(mut self) { + self.finish(); + } + + fn join_with_stdout(mut self) -> Vec { + self.finish(); + let mut output = core::mem::take(self.pending_stdout.get_mut()); + for chunk in self.stdout_rx.try_iter() { + output.extend_from_slice(&chunk); + } + output } } @@ -459,6 +528,7 @@ fn spawn_test_broker_with_mode( let (ready_tx, ready_rx) = std::sync::mpsc::channel(); let (done_tx, done_rx) = std::sync::mpsc::channel(); let (close_object_count_tx, close_object_count_rx) = std::sync::mpsc::channel(); + let (stdout_tx, stdout_rx) = std::sync::mpsc::channel(); let server_control_socket_path = control_socket_path.to_path_buf(); let cleanup_control_socket_path = control_socket_path.to_path_buf(); let broker_thread = std::thread::spawn(move || { @@ -478,6 +548,7 @@ fn spawn_test_broker_with_mode( .expect("failed to create broker test socket provider"), ), std::sync::Arc::new(TestRandomProvider), + std::sync::Arc::new(CapturingStdioProvider { stdout_tx }), ) .expect("failed to create broker core"); ready_tx.send(()).expect("failed to report broker ready"); @@ -520,6 +591,8 @@ fn spawn_test_broker_with_mode( thread: Some(broker_thread), done_rx, close_object_count_rx, + stdout_rx, + pending_stdout: std::cell::RefCell::new(Vec::new()), control_socket_path: cleanup_control_socket_path, } } @@ -899,7 +972,6 @@ fn test_runner_broker_udp_with_rewriter() { #[cfg(all(target_arch = "x86_64", target_os = "linux"))] #[test] fn test_runner_broker_udp_namespace_delivers_after_sender_close() { - use std::io::{BufRead as _, BufReader}; use std::process::Stdio; let target = common::compile( @@ -920,19 +992,8 @@ fn test_runner_broker_udp_namespace_delivers_after_sender_close() { let mut server = Runner::new(&target, "broker_udp_namespace_server_rewriter") .arg("server") .broker_socket(&control_socket_path) - .spawn_with_stdio(Stdio::null(), Stdio::piped(), Stdio::inherit()); - let stdout = server.stdout.take().unwrap(); - let (line_sender, line_receiver) = std::sync::mpsc::channel(); - let reader = std::thread::spawn(move || { - for line in BufReader::new(stdout).lines() { - if line_sender.send(line.unwrap()).is_err() { - return; - } - } - }); - let listen = line_receiver - .recv_timeout(BROKER_HELPER_TIMEOUT) - .expect("timed out waiting for broker UDP server"); + .spawn_with_stdio(Stdio::null(), Stdio::null(), Stdio::inherit()); + let listen = broker.next_stdout_line(); let port = listen .strip_prefix("LISTEN ") .expect("unexpected broker UDP server output") @@ -958,7 +1019,6 @@ fn test_runner_broker_udp_namespace_delivers_after_sender_close() { } std::thread::sleep(std::time::Duration::from_millis(10)); }; - reader.join().unwrap(); assert!(status.success(), "broker UDP server guest failed: {status}"); assert_eq!(broker.next_close_object_count(), 1); assert_eq!(broker.next_close_object_count(), 1); @@ -968,7 +1028,6 @@ fn test_runner_broker_udp_namespace_delivers_after_sender_close() { #[cfg(all(target_arch = "x86_64", target_os = "linux"))] #[test] fn test_runner_broker_tcp_server_with_rewriter() { - use std::io::{BufRead as _, BufReader}; use std::process::Stdio; let target = common::compile( @@ -988,34 +1047,16 @@ fn test_runner_broker_tcp_server_with_rewriter() { ); let mut child = Runner::new(&target, "broker_tcp_server_rewriter") .broker_socket(&control_socket_path) - .spawn_with_stdio(Stdio::null(), Stdio::piped(), Stdio::inherit()); - let stdout = child.stdout.take().unwrap(); - let (line_sender, line_receiver) = std::sync::mpsc::channel(); - let reader = std::thread::spawn(move || { - for line in BufReader::new(stdout).lines() { - if line_sender.send(line.unwrap()).is_err() { - return; - } - } - }); + .spawn_with_stdio(Stdio::null(), Stdio::null(), Stdio::inherit()); let mut output = String::new(); - let mut next_marker = - |prefix: &str| { - let deadline = std::time::Instant::now() + BROKER_HELPER_TIMEOUT; - loop { - let remaining = deadline - .checked_duration_since(std::time::Instant::now()) - .unwrap_or_default(); - let line = line_receiver.recv_timeout(remaining).unwrap_or_else(|error| { - panic!("timed out waiting for guest marker {prefix:?}: {error}; output:\n{output}") - }); - output.push_str(&line); - output.push('\n'); - if line.starts_with(prefix) { - return line; - } - } - }; + let mut next_marker = |prefix: &str| loop { + let line = broker.next_stdout_line(); + output.push_str(&line); + output.push('\n'); + if line.starts_with(prefix) { + return line; + } + }; let listen = next_marker("LISTEN "); assert_ne!( @@ -1066,7 +1107,6 @@ fn test_runner_broker_tcp_server_with_rewriter() { } std::thread::sleep(std::time::Duration::from_millis(10)); }; - reader.join().unwrap(); assert!( status.success(), "broker TCP server guest failed with {status}; output:\n{output}" @@ -1412,14 +1452,14 @@ fn test_broker_with_curl() { 1, ); let url = format!("http://10.0.2.1:{port}/something"); - let output = Runner::new(&curl_path, "curl_rewriter") + Runner::new(&curl_path, "curl_rewriter") .args(["-sS", &url]) .broker_socket(&control_socket_path) - .output(); + .run(); server_thread.join().expect("Server thread panicked"); assert!(broker.next_close_object_count() > 0); - broker.join(); + let output = broker.join_with_stdout(); let output_str = String::from_utf8_lossy(&output); assert!(output_str.contains(RESPONSE_BODY), "Unexpected curl output"); diff --git a/litebox_runner_lvbs/README.md b/litebox_runner_lvbs/README.md index 38ed8cc26f..c2fde3ff2f 100644 --- a/litebox_runner_lvbs/README.md +++ b/litebox_runner_lvbs/README.md @@ -1,6 +1,7 @@ # A LiteBox Runner for running LiteBox in Hyper-V VTL1 kernel space > [!WARNING] -> This crate is work in progress. OP-TEE workloads cannot run on LVBS until a -> kernel broker platform is implemented to provide broker-backed services such -> as cryptographic randomness. +> This crate is work in progress. Broker-backed services are unavailable on +> LVBS until a kernel broker platform is implemented. This includes standard +> output and cryptographic randomness, so workloads that require either service +> cannot run on LVBS. diff --git a/litebox_runner_windows_userland/tests/run.rs b/litebox_runner_windows_userland/tests/run.rs index c30f08d30f..eaefbc56e6 100644 --- a/litebox_runner_windows_userland/tests/run.rs +++ b/litebox_runner_windows_userland/tests/run.rs @@ -3,7 +3,7 @@ #![cfg(all(target_os = "windows", target_arch = "x86_64"))] -/// Runs a hello-world guest PE end to end. +/// Runs a hello-world guest PE end to end through the userland broker. #[test] fn run_hello_world_pe() { let test_dir = std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("kernel32_import"); @@ -20,8 +20,7 @@ fn run_hello_world_pe() { std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("kernel32_import.tar"); create_tar_with_dir(&test_dir, &tar_path); - let mut command = - std::process::Command::new(env!("CARGO_BIN_EXE_litebox_runner_windows_userland")); + let mut command = brokered_windows_runner_command(); // Verbose log for failure triage; not load-bearing for any assertion. command.env("LITEBOX_LOG", "debug"); command.args([ @@ -49,56 +48,7 @@ fn run_hello_world_pe() { ); } -/// Runs a hello-world guest PE through the Windows userland broker. -#[test] -fn run_hello_world_pe_with_broker() { - let test_dir = - std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("kernel32_import_broker"); - let _ = std::fs::remove_dir_all(&test_dir); - std::fs::create_dir_all(&test_dir).unwrap(); - let pe_path = - build_kernel32_import_pe(&test_dir, "kernel32_import", KERNEL32_IMPORT_PE_SOURCE, &[]); - println!( - "Built rewritten kernel32-import PE fixture at `{}`", - pe_path.display() - ); - stage_system_fixtures(&test_dir); - let tar_path = - std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("kernel32_import_broker.tar"); - create_tar_with_dir(&test_dir, &tar_path); - - let (broker, runner) = build_windows_broker(); - let mut command = std::process::Command::new(broker); - command - .env("LITEBOX_LOG", "debug") - .arg("--runner") - .arg(runner) - .args([ - "--initial-files", - tar_path.to_str().unwrap(), - "/kernel32_import.exe", - ]); - println!("Running `{command:?}`"); - let output = command - .output() - .expect("failed to run litebox_runner_windows_userland through the broker"); - let stdout = String::from_utf8_lossy(&output.stdout); - let stderr = String::from_utf8_lossy(&output.stderr); - - assert!( - output.status.success(), - "broker failed to run kernel32-import PE; status {:?}\nstdout:\n{}\nstderr:\n{}", - output.status.code(), - stdout, - stderr - ); - assert!( - stdout.contains("hello world\n"), - "guest output was not captured\nstdout:\n{stdout}\nstderr:\n{stderr}" - ); -} - -/// Runs a guest PE that creates and joins a child thread. +/// Runs a guest PE through the broker that creates and joins a child thread. #[test] fn run_multithreaded_pe() { let test_dir = @@ -120,8 +70,7 @@ fn run_multithreaded_pe() { std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("kernel32_multithread.tar"); create_tar_with_dir(&test_dir, &tar_path); - let mut command = - std::process::Command::new(env!("CARGO_BIN_EXE_litebox_runner_windows_userland")); + let mut command = brokered_windows_runner_command(); command.env("LITEBOX_LOG", "debug"); command.args([ "--initial-files", @@ -155,8 +104,8 @@ fn run_multithreaded_pe() { ); } -/// Runs a guest PE that imports the C runtime (ucrtbase via the CRT api-set -/// contracts), forcing ucrtbase to load and initialize. +/// Runs a guest PE through the broker that imports the C runtime (ucrtbase via +/// the CRT api-set contracts), forcing ucrtbase to load and initialize. #[test] fn run_crt_locale_pe() { let test_dir = std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("crt_locale"); @@ -172,8 +121,7 @@ fn run_crt_locale_pe() { let tar_path = std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("crt_locale.tar"); create_tar_with_dir(&test_dir, &tar_path); - let mut command = - std::process::Command::new(env!("CARGO_BIN_EXE_litebox_runner_windows_userland")); + let mut command = brokered_windows_runner_command(); command.env("LITEBOX_LOG", "debug"); command.args([ "--initial-files", @@ -227,6 +175,13 @@ fn build_windows_broker() -> (std::path::PathBuf, std::path::PathBuf) { (broker, runner) } +fn brokered_windows_runner_command() -> std::process::Command { + let (broker, runner) = build_windows_broker(); + let mut command = std::process::Command::new(broker); + command.arg("--runner").arg(runner); + command +} + /// Stages the guest system DLLs and locale tables the PE fixture needs. fn stage_system_fixtures(test_dir: &std::path::Path) { for dll_name in [ diff --git a/litebox_shim_linux/src/syscalls/tests.rs b/litebox_shim_linux/src/syscalls/tests.rs index beefb1df1d..74bc5fa6fa 100644 --- a/litebox_shim_linux/src/syscalls/tests.rs +++ b/litebox_shim_linux/src/syscalls/tests.rs @@ -8,6 +8,7 @@ use litebox_broker_core::{ random::{RandomProvider, RandomProviderError}, readiness::ReadinessRegistration, socket::{PlatformSocket, SocketProvider}, + stdio::UnsupportedStdioProvider, }; use litebox_broker_local::BrokerLocal; use litebox_broker_protocol::{ @@ -62,6 +63,7 @@ fn test_broker() -> &'static BrokerCore { PolicyEngine::with_unauthenticated_rights(ObjectRights::all()), alloc::sync::Arc::new(PipeOnlySocketProvider), alloc::sync::Arc::new(UnusedRandomProvider), + alloc::sync::Arc::new(UnsupportedStdioProvider), ) .unwrap() })